Advertise and request media streams without naming a producer

Two state records beside the media wire records, travelling as CultMesh
documents through Odin rather than on the media channel:

- gamecult.media_stream_advertisement.v1: a stream a producer can serve
  on request — its video and audio sources, the codecs it can encode,
  audio sample rate and channels, defaults, and the connection id it
  dials a receiver with. Keyed by stream_id. A consumer's picker is built
  from these.
- gamecult.media_stream_request.v1: a consumer asking a producer to start
  or stop serving a stream to a receiver endpoint, with sources, codecs,
  bitrate and latency budget. Keyed by stream_id:receiver_id so a newer
  request supersedes; the producer answers on the same key by rewriting
  state and detail.

Until now the only way to pick a stream was Muninn's own
muninn.obs_stream_catalog and muninn.capture_stream_command, which live
in odin-core and bind every consumer to one producer. Odin persists any
well-formed document, so nothing there has to change to carry these.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

packages/cultnet-rs/src/media_stream_contracts.rs

@@ -46,6 +46,349 @@ pub const GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA: &str = "gamecult.media_video_
 pub const GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA: &str = "gamecult.media_video_parity_shard.v2";
 pub const GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA: &str = "gamecult.media_audio_packet.v1";
 pub const GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA: &str = "gamecult.media_receiver_feedback.v1";
