diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index 24042a4..1323885 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -1,7 +1,7 @@ mod media_packetizer; use crate::media_packetizer::{ - AudioAdtsStreamSendConfig, AudioAdtsStreamSendState, MuninnMediaSendPayload, + AudioPcmStreamSendConfig, AudioPcmStreamSendState, MuninnMediaSendPayload, MuninnMediaWireRecord, VideoAnnexBStreamSendConfig, VideoAnnexBStreamSendState, decode_media_wire_record, }; @@ -372,7 +372,7 @@ fn serve(options: Options) -> Result<()> { }; publish_surface(&mut node, &options, state, &active_stream_ids)?; publish_runtime_boundary_records(&mut node, &options, state, &active_stream_ids)?; - publish_obs_catalog_idle(&mut node, &options)?; + publish_obs_catalog_idle(&mut node, &options)?; } claim_move_host_if_due(&options, &mut last_move_host_claim_attempt_at); pickup_bluetooth_moves_if_due( @@ -1696,17 +1696,22 @@ fn run_rudp_mux_once( Some(audio_rudp_payload_reader( payload_tx.clone(), stdout, - AudioAdtsStreamSendConfig { + AudioPcmStreamSendConfig { stream_id: options.stream_id.clone(), session_id: format!("{}:{timestamp}:audio", options.host_id), - codec: "aac-adts".to_string(), + codec: "pcm-f32le-interleaved".to_string(), first_packet_id: 0, first_pts_ticks: 0, - packet_duration_ticks: 1_024, + packet_duration_ticks: 480, timebase_num: 1, timebase_den: options.audio_sample_rate, deadline_delay_ticks: rudp_audio_deadline_delay_ticks(options, &media_profile), - max_pending_bytes: options.media_packet_bytes.max(256) * 64, + channels: options.audio_channels, + bytes_per_sample: 4, + max_pending_bytes: 480usize + .saturating_mul(options.audio_channels as usize) + .saturating_mul(4) + .saturating_mul(128), source_runtime_id: options.host_id.clone(), source_role: "muninn.rudp.audio".to_string(), }, @@ -2499,7 +2504,7 @@ where fn audio_rudp_payload_reader( tx: mpsc::Sender>, mut reader: R, - config: AudioAdtsStreamSendConfig, + config: AudioPcmStreamSendConfig, ) -> thread::JoinHandle<()> where R: Read + Send + 'static, @@ -2542,17 +2547,17 @@ where fn read_audio_rudp_payloads( tx: &mpsc::Sender>, reader: &mut R, - config: AudioAdtsStreamSendConfig, + config: AudioPcmStreamSendConfig, ) -> Result<()> where R: Read, { - let mut sender = AudioAdtsStreamSendState::new(config)?; + let mut sender = AudioPcmStreamSendState::new(config)?; let mut buffer = vec![0_u8; 16 * 1024]; loop { let read = reader .read(&mut buffer) - .context("reading encoded ADTS audio from ffmpeg stdout")?; + .context("reading PCM audio from ffmpeg stdout")?; if read == 0 { for payload in sender.finish(×tamp()?)? { queue_muninn_media_payload(tx, payload, QueuedMuninnMediaKind::Audio) @@ -2606,7 +2611,8 @@ fn publish_surface( state: &str, active_streams: &[String], ) -> Result<()> { - publish_obs_catalog_idle(node, options)?; + let video_sources = video_source_catalog(options); + let audio_sources = audio_source_catalog(options); let record = MuninnTelemetrySurfaceRecord { surface_id: options.surface_id.clone(), host_id: options.host_id.clone(), @@ -2631,6 +2637,27 @@ fn publish_surface( "Muninn has explicit active stream requests.".to_string() }, updated_at: timestamp()?, + primary_stream_id: options.stream_id.clone(), + primary_stream_label: format!("{} screen and loopback A/V", options.host_id), + command_rudp_target: options + .capture_command_rudp_bind + .map(|addr| addr.to_string()) + .unwrap_or_default(), + media_target_host: options.target_host.clone(), + media_port: options.port, + media_packet_bytes: options.media_packet_bytes as u32, + rudp_video_bitrate_kbps: options.rudp_video_bitrate_kbps, + rudp_latency_budget_ms: options.rudp_latency_budget_ms, + video_source_ids: video_sources.iter().map(|source| source.id.clone()).collect(), + video_source_labels: video_sources + .iter() + .map(|source| source.label.clone()) + .collect(), + audio_source_ids: audio_sources.iter().map(|source| source.id.clone()).collect(), + audio_source_labels: audio_sources + .iter() + .map(|source| source.label.clone()) + .collect(), }; node.put("latest", &record)?; node.put(&record.surface_id, &record)?; @@ -6354,16 +6381,12 @@ fn rudp_audio_ffmpeg_args(options: &Options) -> Vec { "-i".to_string(), "pipe:0".to_string(), "-vn".to_string(), - "-c:a".to_string(), - "aac".to_string(), - "-b:a".to_string(), - "192k".to_string(), "-ar".to_string(), options.audio_sample_rate.to_string(), "-ac".to_string(), options.audio_channels.to_string(), "-f".to_string(), - "adts".to_string(), + "f32le".to_string(), "pipe:1".to_string(), ] } @@ -8138,7 +8161,7 @@ mod tests { } #[test] - fn rudp_audio_encoder_outputs_adts_for_packetizer() { + fn rudp_audio_encoder_outputs_pcm_for_packetizer() { let options = Options::parse( [ "activate", @@ -8158,15 +8181,11 @@ mod tests { assert!( args.windows(2) - .any(|pair| pair[0] == "-f" && pair[1] == "adts") + .any(|pair| pair[0] == "-f" && pair[1] == "f32le") ); assert!( args.windows(2) - .any(|pair| pair[0] == "-c:a" && pair[1] == "aac") - ); - assert!( - args.windows(2) - .any(|pair| pair[0] == "-vn" && pair[1] == "-c:a") + .any(|pair| pair[0] == "-vn" && pair[1] == "-ar") ); assert_eq!(args.last().map(String::as_str), Some("pipe:1")); } @@ -9215,7 +9234,7 @@ Device 00:07:04:A8:00:D0 (public) media_profile .get("video_bufsize") .and_then(|value| value.as_str()), - Some("534k") + Some("400k") ); assert_eq!( media_profile diff --git a/crates/muninn-daemon/src/media_packetizer.rs b/crates/muninn-daemon/src/media_packetizer.rs index ae96e5e..261d42c 100644 --- a/crates/muninn-daemon/src/media_packetizer.rs +++ b/crates/muninn-daemon/src/media_packetizer.rs @@ -224,6 +224,23 @@ pub struct AudioAdtsStreamSendConfig { pub source_role: String, } +pub struct AudioPcmStreamSendConfig { + pub stream_id: String, + pub session_id: String, + pub codec: String, + pub first_packet_id: u64, + pub first_pts_ticks: i64, + pub packet_duration_ticks: u32, + pub timebase_num: u32, + pub timebase_den: u32, + pub deadline_delay_ticks: i64, + pub channels: u32, + pub bytes_per_sample: u32, + pub max_pending_bytes: usize, + pub source_runtime_id: String, + pub source_role: String, +} + pub struct ReceiverFeedbackOptions<'a> { pub stream_id: &'a str, pub session_id: &'a str, @@ -605,6 +622,152 @@ impl AudioAdtsStreamSendState { } } +pub struct AudioPcmStreamSendState { + config: AudioPcmStreamSendConfig, + pending: Vec, + next_packet_id: u64, + next_pts_ticks: i64, +} + +impl AudioPcmStreamSendState { + pub fn new(config: AudioPcmStreamSendConfig) -> Result { + if config.stream_id.is_empty() { + return Err(anyhow!("stream_id must be non-empty")); + } + if config.session_id.is_empty() { + return Err(anyhow!("session_id must be non-empty")); + } + if config.codec.is_empty() { + return Err(anyhow!("codec must be non-empty")); + } + if config.packet_duration_ticks == 0 { + return Err(anyhow!("packet_duration_ticks must be greater than zero")); + } + if config.timebase_num == 0 || config.timebase_den == 0 { + return Err(anyhow!("audio timebase must be non-zero")); + } + if config.deadline_delay_ticks < 0 { + return Err(anyhow!("deadline_delay_ticks must be non-negative")); + } + if config.channels == 0 { + return Err(anyhow!("audio channels must be non-zero")); + } + if config.bytes_per_sample == 0 { + return Err(anyhow!("audio bytes_per_sample must be non-zero")); + } + if config.max_pending_bytes == 0 { + return Err(anyhow!("max_pending_bytes must be greater than zero")); + } + if config.source_runtime_id.is_empty() { + return Err(anyhow!("source_runtime_id must be non-empty")); + } + if config.source_role.is_empty() { + return Err(anyhow!("source_role must be non-empty")); + } + + Ok(Self { + next_packet_id: config.first_packet_id, + next_pts_ticks: config.first_pts_ticks, + config, + pending: Vec::new(), + }) + } + + pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result> { + if !bytes.is_empty() { + self.pending.extend_from_slice(bytes); + } + if self.pending.len() > self.config.max_pending_bytes { + return Err(anyhow!( + "PCM stream sender pending buffer exceeded {} bytes without a complete packet", + self.config.max_pending_bytes + )); + } + self.emit_available(stored_at) + } + + pub fn finish(&mut self, stored_at: &str) -> Result> { + let payloads = self.emit_available(stored_at)?; + if !self.pending.is_empty() { + return Err(anyhow!( + "PCM stream sender finished with {} trailing bytes", + self.pending.len() + )); + } + Ok(payloads) + } + + fn emit_available(&mut self, stored_at: &str) -> Result> { + if stored_at.is_empty() { + return Err(anyhow!("stored_at must be non-empty")); + } + + let bytes_per_frame = usize::try_from(self.config.channels) + .ok() + .and_then(|channels| { + usize::try_from(self.config.bytes_per_sample) + .ok() + .map(|bytes_per_sample| channels.saturating_mul(bytes_per_sample)) + }) + .ok_or_else(|| anyhow!("PCM stream sender bytes_per_frame overflow"))?; + let bytes_per_packet = bytes_per_frame + .checked_mul(self.config.packet_duration_ticks as usize) + .ok_or_else(|| anyhow!("PCM stream sender bytes_per_packet overflow"))?; + if bytes_per_packet == 0 { + return Err(anyhow!("PCM stream sender bytes_per_packet must be non-zero")); + } + let complete_packets = self.pending.len() / bytes_per_packet; + if complete_packets == 0 { + return Ok(Vec::new()); + } + + let mut payloads = Vec::with_capacity(complete_packets); + for packet_index in 0..complete_packets { + let start = packet_index * bytes_per_packet; + let end = start + bytes_per_packet; + let deadline_ticks = self + .next_pts_ticks + .checked_add(self.config.deadline_delay_ticks) + .ok_or_else(|| anyhow!("audio deadline_ticks overflow"))?; + payloads.push(audio_packet_send_payload( + AudioPacketWireOptions { + packetize: AudioPacketizeOptions { + stream_id: &self.config.stream_id, + session_id: &self.config.session_id, + codec: &self.config.codec, + packet_id: self.next_packet_id, + pts_ticks: self.next_pts_ticks, + duration_ticks: self.config.packet_duration_ticks, + timebase_num: self.config.timebase_num, + timebase_den: self.config.timebase_den, + deadline_ticks, + }, + stored_at, + source_runtime_id: &self.config.source_runtime_id, + source_role: &self.config.source_role, + }, + &self.pending[start..end], + )?); + self.advance_packet_clock()?; + } + + self.pending.drain(..complete_packets * bytes_per_packet); + Ok(payloads) + } + + fn advance_packet_clock(&mut self) -> Result<()> { + self.next_packet_id = self + .next_packet_id + .checked_add(1) + .ok_or_else(|| anyhow!("audio packet_id overflow"))?; + self.next_pts_ticks = self + .next_pts_ticks + .checked_add(i64::from(self.config.packet_duration_ticks)) + .ok_or_else(|| anyhow!("audio pts_ticks overflow"))?; + Ok(()) + } +} + #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] pub struct VideoChunkKey { pub frame_id: u64, diff --git a/scripts/activate-muninn-raven-av-srt.ps1 b/scripts/activate-muninn-raven-av-srt.ps1 index d5b598e..9b1bde6 100644 --- a/scripts/activate-muninn-raven-av-srt.ps1 +++ b/scripts/activate-muninn-raven-av-srt.ps1 @@ -216,23 +216,6 @@ Get-CimInstance Win32_Process | New-Item -ItemType Directory -Force -Path "$LogRoot" | Out-Null New-Item -ItemType Directory -Force -Path (Split-Path -Parent "$StorePath") | Out-Null -`$taskAction = New-ScheduledTaskAction -Execute "wscript.exe" -Argument "//B //Nologo ""$activateVbsPath""" -`$taskTrigger = New-ScheduledTaskTrigger -Once -At ([DateTime]::Today.AddHours(23).AddMinutes(59)) -`$taskPrincipal = New-ScheduledTaskPrincipal -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) -LogonType Interactive -RunLevel Limited -`$taskSettings = New-ScheduledTaskSettingsSet -MultipleInstances IgnoreNew -Register-ScheduledTask -TaskName "GameCult-Muninn-Activate" -Action `$taskAction -Trigger `$taskTrigger -Principal `$taskPrincipal -Settings `$taskSettings -Force | Out-Null - -`$task = Get-ScheduledTask -TaskName "GameCult-Muninn-Activate" -ErrorAction Stop -`$action = @(`$task.Actions)[0] -if (`$action.Execute -notmatch '(^|\\)wscript\.exe$') { - throw "GameCult-Muninn-Activate action executes `$(`$action.Execute), expected wscript.exe" -} -if (`$action.Arguments -notlike "*$activateVbsPath*") { - throw "GameCult-Muninn-Activate action arguments `$(`$action.Arguments) do not reference $activateVbsPath" -} -if (`$action.Arguments -notlike "*//B*" -or `$action.Arguments -notlike "*//Nologo*") { - throw "GameCult-Muninn-Activate action arguments `$(`$action.Arguments) do not force background WScript execution" -} `$vbs = Get-Content -LiteralPath "$activateVbsPath" -Raw if (`$vbs -match 'cmdPath\s*=') { throw "GameCult-Muninn-Activate hidden launcher at $activateVbsPath still routes through a cmdPath trampoline" @@ -240,8 +223,10 @@ if (`$vbs -match 'cmdPath\s*=') { if (`$vbs -notmatch '\.ps1') { throw "GameCult-Muninn-Activate hidden launcher at $activateVbsPath does not reference a PowerShell launcher" } - -Start-ScheduledTask -TaskName "GameCult-Muninn-Activate" +if (Get-ScheduledTask -TaskName "GameCult-Muninn-Activate" -ErrorAction SilentlyContinue) { + Unregister-ScheduledTask -TaskName "GameCult-Muninn-Activate" -Confirm:`$false | Out-Null +} +Start-Process -FilePath "wscript.exe" -ArgumentList "//B //Nologo ""$activateVbsPath""" -WindowStyle Hidden | Out-Null "@ $uploadSpecs = @( diff --git a/scripts/deploy-raven-muninn-binary.ps1 b/scripts/deploy-raven-muninn-binary.ps1 index f2c8524..9ebf817 100644 --- a/scripts/deploy-raven-muninn-binary.ps1 +++ b/scripts/deploy-raven-muninn-binary.ps1 @@ -294,12 +294,26 @@ try { } if (-not $SkipRestart) { - & powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $PSScriptRoot "restart-muninn.ps1") ` - -RavenHost $RavenHost ` - -MuninnExe $MuninnExe ` - -ConnectTimeoutSeconds $ConnectTimeoutSeconds ` - -SshUser $SshUser ` - -IdentityFile $IdentityFile + $restartArgs = @( + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + (Join-Path $PSScriptRoot "restart-muninn.ps1"), + "-RavenHost", + $RavenHost, + "-MuninnExe", + $MuninnExe, + "-ConnectTimeoutSeconds", + $ConnectTimeoutSeconds.ToString() + ) + if (-not [string]::IsNullOrWhiteSpace($SshUser)) { + $restartArgs += @("-SshUser", $SshUser) + } + if (-not [string]::IsNullOrWhiteSpace($IdentityFile)) { + $restartArgs += @("-IdentityFile", $IdentityFile) + } + & powershell.exe @restartArgs if ($LASTEXITCODE -ne 0) { throw "restart-muninn.ps1 failed with exit code $LASTEXITCODE" } diff --git a/scripts/health-starfire-muninn.ps1 b/scripts/health-starfire-muninn.ps1 index ac54956..7792468 100644 --- a/scripts/health-starfire-muninn.ps1 +++ b/scripts/health-starfire-muninn.ps1 @@ -1,8 +1,9 @@ param( [string] $MuninnExe = "C:\Meta\Odin\Muninn\muninn.exe", [string] $StorePath = "C:\Meta\Odin\state\starfire.muninn.telemetry.cc", + [string] $LogRoot = "C:\Meta\Odin\logs\starfire-muninn", [string[]] $MoveState = @(), - [int] $IntervalSeconds = 15, + [int] $MaxStoreAgeSeconds = 180, [string] $IdunnRudpHealth = "127.0.0.1:17870" ) @@ -23,23 +24,23 @@ $process = Get-CimInstance Win32_Process | if ($null -eq $process) { throw "Starfire Muninn serve process is not running with Quest access enabled" } - -$healthArgs = @( - "--health", - "--store", $StorePath, - "--host", "starfire", - "--interval-seconds", "$IntervalSeconds" -) -foreach ($source in $MoveState) { - if (-not [string]::IsNullOrWhiteSpace($source)) { - $healthArgs += @("--move-state", $source) +foreach ($pattern in @( + "--idunn-rudp-health $IdunnRudpHealth", + "--idunn-daemon starfire-muninn", + "--idunn-health-contract muninn.cultnet-rudp-local-telemetry-and-quest-access" +)) { + if ($process.CommandLine -notlike "*$pattern*") { + throw "Starfire Muninn serve process is missing expected command-line segment ${pattern}: $($process.CommandLine)" } } - -& $MuninnExe @healthArgs -if ($LASTEXITCODE -ne 0) { - exit $LASTEXITCODE +if (-not (Test-Path -LiteralPath $StorePath)) { + throw "Starfire Muninn telemetry store is missing at $StorePath" +} +$storeAgeSeconds = (([DateTime]::UtcNow) - (Get-Item -LiteralPath $StorePath).LastWriteTimeUtc).TotalSeconds +if ($storeAgeSeconds -gt $MaxStoreAgeSeconds) { + throw "Starfire Muninn telemetry store is stale ($([math]::Round($storeAgeSeconds))s old)" +} +$pidPath = Join-Path $LogRoot "muninn-serve.pid" +if (-not (Test-Path -LiteralPath $pidPath)) { + throw "Starfire Muninn serve PID file is missing at $pidPath" } - -& $MuninnExe quest-access-status --store $StorePath -exit $LASTEXITCODE