From df23b57078e15be162ca554a5a53fe502d25cee9 Mon Sep 17 00:00:00 2001 From: Franz Rolfsvaag Date: Wed, 22 Jul 2026 17:35:27 +0200 Subject: [PATCH] Keep transcription worker failures recoverable --- TODO.md | 5 + companion/scripts/publish-companion.ps1 | 7 +- .../Lumi.Companion.App/CompanionRuntime.cs | 44 +++++++-- .../Lumi.Companion.App.csproj | 2 +- .../Lumi.Companion.Core/CompanionSocket.cs | 11 ++- .../backend/sessions/session_coordinator.js | 32 +++++- .../backend/transcription/provider.js | 97 ++++++++++++++++--- .../transcription/worker-native/src/main.cpp | 10 ++ .../companion_manifest.json | 10 +- plugins/lumi_transcription/index.js | 5 + plugins/lumi_transcription/plugin.json | 2 +- plugins/lumi_transcription/tests/verify.js | 32 +++++- plugins/lumi_transcription/views/settings.ejs | 11 ++- 13 files changed, 224 insertions(+), 44 deletions(-) diff --git a/TODO.md b/TODO.md index 2b9fc63..39ad72a 100644 --- a/TODO.md +++ b/TODO.md @@ -27,6 +27,11 @@ selected-source PCM capture; nested Program-scene evaluation; bounded same-user IPC; bridge health; and native caption submission. Experimental.2 must be replaced manually once to bootstrap the updater. +Experimental.4 contains Windows speech-path reliability fixes: native PCM stdin +is explicitly binary-safe, broken worker pipes no longer terminate Lumi, the +worker restarts and reloads the selected model automatically, and the Companion +test reports the server failure at Speech recognition instead of hanging. + Release-blocking work remains: - Package the native whisper.cpp worker as the Windows CUDA runtime, then benchmark diff --git a/companion/scripts/publish-companion.ps1 b/companion/scripts/publish-companion.ps1 index a2cdca9..b8cfc3c 100644 --- a/companion/scripts/publish-companion.ps1 +++ b/companion/scripts/publish-companion.ps1 @@ -1,5 +1,6 @@ param( - [string]$Version = "0.1.0-experimental.3" + [string]$Version = "0.1.0-experimental.4", + [string]$BridgeVersion = "0.1.0-experimental.3" ) $ErrorActionPreference = "Stop" @@ -10,12 +11,12 @@ $publishRoot = Join-Path $outputRoot "publish" $stageRoot = Join-Path $outputRoot "package" $archive = Join-Path $outputRoot "Lumi.Companion-win-x64.zip" -& (Join-Path $PSScriptRoot "build-obs-bridge.ps1") -BridgeVersion $Version +& (Join-Path $PSScriptRoot "build-obs-bridge.ps1") -BridgeVersion $BridgeVersion Remove-Item $publishRoot, $stageRoot -Recurse -Force -ErrorAction SilentlyContinue New-Item -ItemType Directory -Force -Path $publishRoot, $stageRoot | Out-Null $localDotnet = Join-Path $HOME ".dotnet-sdk\dotnet.exe" $dotnet = if (Test-Path $localDotnet) { $localDotnet } else { (Get-Command dotnet.exe -ErrorAction Stop).Source } -$publishArguments = @("publish", "`"$project`"", "-c", "Release", "-r", "win-x64", "--self-contained", "true", "-o", "`"$publishRoot`"", +$publishArguments = @("publish", "`"$project`"", "-c", "Release", "-r", "win-x64", "--self-contained", "true", "--disable-build-servers", "-o", "`"$publishRoot`"", "-p:PublishSingleFile=true", "-p:IncludeNativeLibrariesForSelfExtract=true", "-p:DebugType=None", "-p:Version=$Version") $published = Start-Process -FilePath $dotnet -ArgumentList $publishArguments -NoNewWindow -Wait -PassThru if ($published.ExitCode) { throw "Companion publish failed." } diff --git a/companion/src/Lumi.Companion.App/CompanionRuntime.cs b/companion/src/Lumi.Companion.App/CompanionRuntime.cs index 809bbad..8986e59 100644 --- a/companion/src/Lumi.Companion.App/CompanionRuntime.cs +++ b/companion/src/Lumi.Companion.App/CompanionRuntime.cs @@ -24,6 +24,8 @@ public sealed class CompanionRuntime : IAsyncDisposable private ObsBridgePipe? _obsBridge; private TaskCompletionSource? _audioSignal; private TaskCompletionSource? _captionSignal; + private TaskCompletionSource? _sessionStartSignal; + private TaskCompletionSource? _testFailure; private bool _disposed; private CompanionUpdate? _availableUpdate; @@ -127,6 +129,7 @@ public sealed class CompanionRuntime : IAsyncDisposable _socket.Disconnected += error => { if (_disposed) return; + _testFailure?.TrySetResult(error is null ? "The Lumi connection closed during the test." : $"The Lumi connection closed during the test: {Friendly(error)}"); SetState(State with { Connected = false, Health = TrayHealth.Degraded, Detail = "The secure Lumi connection closed. Retry when the host is available." }); _ = WriteLogAsync("disconnected", error?.Message ?? "Connection closed."); }; @@ -178,18 +181,29 @@ public sealed class CompanionRuntime : IAsyncDisposable if (!State.Connected) return; _audioSignal = new(TaskCreationOptions.RunContinuationsAsynchronously); _captionSignal = new(TaskCreationOptions.RunContinuationsAsynchronously); + _sessionStartSignal = new(TaskCreationOptions.RunContinuationsAsynchronously); + _testFailure = new(TaskCreationOptions.RunContinuationsAsynchronously); await _socket!.SendAsync("start", new { mode = "test" }, _socket.SessionId, cancellationToken); - MarkStage(stages, 3, TestStageState.Running, "Speak normally into the selected microphone…"); - if (!await WaitForAsync(_audioSignal.Task, TimeSpan.FromSeconds(10), cancellationToken)) + var sessionStart = await WaitForTestSignalAsync(_sessionStartSignal.Task, TimeSpan.FromSeconds(8), cancellationToken); + if (!sessionStart.Completed) { - MarkStage(stages, 3, TestStageState.Blocked, "No active microphone audio reached the companion within 10 seconds."); + MarkStage(stages, 3, TestStageState.Blocked, "Audio capture was not attempted because the server inference session did not start."); + MarkStage(stages, 5, TestStageState.Blocked, sessionStart.Failure ?? "Lumi did not confirm that speech recognition started within 8 seconds."); + return; + } + MarkStage(stages, 3, TestStageState.Running, "Speak normally into the selected microphone…"); + var audio = await WaitForTestSignalAsync(_audioSignal.Task, TimeSpan.FromSeconds(10), cancellationToken); + if (!audio.Completed) + { + MarkStage(stages, 3, TestStageState.Blocked, audio.Failure ?? "No active microphone audio reached the companion within 10 seconds."); return; } MarkStage(stages, 3, TestStageState.Passed, "Live microphone audio reached the companion."); MarkStage(stages, 5, TestStageState.Running, "Waiting for server-hosted speech recognition…"); - if (!await WaitForAsync(_captionSignal.Task, TimeSpan.FromSeconds(30), cancellationToken)) + var caption = await WaitForTestSignalAsync(_captionSignal.Task, TimeSpan.FromSeconds(30), cancellationToken); + if (!caption.Completed) { - MarkStage(stages, 5, TestStageState.Blocked, "No caption returned within 30 seconds. Check the loaded model and host diagnostics."); + MarkStage(stages, 5, TestStageState.Blocked, caption.Failure ?? "No caption returned within 30 seconds. Check the loaded model and host diagnostics."); return; } MarkStage(stages, 5, TestStageState.Passed, "Speech recognition returned a stable result."); @@ -204,6 +218,8 @@ public sealed class CompanionRuntime : IAsyncDisposable try { await _socket.SendAsync("stop", new { reason = "test_complete" }, _socket.SessionId, CancellationToken.None); } catch { } _audioSignal = null; _captionSignal = null; + _sessionStartSignal = null; + _testFailure = null; SetState(State with { TestRunning = false, Detail = TestStages.Any(stage => stage.State == TestStageState.Blocked) ? "The test stopped at the first unavailable real boundary." : State.Detail }); } } @@ -341,6 +357,9 @@ public sealed class CompanionRuntime : IAsyncDisposable private Task OnServerMessageAsync(ServerEnvelope message) { + if (message.Type == "status" && message.Payload.TryGetProperty("kind", out var kind) && kind.GetString() == "session" && + message.Payload.TryGetProperty("state", out var state) && state.GetString() == "running") + _sessionStartSignal?.TrySetResult(true); if (message.Type == "caption") { var text = message.Payload.TryGetProperty("stable_text", out var stable) ? stable.GetString() : null; @@ -355,6 +374,7 @@ public sealed class CompanionRuntime : IAsyncDisposable if (message.Type == "error") { var serverMessage = message.Payload.TryGetProperty("message", out var value) ? value.GetString() : "Lumi reported an error."; + _testFailure?.TrySetResult(serverMessage ?? "Lumi reported an inference error."); SetState(State with { Health = TrayHealth.Degraded, Detail = serverMessage ?? "Lumi reported an error." }); } return Task.CompletedTask; @@ -444,14 +464,16 @@ public sealed class CompanionRuntime : IAsyncDisposable private static bool ReadBoolean(JsonElement value, string name) => value.TryGetProperty(name, out var property) && property.ValueKind == JsonValueKind.True; private static string? ReadString(JsonElement value, string name) => value.TryGetProperty(name, out var property) && property.ValueKind == JsonValueKind.String ? property.GetString() : null; private CancellationToken _lifetimeToken() => _disposed ? new CancellationToken(true) : CancellationToken.None; - private static async Task WaitForAsync(Task task, TimeSpan timeout, CancellationToken cancellationToken) + private async Task<(bool Completed, string? Failure)> WaitForTestSignalAsync(Task signal, TimeSpan timeout, CancellationToken cancellationToken) { + var failure = _testFailure?.Task ?? throw new InvalidOperationException("The test failure signal is unavailable."); using var timeoutSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); var delay = Task.Delay(timeout, timeoutSource.Token); - if (await Task.WhenAny(task, delay) != task) return false; - timeoutSource.Cancel(); - await task; - return true; + var completed = await Task.WhenAny(signal, failure, delay); + if (completed == signal) { timeoutSource.Cancel(); await signal; return (true, null); } + if (completed == failure) { timeoutSource.Cancel(); return (false, await failure); } + cancellationToken.ThrowIfCancellationRequested(); + return (false, null); } private bool DetectBridgeInstallation() => _bridgeManager.Inspect().Valid; @@ -505,6 +527,8 @@ public sealed class CompanionRuntime : IAsyncDisposable private static string Friendly(Exception error) => error switch { InvalidDataException => error.Message, + System.Net.WebSockets.WebSocketException => "The Lumi server connection ended unexpectedly. The host may have restarted; reconnect and review transcription diagnostics.", + EndOfStreamException => error.Message, HttpRequestException => "Check the Lumi address, TLS certificate, and network connection.", TaskCanceledException => "The connection timed out. Check that Lumi is reachable.", _ => error.Message diff --git a/companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj b/companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj index 4cd106f..400a9ae 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.3 + 0.1.0-experimental.4 0.1.0.0 diff --git a/companion/src/Lumi.Companion.Core/CompanionSocket.cs b/companion/src/Lumi.Companion.Core/CompanionSocket.cs index eab1b80..d8066d5 100644 --- a/companion/src/Lumi.Companion.Core/CompanionSocket.cs +++ b/companion/src/Lumi.Companion.Core/CompanionSocket.cs @@ -12,6 +12,7 @@ public sealed class CompanionSocket : IAsyncDisposable private readonly SemaphoreSlim _sendLock = new(1, 1); private readonly List _backgroundTasks = []; private long _droppedFrames; + private int _disconnectNotified; public long DroppedFrames => Interlocked.Read(ref _droppedFrames); public Guid SessionId { get; private set; } public event Func? MessageReceived; @@ -19,6 +20,7 @@ public sealed class CompanionSocket : IAsyncDisposable public async Task ConnectAsync(DeviceCredential credential, string companionVersion, string pluginVersion, string? obsVersion, CancellationToken cancellationToken) { + _disconnectNotified = 0; _socket = new ClientWebSocket(); _socket.Options.SetRequestHeader("Authorization", $"LumiDevice {credential.DeviceId}.{credential.DeviceSecret}"); var host = new Uri(credential.Host); @@ -98,7 +100,7 @@ public sealed class CompanionSocket : IAsyncDisposable catch (Exception error) { _lifetime?.Cancel(); - Disconnected?.Invoke(error); + if (Interlocked.Exchange(ref _disconnectNotified, 1) == 0) Disconnected?.Invoke(error); } } private static async Task ReceiveMessageAsync(ClientWebSocket socket, CancellationToken cancellationToken) @@ -109,7 +111,12 @@ public sealed class CompanionSocket : IAsyncDisposable do { result = await socket.ReceiveAsync(new ArraySegment(buffer), cancellationToken); - if (result.MessageType == WebSocketMessageType.Close) throw new EndOfStreamException("Lumi closed the companion connection."); + if (result.MessageType == WebSocketMessageType.Close) + { + if (socket.State == WebSocketState.CloseReceived) + try { await socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "close_ack", CancellationToken.None); } catch (WebSocketException) { } + throw new EndOfStreamException($"Lumi closed the companion connection ({result.CloseStatus?.ToString() ?? "no status"}: {result.CloseStatusDescription ?? "no reason"})."); + } body.Write(buffer, 0, result.Count); if (body.Length > ProtocolV1.MaxJsonBytes) throw new InvalidDataException("Lumi message exceeds the protocol limit."); } while (!result.EndOfMessage); diff --git a/plugins/lumi_transcription/backend/sessions/session_coordinator.js b/plugins/lumi_transcription/backend/sessions/session_coordinator.js index 3044428..5cf8c43 100644 --- a/plugins/lumi_transcription/backend/sessions/session_coordinator.js +++ b/plugins/lumi_transcription/backend/sessions/session_coordinator.js @@ -11,7 +11,12 @@ class SessionCoordinator { this.graceMs = options.graceMs || 30000; this.sessions = new Map(); this.stabilizer = new CaptionStabilizer({ now: this.now }); - this.provider.on?.("hypothesis", (event) => this.onHypothesis(event)); + this.provider.on?.("hypothesis", (event) => Promise.resolve(this.onHypothesis(event)).catch((error) => { + this.log.append({ kind: "inference_result_error", code: error.code || null, message: error.message }); + })); + this.provider.on?.("provider_error", (error) => Promise.resolve(this.onProviderError(error)).catch((failure) => { + this.log.append({ kind: "provider_failure_handler_error", message: failure.message }); + })); } create(device, send, resumeSessionId = null) { const resumable = resumeSessionId && this.sessions.get(resumeSessionId); @@ -66,7 +71,12 @@ class SessionCoordinator { const tracks = Array.from(session.tracks.values()).filter((track) => track.enabled && !track.source_missing); if (!tracks.length) throw Object.assign(new Error("Select an available OBS audio source first."), { code: "NO_TRACKS" }); const providerHealth = await this.provider.health(); - if (!providerHealth.healthy) throw Object.assign(new Error("The whisper.cpp worker is not ready. Install and load a model first."), { code: "PROVIDER_UNAVAILABLE", details: providerHealth }); + if (!providerHealth.healthy) { + const reason = providerHealth.last_error?.message || + (providerHealth.last_exit ? `The worker last exited with code ${providerHealth.last_exit.code ?? "unknown"}.` : null) || + `Worker state is ${providerHealth.state || "unavailable"}.`; + throw Object.assign(new Error(`Speech recognition is not ready. ${reason}`), { code: "PROVIDER_UNAVAILABLE", details: providerHealth }); + } await this.provider.startSession({ id: session.id, mode }); for (const track of tracks) await this.provider.addTrack(session.id, serializeTrack(track)); await session.delivery.start({ testMode: mode === "test" }); @@ -152,6 +162,24 @@ class SessionCoordinator { const result = await session.delivery.deliver(event); this.log.append({ kind: "caption", session_id: session.id, source_uuid: track.source_uuid, caption_text: event.stable_text, uncertain_text: event.uncertain_text, revision: event.revision, final: event.final, delivery: result.disposition, latency: event.latency, model: event.model }); } + async onProviderError(error) { + const affected = Array.from(this.sessions.values()).filter((session) => session.state === "running" || session.state === "grace"); + for (const session of affected) { + clearTimeout(session.graceTimer); + session.graceTimer = null; + session.graceUntil = 0; + session.state = "idle"; + session.mode = null; + for (const track of session.tracks.values()) track.buffer.clear(); + try { await session.delivery.stop(); } catch { } + session.send("error", { + code: error.code || "PROVIDER_FAILED", + message: `Speech recognition stopped: ${error.message}`, + recoverable: true + }, session.id); + this.log.append({ kind: "session", state: "inference_failed", session_id: session.id, code: error.code || "PROVIDER_FAILED", message: error.message }); + } + } } function speakerLabel(session, track) { const enabled = Array.from(session.tracks.values()).filter((candidate) => candidate.enabled); return enabled.length === 1 && track.single_speaker ? null : track.speaker_label || track.display_name; } diff --git a/plugins/lumi_transcription/backend/transcription/provider.js b/plugins/lumi_transcription/backend/transcription/provider.js index 4972b26..650c542 100644 --- a/plugins/lumi_transcription/backend/transcription/provider.js +++ b/plugins/lumi_transcription/backend/transcription/provider.js @@ -27,6 +27,10 @@ class WhisperWorkerSupervisor extends EventEmitter { this.stopping = false; this.restartTimes = []; this.stdoutBuffer = ""; + this.lastExit = null; + this.lastError = null; + this.lastDiagnostic = null; + this.startedAt = null; } start() { if (this.child) return; @@ -35,9 +39,13 @@ class WhisperWorkerSupervisor extends EventEmitter { this.state = "starting"; const child = this.spawn(this.executable, this.args, { stdio: ["pipe", "pipe", "pipe"], windowsHide: true }); this.child = child; + this.startedAt = Date.now(); child.stdout.on("data", (chunk) => this.onStdout(chunk)); - child.stderr.on("data", (chunk) => this.emit("diagnostic", { level: "warning", message: String(chunk).trim().slice(0, 1000) })); - child.on("error", (error) => this.emit("error", error)); + child.stderr.on("data", (chunk) => this.onDiagnostic(String(chunk).trim().slice(0, 1000))); + child.stdin.on("error", (error) => this.onIoError("stdin", error)); + child.stdout.on("error", (error) => this.onIoError("stdout", error)); + child.stderr.on("error", (error) => this.onIoError("stderr", error)); + child.on("error", (error) => this.onProcessError(error)); child.on("exit", (code, signal) => this.onExit(code, signal)); child.stdin.on("drain", () => this.flush()); this.state = "running"; @@ -59,7 +67,9 @@ class WhisperWorkerSupervisor extends EventEmitter { flush() { if (!this.child || this.state !== "running") return; let packet; - while ((packet = this.queue.shift())) if (!this.child.stdin.write(packet)) break; + try { + while ((packet = this.queue.shift())) if (!this.child.stdin.write(packet)) break; + } catch (error) { this.onIoError("stdin", error); } } async stop() { this.stopping = true; @@ -84,32 +94,71 @@ class WhisperWorkerSupervisor extends EventEmitter { this.stdoutBuffer = this.stdoutBuffer.slice(index + 1); if (!line) continue; try { this.emit("message", JSON.parse(line)); } - catch { this.emit("diagnostic", { level: "warning", message: "Whisper worker emitted malformed output." }); } + catch { this.onDiagnostic("Whisper worker emitted malformed output."); } } } + onDiagnostic(message) { + if (!message) return; + this.lastDiagnostic = { message, at: Date.now() }; + this.emit("diagnostic", { level: "warning", message }); + } + onIoError(stream, error) { + if (this.stopping) return; + this.lastError = { stream, code: error.code || null, message: error.message, at: Date.now() }; + this.emit("error", Object.assign(new Error(`Whisper worker ${stream} failed: ${error.message}`), { code: error.code, stream })); + } + onProcessError(error) { + this.lastError = { stream: "process", code: error.code || null, message: error.message, at: Date.now() }; + this.emit("error", error); + } onExit(code, signal) { this.child = null; + this.lastExit = { code, signal: signal || null, at: Date.now() }; if (this.stopping) { this.state = "stopped"; return; } this.state = "failed"; - this.emit("crash", { code, signal }); + this.emit("state", { state: this.state }); + this.emit("crash", { code, signal, at: this.lastExit.at }); const now = Date.now(); this.restartTimes = this.restartTimes.filter((time) => now - time < this.restartWindowMs); if (this.restartTimes.length >= this.maxRestarts) return; this.restartTimes.push(now); setTimeout(() => { try { this.start(); } catch (error) { this.emit("error", error); } }, 250).unref?.(); } - health() { return { healthy: this.state === "running", state: this.state, queue: this.queue.metrics(), restarts_in_window: this.restartTimes.length }; } + health() { + return { + healthy: this.state === "running", state: this.state, queue: this.queue.metrics(), + restarts_in_window: this.restartTimes.length, started_at: this.startedAt, + last_exit: this.lastExit, last_error: this.lastError, last_diagnostic: this.lastDiagnostic + }; + } } class WhisperCppServerProvider extends TranscriptionProvider { constructor(supervisor) { - super(); this.worker = supervisor; this.model = null; this.sessions = new Set(); + super(); this.worker = supervisor; this.model = null; this.desiredModel = null; this.modelReady = false; this.loading = false; this.recovering = false; this.sessions = new Set(); supervisor.on("message", (message) => this.emit(message.type || "message", message)); - supervisor.on("crash", (event) => this.emit("provider_error", Object.assign(new Error("Whisper worker crashed."), { details: event }))); + supervisor.on("crash", (event) => { + this.modelReady = false; + this.sessions.clear(); + this.emit("provider_error", Object.assign(new Error(`Whisper worker exited unexpectedly${event.code == null ? "" : ` with code ${event.code}`}.`), { code: "WORKER_CRASHED", details: event })); + }); supervisor.on("error", (error) => this.emit("provider_error", error)); + supervisor.on("state", ({ state }) => { + if (state === "running" && this.desiredModel && !this.modelReady && !this.loading && !this.recovering) this.recoverModel(); + }); } async loadModel(model) { - this.worker.start(); + this.desiredModel = model; + this.loading = true; + try { + this.worker.start(); + const result = await this.requestModel(model); + this.model = model; + this.modelReady = true; + return { ...(await this.health()), loaded: result }; + } finally { this.loading = false; } + } + async requestModel(model) { const loaded = new Promise((resolve, reject) => { const timer = setTimeout(() => { cleanup(); reject(new Error("The whisper.cpp worker did not confirm model loading in time.")); }, 30000); const onLoaded = (event) => { cleanup(); resolve(event); }; @@ -120,18 +169,36 @@ class WhisperCppServerProvider extends TranscriptionProvider { this.once("provider_error", onProviderError); }); this.worker.send({ type: "load_model", model }); - const result = await loaded; - this.model = model; - return { ...(await this.health()), loaded: result }; + return loaded; + } + recoverModel() { + this.recovering = true; + this.requestModel(this.desiredModel).then((result) => { + this.model = this.desiredModel; + this.modelReady = true; + this.emit("recovered", { model_id: result.model_id || this.model?.id || null, backend: result.backend || null }); + }).catch((error) => this.emit("provider_error", Object.assign(error, { code: error.code || "MODEL_RECOVERY_FAILED" }))) + .finally(() => { this.recovering = false; }); } async benchmark(options = {}) { this.worker.send({ type: "benchmark", options }); return { accepted: true, model: this.model?.id || null }; } - async startSession(session) { this.sessions.add(session.id); this.worker.send({ type: "start_session", session }); } + async startSession(session) { + if (!this.modelReady) throw coded("PROVIDER_UNAVAILABLE", "The speech model is not ready in the worker."); + this.sessions.add(session.id); this.worker.send({ type: "start_session", session }); + } async addTrack(sessionId, track) { this.worker.send({ type: "add_track", session_id: sessionId, track }); } async pushAudio(sessionId, trackId, frame) { return this.worker.send({ type: "audio", session_id: sessionId, track_id: trackId, sequence: frame.sequence, capture_timestamp_us: frame.capture_timestamp_us, captured_at: Date.now() }, frame.pcm); } async removeTrack(sessionId, trackId) { this.worker.send({ type: "remove_track", session_id: sessionId, track_id: trackId }); } async stopSession(sessionId) { this.sessions.delete(sessionId); this.worker.send({ type: "stop_session", session_id: sessionId }); } - async health() { return { provider: "whisper_cpp_server", model: this.model?.id || null, sessions: this.sessions.size, ...this.worker.health() }; } - async stop() { this.sessions.clear(); await this.worker.stop(); } + async health() { + const worker = this.worker.health(); + return { + ...worker, provider: "whisper_cpp_server", model: this.model?.id || this.desiredModel?.id || null, + model_ready: this.modelReady, sessions: this.sessions.size, + healthy: worker.healthy && this.modelReady, + state: !worker.healthy ? worker.state : this.modelReady ? "running" : this.recovering || this.loading ? "loading_model" : "model_unavailable" + }; + } + async stop() { this.sessions.clear(); this.modelReady = false; await this.worker.stop(); } } function coded(code, message) { return Object.assign(new Error(message), { code }); } diff --git a/plugins/lumi_transcription/backend/transcription/worker-native/src/main.cpp b/plugins/lumi_transcription/backend/transcription/worker-native/src/main.cpp index 45da44f..6a92619 100644 --- a/plugins/lumi_transcription/backend/transcription/worker-native/src/main.cpp +++ b/plugins/lumi_transcription/backend/transcription/worker-native/src/main.cpp @@ -13,6 +13,11 @@ #include #include +#if defined(_WIN32) +#include +#include +#endif + using json = nlohmann::json; using clock_type = std::chrono::steady_clock; @@ -203,6 +208,11 @@ void handle(const json & packet, const std::vector & pcm, bool & r } int main() { +#if defined(_WIN32) + // PCM frames are transported over stdin. Windows text mode treats 0x1A in + // ordinary audio as EOF, which closes the worker and breaks the host pipe. + if (_setmode(_fileno(stdin), _O_BINARY) == -1) return 2; +#endif std::ios::sync_with_stdio(false); whisper_log_set([](enum ggml_log_level, const char *, void *) {}, nullptr); emit({{"type", "ready"}, {"protocol_version", 1}, {"backend", backend}}); diff --git a/plugins/lumi_transcription/companion_manifest.json b/plugins/lumi_transcription/companion_manifest.json index de9c7e0..d4be059 100644 --- a/plugins/lumi_transcription/companion_manifest.json +++ b/plugins/lumi_transcription/companion_manifest.json @@ -1,8 +1,8 @@ { "schema_version": 1, - "version": "0.1.0-experimental.3", + "version": "0.1.0-experimental.4", "signed": false, - "release_notes": "Adds user-approved in-place updates and Companion-managed OBS integration maintenance.", + "release_notes": "Reports server inference failures at Speech recognition and handles abrupt WebSocket disconnects without leaving the test hanging.", "artifacts": [ { "id": "windows-x64-self-contained", @@ -10,9 +10,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.3/Lumi.Companion-win-x64.zip", - "sha256": "913bc719653b7f5b5bbd1b96f136e503654aa6913e8225c12a46ce19d9d91811", - "bytes": 43034535, + "url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.4/Lumi.Companion-win-x64.zip", + "sha256": "20b73e929c6eba816b5ff5737ced053cdcb1e737fac831bad7260f22d3cd2fe3", + "bytes": 41666903, "entrypoint": "Lumi.Companion.App.exe" } ] diff --git a/plugins/lumi_transcription/index.js b/plugins/lumi_transcription/index.js index 9f3b300..a674cc7 100644 --- a/plugins/lumi_transcription/index.js +++ b/plugins/lumi_transcription/index.js @@ -36,7 +36,12 @@ module.exports = { }); supervisor.on("diagnostic", (entry) => diagnosticLog.append({ kind: "worker", ...entry })); supervisor.on("error", (error) => diagnosticLog.append({ kind: "worker", state: "error", message: error.message })); + supervisor.on("crash", (entry) => diagnosticLog.append({ kind: "worker", state: "crashed", exit_code: entry.code, signal: entry.signal, occurred_at: entry.at })); + supervisor.on("state", (entry) => diagnosticLog.append({ kind: "worker", state: entry.state })); const provider = new WhisperCppServerProvider(supervisor); + provider.on("diagnostic", (entry) => diagnosticLog.append({ kind: "worker", state: entry.state || "diagnostic", level: entry.level || null, message: entry.message || null })); + provider.on("provider_error", (error) => diagnosticLog.append({ kind: "worker", state: "provider_error", code: error.code || null, message: error.message, details: error.details || null })); + provider.on("recovered", (entry) => diagnosticLog.append({ kind: "worker", state: "recovered", ...entry })); const sessions = new SessionCoordinator({ provider, deliveryFactory: (send) => new CompanionCaptionDeliveryAdapter(send), diff --git a/plugins/lumi_transcription/plugin.json b/plugins/lumi_transcription/plugin.json index 7419d4d..193842e 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.3", + "version": "0.1.0-experimental.4", "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 663bf28..e7ae355 100644 --- a/plugins/lumi_transcription/tests/verify.js +++ b/plugins/lumi_transcription/tests/verify.js @@ -37,6 +37,7 @@ async function run() { verifyQueues(); verifyStabilization(); await verifySessionLifecycle(); + await verifyProviderFailureFeedback(); await verifyWorkerRestart(); await verifyNativeWorkerBoundary(); await verifyAuthenticatedGateway(); @@ -98,6 +99,7 @@ function verifyLocalhostTransportPolicy() { } function verifyCompanionVersionOrdering() { + assert.equal(plugin.compareVersions("0.1.0-experimental.4", "0.1.0-experimental.3"), 1); assert.equal(plugin.compareVersions("0.1.0-experimental.3", "0.1.0-experimental.2"), 1); assert.equal(plugin.compareVersions("0.1.0-experimental.2", "0.1.0-experimental.2"), 0); assert.equal(plugin.compareVersions("0.1.0", "0.1.0-experimental.9"), 1); @@ -189,6 +191,29 @@ async function verifySessionLifecycle() { await coordinator.close(); } +async function verifyProviderFailureFeedback() { + class Provider extends EventEmitter { + async health() { return { healthy: true, state: "running", model_ready: true }; } + async startSession() {} async addTrack() {} async stopSession() {} + } + const provider = new Provider(); + const sent = []; + const coordinator = new SessionCoordinator({ + provider, + deliveryFactory: () => ({ start: async () => {}, stop: async () => {}, pause: async () => {}, resume: async () => {} }) + }); + const { session } = coordinator.create({ id: "device" }, (type, payload) => sent.push({ type, payload })); + const source = crypto.randomUUID(); + coordinator.updateSource(session.id, { source_uuid: source, display_name: "Mic", primary: true, program_active: true }); + await coordinator.start(session.id, { mode: "test" }); + provider.emit("provider_error", Object.assign(new Error("worker exited with code 3"), { code: "WORKER_CRASHED" })); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(coordinator.status(session.id).state, "idle"); + assert.equal(sent.at(-1).type, "error"); + assert.match(sent.at(-1).payload.message, /worker exited with code 3/); + await coordinator.close(); +} + async function verifyWorkerRestart() { const children = []; const fakeSpawn = () => { @@ -197,12 +222,16 @@ async function verifyWorkerRestart() { children.push(child); return child; }; const supervisor = new WhisperWorkerSupervisor({ executable: "fake-worker", spawn: fakeSpawn, maxRestarts: 1 }); - supervisor.on("error", () => {}); + let streamErrors = 0; + supervisor.on("error", () => { streamErrors += 1; }); supervisor.start(); assert.equal(supervisor.send({ type: "audio", captured_at: Date.now() }, Buffer.alloc(640)), true); children[0].emit("exit", 1, null); await new Promise((resolve) => setTimeout(resolve, 320)); assert.equal(children.length, 2); + children[1].stdin.emit("error", Object.assign(new Error("broken pipe"), { code: "EPIPE" })); + assert.equal(streamErrors, 1); + assert.equal(supervisor.health().last_error.code, "EPIPE"); supervisor.stopping = true; children[1].emit("exit", 0, null); } @@ -223,6 +252,7 @@ async function verifyNativeWorkerBoundary() { const source = fs.readFileSync(path.join(sourceRoot, "src/main.cpp"), "utf8"); assert.match(cmake, /f049fff95a089aa9969deb009cdd4892b3e74916/); assert.match(source, /max_samples = sample_rate \* 6/); + assert.match(source, /_setmode\(_fileno\(stdin\), _O_BINARY\)/); assert.doesNotMatch(source, /ofstream|fwrite|WriteAllBytes/); } diff --git a/plugins/lumi_transcription/views/settings.ejs b/plugins/lumi_transcription/views/settings.ejs index 0abaf9c..cbdb3dc 100644 --- a/plugins/lumi_transcription/views/settings.ejs +++ b/plugins/lumi_transcription/views/settings.ejs @@ -22,7 +22,7 @@
  1. Download and pair CompanionCreate a short-lived, single-use package for the streaming computer.
  2. Install the OBS integrationCompanion owns installation and repair; the bridge has no separate settings.
    3. -
  3. Load and benchmark a modelSmall English is the recommended starting point.
    4. +
  4. Load and benchmark a modelSmall English is the recommended starting point.
  5. Select and test a microphoneThe real safe test must pass before live delivery is enabled.
@@ -38,9 +38,12 @@
Providerwhisper.cpp <%= runtimeManifest.tested_version %>
Worker<%= providerHealth.state %>
-
Selected model<%= providerHealth.model || 'Not loaded' %>
+
Selected model<%= providerHealth.model ? providerHealth.model + (providerHealth.model_ready ? '' : ' · not ready') : 'Not loaded' %>
Active sessions<%= providerHealth.sessions || 0 %>
+ <% if (!providerHealth.healthy) { const workerReason = providerHealth.last_error?.message || (providerHealth.last_exit ? `Last exit code: ${providerHealth.last_exit.code ?? 'unknown'}` : null) || providerHealth.last_diagnostic?.message || 'No worker failure detail has been reported yet.'; %> +
Speech recognition needs attention

<%= workerReason %> Reload the verified model below, then rerun the Companion test.

+ <% } %>
Shared GPU awareness

Lumi warns and benchmarks when Lumi AI already occupies GPU memory. It never unloads AI models automatically.

@@ -64,9 +67,9 @@

<%= model.label %>

<%= model.recommended ? 'Recommended starting point' : 'Fallback option' %> · <%= Math.round(model.bytes / 1048576) %> MiB

- <%= providerHealth.model === model.id ? 'Loaded' : model.status.valid ? 'Verified' : model.status.installed ? 'Checksum failed' : 'Not installed' %> + <%= providerHealth.model === model.id && providerHealth.model_ready ? 'Loaded' : providerHealth.model === model.id && model.status.valid ? 'Reload required' : model.status.valid ? 'Verified' : model.status.installed ? 'Checksum failed' : 'Not installed' %> <% if (!model.status.valid) { %><% } %> - <% if (model.status.valid && providerHealth.model !== model.id) { %><% } %> + <% if (model.status.valid && (providerHealth.model !== model.id || !providerHealth.model_ready)) { %><% } %>
<% }) %>