+pub const GAMECULT_MEDIA_STREAM_ADVERTISEMENT_SCHEMA: &str =
+    "gamecult.media_stream_advertisement.v1";
+pub const GAMECULT_MEDIA_STREAM_REQUEST_SCHEMA: &str = "gamecult.media_stream_request.v1";
+
+/// A media stream a producer can serve on request, as it appears in Odin.
+///
+/// This is the record a consumer's picker is built from: which streams exist,
+/// which video and audio sources each can capture, which codecs it can encode,
+/// and the defaults it would use. It is state, not media — it travels as a
+/// CultMesh document through Odin, never on the media channel — and it names no
+/// consumer. `producer_id` is the runtime that answers requests for it.
+///
+/// Keyed by `stream_id`.
+#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
+#[cultcache(
+    type = "gamecult.media_stream_advertisement",
+    schema = "gamecult.media_stream_advertisement.v1"
+)]
+pub struct GameCultMediaStreamAdvertisementRecord {
+    #[cultcache(key = 0)]
+    pub stream_id: String,
+    #[cultcache(key = 1)]
+    pub producer_id: String,
+    #[cultcache(key = 2)]
+    pub label: String,
+    /// `available`, `streaming`, or `unavailable`.
+    #[cultcache(key = 3)]
+    pub state: String,
+    #[cultcache(key = 4)]
+    pub video_source_ids: Vec<String>,
+    #[cultcache(key = 5)]
+    pub video_source_labels: Vec<String>,
+    #[cultcache(key = 6)]
+    pub audio_source_ids: Vec<String>,
+    #[cultcache(key = 7)]
+    pub audio_source_labels: Vec<String>,
+    /// Codecs the producer can encode video as, e.g. `h264`.
+    #[cultcache(key = 8)]
+    pub video_codecs: Vec<String>,
+    /// Codecs the producer can emit audio as, e.g. `pcm-f32le-interleaved`.
+    #[cultcache(key = 9)]
+    pub audio_codecs: Vec<String>,
+    #[cultcache(key = 10)]
+    pub audio_sample_rate: u32,
+    #[cultcache(key = 11)]
+    pub audio_channels: u32,
+    #[cultcache(key = 12)]
+    pub default_video_bitrate_kbps: u32,
+    #[cultcache(key = 13)]
+    pub default_latency_budget_ms: u32,
+    #[cultcache(key = 14)]
+    pub media_packet_bytes: u32,
+    /// The CultNet RUDP connection id the producer dials the receiver with.
+    #[cultcache(key = 15)]
+    pub media_connection_id: u32,
+    #[cultcache(key = 16)]
+    pub updated_at: String,
+}
+
+/// A consumer asking a producer to start or stop serving a stream to it.
+///
+/// The producer answers by rewriting `state` and `detail` on the same record
+/// key, so a consumer watches one document for the outcome. A newer request for
+/// the same `stream_id` and `receiver_id` supersedes an older one.
+///
+/// Keyed by `media_stream_request_key(stream_id, receiver_id)`.
+#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
+#[cultcache(
+    type = "gamecult.media_stream_request",
+    schema = "gamecult.media_stream_request.v1"
+)]
+pub struct GameCultMediaStreamRequestRecord {
+    #[cultcache(key = 0)]
+    pub request_id: String,
+    #[cultcache(key = 1)]
+    pub stream_id: String,
+    #[cultcache(key = 2)]
+    pub producer_id: String,
+    #[cultcache(key = 3)]
+    pub receiver_id: String,
+    /// `host:port` the producer should dial with the advertised connection id.
+    #[cultcache(key = 4)]
+    pub receiver_endpoint: String,
+    /// `start` or `stop`.
+    #[cultcache(key = 5)]
+    pub action: String,
+    /// `pending`, `running`, `stopped`, or `failed`; the producer owns it after
+    /// `pending`.
+    #[cultcache(key = 6)]
+    pub state: String,
+    /// Empty means no video.
+    #[cultcache(key = 7)]
+    pub video_source_id: String,
+    /// Empty means no audio.
+    #[cultcache(key = 8)]
+    pub audio_source_id: String,
+    #[cultcache(key = 9)]
+    pub video_codec: String,
+    #[cultcache(key = 10)]
+    pub audio_codec: String,
+    #[cultcache(key = 11)]
+    pub video_bitrate_kbps: u32,
+    #[cultcache(key = 12)]
+    pub latency_budget_ms: u32,
+    #[cultcache(key = 13)]
+    pub media_packet_bytes: u32,
+    #[cultcache(key = 14)]
+    pub detail: String,
+    #[cultcache(key = 15)]
+    pub updated_at: String,
+}
+
+pub const MEDIA_STREAM_REQUEST_ACTIONS: [&str; 2] = ["start", "stop"];
+pub const MEDIA_STREAM_REQUEST_STATES: [&str; 4] = ["pending", "running", "stopped", "failed"];
+pub const MEDIA_STREAM_ADVERTISEMENT_STATES: [&str; 3] = ["available", "streaming", "unavailable"];
+
+/// One live request per stream per receiver: a newer one replaces the older.
+pub fn media_stream_request_key(stream_id: &str, receiver_id: &str) -> String {
+    format!("{stream_id}:{receiver_id}")
+}
+
+pub fn validate_media_stream_advertisement(
+    record: &GameCultMediaStreamAdvertisementRecord,
+) -> anyhow::Result<()> {
+    use anyhow::anyhow;
+    for (name, value) in [
+        ("stream_id", &record.stream_id),
+        ("producer_id", &record.producer_id),
+        ("label", &record.label),
+        ("updated_at", &record.updated_at),
+    ] {
+        if value.is_empty() {
+            return Err(anyhow!("media stream advertisement {name} must be non-empty"));
+        }
+    }
+    if !MEDIA_STREAM_ADVERTISEMENT_STATES.contains(&record.state.as_str()) {
+        return Err(anyhow!(
+            "media stream advertisement state {:?} is not one of {:?}",
+            record.state,
+            MEDIA_STREAM_ADVERTISEMENT_STATES
+        ));
+    }
+    if record.video_source_ids.len() != record.video_source_labels.len() {
+        return Err(anyhow!(
+            "media stream advertisement has {} video source ids and {} labels",
+            record.video_source_ids.len(),
+            record.video_source_labels.len()
+        ));
+    }
+    if record.audio_source_ids.len() != record.audio_source_labels.len() {
+        return Err(anyhow!(
+            "media stream advertisement has {} audio source ids and {} labels",
+            record.audio_source_ids.len(),
+            record.audio_source_labels.len()
+        ));
+    }
+    if !record.audio_source_ids.is_empty() && (record.audio_sample_rate == 0 || record.audio_channels == 0) {
+        return Err(anyhow!(
+            "media stream advertisement offers audio without a sample rate and channel count"
+        ));
+    }
+    if record.media_connection_id == 0 {
+        return Err(anyhow!("media stream advertisement media_connection_id must be non-zero"));
+    }
+    Ok(())
+}
+
+pub fn validate_media_stream_request(
+    record: &GameCultMediaStreamRequestRecord,
+) -> anyhow::Result<()> {
+    use anyhow::anyhow;
+    for (name, value) in [
+        ("request_id", &record.request_id),
+        ("stream_id", &record.stream_id),
+        ("producer_id", &record.producer_id),
+        ("receiver_id", &record.receiver_id),
+        ("updated_at", &record.updated_at),
+    ] {
+        if value.is_empty() {
+            return Err(anyhow!("media stream request {name} must be non-empty"));
+        }
+    }
+    if !MEDIA_STREAM_REQUEST_ACTIONS.contains(&record.action.as_str()) {
+        return Err(anyhow!(
+            "media stream request action {:?} is not one of {:?}",
+            record.action,
+            MEDIA_STREAM_REQUEST_ACTIONS
+        ));
+    }
+    if !MEDIA_STREAM_REQUEST_STATES.contains(&record.state.as_str()) {
+        return Err(anyhow!(
+            "media stream request state {:?} is not one of {:?}",
+            record.state,
+            MEDIA_STREAM_REQUEST_STATES
+        ));
+    }
+    if record.action == "start" {
+        if record.receiver_endpoint.parse::<std::net::SocketAddr>().is_err() {
+            return Err(anyhow!(
+                "media stream request receiver_endpoint {:?} must be host:port",
+                record.receiver_endpoint
+            ));
+        }
+        if record.video_source_id.is_empty() && record.audio_source_id.is_empty() {
+            return Err(anyhow!("media stream request asks for neither video nor audio"));
+        }
+        if !record.video_source_id.is_empty() && record.video_codec.is_empty() {
+            return Err(anyhow!("media stream request names a video source without a codec"));
+        }
+        if !record.audio_source_id.is_empty() && record.audio_codec.is_empty() {
+            return Err(anyhow!("media stream request names an audio source without a codec"));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod media_stream_catalog_tests {
+    use super::*;
+
+    fn advertisement() -> GameCultMediaStreamAdvertisementRecord {
+        GameCultMediaStreamAdvertisementRecord {
+            stream_id: "muninn.raven.av.rudp".into(),
+            producer_id: "raven".into(),
+            label: "Raven desktop".into(),
+            state: "available".into(),
+            video_source_ids: vec!["display:0".into()],
+            video_source_labels: vec!["Display 1".into()],
+            audio_source_ids: vec!["wasapi-loopback:Realtek".into()],
+            audio_source_labels: vec!["Realtek loopback".into()],
+            video_codecs: vec!["h264".into()],
+            audio_codecs: vec!["pcm-f32le-interleaved".into()],
+            audio_sample_rate: 48_000,
+            audio_channels: 2,
+            default_video_bitrate_kbps: 12_000,
+            default_latency_budget_ms: 250,
+            media_packet_bytes: 848,
+            media_connection_id: 0x6d75_0001,
+            updated_at: "2026-09-10T00:00:00Z".into(),
+        }
+    }
+
+    fn request() -> GameCultMediaStreamRequestRecord {
+        GameCultMediaStreamRequestRecord {
+            request_id: "req-1".into(),
+            stream_id: "muninn.raven.av.rudp".into(),
+            producer_id: "raven".into(),
+            receiver_id: "starfire.obs".into(),
+            receiver_endpoint: "192.168.178.146:5204".into(),
+            action: "start".into(),
+            state: "pending".into(),
+            video_source_id: "display:0".into(),
+            audio_source_id: "wasapi-loopback:Realtek".into(),
+            video_codec: "h264".into(),
+            audio_codec: "pcm-f32le-interleaved".into(),
+            video_bitrate_kbps: 12_000,
+            latency_budget_ms: 250,
+            media_packet_bytes: 848,
+            detail: String::new(),
+            updated_at: "2026-09-10T00:00:00Z".into(),
+        }
+    }
+
+    #[test]
+    fn a_complete_advertisement_and_request_validate() -> anyhow::Result<()> {
+        validate_media_stream_advertisement(&advertisement())?;
+        validate_media_stream_request(&request())?;
+        Ok(())
+    }
+
+    #[test]
+    fn source_ids_and_labels_must_pair_up() {
+        let mut record = advertisement();
+        record.video_source_labels.clear();
+        let error = validate_media_stream_advertisement(&record).unwrap_err();
+        assert!(error.to_string().contains("video source ids"), "{error}");
+    }
+
+    #[test]
+    fn audio_on_offer_needs_a_sample_rate_and_channels() {
+        let mut record = advertisement();
+        record.audio_channels = 0;
+        assert!(validate_media_stream_advertisement(&record).is_err());
+        record.audio_source_ids.clear();
+        record.audio_source_labels.clear();
+        assert!(validate_media_stream_advertisement(&record).is_ok(), "no audio, no requirement");
+    }
+
+    #[test]
+    fn a_start_needs_somewhere_to_send_and_something_to_send() {
+        let mut record = request();
+        record.receiver_endpoint = "starfire".into();
+        assert!(validate_media_stream_request(&record).unwrap_err().to_string().contains("host:port"));
+        let mut record = request();
+        record.video_source_id.clear();
+        record.audio_source_id.clear();
+        assert!(validate_media_stream_request(&record).unwrap_err().to_string().contains("neither"));
+        let mut record = request();
+        record.video_codec.clear();
+        assert!(validate_media_stream_request(&record).unwrap_err().to_string().contains("codec"));
+    }
+
+    #[test]
+    fn a_stop_needs_no_endpoint_or_sources() -> anyhow::Result<()> {
+        let mut record = request();
+        record.action = "stop".into();
+        record.receiver_endpoint.clear();
+        record.video_source_id.clear();
+        record.audio_source_id.clear();
+        validate_media_stream_request(&record)
+    }
+
+    #[test]
+    fn unknown_actions_and_states_are_refused() {
+        let mut record = request();
+        record.action = "pause".into();
+        assert!(validate_media_stream_request(&record).is_err());
+        let mut record = request();
+        record.state = "done".into();
+        assert!(validate_media_stream_request(&record).is_err());
+    }
+
+    #[test]
+    fn one_request_per_stream_per_receiver() {
+        assert_eq!(media_stream_request_key("s", "starfire.obs"), "s:starfire.obs");
+    }
+
+    /// The records round-trip through the same positional encoding every
+    /// other runtime reads, so a Python or TypeScript consumer sees the same
+    /// slots.
+    #[test]
+    fn records_round_trip_positionally() -> anyhow::Result<()> {
+        let record = advertisement();
+        let encoded = rmp_serde::to_vec(&record)?;
+        let decoded: GameCultMediaStreamAdvertisementRecord = rmp_serde::from_slice(&encoded)?;
+        assert_eq!(decoded, record);
+        let record = request();
+        let encoded = rmp_serde::to_vec(&record)?;
+        let decoded: GameCultMediaStreamRequestRecord = rmp_serde::from_slice(&encoded)?;
+        assert_eq!(decoded, record);
+        Ok(())
+    }
+}
 
 /// One coded video access unit, possibly split across `chunk_count` records
 /// sharing a `frame_id`. A consumer reassembles by `(stream_id, session_id,