diff --git a/Mimir.slnx b/Mimir.slnx index dbd4f1b..a0de91f 100644 --- a/Mimir.slnx +++ b/Mimir.slnx @@ -6,5 +6,6 @@ + diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 0627804..c100387 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -279,14 +279,14 @@ a named invariant that the native runtime cannot protect yet. controls/previews/stats, Yggdrasil publishes to the site, and OBS is compatibility-only. - `src/Mimir.CultMeshMedia` is the first explicit CultMesh media/body bridge: - `relay` hosts the document relay on CultNet UDP `3075`, `send` reads an + `relay` hosts the document relay on CultNet RUDP `3075`, `send` reads an MPEG-TS byte stream from stdin and publishes rolling `mimir.cultmesh_media_frame` documents, and `recv` subscribes to those documents and writes ordered MPEG-TS bytes to a Starfire-local UDP endpoint - for compatibility sinks. Source audit: its C# implementation still uses older - `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` entrypoints; migrate the - bridge to CultLib's explicit C# RUDP helpers before treating it as fully - aligned with the daemon-swarm transport cut. + for compatibility sinks. Its C# implementation now uses explicit CultLib + RUDP client/session helpers; the older + `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` entrypoints no longer own + the media transport lane. - `scripts/start-raven-cultmesh-av-sender.ps1`, `scripts/start-yggdrasil-cultmesh-media-relay.ps1`, and `scripts/start-starfire-cultmesh-av-receiver.ps1` are the CultMesh bridge @@ -322,9 +322,9 @@ a named invariant that the native runtime cannot protect yet. discovery, Idunn owns keepalive decisions, and Mimir-owned dashboard/reference surfaces only report their own observed state. TCP/HTTP/WebSocket stay lowerings or compatibility evidence. -- `Mimir.CultMeshMedia` still needs an explicit RUDP transport cut. Do not add - another TCP bridge, HTTP status shim, or WebSocket-derived service truth while - the C# RUDP helpers exist. +- `Mimir.CultMeshMedia` has completed its explicit RUDP transport cut. Do not + add another TCP bridge, HTTP status shim, or WebSocket-derived service truth + while the RUDP document lane exists. - Process-backed stream sources are only acceptable for network bridge feeds or diagnostics. Six-camera local ingest belongs behind direct capture drivers. - Frame-event process sources are diagnostic only. They prove source cadence and @@ -337,9 +337,10 @@ a named invariant that the native runtime cannot protect yet. 1. Replace the frame-event diagnostic bridge with concrete direct capture drivers for Leap stereo IR first, then the other cameras. -2. Cut `src/Mimir.CultMeshMedia` from older CultMesh node/client entrypoints to - explicit CultNet RUDP helpers, preferably authorized-peer dialing where the - peer catalog exists. Preserve OBS-local UDP as egress only. +2. Cut the remaining dashboard, recorder, and witness service truth paths from + older TCP/WebSocket-derived assumptions to explicit CultNet RUDP records, + preferably authorized-peer dialing where the peer catalog exists. Preserve + OBS-local UDP as egress only for compatibility sinks. 3. Feed those drivers into `MimirVideoCaptureDriverSource` and prove sustained frame cadence in the rolling buffers. 4. Promote the packet-song physical calibration receipt into the runtime diff --git a/notes/current-system-map.md b/notes/current-system-map.md index 8c4ed75..9d3d181 100644 --- a/notes/current-system-map.md +++ b/notes/current-system-map.md @@ -257,13 +257,12 @@ video archive. Yggdrasil relays CultNet reliable UDP over the WireGuard mesh. Starfire lowers the subscribed stream to local UDP for compatibility sinks because OBS is not a CultMesh runtime. -Source audit correction: the desired transport is explicit CultNet RUDP, and -CultLib now provides C# `CultMesh.CreateRudpServer`, -`CultMesh.CreateRudpClient`, and authorized-peer helper APIs. The current -`Mimir.CultMeshMedia` source still uses older `CultMesh.StartNodeAsync` and -`CultMesh.ConnectClient` entrypoints. Treat that as migration debt: the bridge -should be cut to explicit RUDP helpers before it is considered aligned with the -daemon-swarm transport doctrine. +Source audit correction resolved: `Mimir.CultMeshMedia` now uses explicit +CultLib RUDP helpers for the sender/receiver and a relay-owned RUDP +socket/session loop for multi-peer forwarding. The older +`CultMesh.StartNodeAsync` and `CultMesh.ConnectClient` path no longer owns this +media lane. The relay still persists received puts to CultCache as evidence and +for inspection; live network transit is the RUDP schema-document path. Current deployment: Yggdrasil runs the relay from `/opt/gamecult/mimir-cultmesh-media/Mimir.CultMeshMedia` with cache @@ -315,7 +314,8 @@ browser/dashboard WebSocket paths are still renderer/client lowerings. They may remain as lowering adapters, but any provider catalog, retained service state, command boundary, transport profile, or lifecycle health claim must be published as typed CultNet/RUDP records instead of being inferred from those -TCP/WebSocket surfaces. +TCP/WebSocket surfaces. The media-frame bridge is already on that side of the +line. ## Audio Field diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index 5982226..904eb1e 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -25,10 +25,10 @@ Get-Content .\state\evidence.jsonl -Tail 8 - V1 now also has an explicit CultMesh media bridge for Raven program feeds: `src/Mimir.CultMeshMedia` sends rolling `mimir.cultmesh_media_frame` documents over CultNet reliable UDP through Yggdrasil and lowers them on - Starfire to local MPEG-TS UDP for OBS. Source audit warning: the current C# - bridge still uses older `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient`; - migrate it to explicit C# RUDP helpers before treating the media bridge as - fully aligned with the swarm transport cut. + Starfire to local MPEG-TS UDP for OBS. The bridge now uses explicit CultLib + RUDP helpers/session plumbing; the old + `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` path no longer owns this + media lane. - CultLib has moved the daemon-swarm direction to CultNet over RUDP across runtimes. Mimir's typed daemon truth should default to `cultnet.transport.rudp.v0`; TCP/HTTP/WebSocket are client lowerings, @@ -103,9 +103,9 @@ Get-Content .\state\evidence.jsonl -Tail 8 dashboard boundary cut is provider advertisement, retained state, command_boundary, and transport_profile over RUDP so HTTP/WebSocket catalog ingestion can be demoted to client/debug lowering. -- Cut `Mimir.CultMeshMedia` to explicit CultNet RUDP helpers next; no new TCP, - HTTP, or WebSocket surface should be allowed to become daemon or transport - truth. +- Cut the remaining Eve dashboard, recorder, and witness service truth paths to + explicit CultNet RUDP records next; no new TCP, HTTP, or WebSocket surface + should be allowed to become daemon or transport truth. - Kiyo Pro has two UVC extension units. Moving it to a motherboard USB3 port changed the descriptor path to `root_hub30` / `bcdUSB=0x0320` and exposed 720p/1080p YUY2/MJPG/H264/NV12 at 60 fps, but Windows still reports diff --git a/scripts/start-raven-cultmesh-av-sender.ps1 b/scripts/start-raven-cultmesh-av-sender.ps1 new file mode 100644 index 0000000..7dda8a9 --- /dev/null +++ b/scripts/start-raven-cultmesh-av-sender.ps1 @@ -0,0 +1,89 @@ +param( + [string]$RelayHost = "10.77.0.1", + [int]$RelayPort = 3075, + [string]$Stream = "raven-primary-av", + [string]$FfmpegPath = "ffmpeg", + [string]$VideoInput = "desktop", + [ValidateSet("WasapiLoopback", "DirectShow")] + [string]$AudioMode = "WasapiLoopback", + [string]$AudioInput = "Stereo Mix (Realtek(R) Audio)", + [ValidateSet("Console", "Multimedia", "Communications")] + [string]$LoopbackRole = "Console", + [int]$Width = 1920, + [int]$Height = 1080, + [int]$Framerate = 30, + [string]$VideoBitrate = "8000k", + [string]$AudioBitrate = "192k", + [string]$MediaToolPath = "" +) + +$ErrorActionPreference = "Stop" +$project = Join-Path $PSScriptRoot "..\src\Mimir.CultMeshMedia\Mimir.CultMeshMedia.csproj" +$publishedTool = Join-Path $PSScriptRoot "..\artifacts\mimir-cultmesh-media-win-x64-selfcontained\Mimir.CultMeshMedia.exe" +$loopbackScript = Join-Path $PSScriptRoot "wasapi-loopback-capture.ps1" + +if ([string]::IsNullOrWhiteSpace($MediaToolPath) -and (Test-Path $publishedTool)) { + $MediaToolPath = $publishedTool +} + +$ffmpegArgs = @( + "-hide_banner", + "-loglevel", "warning", + "-f", "gdigrab", + "-framerate", $Framerate, + "-video_size", "${Width}x${Height}", + "-i", $VideoInput +) + +if ($AudioMode -eq "WasapiLoopback") { + $ffmpegArgs += @( + "-f", "f32le", + "-ar", "48000", + "-ac", "2", + "-i", "pipe:0" + ) +} else { + $ffmpegArgs += @( + "-f", "dshow", + "-i", "audio=$AudioInput" + ) +} + +$ffmpegArgs += @( + "-map", "0:v:0", + "-map", "1:a:0", + "-c:v", "h264_nvenc", + "-preset", "p3", + "-tune", "ll", + "-b:v", $VideoBitrate, + "-maxrate", $VideoBitrate, + "-bufsize", "16000k", + "-g", ($Framerate * 2), + "-pix_fmt", "yuv420p", + "-c:a", "aac", + "-b:a", $AudioBitrate, + "-ar", "48000", + "-ac", "2", + "-f", "mpegts", + "pipe:1" +) + +$senderArgs = @("send", "--host", $RelayHost, "--port", $RelayPort, "--stream", $Stream, "--producer", "raven", "--chunk-bytes", "16384", "--slots", "96") + +if ($AudioMode -eq "WasapiLoopback") { + if (-not (Test-Path $loopbackScript)) { + throw "WASAPI loopback script not found: $loopbackScript" + } + + if ([string]::IsNullOrWhiteSpace($MediaToolPath)) { + & $loopbackScript -Output stdout -Role $LoopbackRole | & $FfmpegPath @ffmpegArgs | dotnet run --project $project -- @senderArgs + } else { + & $loopbackScript -Output stdout -Role $LoopbackRole | & $FfmpegPath @ffmpegArgs | & $MediaToolPath @senderArgs + } +} else { + if ([string]::IsNullOrWhiteSpace($MediaToolPath)) { + & $FfmpegPath @ffmpegArgs | dotnet run --project $project -- @senderArgs + } else { + & $FfmpegPath @ffmpegArgs | & $MediaToolPath @senderArgs + } +} diff --git a/scripts/start-starfire-cultmesh-av-receiver.ps1 b/scripts/start-starfire-cultmesh-av-receiver.ps1 new file mode 100644 index 0000000..67bad67 --- /dev/null +++ b/scripts/start-starfire-cultmesh-av-receiver.ps1 @@ -0,0 +1,10 @@ +param( + [string]$RelayHost = "10.77.0.1", + [int]$RelayPort = 3075, + [string]$Stream = "raven-primary-av", + [string]$Udp = "127.0.0.1:5200" +) + +$ErrorActionPreference = "Stop" +$project = Join-Path $PSScriptRoot "..\src\Mimir.CultMeshMedia\Mimir.CultMeshMedia.csproj" +dotnet run --project $project -- recv --host $RelayHost --port $RelayPort --stream $Stream --udp $Udp diff --git a/scripts/start-yggdrasil-cultmesh-media-relay.ps1 b/scripts/start-yggdrasil-cultmesh-media-relay.ps1 new file mode 100644 index 0000000..77c4126 --- /dev/null +++ b/scripts/start-yggdrasil-cultmesh-media-relay.ps1 @@ -0,0 +1,8 @@ +param( + [string]$Cache = "C:\Meta\Mimir\state\mimir-cultmesh-media.cc" +) + +$ErrorActionPreference = "Stop" +$project = Join-Path $PSScriptRoot "..\src\Mimir.CultMeshMedia\Mimir.CultMeshMedia.csproj" +New-Item -ItemType Directory -Force -Path (Split-Path -Parent $Cache) | Out-Null +dotnet run --project $project -- relay --cache $Cache diff --git a/src/Mimir.CultMeshMedia/Mimir.CultMeshMedia.csproj b/src/Mimir.CultMeshMedia/Mimir.CultMeshMedia.csproj new file mode 100644 index 0000000..cb154c1 --- /dev/null +++ b/src/Mimir.CultMeshMedia/Mimir.CultMeshMedia.csproj @@ -0,0 +1,18 @@ + + + + Exe + net10.0 + Mimir.CultMeshMedia + Mimir.CultMeshMedia + enable + enable + + + + + + + + + diff --git a/src/Mimir.CultMeshMedia/Program.cs b/src/Mimir.CultMeshMedia/Program.cs new file mode 100644 index 0000000..a497083 --- /dev/null +++ b/src/Mimir.CultMeshMedia/Program.cs @@ -0,0 +1,568 @@ +using System.Diagnostics; +using System.Globalization; +using System.Net; +using System.Net.Sockets; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using GameCult.Mesh; +using GameCult.Networking; +using Mimir.Runtime.Synchronization; + +const string SchemaName = "mimir.cultmesh_media_frame"; +const string DefaultStreamId = "raven-primary-av"; +const uint DefaultConnectionId = 0x4d4d_0101; +const int DefaultPort = 3075; +const int DefaultMaxFragmentBytes = 1200; +const int DefaultMaxPendingReliablePackets = 8192; + +var command = args.Length > 0 ? args[0].Trim().ToLowerInvariant() : string.Empty; +var options = CliOptions.Parse(args.Skip(1)); + +return command switch +{ + "relay" => await RunRelayAsync(options), + "send" => await RunSenderAsync(options), + "recv" or "receive" => await RunReceiverAsync(options), + _ => Usage() +}; + +static int Usage() +{ + Console.Error.WriteLine(""" +Usage: + Mimir.CultMeshMedia relay --cache [--bind 0.0.0.0:3075] + Mimir.CultMeshMedia send --host [--port 3075] [--stream raven-primary-av] [--chunk-bytes 16384] [--slots 96] + Mimir.CultMeshMedia recv --host [--port 3075] [--stream raven-primary-av] [--udp 127.0.0.1:5200] + +send reads an MPEG-TS byte stream from stdin and publishes rolling CultMesh media-frame documents over CultNet RUDP. +recv subscribes to those RUDP document changes and writes ordered payload bytes to a local UDP MPEG-TS endpoint for OBS. +"""); + return 2; +} + +static async Task RunRelayAsync(CliOptions options) +{ + var cachePath = options.Get("cache") ?? Path.Combine(Environment.CurrentDirectory, "mimir-cultmesh-media.cc"); + var bind = ParseEndpoint(options.Get("bind") ?? $"0.0.0.0:{options.GetInt("port", DefaultPort)}"); + var registry = CreateDocumentRegistry(); + var schemaId = ResolveMediaSchemaId(registry); + using var cache = await CultCacheMessagePack.OpenAsync(cachePath).ConfigureAwait(false); + using var relay = new RudpMediaRelay( + bind, + cache, + registry, + schemaId, + options.GetUInt("connection-id", DefaultConnectionId), + options.GetInt("max-fragment-bytes", DefaultMaxFragmentBytes), + options.GetInt("max-pending-reliable", DefaultMaxPendingReliablePackets)); + + Console.WriteLine($"CultMesh media relay listening on rudp://{bind.Address}:{bind.Port}; cache={Path.GetFullPath(cachePath)}"); + await relay.RunAsync().ConfigureAwait(false); + await cache.FlushAsync().ConfigureAwait(false); + return 0; +} + +static async Task RunSenderAsync(CliOptions options) +{ + var endpoint = ResolveRudpEndpoint(options); + var streamId = options.Get("stream") ?? DefaultStreamId; + var chunkBytes = Math.Clamp(options.GetInt("chunk-bytes", 16 * 1024), 1316, 60 * 1024); + var slots = Math.Clamp(options.GetInt("slots", 96), 8, 4096); + var producer = options.Get("producer") ?? Environment.MachineName.ToLowerInvariant(); + var registry = CreateDocumentRegistry(); + + using var transport = CreateRudpClient( + options.Get("runtime") ?? $"{producer}-mimir-media-sender", + endpoint, + options); + ConnectOrThrow(transport, endpoint.Uri); + + Console.Error.WriteLine($"Sending stdin MPEG-TS to {endpoint.Uri}/{streamId}; chunkBytes={chunkBytes}; slots={slots}"); + + var input = Console.OpenStandardInput(); + var buffer = new byte[chunkBytes]; + long sequence = 0; + var started = Stopwatch.StartNew(); + + while (true) + { + var read = await input.ReadAsync(buffer.AsMemory(0, buffer.Length)).ConfigureAwait(false); + if (read <= 0) + { + break; + } + + var payload = buffer.AsSpan(0, read).ToArray(); + var document = new MimirCultMeshMediaFrameDocument( + FrameId: $"{streamId}:{sequence:D20}", + StreamId: streamId, + Sequence: sequence, + ProducedAtUtc: DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture), + TimestampNanoseconds: started.ElapsedTicks * 1_000_000_000L / Stopwatch.Frequency, + PayloadKind: "mpegts-chunk", + Container: "mpegts", + VideoCodec: options.Get("video-codec") ?? "h264", + AudioCodec: options.Get("audio-codec") ?? "aac", + PayloadBytes: payload.Length, + Payload: payload, + ProducerNode: producer, + ClockDomainId: options.Get("clock-domain") ?? "raven-monotonic", + Tags: ["raven", "program", "obs", "cultmesh-media", "cultnet.transport.rudp.v0"]); + + var handle = new CultRecordHandle( + new CultRecordKey($"{streamId}:slot:{sequence % slots:D4}")); + var message = registry.CreateRawDocumentPutMessage( + Guid.NewGuid().ToString("N"), + handle, + document, + new CultNetDocumentMessageOptions + { + SourceRuntimeId = producer, + SourceRole = "mimir-cultmesh-media-sender", + Tags = ["mimir", "media", streamId, "cultnet.transport.rudp.v0"] + }); + + transport.SendSchema(CultNetSchemaMessageSerialization.Serialize(message)); + PumpClient(transport, TimeSpan.FromMilliseconds(5)); + sequence++; + if (sequence % 120 == 0) + { + Console.Error.WriteLine($"sent sequence={sequence - 1}"); + } + } + + Console.Error.WriteLine($"sent chunks={sequence}; draining"); + PumpClient(transport, TimeSpan.FromMilliseconds(options.GetInt("drain-ms", 1000))); + return 0; +} + +static async Task RunReceiverAsync(CliOptions options) +{ + var endpoint = ResolveRudpEndpoint(options); + var streamId = options.Get("stream") ?? DefaultStreamId; + var udp = options.Get("udp") ?? "127.0.0.1:5200"; + var udpEndpoint = ParseEndpoint(udp); + var registry = CreateDocumentRegistry(); + var schemaId = ResolveMediaSchemaId(registry); + var lastSequence = -1L; + long frames = 0; + long bytes = 0; + var loggedFirstFrame = false; + + using var socket = new UdpClient(); + using var transport = CreateRudpClient( + options.Get("runtime") ?? $"{Environment.MachineName.ToLowerInvariant()}-mimir-media-receiver", + endpoint, + options); + ConnectOrThrow(transport, endpoint.Uri); + transport.SendSchema(CultNetSchemaMessageSerialization.Serialize(new CultNetDatabaseSubscribeMessage + { + MessageId = Guid.NewGuid().ToString("N"), + SubscriptionId = $"mimir-media-{streamId}", + SchemaIds = [schemaId], + IncludeSnapshot = false + })); + + Console.WriteLine($"Receiving {endpoint.Uri}/{streamId}; writing MPEG-TS UDP to udp://{udp}"); + while (true) + { + transport.PollResends(); + if (transport.ReceiveOnce() is not { } frame) + { + await Task.Delay(2).ConfigureAwait(false); + continue; + } + + if (!string.Equals(frame.ChannelId, "schema", StringComparison.Ordinal)) + { + continue; + } + + if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is not CultNetDatabaseChangeRawMessage change || + change.Document is not { } rawDocument || + !string.Equals(rawDocument.SchemaId, schemaId, StringComparison.Ordinal)) + { + continue; + } + + var document = (MimirCultMeshMediaFrameDocument)registry + .GetBySchemaId(schemaId)! + .PayloadDeserializer(rawDocument.Payload); + + if (!string.Equals(document.StreamId, streamId, StringComparison.Ordinal) || + document.Sequence <= lastSequence) + { + continue; + } + + lastSequence = document.Sequence; + socket.Send(document.Payload, document.Payload.Length, udpEndpoint); + frames++; + bytes += document.Payload.Length; + + if (!loggedFirstFrame) + { + loggedFirstFrame = true; + Console.WriteLine($"received first sequence={document.Sequence} bytes={document.PayloadBytes}"); + } + + if (frames % 120 == 0) + { + Console.WriteLine($"received sequence={document.Sequence} frames={frames} bytes={bytes}"); + } + } +} + +static CultMeshRudpEndpoint ResolveRudpEndpoint(CliOptions options) +{ + if (options.Get("endpoint") is { } endpoint) + { + return CultMesh.ParseRudpEndpoint(endpoint); + } + + var host = options.Get("host") ?? "10.77.0.1"; + var port = options.GetInt("port", DefaultPort); + return CultMesh.ParseRudpEndpoint($"rudp://{host}:{port}"); +} + +static CultNetRudpSocketTransportConnection CreateRudpClient( + string runtimeId, + CultMeshRudpEndpoint endpoint, + CliOptions options) +{ + return CultMesh.CreateRudpClient( + runtimeId, + options.GetUInt("connection-id", DefaultConnectionId), + endpoint, + new CultMeshRudpSocketOptions + { + BindHost = options.Get("bind-host") ?? "0.0.0.0", + BindPort = options.GetInt("bind-port", 0), + ResendDelayMs = options.GetInt("resend-ms", 100), + MaxFragmentBytes = options.GetInt("max-fragment-bytes", DefaultMaxFragmentBytes), + MaxPendingReliablePackets = options.GetInt("max-pending-reliable", DefaultMaxPendingReliablePackets) + }); +} + +static void ConnectOrThrow(CultNetRudpSocketTransportConnection transport, string endpoint) +{ + if (!transport.ConnectAndWait(Array.Empty(), TimeSpan.FromSeconds(10), TimeSpan.FromMilliseconds(5))) + { + throw new TimeoutException($"timed out connecting RUDP media transport to {endpoint}"); + } +} + +static void PumpClient(CultNetRudpSocketTransportConnection transport, TimeSpan duration) +{ + var until = Stopwatch.StartNew(); + while (until.Elapsed < duration) + { + transport.PollResends(); + transport.ReceiveOnce(); + Thread.Sleep(1); + } +} + +static string ResolveMediaSchemaId(CultNetDocumentRegistry registry) +{ + var binding = registry.GetByDocumentType(typeof(MimirCultMeshMediaFrameDocument)) ?? + throw new InvalidOperationException($"CultNet document binding missing for {SchemaName}."); + return binding.SchemaId; +} + +static CultNetDocumentRegistry CreateDocumentRegistry() +{ + var documents = CultDocumentRegistry.Shared; + documents.GetRequired(); + return new CultNetDocumentRegistry(documents) + .Register(CultNetDocumentBinding.ForDocument(documents)); +} + +static IPEndPoint ParseEndpoint(string value) +{ + var split = value.LastIndexOf(':'); + if (split <= 0 || split == value.Length - 1) + { + throw new ArgumentException($"Endpoint must be host:port, not '{value}'."); + } + + var host = value[..split]; + var port = int.Parse(value[(split + 1)..], CultureInfo.InvariantCulture); + var addresses = Dns.GetHostAddresses(host); + return new IPEndPoint(addresses.First(address => address.AddressFamily == AddressFamily.InterNetwork), port); +} + +sealed class RudpMediaRelay : IDisposable +{ + private readonly Socket socket; + private readonly CultCache cache; + private readonly CultNetDocumentRegistry registry; + private readonly string schemaId; + private readonly uint connectionId; + private readonly int maxFragmentBytes; + private readonly int maxPendingReliablePackets; + private readonly Dictionary peers = new(StringComparer.Ordinal); + + public RudpMediaRelay( + IPEndPoint bind, + CultCache cache, + CultNetDocumentRegistry registry, + string schemaId, + uint connectionId, + int maxFragmentBytes, + int maxPendingReliablePackets) + { + this.cache = cache; + this.registry = registry; + this.schemaId = schemaId; + this.connectionId = connectionId; + this.maxFragmentBytes = maxFragmentBytes; + this.maxPendingReliablePackets = maxPendingReliablePackets; + socket = new Socket(bind.AddressFamily, SocketType.Dgram, ProtocolType.Udp); + socket.Bind(bind); + socket.Blocking = false; + } + + public async Task RunAsync() + { + while (true) + { + ReceiveAvailable(); + PumpReliableResends(); + await Task.Delay(2).ConfigureAwait(false); + } + } + + public void Dispose() + { + socket.Dispose(); + } + + private void ReceiveAvailable() + { + while (socket.Poll(0, SelectMode.SelectRead)) + { + var buffer = new byte[65535]; + EndPoint remote = new IPEndPoint(IPAddress.Any, 0); + int received; + try + { + received = socket.ReceiveFrom(buffer, ref remote); + } + catch (SocketException error) when (error.SocketErrorCode == SocketError.WouldBlock) + { + return; + } + + var wire = buffer.AsSpan(0, received).ToArray(); + var packet = CultNetRudpPacketCodec.Decode(wire); + if (packet.ConnectionId != connectionId) + { + continue; + } + + var peer = GetPeer((IPEndPoint)remote); + if (packet.PacketType == CultNetRudpPacketType.Connect) + { + SendPacket(peer, peer.Session.AcceptConnect(packet, NowMs())); + continue; + } + + var result = peer.Session.Receive(packet, NowMs()); + if (result.Reply != null) + { + SendPacket(peer, result.Reply); + } + + if (result.Disconnected) + { + peers.Remove(peer.Key); + continue; + } + + foreach (var frame in result.Delivered.Where(static frame => string.Equals(frame.ChannelId, "schema", StringComparison.Ordinal))) + { + HandleSchemaFrameAsync(peer, frame.Payload).GetAwaiter().GetResult(); + } + + if (packet.PacketType == CultNetRudpPacketType.Accept || result.Delivered.Count > 0) + { + SendPacket(peer, peer.Session.CreateAck()); + } + } + } + + private async Task HandleSchemaFrameAsync(RelayPeer peer, byte[] payload) + { + var message = CultNetSchemaMessageSerialization.Deserialize(payload); + switch (message) + { + case CultNetDocumentPutRawMessage put: + await ApplyPutAsync(put).ConfigureAwait(false); + break; + case CultNetDatabaseSubscribeMessage subscribe: + peer.Subscriptions[subscribe.SubscriptionId] = subscribe; + Console.WriteLine($"relay RUDP subscribe peer={peer.Key} subscription={subscribe.SubscriptionId} schemas={string.Join(",", subscribe.SchemaIds ?? [])}"); + break; + case CultNetDatabaseUnsubscribeMessage unsubscribe: + peer.Subscriptions.Remove(unsubscribe.SubscriptionId); + break; + } + } + + private async Task ApplyPutAsync(CultNetDocumentPutRawMessage put) + { + if (put.Document == null || + !string.Equals(put.Document.SchemaId, schemaId, StringComparison.Ordinal)) + { + return; + } + + await registry.ApplyRawDocumentPutMessageAsync(cache, put).ConfigureAwait(false); + await cache.FlushAsync(soft: true).ConfigureAwait(false); + var change = new CultNetDatabaseChangeRawMessage + { + MessageId = Guid.NewGuid().ToString("N"), + ChangeKind = "updated", + Document = put.Document + }; + + var frame = CultNetSchemaMessageSerialization.Serialize(change); + foreach (var peer in peers.Values) + { + foreach (var subscription in peer.Subscriptions.Values) + { + if (!Matches(subscription, put.Document.SchemaId, put.Document.RecordKey)) + { + continue; + } + + change.SubscriptionId = subscription.SubscriptionId; + SendSchema(peer, CultNetSchemaMessageSerialization.Serialize(change)); + if (!peer.HasForwardedFrame) + { + peer.HasForwardedFrame = true; + Console.WriteLine($"relay RUDP forward peer={peer.Key} subscription={subscription.SubscriptionId} schema={put.Document.SchemaId}"); + } + ReceiveAvailable(); + } + } + + var document = (MimirCultMeshMediaFrameDocument)registry + .GetBySchemaId(schemaId)! + .PayloadDeserializer(put.Document.Payload); + if (document.Sequence % 120 == 0) + { + Console.WriteLine($"relay RUDP put stream={document.StreamId} sequence={document.Sequence} bytes={document.PayloadBytes}"); + } + } + + private static bool Matches(CultNetDatabaseSubscribeMessage request, string schemaId, string recordKey) + { + var schemaMatches = request.SchemaIds == null || + request.SchemaIds.Length == 0 || + request.SchemaIds.Contains(schemaId, StringComparer.Ordinal); + var keyMatches = request.RecordKeys == null || + request.RecordKeys.Length == 0 || + request.RecordKeys.Contains(recordKey, StringComparer.Ordinal); + return schemaMatches && keyMatches; + } + + private RelayPeer GetPeer(IPEndPoint endpoint) + { + var key = endpoint.ToString(); + if (peers.TryGetValue(key, out var existing)) + { + return existing; + } + + var peer = new RelayPeer( + key, + endpoint, + new CultNetRudpSession(new CultNetRudpSessionOptions + { + ConnectionId = connectionId, + InitialSequence = (uint)(peers.Count + 1), + ResendDelayMs = 100, + MaxPendingReliablePackets = maxPendingReliablePackets + })); + peers[key] = peer; + return peer; + } + + private void SendSchema(RelayPeer peer, byte[] payload) + { + foreach (var packet in peer.Session.SendMany( + "schema", + payload, + new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = NowMs() }, + maxFragmentBytes)) + { + SendPacket(peer, packet); + } + } + + private void SendPacket(RelayPeer peer, CultNetRudpPacket packet) + { + var wire = CultNetRudpPacketCodec.Encode(packet); + socket.SendTo(wire, peer.Endpoint); + } + + private void PumpReliableResends() + { + foreach (var peer in peers.Values) + { + foreach (var packet in peer.Session.DueResends(NowMs())) + { + SendPacket(peer, packet); + } + } + } + + private static long NowMs() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); +} + +sealed record RelayPeer( + string Key, + IPEndPoint Endpoint, + CultNetRudpSession Session) +{ + public Dictionary Subscriptions { get; } = new(StringComparer.Ordinal); + public bool HasForwardedFrame { get; set; } +} + +sealed class CliOptions +{ + private readonly Dictionary values = new(StringComparer.OrdinalIgnoreCase); + + public static CliOptions Parse(IEnumerable args) + { + var parsed = new CliOptions(); + var pendingKey = string.Empty; + foreach (var arg in args) + { + if (arg.StartsWith("--", StringComparison.Ordinal)) + { + pendingKey = arg[2..]; + parsed.values[pendingKey] = "true"; + continue; + } + + if (!string.IsNullOrWhiteSpace(pendingKey)) + { + parsed.values[pendingKey] = arg; + pendingKey = string.Empty; + } + } + + return parsed; + } + + public string? Get(string key) => values.TryGetValue(key, out var value) ? value : null; + + public int GetInt(string key, int fallback) => + int.TryParse(Get(key), NumberStyles.Integer, CultureInfo.InvariantCulture, out var value) ? value : fallback; + + public uint GetUInt(string key, uint fallback) => + uint.TryParse(Get(key), NumberStyles.Integer, CultureInfo.InvariantCulture, out var value) ? value : fallback; +} diff --git a/src/Mimir.Runtime/Synchronization/MimirCultMeshMediaFrameDocument.cs b/src/Mimir.Runtime/Synchronization/MimirCultMeshMediaFrameDocument.cs new file mode 100644 index 0000000..2f77aed --- /dev/null +++ b/src/Mimir.Runtime/Synchronization/MimirCultMeshMediaFrameDocument.cs @@ -0,0 +1,24 @@ +using GameCult.Caching; +using MessagePack; + +namespace Mimir.Runtime.Synchronization; + +[CultDocument("mimir.cultmesh_media_frame", "mimir.cultmesh_media_frame.v1")] +[MessagePackObject] +public sealed record MimirCultMeshMediaFrameDocument( + [property: Key(0)] + [property: CultName] + string FrameId, + [property: Key(1)] string StreamId, + [property: Key(2)] long Sequence, + [property: Key(3)] string ProducedAtUtc, + [property: Key(4)] long TimestampNanoseconds, + [property: Key(5)] string PayloadKind, + [property: Key(6)] string Container, + [property: Key(7)] string VideoCodec, + [property: Key(8)] string AudioCodec, + [property: Key(9)] int PayloadBytes, + [property: Key(10)] byte[] Payload, + [property: Key(11)] string ProducerNode, + [property: Key(12)] string ClockDomainId, + [property: Key(13)] string[] Tags); diff --git a/src/Mimir.Runtime/Synchronization/MimirNetworkTransportConfiguration.cs b/src/Mimir.Runtime/Synchronization/MimirNetworkTransportConfiguration.cs index 13b2636..e2e48ed 100644 --- a/src/Mimir.Runtime/Synchronization/MimirNetworkTransportConfiguration.cs +++ b/src/Mimir.Runtime/Synchronization/MimirNetworkTransportConfiguration.cs @@ -6,12 +6,14 @@ public enum MimirNetworkPayloadKind RawAudioDebugWindow, CompressedAudioProgramFeed, CompressedVideoProgramFeed, - RawVideoDiagnosticWindow + RawVideoDiagnosticWindow, + MuxedAudioVideoProgramFeed } public enum MimirNetworkTransportKind { CultMeshTypedState, + CultMeshReliableUdpMedia, LanUdpDatagrams, SrtDiagnosticMedia, WebRtcExperimental, @@ -68,6 +70,18 @@ public static class MimirNetworkTransportConfigurations RejectionConditions: ["used-as-clock-authority", "used-for-local-hot-path"], "Useful bridge. Not the Perfect Machine's synchronization core."); + public static MimirNetworkTransportConfiguration CultMeshProgramBridge { get; } = new( + "cultmesh-program-bridge", + "Reliable-UDP CultMesh media bridge for muxed program frames that OBS consumes through a local Starfire endpoint.", + MimirNetworkTransportKind.CultMeshReliableUdpMedia, + MimirNetworkPayloadKind.MuxedAudioVideoProgramFeed, + MayAffectClock: false, + CarriesRawMedia: true, + TargetLatencyMilliseconds: 500.0, + RequiredDocuments: ["mimir.cultmesh_media_frame"], + RejectionConditions: ["used-as-clock-authority", "unbounded-media-cache", "non-local-obs-egress"], + "Raven publishes rolling media-frame documents over CultNet reliable UDP; Starfire lowers them to a local OBS MPEG-TS endpoint."); + public static MimirNetworkTransportConfiguration WebRtcExperiment { get; } = CultMeshTimingState with { Id = "webrtc-experimental-media", @@ -84,6 +98,7 @@ public static class MimirNetworkTransportConfigurations [ CultMeshTimingState, RawAudioDebugWindow, + CultMeshProgramBridge, SrtProgramBridge, WebRtcExperiment ]; diff --git a/state/evidence.jsonl b/state/evidence.jsonl index 6ea427a..065e153 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -142,3 +142,4 @@ {"ts":"2026-06-16T00:00:00+01:00","type":"correction","status":"ok","note":"Reoriented Mimir's daemon-swarm map after Odin/CultLib RUDP work: Raven media body transport remains src/Mimir.CultMeshMedia over CultNet/CultMesh reliable UDP with OBS-local UDP as egress only, while Eve dashboard/browser reference health now publishes idunn.daemon_health over cultnet.transport.rudp.v0 as self-witness state. Odin owns accepted Verse/service discovery and Idunn owns keepalive decisions; Mimir health publishers do not own lifecycle."} {"ts":"2026-06-16T00:05:00+01:00","type":"correction","status":"ok","note":"Corrected the transport model: CultLib has added authorized RUDP helpers across runtimes and Odin/Idunn now treats daemon truth as typed CultCache/CultMesh state carried over cultnet.transport.rudp.v0. Mimir should not assume TCP is needed for any daemon-truth path once the runtime has RUDP support; HTTP/WebSocket/TCP are client lowerings, compatibility probes, or migration debt. Health is separate from media by authority, not because it needs a separate TCP transport."} {"ts":"2026-06-16T00:12:00+01:00","type":"correction","status":"ok","note":"Source audit found Mimir.CultMeshMedia still using older C# CultMesh.StartNodeAsync/CultMesh.ConnectClient entrypoints while CultLib now exposes C# CultMesh.CreateRudpServer/CreateRudpClient/CreateRudpClientForAuthorizedPeer. Updated maps to treat the media bridge's current transport path as migration debt: typed program/media documents should move to explicit CultNet RUDP helpers, with OBS-local UDP preserved only as compatibility egress."} +{"ts": "2026-06-16T18:05:56+01:00", "type": "verification", "status": "ok", "note": "Mimir.CultMeshMedia migrated from old CultMesh node/client transport to explicit CultNet RUDP relay/send/recv path; local smoke artifacts/rudp-media-smoke-20260616-180404 proved 32 MPEG-TS chunks reached Starfire-local UDP sink over RUDP (packets=32 bytes=42112)."} diff --git a/state/map.yaml b/state/map.yaml index 89ccbb1..e3baf7e 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -383,12 +383,10 @@ accepted_design: `mimir.cultmesh_media_frame` documents over CultNet reliable UDP, Yggdrasil relays them on `10.77.0.1:3075`, and Starfire lowers the subscribed stream to `udp://127.0.0.1:5200` for the OBS source - `Raven Monitor + Realtek`. Source audit warning: the current C# bridge - still calls older `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` - entrypoints; now that CultLib exposes C# RUDP facade helpers, the next - transport cut should move this bridge onto explicit RUDP server/client or - authorized-peer helpers and demote any TCP-like node/client path to - compatibility evidence. + `Raven Monitor + Realtek`. The bridge now uses explicit CultLib RUDP + socket/session helpers for relay, sender, and receiver. The old + `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` node/client path is no + longer a transport authority for this media lane. - `src/Mimir.EveDashboard` publishes `mimir.eve_dashboard_state` through CultNet/CultMesh and can push `idunn.daemon_health` records over `cultnet.transport.rudp.v0` for itself and a paired service daemon. Its @@ -433,9 +431,11 @@ active_subgoals: keepalive, and Mimir surfaces publish typed RUDP self-health rather than private watchdog loops or TCP/HTTP status truth. - Cut remaining older CultMesh node/client transport assumptions in - `src/Mimir.CultMeshMedia`, `src/Mimir.EveDashboard`, and witness tooling: - daemon truth and typed program/media document paths should use explicit - CultNet RUDP helpers; HTTP/WebSocket remain render/client lowerings only. + `src/Mimir.EveDashboard` and witness tooling: daemon truth, provider + catalogs, command boundaries, and typed program document paths should use + explicit CultNet RUDP helpers; HTTP/WebSocket remain render/client + lowerings only. `src/Mimir.CultMeshMedia` has completed this cut for the + Raven/Yggdrasil/Starfire media-frame lane. current_status: summary: >