diff --git a/TODO.md b/TODO.md index a9a761c..0f9992a 100644 --- a/TODO.md +++ b/TODO.md @@ -54,6 +54,10 @@ Experimental.8 makes the installed single-file build discover and verify its OBS payload deterministically, adds an executable-level package diagnostic, and shows OBS maintenance failures directly instead of leaving them only in diagnostics. +Experimental.9 coalesces OBS callbacks into fixed 20 ms network frames and gives +audio its own soft gateway budget, so normal or pathological capture cadence can +never rate-limit and disconnect the Companion control connection. + Release-blocking work remains: - Package the native whisper.cpp worker as the Windows CUDA runtime, then benchmark diff --git a/companion/installer/Lumi.Companion.iss b/companion/installer/Lumi.Companion.iss index d09d8e6..e4ece72 100644 --- a/companion/installer/Lumi.Companion.iss +++ b/companion/installer/Lumi.Companion.iss @@ -1,5 +1,5 @@ #ifndef AppVersion - #define AppVersion "0.1.0-experimental.8" + #define AppVersion "0.1.0-experimental.9" #endif #ifndef SourceRoot #error SourceRoot must point at the self-contained Companion publish directory. diff --git a/companion/scripts/publish-companion.ps1 b/companion/scripts/publish-companion.ps1 index dbff2cc..1e440fd 100644 --- a/companion/scripts/publish-companion.ps1 +++ b/companion/scripts/publish-companion.ps1 @@ -1,5 +1,5 @@ param( - [string]$Version = "0.1.0-experimental.8", + [string]$Version = "0.1.0-experimental.9", [string]$BridgeVersion = "0.1.0-experimental.5" ) diff --git a/companion/src/Lumi.Companion.App/CompanionRuntime.cs b/companion/src/Lumi.Companion.App/CompanionRuntime.cs index fa41459..74f9d02 100644 --- a/companion/src/Lumi.Companion.App/CompanionRuntime.cs +++ b/companion/src/Lumi.Companion.App/CompanionRuntime.cs @@ -600,7 +600,8 @@ public sealed class CompanionRuntime : IAsyncDisposable { if (State.BenchmarkRunning) _benchmarkLastVoiceAt = DateTimeOffset.UtcNow; } - if (_socket is not null && State.Connected && !_socket.QueueBridgeAudio(frame)) + var audioNeeded = State.TestRunning || State.BenchmarkRunning || State.ObsStreaming; + if (audioNeeded && _socket is not null && State.Connected && !_socket.QueueBridgeAudio(frame)) _ = WriteLogAsync("audio_dropped", "An obsolete audio frame was dropped before network delivery."); return Task.CompletedTask; } diff --git a/companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj b/companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj index ffc86bd..c22cecc 100644 --- a/companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj +++ b/companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj @@ -5,7 +5,7 @@ enable enable app.manifest - 0.1.0-experimental.8 + 0.1.0-experimental.9 0.1.0.0 diff --git a/companion/src/Lumi.Companion.Core/CompanionSocket.cs b/companion/src/Lumi.Companion.Core/CompanionSocket.cs index d9cd888..f0b9f66 100644 --- a/companion/src/Lumi.Companion.Core/CompanionSocket.cs +++ b/companion/src/Lumi.Companion.Core/CompanionSocket.cs @@ -7,7 +7,13 @@ namespace Lumi.Companion.Core; public sealed class CompanionSocket : IAsyncDisposable { + private const int NetworkAudioPayloadBytes = 640; // 20 ms of 16 kHz mono PCM16. private readonly Channel> _audio = Channel.CreateBounded>(new BoundedChannelOptions(250) { FullMode = BoundedChannelFullMode.Wait, SingleReader = true }); + private readonly object _bridgeAudioGate = new(); + private readonly byte[] _bridgeAudioBuffer = new byte[ProtocolV1.MaxAudioPayloadBytes + NetworkAudioPayloadBytes]; + private readonly byte[] _bridgeAudioHeader = new byte[ProtocolV1.AudioHeaderBytes]; + private int _bridgeAudioBuffered; + private uint _bridgeAudioSequence; private ClientWebSocket? _socket; private CancellationTokenSource? _lifetime; private readonly SemaphoreSlim _sendLock = new(1, 1); @@ -56,11 +62,36 @@ public sealed class CompanionSocket : IAsyncDisposable } public bool QueueBridgeAudio(ReadOnlyMemory encoded) { - if (encoded.Length < ProtocolV1.AudioHeaderBytes || !encoded.Span[..4].SequenceEqual("LACP"u8)) return false; - var normalized = encoded.ToArray(); - SessionId.TryWriteBytes(normalized.AsSpan(20, 16), bigEndian: true, out _); - BinaryPrimitives.WriteUInt64LittleEndian(normalized.AsSpan(12, 8), (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1000UL); - return QueueEncodedAudio(normalized); + var input = encoded.Span; + if (input.Length < ProtocolV1.AudioHeaderBytes || input.Length > ProtocolV1.AudioHeaderBytes + ProtocolV1.MaxAudioPayloadBytes || + !input[..4].SequenceEqual("LACP"u8) || BinaryPrimitives.ReadUInt16LittleEndian(input[6..8]) != ProtocolV1.AudioHeaderBytes) return false; + var payloadBytes = (int)BinaryPrimitives.ReadUInt32LittleEndian(input[60..64]); + if (payloadBytes < 0 || payloadBytes % 2 != 0 || input.Length != ProtocolV1.AudioHeaderBytes + payloadBytes) return false; + lock (_bridgeAudioGate) + { + var sourceChanged = !_bridgeAudioHeader.AsSpan(36, 16).SequenceEqual(input[36..52]); + var stateChanged = _bridgeAudioHeader[5] != input[5]; + if (_bridgeAudioBuffered > 0 && (sourceChanged || stateChanged)) _bridgeAudioBuffered = 0; + input[..ProtocolV1.AudioHeaderBytes].CopyTo(_bridgeAudioHeader); + input[ProtocolV1.AudioHeaderBytes..].CopyTo(_bridgeAudioBuffer.AsSpan(_bridgeAudioBuffered)); + _bridgeAudioBuffered += payloadBytes; + var accepted = true; + while (_bridgeAudioBuffered >= NetworkAudioPayloadBytes) + { + var normalized = new byte[ProtocolV1.AudioHeaderBytes + NetworkAudioPayloadBytes]; + _bridgeAudioHeader.CopyTo(normalized, 0); + BinaryPrimitives.WriteUInt32LittleEndian(normalized.AsSpan(8, 4), _bridgeAudioSequence++); + BinaryPrimitives.WriteUInt64LittleEndian(normalized.AsSpan(12, 8), (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1000UL); + SessionId.TryWriteBytes(normalized.AsSpan(20, 16), bigEndian: true, out _); + BinaryPrimitives.WriteUInt32LittleEndian(normalized.AsSpan(60, 4), NetworkAudioPayloadBytes); + _bridgeAudioBuffer.AsSpan(0, NetworkAudioPayloadBytes).CopyTo(normalized.AsSpan(ProtocolV1.AudioHeaderBytes)); + _bridgeAudioBuffered -= NetworkAudioPayloadBytes; + if (_bridgeAudioBuffered > 0) + _bridgeAudioBuffer.AsSpan(NetworkAudioPayloadBytes, _bridgeAudioBuffered).CopyTo(_bridgeAudioBuffer); + accepted &= QueueEncodedAudio(normalized); + } + return accepted; + } } public async Task SendAsync(string type, object payload, Guid? sessionId, CancellationToken cancellationToken) { diff --git a/plugins/lumi_transcription/backend/companion/gateway.js b/plugins/lumi_transcription/backend/companion/gateway.js index d40d1f7..e542cd1 100644 --- a/plugins/lumi_transcription/backend/companion/gateway.js +++ b/plugins/lumi_transcription/backend/companion/gateway.js @@ -2,6 +2,9 @@ const { WebSocketServer, WebSocket } = require("ws"); const { MAX_JSON_BYTES, MAX_AUDIO_PAYLOAD_BYTES, AUDIO_HEADER_BYTES, parseEnvelope, validateHello, parseAudioFrame, envelope } = require("./protocol"); const { insecureDeviceAllowed } = require("./device_store"); +const MAX_CONTROL_MESSAGES_PER_SECOND = 120; +const MAX_AUDIO_MESSAGES_PER_SECOND = 200; + class CompanionGateway { constructor(options) { this.devices = options.devices; @@ -27,7 +30,9 @@ class CompanionGateway { let helloComplete = false; let lastPong = Date.now(); let windowStarted = Date.now(); - let messagesInWindow = 0; + let controlMessagesInWindow = 0; + let audioMessagesInWindow = 0; + let audioMessagesDropped = 0; let messageChain = Promise.resolve(); const send = (type, payload, sessionId = session?.id || null) => { if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify(envelope(type, payload, sessionId))); @@ -42,15 +47,20 @@ class CompanionGateway { const body = Buffer.from(data); messageChain = messageChain.then(async () => { try { - if (Date.now() - windowStarted >= 1000) { windowStarted = Date.now(); messagesInWindow = 0; } - messagesInWindow += 1; - if (messagesInWindow > 300) throw coded("RATE_LIMIT", "Companion message rate exceeded its limit."); + if (Date.now() - windowStarted >= 1000) { + if (audioMessagesDropped) this.log.append({ kind: "audio_rate_limited", device_id: device.id, session_id: session?.id, dropped_frames: audioMessagesDropped }); + windowStarted = Date.now(); controlMessagesInWindow = 0; audioMessagesInWindow = 0; audioMessagesDropped = 0; + } if (isBinary) { if (!helloComplete || !session) throw coded("HELLO_REQUIRED", "Complete the handshake before sending audio."); + audioMessagesInWindow += 1; + if (audioMessagesInWindow > MAX_AUDIO_MESSAGES_PER_SECOND) { audioMessagesDropped += 1; return; } const result = await this.sessions.audio(session.id, parseAudioFrame(body)); if (result.gap) send("metric", { kind: "sequence_gap", missing_frames: result.gap }); return; } + controlMessagesInWindow += 1; + if (controlMessagesInWindow > MAX_CONTROL_MESSAGES_PER_SECOND) throw coded("RATE_LIMIT", "Companion control message rate exceeded its limit."); const message = parseEnvelope(body); if (!helloComplete) { const hello = validateHello(message); @@ -121,4 +131,4 @@ function sameHostOrigin(origin, host) { try { return new URL(origin).host === ho function coded(code, message) { return Object.assign(new Error(message), { code }); } function cleanReason(value) { return ["requested", "test_complete", "benchmark_complete", "silence_timeout", "disconnect"].includes(String(value)) ? String(value) : "requested"; } -module.exports = { CompanionGateway, sameHostOrigin }; +module.exports = { CompanionGateway, sameHostOrigin, MAX_CONTROL_MESSAGES_PER_SECOND, MAX_AUDIO_MESSAGES_PER_SECOND }; diff --git a/plugins/lumi_transcription/companion_manifest.json b/plugins/lumi_transcription/companion_manifest.json index 53620d4..2ca57c8 100644 --- a/plugins/lumi_transcription/companion_manifest.json +++ b/plugins/lumi_transcription/companion_manifest.json @@ -1,17 +1,17 @@ { "schema_version": 1, - "version": "0.1.0-experimental.8", + "version": "0.1.0-experimental.9", "signed": false, - "release_notes": "Fixes OBS integration package discovery in the installed single-file app, validates the packaged payload directly, and displays actionable installation errors.", + "release_notes": "Prevents OBS audio callback bursts from rate-limiting the Companion connection by sending steady 20 ms audio frames and keeping excess audio traffic separate from control messages.", "installer": { "id": "windows-x64-installer", "platform": "win32", "architecture": "x64", "label": "Windows x64 per-user installer", "filename": "Lumi.Companion-Setup.exe", - "url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.8/Lumi.Companion-Setup.exe", - "sha256": "075fb4dd9a84ab10655d75899a6b4944269c9fe6371573778740d80b16ca19d0", - "bytes": 32041200 + "url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.9/Lumi.Companion-Setup.exe", + "sha256": "22bdb0f351e5b4960e2c1ae40e3bdbd8de8a8d6326b30714defb5eb202c0efcf", + "bytes": 32039974 }, "artifacts": [ { @@ -20,9 +20,9 @@ "architecture": "x64", "label": "Windows x64 self-contained", "filename": "Lumi.Companion-win-x64.zip", - "url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.8/Lumi.Companion-win-x64.zip", - "sha256": "46c58b044a213e328fa96fe86e48b9ea0b67e327439fed14fc485c711f03eb3e", - "bytes": 41757425, + "url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.9/Lumi.Companion-win-x64.zip", + "sha256": "70d9a780859824aa7b8a12e9f25ae8871778b83458889237f171484b2a200dbc", + "bytes": 41757969, "entrypoint": "Lumi.Companion.App.exe" } ] diff --git a/plugins/lumi_transcription/plugin.json b/plugins/lumi_transcription/plugin.json index ad0e339..21bb26b 100644 --- a/plugins/lumi_transcription/plugin.json +++ b/plugins/lumi_transcription/plugin.json @@ -1,7 +1,7 @@ { "id": "lumi_transcription", "name": "Lumi Transcription", - "version": "0.1.0-experimental.8", + "version": "0.1.0-experimental.9", "description": "Server-hosted whisper.cpp transcription for Lumi Companion and OBS closed captions.", "main": "index.js", "channel": "experimental", diff --git a/plugins/lumi_transcription/tests/verify.js b/plugins/lumi_transcription/tests/verify.js index 408a951..7452803 100644 --- a/plugins/lumi_transcription/tests/verify.js +++ b/plugins/lumi_transcription/tests/verify.js @@ -117,6 +117,7 @@ function verifyLocalhostTransportPolicy() { } function verifyCompanionVersionOrdering() { + assert.equal(plugin.compareVersions("0.1.0-experimental.9", "0.1.0-experimental.8"), 1); assert.equal(plugin.compareVersions("0.1.0-experimental.8", "0.1.0-experimental.7"), 1); assert.equal(plugin.compareVersions("0.1.0-experimental.7", "0.1.0-experimental.6"), 1); assert.equal(plugin.compareVersions("0.1.0-experimental.6", "0.1.0-experimental.5"), 1); @@ -357,10 +358,11 @@ async function verifyAuthenticatedGateway() { const devices = new DeviceStore(db); const sessionId = crypto.randomUUID(); let disconnected = false; + let audioMessages = 0; const sessions = { create: (_device, send) => ({ session: { id: sessionId, state: "idle", send }, resumed: false }), disconnect: () => { disconnected = true; }, - audio: async () => ({ accepted: true }), updateSource: () => ({}), updateObsState: async () => ({}), start: async () => ({}), stop: async () => ({}) + audio: async () => { audioMessages += 1; return { accepted: true }; }, updateSource: () => ({}), updateObsState: async () => ({}), start: async () => ({}), stop: async () => ({}) }; const gateway = new CompanionGateway({ devices, sessions }); const registry = createWebUpgradeRegistry(); @@ -379,6 +381,12 @@ async function verifyAuthenticatedGateway() { assert.equal(response.type, "hello_ack"); assert.equal(response.session_id, sessionId); assert.equal(devices.list().find((device) => device.id === credential.device_id).metadata.companion_version, "0.1.0-experimental.6"); + const sourceUuid = crypto.randomUUID(); + const audioFrame = protocol.encodeAudioFrame({ session_id: sessionId, source_uuid: sourceUuid, sequence: 1, capture_timestamp_us: Date.now() * 1000, pcm: Buffer.alloc(640) }); + for (let index = 0; index < 350; index += 1) client.send(audioFrame); + await new Promise((resolve) => setTimeout(resolve, 250)); + assert.equal(client.readyState, WebSocket.OPEN, "Audio bursts must be bounded without closing the authenticated control connection"); + assert.ok(audioMessages > 0 && audioMessages < 350, "Excess audio frames were not softly rate-limited"); await new Promise((resolve) => { client.once("close", resolve); client.close(); }); for (let attempt = 0; attempt < 20 && !disconnected; attempt += 1) await new Promise((resolve) => setTimeout(resolve, 5)); assert.equal(disconnected, true);