using System.Text.Json; using Jibo.Cloud.Application.Abstractions; using Jibo.Cloud.Application.Services; using Jibo.Cloud.Domain.Models; using Jibo.Cloud.Infrastructure.Telemetry; using Jibo.Runtime.Abstractions; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Moq; namespace Jibo.Cloud.Tests.Turn; public sealed class FileTurnTelemetrySinkTests { [Fact] public async Task RecordsTurnDiagnosticSnapshot() { var directoryPath = Path.Combine(Path.GetTempPath(), "OpenJibo.Tests", Guid.NewGuid().ToString("N")); var sink = new FileTurnTelemetrySink( NullLogger.Instance, Options.Create(new TurnTelemetryOptions { Enabled = true, DirectoryPath = directoryPath })); await sink.RecordTurnDiagnosticAsync("yes_no_turn_received", new Dictionary { ["transID"] = "trans-1", ["bufferedAudioBytes"] = 1234, ["listenRules"] = new[] { "shared/yes_no", "globals/gui_nav" }, ["awaitingTurnCompletion"] = true }); var filePath = Directory.GetFiles(directoryPath, "*.events.ndjson").Single(); var payload = JsonDocument.Parse(await File.ReadAllTextAsync(filePath)).RootElement; Assert.Equal("yes_no_turn_received", payload.GetProperty("type").GetString()); Assert.Equal("trans-1", payload.GetProperty("details").GetProperty("transID").GetString()); Assert.Equal(1234, payload.GetProperty("details").GetProperty("bufferedAudioBytes").GetInt32()); } [Fact] public async Task RecordsCaptureIndexForTurnDiagnosticsAndErrors() { var directoryPath = Path.Combine(Path.GetTempPath(), "OpenJibo.Tests", Guid.NewGuid().ToString("N")); var sink = new FileTurnTelemetrySink( NullLogger.Instance, Options.Create(new TurnTelemetryOptions { Enabled = true, DirectoryPath = directoryPath })); await sink.RecordTurnDiagnosticAsync("yes_no_turn_received", new Dictionary { ["transID"] = "trans-1", ["bufferedAudioBytes"] = 1234 }); await sink.RecordTranscriptError(new InvalidOperationException("boom"), "turn error"); var indexPath = Path.Combine(directoryPath, "capture-index.ndjson"); var lines = await File.ReadAllLinesAsync(indexPath); Assert.Contains(lines, line => line.Contains("\"eventType\":\"yes_no_turn_received\"", StringComparison.Ordinal)); Assert.Contains(lines, line => line.Contains("\"eventType\":\"transcript_error\"", StringComparison.Ordinal)); Assert.Contains(lines, line => line.Contains("\"message\":\"Turn telemetry diagnostic\"", StringComparison.Ordinal)); Assert.Contains(lines, line => line.Contains("\"message\":\"Turn telemetry error\"", StringComparison.Ordinal)); } [Fact] public async Task RecordsTranscriptErrorOnTurnError() { var sink = new Mock(); var sttStrategySelector = new Mock(); sttStrategySelector.Setup(s => s.SelectAsync(It.IsAny(), It.IsAny())) .ThrowsAsync(new Exception("dummy")); var turnService = new WebSocketTurnFinalizationService(Mock.Of(), sttStrategySelector.Object, sink.Object ); await turnService.HandleTurnAsync(new CloudSession { TurnState = { BufferedAudioBytes = 100 } }, new WebSocketMessageEnvelope(), "dummy", CancellationToken.None); sink.Verify( s => s.RecordTranscriptError(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once()); } [Fact] public async Task AutoFinalize_DoesNotFallbackImmediately_WhenSttThrows() { var sink = new Mock(); var sttStrategySelector = new Mock(); sttStrategySelector.Setup(s => s.SelectAsync(It.IsAny(), It.IsAny())) .ThrowsAsync(new InvalidOperationException("ffmpeg failed")); var turnService = new WebSocketTurnFinalizationService(Mock.Of(), sttStrategySelector.Object, sink.Object ); var session = new CloudSession { TurnState = { AwaitingTurnCompletion = true, SawListen = true, SawContext = true, BufferedAudioBytes = 15000, BufferedAudioChunkCount = 5, FirstAudioReceivedUtc = DateTimeOffset.UtcNow - TimeSpan.FromSeconds(2) } }; var replies = await turnService.HandleContextAsync( session, new WebSocketMessageEnvelope { Text = """{"type":"CONTEXT","data":{"topic":"conversation"}}""" }, CancellationToken.None); Assert.Empty(replies); Assert.True(session.TurnState.AwaitingTurnCompletion); Assert.Equal(15000, session.TurnState.BufferedAudioBytes); Assert.Equal("ffmpeg failed", session.TurnState.LastSttError); sink.Verify( s => s.RecordTranscriptError(It.IsAny(), It.IsAny(), It.IsAny()), Times.Once()); } [Fact] public async Task HandleContext_EmitsGlsmPhaseTransitionDiagnostic() { var sink = new Mock(); sink.Setup(s => s.RecordTurnDiagnosticAsync(It.IsAny(), It.IsAny>(), It.IsAny())) .Returns(Task.CompletedTask); var turnService = new WebSocketTurnFinalizationService( Mock.Of(), Mock.Of(), sink.Object); var session = new CloudSession { Token = "glsm-phase-token", TurnState = { TransId = "trans-glsm", AwaitingTurnCompletion = true, SawListen = true, ListenOpenedUtc = DateTimeOffset.UtcNow - TimeSpan.FromSeconds(1) } }; session.Metadata["glsmPhase"] = "HJ_LISTENING"; await turnService.HandleContextAsync( session, new WebSocketMessageEnvelope { HostName = "neo-hub.jibo.com", Path = "/listen", Kind = "neo-hub-listen", Text = """{"type":"CONTEXT","transID":"trans-glsm","data":{"topic":"conversation"}}""" }, CancellationToken.None); sink.Verify( s => s.RecordTurnDiagnosticAsync( "glsm_phase_transition", It.Is>(details => details.ContainsKey("state") && string.Equals(details["state"] == null ? null : details["state"]!.ToString(), "LISTENING", StringComparison.OrdinalIgnoreCase)), It.IsAny()), Times.AtLeastOnce()); } }