Add native and process stream sources

This commit is contained in:
Meta Vens 2026-05-21 18:27:39 +01:00
commit a22fa0745c
14 changed files with 413 additions and 3 deletions

View file

@ -0,0 +1,38 @@
{
"bufferSeconds": 5.0,
"streams": [
{
"sourceId": "kiyo-primary",
"kind": "Video",
"origin": "LocalDevice",
"enabled": true,
"adapter": "native"
},
{
"sourceId": "host-focusrite",
"kind": "Audio",
"origin": "LocalDevice",
"enabled": true,
"adapter": "native"
},
{
"sourceId": "neighbor-video-srt",
"kind": "Video",
"origin": "Network",
"enabled": false,
"adapter": "process",
"command": "ffmpeg",
"arguments": [
"-hide_banner",
"-loglevel",
"warning",
"-i",
"srt://127.0.0.1:5100?mode=caller",
"-f",
"rawvideo",
"pipe:1"
],
"chunkBytes": 65536
}
]
}

View file

@ -54,6 +54,9 @@ latency, and confidence.
`MimirRollingStreamBuffer`, stream descriptors, and the `IMimirStreamSource`
adapter seam. It initializes one bounded rolling buffer per configured
audio/video stream and polls registered local/network sources each frame.
Local devices should feed through native push adapters; process-backed
sources are compatibility edges for network/diagnostic command ingest, not
the six-camera capture foundation.
- `config/perfect-machine.example.json` declares the contract shape for six cameras, six microphones, reservoir typed views, native authorities, outputs, and the demotion of bridge scripts to tooling.
- `native/reservoir` is the first native Rust crate. It implements one
shared-edge five-second rolling buffer with typed views and proves that

View file

@ -94,6 +94,10 @@ The first cut should be deliberately boring:
- `MimirRollingStreamBuffer` owns one stream's short history and evicts samples
outside the configured window.
- `IMimirStreamSource` is the adapter seam for local devices and network feeds.
- `MimirNativeIngestStreamSource` is the low-overhead push seam for native
camera/audio adapters that already own buffers or handles.
- `MimirProcessStreamSource` is a compatibility adapter for network or
diagnostic command sources. It is not the six-camera local capture foundation.
The default buffer duration is five seconds. It can be overridden with
`MIMIR_SYNC_BUFFER_SECONDS` or `MIMIR_RESERVOIR_SECONDS`.
@ -109,6 +113,11 @@ environment variables:
These create buffers immediately. Capture adapters still need to register
`IMimirStreamSource` implementations to feed live samples into those buffers.
For local cameras, prefer native adapters that push sample handles directly
through `MimirNativeIngestStreamSource`. Process-backed FFmpeg ingest is allowed
for remote SRT/diagnostic edges, but local six-camera ingest should stay close
to the driver, GPU, and capture APIs.
## Cut Line
Do not add a new registry, bridge daemon, JSON live store, OBS plugin, or

View file

