From 4c75abcada6eaabe614c1385d2029ff4ab00c4b2 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 3 Jun 2026 05:25:50 +0100 Subject: [PATCH] Expose Well sender pressure --- src/Mimir.Well/Program.cs | 51 +++++++++++++++++++++++++++++++++-- state/evidence.jsonl | 1 + tools/mimir_well_shakedown.py | 26 ++++++++++++++++-- 3 files changed, 74 insertions(+), 4 deletions(-) diff --git a/src/Mimir.Well/Program.cs b/src/Mimir.Well/Program.cs index 315b2dd..544cd9a 100644 --- a/src/Mimir.Well/Program.cs +++ b/src/Mimir.Well/Program.cs @@ -141,7 +141,7 @@ while (!stopping.IsCancellationRequested) featureSignalFrame, exposureController.Statuses, sourceErrors, - streamTelemetry.Snapshot(now, subscribersAreExternal: true), + streamTelemetry.Snapshot(now, subscribersAreExternal: true, publisher.SenderPressureSnapshot()), ++sequence, startedAt); streamTelemetry.ObservePublish( @@ -681,7 +681,7 @@ internal sealed class MimirWellStreamTelemetry lastPublishMilliseconds = receipt.TotalMilliseconds; } - public object Snapshot(DateTimeOffset now, bool subscribersAreExternal) => new + public object Snapshot(DateTimeOffset now, bool subscribersAreExternal, object senderPressure) => new { document = "mimir.well_stream_pressure.v1", status = "transitional-websocket-json", @@ -709,6 +709,7 @@ internal sealed class MimirWellStreamTelemetry lastMilliseconds = lastPublishMilliseconds, subscribersAreExternal, }, + sender = senderPressure, nextOrgan = new { controlLane = "CultMesh typed state and compact stream cursors", @@ -834,8 +835,16 @@ internal sealed class MimirWellPublisher(Uri url) : IAsyncDisposable private readonly ConcurrentQueue bodyQueue = new(); private readonly SemaphoreSlim signal = new(0); private readonly CancellationTokenSource senderStopping = new(); + private readonly object sendTelemetryLock = new(); private Task? senderTask; private int queueDepth; + private long sentDocuments; + private long sentBytes; + private double totalSendMilliseconds; + private double maxSendMilliseconds; + private string lastSentDocument = ""; + private int lastSentBytes; + private double lastSendMilliseconds; private const int MaxQueuedDocuments = 256; private const int StreamFrameDropThreshold = 128; @@ -909,7 +918,10 @@ internal sealed class MimirWellPublisher(Uri url) : IAsyncDisposable Interlocked.Decrement(ref queueDepth); try { + var stopwatch = Stopwatch.StartNew(); await socket.SendAsync(item.Bytes, WebSocketMessageType.Text, endOfMessage: true, stopping).ConfigureAwait(false); + stopwatch.Stop(); + ObserveSend(item, stopwatch.Elapsed); } catch (OperationCanceledException) { @@ -927,6 +939,41 @@ internal sealed class MimirWellPublisher(Uri url) : IAsyncDisposable private bool TryDequeueNext(out QueuedPublish item) => controlQueue.TryDequeue(out item!) || bodyQueue.TryDequeue(out item!); + public object SenderPressureSnapshot() + { + lock (sendTelemetryLock) + { + return new + { + document = "mimir.well_sender_pressure.v1", + authority = "Mimir.Well websocket sender loop observes wire send pressure after enqueue.", + sentDocuments, + sentBytes, + averageMilliseconds = sentDocuments == 0 ? 0.0 : totalSendMilliseconds / sentDocuments, + maxMilliseconds = maxSendMilliseconds, + lastDocument = lastSentDocument, + lastBytes = lastSentBytes, + lastMilliseconds = lastSendMilliseconds, + currentQueueDepth = Volatile.Read(ref queueDepth), + }; + } + } + + private void ObserveSend(QueuedPublish item, TimeSpan elapsed) + { + var milliseconds = elapsed.TotalMilliseconds; + lock (sendTelemetryLock) + { + sentDocuments++; + sentBytes += item.Bytes.Length; + totalSendMilliseconds += milliseconds; + maxSendMilliseconds = Math.Max(maxSendMilliseconds, milliseconds); + lastSentDocument = item.Document; + lastSentBytes = item.Bytes.Length; + lastSendMilliseconds = milliseconds; + } + } + public async ValueTask DisposeAsync() { senderStopping.Cancel(); diff --git a/state/evidence.jsonl b/state/evidence.jsonl index 88f2875..a2420b4 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -372,3 +372,4 @@ {"ts": "2026-06-03T05:10:12+01:00", "type": "observation", "status": "ok", "note": "Kiyo Pro source-health diagnosis: after stopping Well, direct native KS smoke through mimir_camera_capture.dll produced 0 frames for Kiyo Pro across 640x480 YUY2, 640x480 MJPG, and 1920x1080 MJPG, while Kiyo basic produced 107 frames in 3s through the same DLL. MF/GPU fallback probe hung under the live stack. Added source factory diagnostics to Well snapshots so Odin sees adapter/path/mode and starved KS-camera reason; live restart showed Kiyo Pro waiting-device-samples with reason 'KS camera opened but no frames...' for 640x480 YUY2, while Leap and Kiyo basic stayed live and Well held 6/6 Ready at 40 ms."} {"ts": "2026-06-03T05:13:42+01:00", "type": "implementation", "status": "ok", "note": "Presentation availability cut: MimirPresentationControlState now keeps user Enabled/Solo intent separate from derived Available/SampleCount. IncludesVideo requires live samples, so configured-but-starved video sources no longer occupy program output. Live Well verification after restart showed Kiyo Pro, Raven display, and Eve camera available=false/included=false, while Leap and Kiyo basic were available=true/included=true and synchronized frame stayed 6/6 Ready at 40 ms."} {"ts": "2026-06-03T05:20:09+01:00", "type": "implementation", "status": "ok", "note": "Mimir Well shakedown now defaults to the latest monotonic Well run instead of aggregating restarted ledgers; verified 20260602-233914 latest run at 1609 snapshots, 6/6 Ready after 120s and after 300s, recent presentation delay 40 ms. Remaining unavailable sources are source-health/device/network issues, not synchronized-frame collapse."} +{"ts": "2026-06-03T05:25:36+01:00", "type": "implementation", "status": "ok", "note": "Well stream pressure now includes Mimir.Well sender-loop telemetry at the websocket send authority. Live restart of 20260602-233914 reached after-120 readiness 1.0 with 6/6 Ready, recent presentation delay 40 ms, sender queue depth 0, sender max about 10 ms, and no drops; historical enqueue max is not active wire pressure."} diff --git a/tools/mimir_well_shakedown.py b/tools/mimir_well_shakedown.py index 87e1e48..0162485 100644 --- a/tools/mimir_well_shakedown.py +++ b/tools/mimir_well_shakedown.py @@ -176,6 +176,7 @@ def summarize_well(path: Path, all_runs: bool = False) -> dict[str, Any]: pressure = last.get("streamPressure") if isinstance(last.get("streamPressure"), dict) else {} publish = pressure.get("publish") if isinstance(pressure.get("publish"), dict) else {} poll = pressure.get("poll") if isinstance(pressure.get("poll"), dict) else {} + sender = pressure.get("sender") if isinstance(pressure.get("sender"), dict) else {} visual = last.get("visualCalibration", {}).get("cameras", []) return { @@ -224,6 +225,16 @@ def summarize_well(path: Path, all_runs: bool = False) -> dict[str, Any]: "lastDocument": publish.get("lastDocument", ""), "lastBytes": int(number(publish.get("lastBytes"))), }, + "senderPressure": { + "sentDocuments": int(number(sender.get("sentDocuments"))), + "sentBytes": int(number(sender.get("sentBytes"))), + "averageMilliseconds": number(sender.get("averageMilliseconds")), + "maxMilliseconds": number(sender.get("maxMilliseconds")), + "lastDocument": sender.get("lastDocument", ""), + "lastBytes": int(number(sender.get("lastBytes"))), + "lastMilliseconds": number(sender.get("lastMilliseconds")), + "currentQueueDepth": int(number(sender.get("currentQueueDepth"))), + }, "captureBodies": { "medianSampleCount": statistics.median(capture_sample_counts) if capture_sample_counts else 0.0, "bodyStatuses": dict(body_statuses), @@ -425,16 +436,24 @@ def hypotheses(summary: dict[str, Any]) -> list[str]: readiness = summary["readiness"] overlap = summary["overlapAfter120Ms"] pressure = summary["streamPressure"] + sender = summary.get("senderPressure", {}) after300_full_frame_rate = readiness.get("after300FullFrameRate") if isinstance(after300_full_frame_rate, (int, float)) and after300_full_frame_rate >= 0.95: result.append( "After warm-up, frame readiness is high; fixed 2500 ms delay should be reduced by an adaptive controller until misses reappear." ) - if overlap["p10MinOverlap"] <= 25.0: + if readiness.get("after120Samples", 0) > 0 and overlap["p10MinOverlap"] <= 25.0: result.append( "ASIO blocks expose a very thin per-domain overlap; audio should be aligned by Faust delay/resampling, not by asking the presentation planner for seconds of holdback." ) - if pressure["publishMaxMs"] > 50.0 or pressure["lastBytes"] > 1_000_000: + sender_queue_depth = sender.get("currentQueueDepth") if isinstance(sender, dict) else None + sender_max_ms = sender.get("maxMilliseconds") if isinstance(sender, dict) else None + if ( + pressure["publishMaxMs"] > 50.0 + and (not isinstance(sender_queue_depth, (int, float)) or sender_queue_depth > 0) + ) or ( + isinstance(sender_max_ms, (int, float)) and sender_max_ms > 50.0 + ) or pressure["lastBytes"] > 1_000_000: result.append( "Capture-page publication creates burst pressure; body paging/stream frames need their own compact lane before latency experiments can trust publish timing." ) @@ -483,6 +502,9 @@ def main() -> int: print("Stream pressure:") for key, value in summary["streamPressure"].items(): print(f" {key}: {value}") + print("Sender pressure:") + for key, value in summary["senderPressure"].items(): + print(f" {key}: {value}") if move: print(f"Move score: bpm={move['bpm']:.1f} c={move['bpmConfidence']:.2f} key={move['key']} chord={move['chord']} voices={move['activeVoiceCount']}/{move['voiceCount']}") print(f" tail score confidence median/max: {move['scoreConfidenceMedian']:.3f}/{move['scoreConfidenceMax']:.3f}")