diff --git a/src-tauri/src/audio/engine.rs b/src-tauri/src/audio/engine.rs index 48d7b6ee..8480d73c 100644 --- a/src-tauri/src/audio/engine.rs +++ b/src-tauri/src/audio/engine.rs @@ -6,13 +6,15 @@ use std::sync::mpsc::{Receiver, Sender}; -use tauri::AppHandle; +use tauri::{AppHandle, Emitter}; use tracing::{error, info, warn}; use crate::audio::graph::ValidGraph; use crate::audio::pipeline::{self, ActivePipeline}; use crate::error::{AppError, AppResult}; +const STATE_EVENT: &str = "audio://state"; + pub enum Command { Start { graph: ValidGraph, @@ -98,21 +100,22 @@ pub fn run(rx: Receiver) { } Command::Stop { reply } => { if active.take().is_none() { - warn!("stop ignored: pipeline not running"); - let _ = reply.send(Err(AppError::NotRunning)); + info!("stop: pipeline was already idle"); } else { info!("pipeline torn down"); - let _ = reply.send(Ok(())); } + let _ = reply.send(Ok(())); } Command::Reconcile { graph, app, reply } => match active.as_mut() { None => { let _ = reply.send(Err(AppError::NotRunning)); } Some(p) => { - let r = p.reconcile(&graph, app); + let r = p.reconcile(&graph, app.clone()); if let Err(e) = &r { - error!(error = %e, "reconcile failed"); + error!(error = %e, "reconcile failed, clearing active pipeline"); + active = None; + let _ = app.emit(STATE_EVENT, serde_json::json!({ "kind": "stopped" })); } let _ = reply.send(r); } diff --git a/src-tauri/src/audio/graph.rs b/src-tauri/src/audio/graph.rs index 20d4a4cf..2c860186 100644 --- a/src-tauri/src/audio/graph.rs +++ b/src-tauri/src/audio/graph.rs @@ -864,9 +864,10 @@ impl GraphSpec { .intersection(&reachable_from_terminals) .copied() .collect(); - // Keep unrouted input nodes too, so their capture + level meter run - // before they're wired anywhere; unresolvable ones drop in resolve_inputs. - let mut keep = routed.clone(); + // Keep unrouted input nodes (so their capture + level meter run) + // as well as nodes reachable from terminals (outputs, analyzers, and + // their upstream effect chains, which stream silence if inputs disconnect). + let mut keep = reachable_from_terminals; for n in &nodes { if n.role == NodeCategory::Input { keep.insert(n.id.as_str()); @@ -874,7 +875,7 @@ impl GraphSpec { } let inputs = resolve_inputs(&nodes, &keep, &routed)?; - let outputs = resolve_outputs(&nodes, &keep)?; + let outputs = resolve_outputs(&nodes, &keep, &routed)?; let effects = resolve_effects(&nodes, &keep)?; let edges: Vec = edges @@ -984,129 +985,139 @@ fn resolve_inputs( Ok(result) } -fn resolve_outputs(nodes: &[RoleNode<'_>], keep: &HashSet<&str>) -> AppResult> { +fn resolve_outputs( + nodes: &[RoleNode<'_>], + keep: &HashSet<&str>, + routed: &HashSet<&str>, +) -> AppResult> { let mut result = Vec::new(); for n in nodes { if n.role != NodeCategory::Output || !keep.contains(n.id.as_str()) { continue; } - let spec = match n.kind { - NodeKind::Speaker => { - let data: SpeakerData = parse(n.data, "Speaker")?; - OutputSpec::Speaker { - device_id: data - .device_id - .ok_or_else(|| miss(&n.id, "Speaker has no device selected"))?, - } - } - NodeKind::FileRecording => { - let data: FileRecordingData = parse(n.data, "FileRecording")?; - let file_path = data - .file_path - .ok_or_else(|| miss(&n.id, "File Recording has no path"))?; - let path = std::path::Path::new(&file_path); - let parent = path.parent().unwrap_or(std::path::Path::new(".")); - if !parent.exists() { - return Err(choose_file_err(&n.id, "directory does not exist")); + let resolved = (|| -> AppResult { + Ok(match n.kind { + NodeKind::Speaker => { + let data: SpeakerData = parse(n.data, "Speaker")?; + OutputSpec::Speaker { + device_id: data + .device_id + .ok_or_else(|| miss(&n.id, "Speaker has no device selected"))?, + } } - match data.mode { - RecordingMode::New => { - if path.exists() { - return Err(choose_file_err(&n.id, "file already exists")); + NodeKind::FileRecording => { + let data: FileRecordingData = parse(n.data, "FileRecording")?; + let file_path = data + .file_path + .ok_or_else(|| miss(&n.id, "File Recording has no path"))?; + let path = std::path::Path::new(&file_path); + let parent = path.parent().unwrap_or(std::path::Path::new(".")); + if !parent.exists() { + return Err(choose_file_err(&n.id, "directory does not exist")); + } + match data.mode { + RecordingMode::New => { + if path.exists() { + return Err(choose_file_err(&n.id, "file already exists")); + } + } + RecordingMode::Overwrite => {} + RecordingMode::Append => { + if !matches!( + data.format, + RecordingFormat::Wav { .. } | RecordingFormat::Aiff { .. } + ) { + return Err(AppError::Validation(format!( + "append recording is only supported for WAV/AIFF (node {})", + n.id + ))); + } } } - RecordingMode::Overwrite => {} - RecordingMode::Append => { - if !matches!( - data.format, - RecordingFormat::Wav { .. } | RecordingFormat::Aiff { .. } - ) { + #[cfg(not(target_os = "macos"))] + if matches!(data.format, RecordingFormat::Aac { .. }) { + return Err(AppError::Validation(format!( + "AAC recording is only supported on macOS (node {})", + n.id + ))); + } + let max = data.format.max_channels(); + if data.channels == 0 || data.channels > max { + return Err(AppError::Validation(format!( + "recording node {} asks for {} channels; format allows 1..{max}", + n.id, data.channels + ))); + } + if let Some(sr) = data.sample_rate { + // FLAC's format tops out at 655350 Hz (20-bit rate field); + // every other recording format caps at 384000. + let max = if matches!(data.format, RecordingFormat::Flac { .. }) { + 655_350 + } else { + 384_000 + }; + if !(8000..=max).contains(&sr) { return Err(AppError::Validation(format!( - "append recording is only supported for WAV/AIFF (node {})", + "recording node {} pins sample rate {sr}; expected 8000..{max}", n.id ))); } } - } - #[cfg(not(target_os = "macos"))] - if matches!(data.format, RecordingFormat::Aac { .. }) { - return Err(AppError::Validation(format!( - "AAC recording is only supported on macOS (node {})", - n.id - ))); - } - let max = data.format.max_channels(); - if data.channels == 0 || data.channels > max { - return Err(AppError::Validation(format!( - "recording node {} asks for {} channels; format allows 1..{max}", - n.id, data.channels - ))); - } - if let Some(sr) = data.sample_rate { - // FLAC's format tops out at 655350 Hz (20-bit rate field); - // every other recording format caps at 384000. - let max = if matches!(data.format, RecordingFormat::Flac { .. }) { - 655_350 - } else { - 384_000 - }; - if !(8000..=max).contains(&sr) { - return Err(AppError::Validation(format!( - "recording node {} pins sample rate {sr}; expected 8000..{max}", - n.id - ))); + OutputSpec::FileRecording { + file_path, + format: data.format, + channels: data.channels, + mode: data.mode, + sample_rate: data.sample_rate.filter(|_| { + !matches!( + data.format, + RecordingFormat::Opus { .. } | RecordingFormat::Mp3 { .. } + ) + }), } } - OutputSpec::FileRecording { - file_path, - format: data.format, - channels: data.channels, - mode: data.mode, - sample_rate: data.sample_rate.filter(|_| { - !matches!( - data.format, - RecordingFormat::Opus { .. } | RecordingFormat::Mp3 { .. } - ) - }), - } - } - NodeKind::NetSender => { - let data: NetSenderData = parse(n.data, "NetSender")?; - let ip: IpAddr = data - .target_ip - .trim() - .parse() - .map_err(|_| miss(&n.id, "Net Sender has an invalid target IP"))?; - OutputSpec::NetSender { - node_id: n.id.clone(), - target: SocketAddr::new(ip, data.port), - channels: data - .channels - .clamp(1, crate::audio::netaudio::MAX_CHANNELS as u32), - codec: data.codec, - opus_bitrate: data.opus_bitrate, - opus_application: data.opus_application, + NodeKind::NetSender => { + let data: NetSenderData = parse(n.data, "NetSender")?; + let ip: IpAddr = data + .target_ip + .trim() + .parse() + .map_err(|_| miss(&n.id, "Net Sender has an invalid target IP"))?; + OutputSpec::NetSender { + node_id: n.id.clone(), + target: SocketAddr::new(ip, data.port), + channels: data + .channels + .clamp(1, crate::audio::netaudio::MAX_CHANNELS as u32), + codec: data.codec, + opus_bitrate: data.opus_bitrate, + opus_application: data.opus_application, + } } - } - // Send half of a collaborator: audio wired in goes to peers, - // which is a destination like any other sender. - NodeKind::WebRtcCollaborator => { - let data: WebRtcCollaboratorData = parse(n.data, "WebRtcCollaborator")?; - OutputSpec::WebRtcSend { - node_id: n.id.clone(), - channels: data - .channels - .clamp(1, crate::audio::netaudio::MAX_CHANNELS as u32), - opus_bitrate: data.opus_bitrate, - opus_application: data.opus_application, + // Send half of a collaborator: audio wired in goes to peers, + // which is a destination like any other sender. + NodeKind::WebRtcCollaborator => { + let data: WebRtcCollaboratorData = parse(n.data, "WebRtcCollaborator")?; + OutputSpec::WebRtcSend { + node_id: n.id.clone(), + channels: data + .channels + .clamp(1, crate::audio::netaudio::MAX_CHANNELS as u32), + opus_bitrate: data.opus_bitrate, + opus_application: data.opus_application, + } } - } - _ => unreachable!(), - }; - result.push(ValidOutput { - id: n.id.clone(), - spec, - }); + _ => unreachable!(), + }) + })(); + match resolved { + Ok(spec) => result.push(ValidOutput { + id: n.id.clone(), + spec, + }), + Err(e) if routed.contains(n.id.as_str()) => return Err(e), + Err(_) => continue, + } } Ok(result) } @@ -1385,4 +1396,37 @@ mod tests { assert!(v.inputs.is_empty()); assert!(v.outputs.is_empty()); } + + #[test] + fn unrouted_output_is_valid_and_streams_silence() { + let g = GraphSpec { + nodes: vec![speaker("s")], + edges: vec![], + }; + let v = g.validate().expect("unrouted output is valid"); + assert!(v.inputs.is_empty()); + assert_eq!(v.outputs.len(), 1); + assert_eq!(v.outputs[0].id, "s"); + } + + #[test] + fn effect_leading_to_output_survives_when_input_disconnects() { + fn gain(id: &str) -> NodeSpec { + node(id, NodeKind::Gain, serde_json::json!({ "gainDb": 0.0 })) + } + + let g = GraphSpec { + nodes: vec![gain("g"), speaker("s")], + edges: vec![edge("e", "g", None, "s", None)], + }; + let v = g + .validate() + .expect("effect + output without inputs is valid"); + assert!(v.inputs.is_empty()); + assert_eq!(v.effects.len(), 1); + assert_eq!(v.effects[0].id, "g"); + assert_eq!(v.outputs.len(), 1); + assert_eq!(v.outputs[0].id, "s"); + assert_eq!(v.edges.len(), 1); + } } diff --git a/src-tauri/src/audio/macos_hal.rs b/src-tauri/src/audio/macos_hal.rs index 1c1d3cd6..deddf485 100644 --- a/src-tauri/src/audio/macos_hal.rs +++ b/src-tauri/src/audio/macos_hal.rs @@ -40,6 +40,10 @@ const K_AUDIO_DEVICE_PROPERTY_VOLUME_DECIBELS: AudioObjectPropertySelector = fou const K_AUDIO_DEVICE_PROPERTY_MUTE: AudioObjectPropertySelector = fourcc(b"mute"); const K_AUDIO_OBJECT_PROPERTY_NAME: AudioObjectPropertySelector = fourcc(b"lnam"); const K_AUDIO_DEVICE_PROPERTY_DEVICE_UID: AudioObjectPropertySelector = fourcc(b"uid "); +const K_AUDIO_DEVICE_PROPERTY_TRANSPORT_TYPE: AudioObjectPropertySelector = fourcc(b"trpt"); +const K_AUDIO_DEVICE_PROPERTY_DEVICE_IS_ALIVE: AudioObjectPropertySelector = fourcc(b"livn"); +const K_AUDIO_DEVICE_TRANSPORT_TYPE_BLUETOOTH: u32 = fourcc(b"blue"); +const K_AUDIO_DEVICE_TRANSPORT_TYPE_BLUETOOTH_LE: u32 = fourcc(b"blea"); /// UID prefix of the private aggregate we create for CATap app-audio capture /// (see `native/CATapCapture.swift`). CoreAudio still lists a process's own @@ -330,6 +334,51 @@ unsafe fn nominal_sample_rate(device_id: AudioObjectID) -> Option { } } +unsafe fn is_bluetooth_device(device_id: AudioObjectID) -> bool { + let addr = AudioObjectPropertyAddress { + selector: K_AUDIO_DEVICE_PROPERTY_TRANSPORT_TYPE, + scope: K_AUDIO_OBJECT_PROPERTY_SCOPE_GLOBAL, + element: K_AUDIO_OBJECT_PROPERTY_ELEMENT_MAIN, + }; + let mut transport: u32 = 0; + let mut size: u32 = mem::size_of::() as u32; + if AudioObjectGetPropertyData( + device_id, + &addr, + 0, + ptr::null(), + &mut size, + &mut transport as *mut _ as *mut c_void, + ) != 0 + { + return false; + } + transport == K_AUDIO_DEVICE_TRANSPORT_TYPE_BLUETOOTH + || transport == K_AUDIO_DEVICE_TRANSPORT_TYPE_BLUETOOTH_LE +} + +unsafe fn is_device_alive(device_id: AudioObjectID) -> bool { + let addr = AudioObjectPropertyAddress { + selector: K_AUDIO_DEVICE_PROPERTY_DEVICE_IS_ALIVE, + scope: K_AUDIO_OBJECT_PROPERTY_SCOPE_GLOBAL, + element: K_AUDIO_OBJECT_PROPERTY_ELEMENT_MAIN, + }; + let mut alive: u32 = 0; + let mut size: u32 = mem::size_of::() as u32; + if AudioObjectGetPropertyData( + device_id, + &addr, + 0, + ptr::null(), + &mut size, + &mut alive as *mut _ as *mut c_void, + ) != 0 + { + return false; + } + alive != 0 +} + fn list_by_scope(scope: AudioObjectPropertyScope) -> Vec { let mut out = Vec::new(); unsafe { @@ -340,16 +389,29 @@ fn list_by_scope(scope: AudioObjectPropertyScope) -> Vec { if !has_streams_in_scope(id, scope) { continue; } - let channels = channel_count_in_scope(id, scope); - if channels == 0 { + let is_bt = is_bluetooth_device(id); + if is_bt && !is_device_alive(id) { continue; } + + let mut channels = channel_count_in_scope(id, scope); + if channels == 0 { + if is_bt { + // Inactive Bluetooth headsets default to stereo output or mono input. + channels = match scope { + K_AUDIO_OBJECT_PROPERTY_SCOPE_OUTPUT => 2, + K_AUDIO_OBJECT_PROPERTY_SCOPE_INPUT => 1, + _ => 0, + }; + } + if channels == 0 { + continue; + } + } let Some(name) = device_name(id) else { continue; }; - let Some(sample_rate) = nominal_sample_rate(id) else { - continue; - }; + let sample_rate = nominal_sample_rate(id).unwrap_or(48_000); out.push(HalDevice { name, sample_rate, @@ -392,6 +454,10 @@ fn find_device_id(name: &str, scope: AudioObjectPropertyScope) -> Option { + let dead = Arc::new(AtomicBool::new(false)); + let dead_cb = dead.clone(); let app_err = app.clone(); + let node_id_cb = node_id.to_string(); let err_cb = move |e: cpal::StreamError| { + if dead_cb.swap(true, Ordering::Relaxed) { + return; + } health::bump(&health::STREAM_ERRORS, 1); - error!(error = %e, "input stream error"); + error!(node_id = %node_id_cb, error = %e, "input stream error"); let _ = app_err.emit( - STATE_EVENT, - json!({ "kind": "error", "message": format!("input: {e}") }), + "audio://input_error", + json!({ "nodeId": node_id_cb, "error": format!("{e}") }), ); }; let stream = streams::build_input_stream( diff --git a/src-tauri/src/audio/pipeline/input/windows.rs b/src-tauri/src/audio/pipeline/input/windows.rs index 356bcf34..eb2430c2 100644 --- a/src-tauri/src/audio/pipeline/input/windows.rs +++ b/src-tauri/src/audio/pipeline/input/windows.rs @@ -1,4 +1,4 @@ -use std::sync::atomic::AtomicBool; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use serde_json::json; @@ -13,7 +13,6 @@ use crate::audio::streams; use crate::error::AppResult; use super::super::native::native_config; -use super::super::STATE_EVENT; use super::{resolve_audio_file, start_audio_file, InputHandle, ResolvedInput, SCK_SR}; const LOOPBACK_CHANNELS: usize = 2; @@ -64,13 +63,19 @@ pub(in crate::audio::pipeline) fn start_input_stream( src_channels, .. } => { + let dead = Arc::new(AtomicBool::new(false)); + let dead_cb = dead.clone(); let app_err = app.clone(); + let node_id_cb = node_id.to_string(); let err_cb = move |e: cpal::StreamError| { + if dead_cb.swap(true, Ordering::Relaxed) { + return; + } health::bump(&health::STREAM_ERRORS, 1); - error!(error = %e, "input stream error"); + error!(node_id = %node_id_cb, error = %e, "input stream error"); let _ = app_err.emit( - STATE_EVENT, - json!({ "kind": "error", "message": format!("input: {e}") }), + "audio://input_error", + json!({ "nodeId": node_id_cb, "error": format!("{e}") }), ); }; let stream = streams::build_input_stream( diff --git a/src-tauri/src/audio/pipeline/mod.rs b/src-tauri/src/audio/pipeline/mod.rs index 430ec2b0..8847cb51 100644 --- a/src-tauri/src/audio/pipeline/mod.rs +++ b/src-tauri/src/audio/pipeline/mod.rs @@ -54,9 +54,6 @@ use output::{ use sig::{compute_output_sig, OutputSig, MONITOR_KEY}; use worker::WorkerCtrl; -#[cfg(any(target_os = "macos", target_os = "windows"))] -pub(super) const STATE_EVENT: &str = "audio://state"; - /// Overlap between a hot-swapped output's old and new bridges: one DSP block at /// 48 kHz plus slack, so the incoming sub-graph starts with its rings primed. const SWAP_PREFILL: std::time::Duration = std::time::Duration::from_millis(25); diff --git a/src-tauri/src/audio/pipeline/output/macos.rs b/src-tauri/src/audio/pipeline/output/macos.rs index 35dfd82b..aab2f84b 100644 --- a/src-tauri/src/audio/pipeline/output/macos.rs +++ b/src-tauri/src/audio/pipeline/output/macos.rs @@ -121,10 +121,15 @@ pub(in crate::audio::pipeline) fn start_speaker_stream( let dead_cb = dead.clone(); let node_id_cb = node_id.to_string(); let err_cb = move |e: cpal::StreamError| { - dead_cb.store(true, Ordering::Relaxed); + if dead_cb.swap(true, Ordering::Relaxed) { + return; + } health::bump(&health::STREAM_ERRORS, 1); error!(node_id = %node_id_cb, error = %e, "speaker stream error"); - let _ = app_err.emit("audio://speaker_error", json!({ "nodeId": node_id_cb })); + let _ = app_err.emit( + "audio://speaker_error", + json!({ "nodeId": node_id_cb, "error": format!("{e}") }), + ); }; match streams::build_output_stream( &spec.device, diff --git a/src-tauri/src/audio/pipeline/output/windows.rs b/src-tauri/src/audio/pipeline/output/windows.rs index 16634808..cde36a13 100644 --- a/src-tauri/src/audio/pipeline/output/windows.rs +++ b/src-tauri/src/audio/pipeline/output/windows.rs @@ -79,10 +79,15 @@ pub(in crate::audio::pipeline) fn start_speaker_stream( let dead_cb = dead.clone(); let node_id_cb = node_id.to_string(); let err_cb = move |e: cpal::StreamError| { - dead_cb.store(true, Ordering::Relaxed); + if dead_cb.swap(true, Ordering::Relaxed) { + return; + } health::bump(&health::STREAM_ERRORS, 1); error!(node_id = %node_id_cb, error = %e, "speaker stream error"); - let _ = app_err.emit("audio://speaker_error", json!({ "nodeId": node_id_cb })); + let _ = app_err.emit( + "audio://speaker_error", + json!({ "nodeId": node_id_cb, "error": format!("{e}") }), + ); }; let stream = streams::build_output_stream( diff --git a/src/lib/modules/audio/methods.ts b/src/lib/modules/audio/methods.ts index cdc7c8dc..a89a62bd 100644 --- a/src/lib/modules/audio/methods.ts +++ b/src/lib/modules/audio/methods.ts @@ -77,7 +77,10 @@ export const methods = { /** Throws when not settable. */ setDeviceVolume: (kind: 'input' | 'output', name: string, scalar: number): Promise => invoke('set_device_volume', { kind, name, scalar }), onState: (cb: (e: AudioStateEvent) => void): Promise => listen(AUDIO_STATE_EVENT, (evt) => cb(evt.payload)), - onSpeakerError: (cb: () => void): Promise => listen('audio://speaker_error', () => cb()), + onInputError: (cb: (e: { nodeId: string; error: string }) => void): Promise => + listen<{ nodeId: string; error: string }>('audio://input_error', (evt) => cb(evt.payload)), + onSpeakerError: (cb: (e: { nodeId: string; error?: string }) => void): Promise => + listen<{ nodeId: string; error?: string }>('audio://speaker_error', (evt) => cb(evt.payload)), virtualDriverStatus: (): Promise => invoke('virtual_driver_status'), windowsVirtualCableStatus: (): Promise => invoke('windows_virtual_cable_status'), installWindowsVirtualCable: (): Promise => invoke('install_windows_virtual_cable'), diff --git a/src/lib/modules/audio/stores.svelte.ts b/src/lib/modules/audio/stores.svelte.ts index e4998264..8b9811c9 100644 --- a/src/lib/modules/audio/stores.svelte.ts +++ b/src/lib/modules/audio/stores.svelte.ts @@ -7,6 +7,7 @@ import { methods as pipelineMethods } from '$lib/modules/pipeline/methods'; import { pipelineStore } from '$lib/modules/pipeline/stores.svelte'; import { isFromFuture } from '$lib/modules/pipeline/migrations'; import type { FileRecordingNodeData, PipelineNode, RecordingFormat } from '$lib/modules/pipeline/types'; +import { appSettings } from '$lib/modules/settings/stores.svelte'; // Mirrors `extension()` in the File Recording node: the dialog filter must // match the encoder the node will actually write. @@ -41,8 +42,10 @@ class AudioStore { private fullGraph: StartPipelinePayload | null = null; private reconnectTimer: ReturnType | undefined; private speakerRecovering = false; + private inputRecovering = false; private unlisten: UnlistenFn | undefined; private unlistenSpeakerError: UnlistenFn | undefined; + private unlistenInputError: UnlistenFn | undefined; async refreshInputDevices(): Promise { this.inputDevices = await methods.listInputDevices(); @@ -74,36 +77,110 @@ class AudioStore { this.isRunning = true; this.startedAt = Date.now(); } else if (e.kind === 'stopped') { - this.stopPendingReconnectLoop(); - this.isRunning = false; - this.runningPipelineId = null; - this.startedAt = null; + if (this.pendingNodeIds.size === 0) { + this.stopPendingReconnectLoop(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + } } else if (e.kind === 'error') { this.stopPendingReconnectLoop(); + this.pendingNodeIds.clear(); this.isRunning = false; this.runningPipelineId = null; this.startedAt = null; + void methods.stopPipeline().catch(() => {}); this.reportError(e.message); } }); + + methods + .onInputError(async (payload) => { + if (this.inputRecovering || !this.isRunning) return; + this.inputRecovering = true; + const msg = payload.error || 'Input device disconnected'; + try { + if (!appSettings.keepRunningOnDisconnect) { + this.stopPendingReconnectLoop(); + this.pendingNodeIds.clear(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + await methods.stopPipeline().catch(() => {}); + this.reportError(msg); + return; + } + // If we have full graph, try to exclude the disconnected node and keep running + if (this.fullGraph && payload.nodeId) { + this.pendingNodeIds.add(payload.nodeId); + const reduced = this.buildReducedGraph(this.fullGraph, this.pendingNodeIds); + if (reduced.nodes.length > 0) { + try { + await methods.reconcilePipeline(reduced); + this.lastGraph = reduced; + this.startPendingReconnectLoop(); + this.reportError(msg); + return; + } catch { + // Reconcile failed or no working route left, fall through + } + } + } + // Stop backend to avoid orphan threads, but keep frontend active & reconnecting + await methods.stopPipeline().catch(() => {}); + this.startPendingReconnectLoop(); + this.reportError(msg); + } catch (e: unknown) { + this.reportError(msg); + } finally { + this.inputRecovering = false; + } + }) + .then((fn) => { + this.unlistenInputError = fn; + }) + .catch(() => {}); + methods - .onSpeakerError(() => { - if (this.speakerRecovering || !this.lastGraph || !this.isRunning) return; + .onSpeakerError(async (payload) => { + if (this.speakerRecovering || !this.isRunning) return; this.speakerRecovering = true; - methods - .reconcilePipeline(this.lastGraph) - .catch((e: unknown) => { - const msg = e instanceof Error ? e.message : String(e); - if (!msg.includes('not running')) { - this.isRunning = false; - this.runningPipelineId = null; - this.startedAt = null; - this.reportError(msg); + const msg = payload?.error || 'Speaker device disconnected'; + try { + if (!appSettings.keepRunningOnDisconnect) { + this.stopPendingReconnectLoop(); + this.pendingNodeIds.clear(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + await methods.stopPipeline().catch(() => {}); + this.reportError(msg); + return; + } + if (this.fullGraph && payload?.nodeId) { + this.pendingNodeIds.add(payload.nodeId); + const reduced = this.buildReducedGraph(this.fullGraph, this.pendingNodeIds); + if (reduced.nodes.length > 0) { + try { + await methods.reconcilePipeline(reduced); + this.lastGraph = reduced; + this.startPendingReconnectLoop(); + this.reportError(msg); + return; + } catch { + // Reconcile failed (e.g. no valid route left) + } } - }) - .finally(() => { - this.speakerRecovering = false; - }); + } + // Stop backend to avoid orphan threads, but keep frontend active & reconnecting + await methods.stopPipeline().catch(() => {}); + this.startPendingReconnectLoop(); + this.reportError(msg); + } catch (e: unknown) { + this.reportError(msg); + } finally { + this.speakerRecovering = false; + } }) .then((fn) => { this.unlistenSpeakerError = fn; @@ -113,10 +190,24 @@ class AudioStore { async activatePipeline(pipelineId: string, graph: StartPipelinePayload): Promise { this.lastGraph = graph; + this.fullGraph = graph; + const excluded = appSettings.keepRunningOnDisconnect ? await this.unresolvedInputIds(graph) : new Set(); + this.pendingNodeIds = excluded; + const toStart = excluded.size > 0 ? this.buildReducedGraph(graph, excluded) : graph; try { - await methods.startPipeline(graph); + await methods.startPipeline(toStart); + this.lastGraph = toStart; + if (excluded.size > 0) { + this.startPendingReconnectLoop(); + } } catch (e) { if (await this.handleStartError(e, pipelineId, graph)) return; + this.stopPendingReconnectLoop(); + this.pendingNodeIds.clear(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + await methods.stopPipeline().catch(() => {}); throw e; } this.runningPipelineId = pipelineId; @@ -188,7 +279,11 @@ class AudioStore { * so a future launch doesn't try to auto-activate it again. */ async deactivatePipeline(): Promise { this.stopPendingReconnectLoop(); - await methods.stopPipeline(); + this.pendingNodeIds.clear(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + await methods.stopPipeline().catch(() => {}); await pipelineMethods.setActivePipelineId(null).catch(() => {}); } @@ -203,7 +298,7 @@ class AudioStore { if (!p || isFromFuture(p)) return; const full: StartPipelinePayload = { nodes: p.nodes, edges: p.edges }; this.fullGraph = full; - const excluded = await this.unresolvedInputIds(full); + const excluded = appSettings.keepRunningOnDisconnect ? await this.unresolvedInputIds(full) : new Set(); this.pendingNodeIds = excluded; const reduced = this.buildReducedGraph(full, excluded); try { @@ -214,7 +309,9 @@ class AudioStore { } this.lastGraph = reduced; this.runningPipelineId = id; - this.startPendingReconnectLoop(); + if (excluded.size > 0) { + this.startPendingReconnectLoop(); + } } /** Which input nodes in `full` can't resolve right now: an App Audio node @@ -222,14 +319,22 @@ class AudioStore { * doesn't exist on disk. Building this list also refreshes the app list, * since a stale snapshot would wrongly exclude/include nodes. */ private async unresolvedInputIds(full: StartPipelinePayload): Promise> { - await this.refreshAudioApplications(); + await Promise.all([this.refreshInputDevices(), this.refreshOutputDevices(), this.refreshAudioApplications()]); const running = new Set(this.audioApplications.map((a) => a.bundleId)); + const inputIds = new Set(this.inputDevices.map((d) => d.id)); + const outputIds = new Set(this.outputDevices.map((d) => d.id)); const unresolved = new Set(); const filePaths = new Set(); for (const n of full.nodes) { if (n.kind === 'appAudio') { const bundleId = (n.data as { bundleId: string | null }).bundleId; if (bundleId && !running.has(bundleId)) unresolved.add(n.id); + } else if (n.kind === 'microphone') { + const deviceId = (n.data as { deviceId: string | null }).deviceId; + if (deviceId && !inputIds.has(deviceId)) unresolved.add(n.id); + } else if (n.kind === 'speaker') { + const deviceId = (n.data as { deviceId: string | null }).deviceId; + if (deviceId && !outputIds.has(deviceId)) unresolved.add(n.id); } else if (n.kind === 'audioFile') { const filePath = (n.data as { filePath: string | null }).filePath; if (filePath) filePaths.add(filePath); @@ -273,18 +378,30 @@ class AudioStore { } private async tryReconnectPending(): Promise { - if (this.pendingNodeIds.size === 0 || !this.fullGraph || !this.isRunning) { + if (!appSettings.keepRunningOnDisconnect || this.pendingNodeIds.size === 0 || !this.fullGraph || !this.isRunning) { this.stopPendingReconnectLoop(); return; } const stillUnresolved = await this.unresolvedInputIds(this.fullGraph); - if (stillUnresolved.size === this.pendingNodeIds.size) return; + const isUnchanged = stillUnresolved.size === this.pendingNodeIds.size && [...stillUnresolved].every((id) => this.pendingNodeIds.has(id)); + if (isUnchanged) return; this.pendingNodeIds = stillUnresolved; const reduced = this.buildReducedGraph(this.fullGraph, stillUnresolved); try { - await this.restartPipeline(reduced); - } catch { - return; + await methods.reconcilePipeline(reduced); + this.lastGraph = reduced; + } catch (e: unknown) { + const msg = e instanceof Error ? e.message : String(e); + if (msg.includes('not running')) { + try { + await methods.startPipeline(reduced); + this.lastGraph = reduced; + } catch { + return; + } + } else { + return; + } } if (stillUnresolved.size === 0) this.stopPendingReconnectLoop(); } @@ -294,10 +411,20 @@ class AudioStore { * streams stay alive across edits when their spec is unchanged. * Falls back to stop + start if the pipeline isn't running. */ async restartPipeline(graph: StartPipelinePayload): Promise { - this.lastGraph = graph; + this.fullGraph = graph; + for (const id of this.pendingNodeIds) { + if (!graph.nodes.some((n) => n.id === id)) { + this.pendingNodeIds.delete(id); + } + } + if (this.pendingNodeIds.size === 0) { + this.stopPendingReconnectLoop(); + } + const toRun = appSettings.keepRunningOnDisconnect && this.pendingNodeIds.size > 0 ? this.buildReducedGraph(graph, this.pendingNodeIds) : graph; + this.lastGraph = toRun; let reconcileErr: unknown; try { - await methods.reconcilePipeline(graph); + await methods.reconcilePipeline(toRun); return; } catch (e) { reconcileErr = e; @@ -305,13 +432,23 @@ class AudioStore { const msg = reconcileErr instanceof Error ? reconcileErr.message : String(reconcileErr); if (msg.includes('not running')) { try { - await methods.startPipeline(graph); + await methods.startPipeline(toRun); } catch (e) { if (this.routeStartError(e)) return; + this.stopPendingReconnectLoop(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + await methods.stopPipeline().catch(() => {}); throw e; } } else { if (this.routeStartError(reconcileErr)) return; + this.stopPendingReconnectLoop(); + this.isRunning = false; + this.runningPipelineId = null; + this.startedAt = null; + await methods.stopPipeline().catch(() => {}); throw reconcileErr; } } @@ -333,6 +470,8 @@ class AudioStore { this.unlisten = undefined; this.unlistenSpeakerError?.(); this.unlistenSpeakerError = undefined; + this.unlistenInputError?.(); + this.unlistenInputError = undefined; } } diff --git a/src/lib/modules/settings/stores.svelte.ts b/src/lib/modules/settings/stores.svelte.ts index 3af58844..3c3feba5 100644 --- a/src/lib/modules/settings/stores.svelte.ts +++ b/src/lib/modules/settings/stores.svelte.ts @@ -10,6 +10,7 @@ interface Stored { gridSize: number; launchOnStartup: boolean; confirmOverwriteChanges: boolean; + keepRunningOnDisconnect: boolean; } const DEFAULTS: Stored = { @@ -18,7 +19,8 @@ const DEFAULTS: Stored = { snapToGrid: false, gridSize: 20, launchOnStartup: false, - confirmOverwriteChanges: true + confirmOverwriteChanges: true, + keepRunningOnDisconnect: true }; export const SNAPSHOT_LIMITS = [10, 20, 50, 100] as const; @@ -41,13 +43,14 @@ class AppSettings { gridSize = $state(this.#initial.gridSize); launchOnStartup = $state(this.#initial.launchOnStartup); confirmOverwriteChanges = $state(this.#initial.confirmOverwriteChanges); + keepRunningOnDisconnect = $state(this.#initial.keepRunningOnDisconnect); persist(): void { if (!browser) return; - const { checkUpdatesOnLaunch, maxSnapshots, snapToGrid, gridSize, launchOnStartup, confirmOverwriteChanges } = this; + const { checkUpdatesOnLaunch, maxSnapshots, snapToGrid, gridSize, launchOnStartup, confirmOverwriteChanges, keepRunningOnDisconnect } = this; window.localStorage.setItem( KEY, - JSON.stringify({ checkUpdatesOnLaunch, maxSnapshots, snapToGrid, gridSize, launchOnStartup, confirmOverwriteChanges }) + JSON.stringify({ checkUpdatesOnLaunch, maxSnapshots, snapToGrid, gridSize, launchOnStartup, confirmOverwriteChanges, keepRunningOnDisconnect }) ); } diff --git a/src/routes/settings/+page.svelte b/src/routes/settings/+page.svelte index 7c9c1c1c..a5f66963 100644 --- a/src/routes/settings/+page.svelte +++ b/src/routes/settings/+page.svelte @@ -39,7 +39,7 @@ void disableAutostart(); } - function setApp( + function setApp( key: K, value: (typeof appSettings)[K] ) { @@ -214,6 +214,19 @@ onChange={() => setApp('confirmOverwriteChanges', !appSettings.confirmOverwriteChanges)} /> +
+
+

Device disconnection

+

What happens when an audio device disconnects while a pipeline is running.

+
+ + setApp('keepRunningOnDisconnect', !appSettings.keepRunningOnDisconnect)} /> +
+

Startup