Publish Muninn startup respect to Odin

crates/muninn-daemon/src/main.rs

@@ -8,7 +8,8 @@ use crate::media_packetizer::{
 use anyhow::{Context, Result, anyhow};
 use cultmesh_rs::{
     CultMesh, CultMeshNodeOptions, CultMeshSharedMemoryFrameRing, CultMeshStreamBodyTransport,
-    CultMeshStreamCatalog, CultMeshStreamClock, CultMeshStreamDescriptor, CultMeshStreamKind,
+    CultMeshRudpDocumentPublishOptions, CultMeshStreamCatalog, CultMeshStreamClock,
+    CultMeshStreamDescriptor, CultMeshStreamKind,
 };
 use cultnet_rs::{
     CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding,
@@ -156,6 +157,7 @@ struct Options {
     quest_pose_stream_id: Option<String>,
     quest_video_input_stream_id: Option<String>,
     idunn_rudp_health: Option<IdunnRudpHealthOptions>,
+    odin_cultmesh_rudp: Option<SocketAddr>,
     capture_command_rudp_bind: Option<SocketAddr>,
     capture_command_rudp_target: Option<SocketAddr>,
     obs_catalog_rudp_target: Option<SocketAddr>,
@@ -327,6 +329,7 @@ fn serve(options: Options) -> Result<()> {
     let mut active_move_lights = Vec::new();
     let mut last_default_move_light_write_at = None;
     let mut last_idunn_health_publish_attempt_at = None;
+    let mut odin_respects_paid = false;
     let mut last_move_host_claim_attempt_at = None;
     let mut last_move_bluetooth_pickup_attempt_at = None;
     let move_runtime_enabled = serve_should_manage_move_runtime(&options);
@@ -372,6 +375,14 @@ 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)?;
+            if !odin_respects_paid {
+                odin_respects_paid = true;
+                if let Err(error) = publish_odin_startup_respect(&node, &options) {
+                    eprintln!(
+                        "Muninn could not publish startup respect to Odin: {error:#}"
+                    );
+                }
+            }
             publish_obs_catalog_idle(&mut node, &options)?;
         }
         claim_move_host_if_due(&options, &mut last_move_host_claim_attempt_at);
@@ -2857,6 +2868,31 @@ fn publish_runtime_boundary_records(
     Ok(())
 }
 
+fn publish_odin_startup_respect(
+    node: &cultmesh_rs::CultMeshNode,
+    options: &Options,
+) -> Result<()> {
+    let Some(target) = options.odin_cultmesh_rudp else {
+        return Ok(());
+    };
+    let provider_id = muninn_provider_id(options);
+    let provider = node.get_required::<EveProviderAdvertisementCompatRecord>(&provider_id)?;
+    node.publish_document_to_rudp_catalog(
+        &provider_id,
+        &provider,
+        CultMeshRudpDocumentPublishOptions {
+            target,
+            runtime_id: muninn_daemon_id(options),
+            source_role: Some("muninn.telemetry-provider".to_string()),
+            tags: vec![
+                "startup-respect".to_string(),
+                "odin-verse-discovery".to_string(),
+            ],
+            ..CultMeshRudpDocumentPublishOptions::default()
+        },
+    )
+}
+
 fn muninn_provider_id(options: &Options) -> String {
     format!("muninn.telemetry.{}", options.host_id)
 }
@@ -6592,6 +6628,7 @@ impl Options {
             quest_pose_stream_id: None,
             quest_video_input_stream_id: None,
             idunn_rudp_health: None,
+            odin_cultmesh_rudp: None,
             capture_command_rudp_bind: None,
             capture_command_rudp_target: None,
             obs_catalog_rudp_target: None,
@@ -6773,6 +6810,13 @@ impl Options {
                 "--idunn-health-contract" => {
                     idunn_health_contract = Some(take_value(&mut args, "--idunn-health-contract")?)
                 }
+                "--odin-cultmesh-rudp" => {
+                    options.odin_cultmesh_rudp = Some(
+                        take_value(&mut args, "--odin-cultmesh-rudp")?
+                            .parse()
+                            .context("--odin-cultmesh-rudp must be a socket address")?,
+                    )
+                }
                 "--capture-command-rudp-bind" => {
                     options.capture_command_rudp_bind = Some(
                         take_value(&mut args, "--capture-command-rudp-bind")?
@@ -6962,7 +7006,7 @@ fn parse_catalog_source(value: &str) -> Result<CatalogSource> {
 }
 
 fn help_text() -> &'static str {
-    "Usage: muninn [serve|activate|request-stream|capture-stream-status|request-move-light|move-light-status|move-identity-status|move-source-status|move-state-status|claim-move-host|quest-access-status] [--store <path>] [--activate-store <path>] [--stream-action <start|stop>] [--target-host <host>] [--port <port>] [--obs-target-host <host>] [--obs-port <port>] [--media-transport <srt|rudp>] [--media-packet-bytes <bytes>] [--rudp-video-bitrate-kbps <kbps>] [--rudp-latency-budget-ms <ms>] [--video-source <source-id=label>] [--audio-source <source-id=label>] [--audio-source-id <source-id>] [--no-video] [--no-audio] [--loopback-script <path>] [--ffmpeg <path>] [--capture-command-rudp-bind <addr>] [--capture-command-rudp-target <addr>] [--obs-catalog-rudp-target <addr>] [--move-state <move-id>=<hidraw-path>] [--move-host <bt-addr>] [--move-evidence-stream <stream-id>] [--move-evidence-verse <verse-id>] [--quest-adb] [--quest-serial <serial>] [--quest-input-stream <stream-id>] [--quest-pose-stream <stream-id>] [--quest-video-input-stream <stream-id>] [--idunn-rudp-health <addr>] [--idunn-daemon <id>] [--idunn-health-contract <contract>] [--dry-run] [--health]\n\nMuninn is Odin's portable telemetry Verse assembler. serve publishes cheap typed telemetry affordances, optional Quest USB access surfaces, and the explicitly configured Move runtime; when serve receives --move-state, --move-host, or --move-evidence-stream it may publish source-local Move controller state, typed Move identity records, a CultMesh Move evidence stream, and keep USB-attached PS Moves claimed to that explicit Bluetooth host; serve also consumes typed capture stream commands from --activate-store or --capture-command-rudp-bind and owns the local ffmpeg/loopback activation child lifecycle, and may project the OBS stream catalog to a local OBS plugin over --obs-catalog-rudp-target; activate starts an explicitly requested local stream over SRT or CultNet RUDP as a daemon child; request-stream publishes a typed capture stream command for Muninn serve to execute, either into the activation store or over --capture-command-rudp-target; capture-stream-status reads typed capture stream command receipts; use --no-video or --no-audio to request one leg without the other when the transport is CultNet RUDP; request-move-light publishes a typed Move light command for Muninn serve to execute; move-light-status reads typed command receipts; move-identity-status reads typed Move identity records; move-source-status prints live Move source discovery; move-state-status reads typed controller-state records; claim-move-host assigns USB-attached PS Moves to a Bluetooth host; quest-access-status reads typed Quest access state. In --health mode, the Idunn RUDP flags publish typed daemon health to Idunn while preserving command-probe exit semantics."
+    "Usage: muninn [serve|activate|request-stream|capture-stream-status|request-move-light|move-light-status|move-identity-status|move-source-status|move-state-status|claim-move-host|quest-access-status] [--store <path>] [--activate-store <path>] [--stream-action <start|stop>] [--target-host <host>] [--port <port>] [--obs-target-host <host>] [--obs-port <port>] [--media-transport <srt|rudp>] [--media-packet-bytes <bytes>] [--rudp-video-bitrate-kbps <kbps>] [--rudp-latency-budget-ms <ms>] [--video-source <source-id=label>] [--audio-source <source-id=label>] [--audio-source-id <source-id>] [--no-video] [--no-audio] [--loopback-script <path>] [--ffmpeg <path>] [--capture-command-rudp-bind <addr>] [--capture-command-rudp-target <addr>] [--obs-catalog-rudp-target <addr>] [--odin-cultmesh-rudp <addr>] [--move-state <move-id>=<hidraw-path>] [--move-host <bt-addr>] [--move-evidence-stream <stream-id>] [--move-evidence-verse <verse-id>] [--quest-adb] [--quest-serial <serial>] [--quest-input-stream <stream-id>] [--quest-pose-stream <stream-id>] [--quest-video-input-stream <stream-id>] [--idunn-rudp-health <addr>] [--idunn-daemon <id>] [--idunn-health-contract <contract>] [--dry-run] [--health]\n\nMuninn is Odin's portable telemetry Verse assembler. serve publishes cheap typed telemetry affordances, optional Quest USB access surfaces, and the explicitly configured Move runtime; when serve receives --move-state, --move-host, or --move-evidence-stream it may publish source-local Move controller state, typed Move identity records, a CultMesh Move evidence stream, and keep USB-attached PS Moves claimed to that explicit Bluetooth host; serve also consumes typed capture stream commands from --activate-store or --capture-command-rudp-bind and owns the local ffmpeg/loopback activation child lifecycle, may project the OBS stream catalog to a local OBS plugin over --obs-catalog-rudp-target, and pays startup respect to Odin over --odin-cultmesh-rudp by publishing its provider advertisement once; activate starts an explicitly requested local stream over SRT or CultNet RUDP as a daemon child; request-stream publishes a typed capture stream command for Muninn serve to execute, either into the activation store or over --capture-command-rudp-target; capture-stream-status reads typed capture stream command receipts; use --no-video or --no-audio to request one leg without the other when the transport is CultNet RUDP; request-move-light publishes a typed Move light command for Muninn serve to execute; move-light-status reads typed command receipts; move-identity-status reads typed Move identity records; move-source-status prints live Move source discovery; move-state-status reads typed controller-state records; claim-move-host assigns USB-attached PS Moves to a Bluetooth host; quest-access-status reads typed Quest access state. In --health mode, the Idunn RUDP flags publish typed daemon health to Idunn while preserving command-probe exit semantics."
 }
 
 fn parse_move_state_source(value: &str) -> Result<MoveStateSource> {
@@ -7293,6 +7337,21 @@ mod tests {
         assert!(error.contains("must be provided together"));
     }
 
+    #[test]
+    fn serve_accepts_odin_cultmesh_rudp_startup_target() {
+        let options = Options::parse(
+            ["serve", "--odin-cultmesh-rudp", "10.77.0.2:17871"]
+                .into_iter()
+                .map(String::from),
+        )
+        .unwrap();
+
+        assert_eq!(
+            options.odin_cultmesh_rudp,
+            Some("10.77.0.2:17871".parse().unwrap())
+        );
+    }
+
     #[test]
     fn idunn_rudp_health_publisher_sends_raw_daemon_health_document() {
         let receiver = UdpSocket::bind("127.0.0.1:0").unwrap();

scripts/restart-muninn.ps1

@@ -18,6 +18,7 @@ param(
   [string] $IdunnRudpHealth = "10.77.0.2:17870",
   [string] $IdunnDaemon = "muninn",
   [string] $IdunnHealthContract = "muninn.cultnet-rudp-remote-telemetry-health",
+  [string] $OdinCultMeshRudp = "10.77.0.2:17871",
   [string] $CaptureCommandRudpBind = "0.0.0.0:17883",
   [string] $CaptureCommandRudpTarget = "127.0.0.1:17883",
   [string] $ObsCatalogRudpTarget = "10.77.0.2:17874",
@@ -373,6 +374,9 @@ $serveArguments = @(
   "--port", $Port.ToString(),
   "--media-transport", $MediaTransport
 )
+if (-not [string]::IsNullOrWhiteSpace($OdinCultMeshRudp)) {
+  $serveArguments += @("--odin-cultmesh-rudp", $OdinCultMeshRudp)
+}
 foreach ($videoSource in $VideoSources) {
   $serveArguments += @("--video-source", $videoSource)
 }

scripts/restart-nightwing-muninn.ps1

@@ -10,7 +10,8 @@ param(
   [int] $IntervalSeconds = 15,
   [string] $IdunnRudpHealth = "10.77.0.2:17870",
   [string] $IdunnDaemon = "nightwing-muninn",
-  [string] $IdunnHealthContract = "muninn.cultnet-rudp-remote-telemetry-and-move-hid"
+  [string] $IdunnHealthContract = "muninn.cultnet-rudp-remote-telemetry-and-move-hid",
+  [string] $OdinCultMeshRudp = "10.77.0.2:17871"
 )
 
 $ErrorActionPreference = "Stop"
@@ -84,6 +85,10 @@ if ($moveStateSpecs.Count -gt 0) {
   $moveRuntimeSetLines += "set -- ""`$@"" --move-evidence-stream $(Quote-ShSingle $MoveEvidenceStream)"
 }
 $moveRuntimeSetBlock = ($moveRuntimeSetLines -join "`n")
+$odinCultMeshRudpSetBlock = ""
+if (-not [string]::IsNullOrWhiteSpace($OdinCultMeshRudp)) {
+  $odinCultMeshRudpSetBlock = "set -- ""`$@"" --odin-cultmesh-rudp $(Quote-ShSingle $OdinCultMeshRudp)"
+}
 
 $remoteScript = @"
 set -eu
@@ -109,6 +114,7 @@ set -- "`$@" \
   --idunn-rudp-health '$IdunnRudpHealth' \
   --idunn-daemon '$IdunnDaemon' \
   --idunn-health-contract '$IdunnHealthContract'
+$odinCultMeshRudpSetBlock
 nohup '$MuninnExe' "`$@" \
   > '$LogRoot/muninn-serve.out.log' \
   2> '$LogRoot/muninn-serve.err.log' \

scripts/restart-starfire-muninn.ps1

@@ -8,7 +8,8 @@ param(
   [switch] $EnableUsbMoveState,
   [string] $IdunnRudpHealth = "127.0.0.1:17870",
   [string] $IdunnDaemon = "starfire-muninn",
-  [string] $IdunnHealthContract = "muninn.cultnet-rudp-local-telemetry-and-quest-access"
+  [string] $IdunnHealthContract = "muninn.cultnet-rudp-local-telemetry-and-quest-access",
+  [string] $OdinCultMeshRudp = "127.0.0.1:17871"
 )
 
 $ErrorActionPreference = "Stop"
@@ -49,6 +50,9 @@ $arguments = @(
   "--idunn-daemon", $IdunnDaemon,
   "--idunn-health-contract", $IdunnHealthContract
 )
+if (-not [string]::IsNullOrWhiteSpace($OdinCultMeshRudp)) {
+  $arguments += @("--odin-cultmesh-rudp", $OdinCultMeshRudp)
+}
 if (-not [string]::IsNullOrWhiteSpace($MoveBluetoothHost)) {
   $arguments += @("--move-host", $MoveBluetoothHost)
 }

vendor/cultmesh-rs/src/lib.rs

@@ -2,9 +2,22 @@ use anyhow::Result;
 use cultcache_rs::CultCache;
 use cultcache_rs::DatabaseEntry;
 use cultcache_rs::SingleFileMessagePackBackingStore;
+use cultnet_rs::CultNetDocumentPutOptions;
 use cultnet_rs::CultNetDocumentRegistry;
+use cultnet_rs::CultNetRudpSocketTransportConnection;
+use cultnet_rs::CultNetRudpSocketTransportOptions;
+use cultnet_rs::CultNetWireContract;
+use cultnet_rs::encode_cultnet_message_to_vec;
+use serde::Serialize;
 use std::collections::BTreeMap;
+use std::net::SocketAddr;
+use std::net::UdpSocket;
 use std::path::Path;
+use std::thread;
+use std::time::Duration;
+use std::time::Instant;
+
+pub const CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID: u32 = 0x0d1d_0002;
 
 pub trait CultMeshDocumentSet: Clone + Send + Sync + 'static {
     fn register_cache(&self, cache: &mut CultCache) -> Result<()>;
@@ -52,6 +65,47 @@ pub struct CultMeshNodeOptions {
     pub pull_on_start: bool,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CultMeshRudpDocumentPublishOptions {
+    pub target: SocketAddr,
+    pub runtime_id: String,
+    pub connection_id: u32,
+    pub connect_timeout: Duration,
+    pub flush_timeout: Duration,
+    pub poll_interval: Duration,
+    pub resend_delay_ms: u64,
+    pub source_agent_id: Option<String>,
+    pub source_role: Option<String>,
+    pub tags: Vec<String>,
+}
+
+impl CultMeshRudpDocumentPublishOptions {
+    pub fn odin(target: SocketAddr, runtime_id: impl Into<String>) -> Self {
+        Self {
+            target,
+            runtime_id: runtime_id.into(),
+            ..Self::default()
+        }
+    }
+}
+
+impl Default for CultMeshRudpDocumentPublishOptions {
+    fn default() -> Self {
+        Self {
+            target: SocketAddr::from(([127, 0, 0, 1], 17871)),
+            runtime_id: "cultmesh-rudp-document-publisher".to_string(),
+            connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID,
+            connect_timeout: Duration::from_secs(1),
+            flush_timeout: Duration::from_millis(150),
+            poll_interval: Duration::from_millis(5),
+            resend_delay_ms: 100,
+            source_agent_id: None,
+            source_role: None,
+            tags: Vec::new(),
+        }
+    }
+}
+
 impl Default for CultMeshNodeOptions {
     fn default() -> Self {
         Self {
@@ -532,6 +586,91 @@ impl CultMeshNode {
     pub fn flush(&mut self) -> Result<()> {
         Ok(())
     }
+
+    pub fn publish_document_to_rudp_catalog<T>(
+        &self,
+        key: impl Into<String>,
+        value: &T,
+        options: CultMeshRudpDocumentPublishOptions,
+    ) -> Result<()>
+    where
+        T: DatabaseEntry + Serialize,
+    {
+        let key = key.into();
+        let message = self.documents.create_raw_document_put_message(
+            format!("{}:{}:{}", options.runtime_id, T::TYPE, key),
+            key,
+            value,
+            CultNetDocumentPutOptions {
+                source_runtime_id: Some(options.runtime_id.clone()),
+                source_agent_id: options.source_agent_id.clone(),
+                source_role: options.source_role.clone(),
+                tags: if options.tags.is_empty() {
+                    None
+                } else {
+                    Some(options.tags.clone())
+                },
+                ..CultNetDocumentPutOptions::default()
+            },
+        )?;
+        publish_cultnet_message_to_rudp_catalog(&message, options)
+    }
+}
+
+fn publish_cultnet_message_to_rudp_catalog(
+    message: &cultnet_rs::CultNetMessage,
+    options: CultMeshRudpDocumentPublishOptions,
+) -> Result<()> {
+    if options.runtime_id.trim().is_empty() {
+        anyhow::bail!("runtime_id must be non-empty");
+    }
+    let bind_addr = if options.target.is_ipv4() {
+        SocketAddr::from(([0, 0, 0, 0], 0))
+    } else {
+        SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], 0))
+    };
+    let socket = UdpSocket::bind(bind_addr)?;
+    socket.set_read_timeout(Some(options.poll_interval))?;
+    let mut client = CultNetRudpSocketTransportConnection::new(
+        CultNetRudpSocketTransportOptions {
+            runtime_id: options.runtime_id.clone(),
+            socket,
+            mode: cultnet_rs::CultNetRudpSocketMode::Client,
+            remote_addr: Some(options.target),
+            connection_id: options.connection_id,
+            initial_sequence: 1,
+            resend_delay_ms: options.resend_delay_ms,
+            transport_id: Some("cultmesh-rudp-document-publisher".to_string()),
+            max_payload_bytes: None,
+            max_fragment_bytes: Some(1200),
+            max_pending_reliable_packets: Some(64),
+            media_reliable_expire_after_ms: None,
+        },
+    )?;
+    client.connect(Vec::new())?;
+    let connect_deadline = Instant::now() + options.connect_timeout;
+    while !client.connected() && Instant::now() < connect_deadline {
+        let _ = client.receive_once()?;
+        client.poll_resends()?;
+        thread::sleep(options.poll_interval);
+    }
+    if !client.connected() {
+        anyhow::bail!(
+            "Timed out connecting CultMesh RUDP document publisher {} to {}",
+            options.runtime_id,
+            options.target
+        );
+    }
+
+    let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
+    client.send("schema", payload)?;
+    let flush_deadline = Instant::now() + options.flush_timeout;
+    while Instant::now() < flush_deadline {
+        let _ = client.receive_once()?;
+        client.poll_resends()?;
+        thread::sleep(options.poll_interval);
+    }
+    Ok(())
 }
 
 #[derive(Clone, Copy, Debug, Default)]
@@ -653,6 +792,71 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn node_publishes_registered_document_to_rudp_catalog() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let store_path = temp.path().join("cultmesh.cc");
+        let note = Note {
+            body: "respect odin once".to_string(),
+        };
+        let node = CultMesh::create_node(
+            &store_path,
+            TestDocuments,
+            CultMeshNodeOptions {
+                runtime_id: "muninn-test".to_string(),
+                ..CultMeshNodeOptions::default()
+            },
+        )?;
+
+        let socket = UdpSocket::bind("127.0.0.1:0")?;
+        socket.set_read_timeout(Some(Duration::from_millis(10)))?;
+        let target = socket.local_addr()?;
+        let (sender, receiver) = std::sync::mpsc::channel();
+        let server = thread::spawn(move || -> Result<()> {
+            let mut server = CultNetRudpSocketTransportConnection::new(
+                CultNetRudpSocketTransportOptions::server(
+                    "odin-test-catalog",
+                    socket,
+                    CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID,
+                ),
+            )?;
+            let deadline = Instant::now() + Duration::from_secs(2);
+            while Instant::now() < deadline {
+                if let Some(frame) = server.receive_once()? {
+                    let message = cultnet_rs::decode_cultnet_message_from_slice(
+                        &frame.payload,
+                        CultNetWireContract::CultNetSchemaV0,
+                    )?;
+                    if let cultnet_rs::CultNetMessage::DocumentPutRaw { document, .. } = message {
+                        sender
+                            .send((
+                                document.schema_id,
+                                document.record_key,
+                                document.source_runtime_id,
+                            ))
+                            .ok();
+                        return Ok(());
+                    }
+                }
+                server.poll_resends()?;
+            }
+            anyhow::bail!("timed out waiting for RUDP document put")
+        });
+
+        node.publish_document_to_rudp_catalog(
+            "note",
+            &note,
+            CultMeshRudpDocumentPublishOptions::odin(target, "muninn-test"),
+        )?;
+        let received = receiver.recv_timeout(Duration::from_secs(2))?;
+        server.join().expect("server thread should not panic")?;
+
+        assert_eq!(received.0, "cultmesh.test.note.v0");
+        assert_eq!(received.1, "note");
+        assert_eq!(received.2.as_deref(), Some("muninn-test"));
+        Ok(())
+    }
+
     #[test]
     fn stream_catalog_negotiates_and_publishes_latest_shared_memory_frame() -> Result<()> {
         let mut catalog = CultMesh::create_stream_catalog();