@ -68,6 +68,9 @@ and D3D12 bridge, while `src/Mimir.Runtime` provides the Aquarium client runtime
`Mimir.Runtime` owns the first synchronization hub: one five-second default
rolling buffer per configured audio/video stream, source polling through
`IMimirStreamSource`, and runtime UI readouts for stream/buffer health.
Local six-camera ingest should push native handles through
`MimirNativeIngestStreamSource`; process-backed ingest remains a compatibility
edge for network SRT or diagnostics.
```mermaid
flowchart TD

View file

@ -49,7 +49,9 @@ Get-Content .\state\evidence.jsonl -Tail 8
buffer per audio/video stream, defaulting to five seconds. Stream ids can be
declared with `MIMIR_LOCAL_VIDEO_STREAMS`, `MIMIR_LOCAL_AUDIO_STREAMS`,
`MIMIR_NETWORK_VIDEO_STREAMS`, and `MIMIR_NETWORK_AUDIO_STREAMS`; real capture
adapters feed buffers through `IMimirStreamSource`.
adapters feed buffers through `IMimirStreamSource`. Local six-camera ingest
should use native push adapters such as `MimirNativeIngestStreamSource`, not
FFmpeg stdout/process capture except as a network or diagnostic edge.
- The reservoir is now one native time-ordered rolling buffer with typed
indexes/views. The previous typed-ring crate is gone.
- Native reservoir samples carry provenance flags; diagnostic/fallback samples

View file

@ -16,7 +16,12 @@ public sealed class MimirRuntime : IAquariumRuntime
private int lastPollCount;
public MimirRuntime(AquariumRuntimeOptions options)
: this(options, MimirSynchronizationSettings.FromEnvironment())
: this(options, MimirRuntimeConfiguration.Load())
{
}
public MimirRuntime(AquariumRuntimeOptions options, MimirRuntimeConfiguration configuration)
: this(options, configuration.Settings, configuration.Sources)
{
}

View file

@ -0,0 +1,44 @@
using System.Collections.Concurrent;
namespace Mimir.Runtime.Synchronization;
public sealed class MimirNativeIngestStreamSource : IMimirStreamSource
{
private readonly ConcurrentQueue<MimirStreamSample> samples = new();
private ulong sequence;
public MimirNativeIngestStreamSource(MimirStreamDescriptor descriptor)
{
Descriptor = descriptor;
}
public MimirStreamDescriptor Descriptor { get; }
public void Push(
long timestampNs,
long arrivalNs,
ulong payloadHandle,
int byteLength = 0,
ReadOnlyMemory<byte> data = default)
{
samples.Enqueue(new MimirStreamSample(
Descriptor.SourceId,
Descriptor.Kind,
Descriptor.Origin,
timestampNs,
arrivalNs,
sequence++,
payloadHandle,
byteLength,
data));
}
public bool TryRead(out MimirStreamSample sample)
{
return samples.TryDequeue(out sample);
}
public void Dispose()
{
}
}

View file

@ -0,0 +1,134 @@
using System.Collections.Concurrent;
using System.Diagnostics;
namespace Mimir.Runtime.Synchronization;
public sealed record MimirProcessStreamSourceOptions(
string Command,
IReadOnlyList<string> Arguments,
int ChunkBytes = 65_536);
public sealed class MimirProcessStreamSource : IMimirStreamSource
{
private readonly ConcurrentQueue<MimirStreamSample> samples = new();
private readonly CancellationTokenSource cancellation = new();
private readonly Process process;
private readonly Task stdoutTask;
private readonly Task stderrTask;
private ulong sequence;
private bool disposed;
public MimirProcessStreamSource(
MimirStreamDescriptor descriptor,
MimirProcessStreamSourceOptions options)
{
Descriptor = descriptor;
Options = options;
process = StartProcess(options);
stdoutTask = Task.Run(() => ReadStdout(options.ChunkBytes, cancellation.Token));
stderrTask = Task.Run(() => DrainStderr(cancellation.Token));
}
public MimirStreamDescriptor Descriptor { get; }
public MimirProcessStreamSourceOptions Options { get; }
public bool TryRead(out MimirStreamSample sample)
{
return samples.TryDequeue(out sample);
}
private Process StartProcess(MimirProcessStreamSourceOptions options)
{
var startInfo = new ProcessStartInfo(options.Command)
{
UseShellExecute = false,
RedirectStandardOutput = true,
RedirectStandardError = true,
CreateNoWindow = true,
};
foreach (var argument in options.Arguments)
{
startInfo.ArgumentList.Add(argument);
}
return Process.Start(startInfo)
?? throw new InvalidOperationException($"Could not start stream source process: {options.Command}");
}
private async Task ReadStdout(int chunkBytes, CancellationToken token)
{
var stream = process.StandardOutput.BaseStream;
while (!token.IsCancellationRequested)
{
var buffer = new byte[chunkBytes];
var read = await stream.ReadAsync(buffer.AsMemory(0, buffer.Length), token).ConfigureAwait(false);
if (read <= 0)
{
break;
}
if (read != buffer.Length)
{
Array.Resize(ref buffer, read);
}
var now = Stopwatch.GetTimestamp();
var timestampNs = StopwatchTicksToNs(now);
samples.Enqueue(new MimirStreamSample(
Descriptor.SourceId,
Descriptor.Kind,
Descriptor.Origin,
timestampNs,
timestampNs,
sequence++,
0,
buffer.Length,
buffer));
}
}
private async Task DrainStderr(CancellationToken token)
{
while (!token.IsCancellationRequested)
{
var line = await process.StandardError.ReadLineAsync(token).ConfigureAwait(false);
if (line == null)
{
break;
}
}
}
private static long StopwatchTicksToNs(long ticks)
{
return checked((long)(ticks * (1_000_000_000.0 / Stopwatch.Frequency)));
}
public void Dispose()
{
if (disposed)
{
return;
}
disposed = true;
cancellation.Cancel();
try
{
if (!process.HasExited)
{
process.Kill(entireProcessTree: true);
}
}
catch
{
}
process.Dispose();
cancellation.Dispose();
GC.SuppressFinalize(stdoutTask);
GC.SuppressFinalize(stderrTask);
}
}

View file

@ -0,0 +1,153 @@
using System.Text.Json;
namespace Mimir.Runtime.Synchronization;
public sealed class MimirRuntimeConfiguration
{
public MimirSynchronizationSettings Settings { get; init; } = new();
public IReadOnlyList<IMimirStreamSource> Sources { get; init; } = [];
public static MimirRuntimeConfiguration Load()
{
var path = ResolveConfigPath();
if (path == null)
{
return new MimirRuntimeConfiguration
{
Settings = MimirSynchronizationSettings.FromEnvironment(),
};
}
var model = JsonSerializer.Deserialize<MimirRuntimeConfigFile>(
File.ReadAllText(path),
new JsonSerializerOptions
{
PropertyNameCaseInsensitive = true,
}) ?? new MimirRuntimeConfigFile();
var bufferDuration = model.BufferSeconds > 0.0
? TimeSpan.FromSeconds(Math.Clamp(model.BufferSeconds, 0.25, 60.0))
: MimirSynchronizationSettings.FromEnvironment().BufferDuration;
var sourceModels = model.Streams
.Where(stream => stream.Enabled)
.ToArray();
var streams = sourceModels
.Select(ToDescriptor)
.ToArray();
var sources = sourceModels
.Select(TryCreateSource)
.Where(source => source != null)
.Cast<IMimirStreamSource>()
.ToArray();
return new MimirRuntimeConfiguration
{
Settings = new MimirSynchronizationSettings
{
BufferDuration = bufferDuration,
Streams = streams,
},
Sources = sources,
};
}
private static string? ResolveConfigPath()
{
var configured = Environment.GetEnvironmentVariable("MIMIR_RUNTIME_CONFIG");
if (!string.IsNullOrWhiteSpace(configured) && File.Exists(configured))
{
return configured;
}
var current = AppContext.BaseDirectory;
for (var index = 0; index < 8; index++)
{
var candidate = Path.Combine(current, "config", "mimir-runtime.json");
if (File.Exists(candidate))
{
return candidate;
}
var parent = Directory.GetParent(current);
if (parent == null)
{
break;
}
current = parent.FullName;
}
return null;
}
private static MimirStreamDescriptor ToDescriptor(MimirStreamConfig stream)
{
return new MimirStreamDescriptor(
stream.SourceId,
ParseKind(stream.Kind),
ParseOrigin(stream.Origin),
stream.Enabled);
}
private static IMimirStreamSource? TryCreateSource(MimirStreamConfig stream)
{
if (!string.Equals(stream.Adapter, "process", StringComparison.OrdinalIgnoreCase))
{
return null;
}
if (string.IsNullOrWhiteSpace(stream.Command))
{
return null;
}
return new MimirProcessStreamSource(
ToDescriptor(stream),
new MimirProcessStreamSourceOptions(
stream.Command,
stream.Arguments,
Math.Max(1024, stream.ChunkBytes)));
}
private static MimirStreamKind ParseKind(string value)
{
return Enum.TryParse<MimirStreamKind>(value, ignoreCase: true, out var kind)
? kind
: throw new InvalidOperationException($"Unknown Mimir stream kind: {value}");
}
private static MimirStreamOrigin ParseOrigin(string value)
{
return Enum.TryParse<MimirStreamOrigin>(value, ignoreCase: true, out var origin)
? origin
: throw new InvalidOperationException($"Unknown Mimir stream origin: {value}");
}
}
public sealed class MimirRuntimeConfigFile
{
public double BufferSeconds { get; set; } = 5.0;
public List<MimirStreamConfig> Streams { get; set; } = [];
}
public sealed class MimirStreamConfig
{
public string SourceId { get; set; } = "";
public string Kind { get; set; } = "Video";
public string Origin { get; set; } = "LocalDevice";
public bool Enabled { get; set; } = true;
public string Adapter { get; set; } = "process";
public string Command { get; set; } = "";
public string[] Arguments { get; set; } = [];
public int ChunkBytes { get; set; } = 65_536;
}

View file

@ -8,4 +8,5 @@ public readonly record struct MimirStreamSample(
long ArrivalNs,
ulong Sequence,
ulong PayloadHandle,
int ByteLength = 0);
int ByteLength = 0,
ReadOnlyMemory<byte> Data = default);

View file

@ -98,3 +98,4 @@
{"ts": "2026-05-21T12:09:51+00:00", "type": "decision", "status": "ok", "note": "Viable stream app cut pinned: Aquarium hosts one five-second in-memory LocalcastRuntime, local/networked feeds append as producers, Aquarium owns runtime UI/debug/settings/output management, Faust/native audio builds synchronized stems, and OBS receives program outputs rather than raw feed sync."}
{"ts": "2026-05-21T12:14:26+00:00", "type": "implementation", "status": "ok", "note": "Mimir now owns a C# app solution: Mimir.slnx contains src/Mimir.App, which launches AquariumHost from Aquarium Engine, and src/Mimir.Runtime, which provides the Aquarium runtime factory over Aquarium.LocalCast. Aquarium Engine remains the windowing/rendering/D3D12 library reference."}
{"ts": "2026-05-21T12:33:01+00:00", "type": "implementation", "status": "ok", "note": "Mimir.Runtime now owns the first synchronization spine: MimirRuntime wraps Aquarium LocalCast rendering, initializes per-stream rolling buffers through MimirSynchronizationHub, polls registered IMimirStreamSource adapters each update, and defaults buffer duration to five seconds with env overrides for local/network audio/video stream ids."}
{"ts": "2026-05-21T17:27:12+00:00", "type": "decision", "status": "ok", "note": "For six local cameras, Mimir ingest should stay close to driver/GPU capture surfaces. MimirNativeIngestStreamSource is the preferred local adapter seam carrying payload handles and optional memory into rolling buffers; MimirProcessStreamSource remains a network/diagnostic compatibility edge, not the local camera foundation."}

View file

@ -76,6 +76,11 @@ accepted_design:
and `IMimirStreamSource`. It initializes one bounded rolling buffer per
configured audio/video stream and drains registered local/network sources
into those buffers each Aquarium update.
- Local six-camera ingest should use native push sources such as
`MimirNativeIngestStreamSource`, carrying native payload handles and
optional memory spans into Mimir buffers. `MimirProcessStreamSource` is a
compatibility path for network SRT or diagnostics, not the local camera
foundation.
- `native/reservoir` is the first native Rust crate. It now uses one
time-ordered rolling buffer with typed views, so retention has exactly one
owner while Aquarium/Faust still get kind-scoped queries.
@ -119,6 +124,9 @@ active_subgoals:
expose runtime UI, and emit synchronized OBS program outputs.
- Implement concrete local device and network `IMimirStreamSource` adapters
that feed the Mimir runtime buffers with live audio/video samples.
- Build the local camera adapter close to the driver/GPU capture surface;
avoid FFmpeg stdout for the six local camera streams unless used only as a
temporary diagnostic witness.
- Bind Aquarium/Faust to the rolling-buffer `LocalcastRuntime` and
`LocalcastProducer` C ABI described by
`native/reservoir/include/localcast_reservoir.h`.

View file

@ -14,5 +14,8 @@ No active scratch subgoal.
- Synchronization ownership started moving into `Mimir.Runtime`: hub, stream
descriptors, per-stream rolling buffers, settings, and source adapter seam are
in C# now; concrete capture adapters are next.
- Local six-camera ingest should stay close to the metal: native push sources
carry payload handles into buffers; process sources are network/diagnostic
compatibility, not the main local camera path.
- Keep scratch short and disposable.
- Promote only durable lessons into `state/evidence.jsonl` or the map.

View file

@ -47,6 +47,12 @@ class MimirAppContractTests(unittest.TestCase):
self.assertIn("TimeSpan.FromSeconds(5)", settings)
self.assertIn("MIMIR_NETWORK_VIDEO_STREAMS", settings)
self.assertIn("MIMIR_LOCAL_AUDIO_STREAMS", settings)
native_source = (ROOT / "src" / "Mimir.Runtime" / "Synchronization" / "MimirNativeIngestStreamSource.cs").read_text(encoding="utf-8")
process_source = (ROOT / "src" / "Mimir.Runtime" / "Synchronization" / "MimirProcessStreamSource.cs").read_text(encoding="utf-8")
self.assertIn("payloadHandle", native_source)
self.assertIn("ReadOnlyMemory<byte>", native_source)
self.assertIn("RedirectStandardOutput", process_source)
if __name__ == "__main__":