Jibo-Revival-Group/JiboExperiments
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new pass at websocket telemetry and prepare for capture

Browse Source
This commit is contained in:
Jacob Dubin
2026-04-12 09:00:17 -05:00
parent 70d36bf563
commit 8e060a1ded
15 changed files with 603 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
OpenJibo/src/Jibo.Cloud/dotnet/README.md
View File

@@ -72,6 +72,7 @@ Current websocket scope is still intentionally narrow:
- synthetic `LISTEN` result shaping for `LISTEN`, `CLIENT_NLU`, and `CLIENT_ASR`
- buffered audio state tracking behind a dedicated turn-finalization layer
- synthetic STT strategy selection for fixture-driven audio turn completion
- structured websocket telemetry and live-run fixture export
- `CONTEXT` capture and follow-up turn state
- `EOS` completion
- first skill vertical for joke/chat `SKILL_ACTION` playback

54
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Api/Program.cs
View File

@@ -1,12 +1,13 @@
using System.Net.WebSockets;
using System.Text;
using Jibo.Cloud.Application.Abstractions;
using Jibo.Cloud.Application.Services;
using Jibo.Cloud.Domain.Models;
using Jibo.Cloud.Infrastructure.DependencyInjection;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddOpenJiboCloud();
builder.Services.AddOpenJiboCloud(builder.Configuration);
var app = builder.Build();
@@ -21,10 +22,21 @@ app.Use(async (context, next) =>
}
var webSocketService = context.RequestServices.GetRequiredService<JiboWebSocketService>();
var telemetrySink = context.RequestServices.GetRequiredService<IWebSocketTelemetrySink>();
using var socket = await context.WebSockets.AcceptWebSocketAsync();
var kind = ResolveSocketKind(context.Request.Host.Host, context.Request.Path);
var token = ResolveToken(context.Request);
var openEnvelope = new WebSocketMessageEnvelope
{
ConnectionId = Guid.NewGuid().ToString("N"),
HostName = context.Request.Host.Host,
Path = context.Request.Path.Value ?? "/",
Kind = kind,
Token = token
};
var openSession = ResolveSession(webSocketService, openEnvelope);
await telemetrySink.RecordConnectionOpenedAsync(openEnvelope, openSession, context.RequestAborted);
while (socket.State == WebSocketState.Open)
{
@@ -47,6 +59,8 @@ app.Use(async (context, next) =>
};
var replies = await webSocketService.HandleMessageAsync(envelope, context.RequestAborted);
var session = ResolveSession(webSocketService, envelope);
await telemetrySink.RecordInboundAsync(envelope, session, ReadMessageType(envelope.Text), context.RequestAborted);
foreach (var reply in replies)
{
if (string.IsNullOrWhiteSpace(reply.Text))
@@ -57,7 +71,20 @@ app.Use(async (context, next) =>
var payload = Encoding.UTF8.GetBytes(reply.Text);
await socket.SendAsync(payload, WebSocketMessageType.Text, true, context.RequestAborted);
}
await telemetrySink.RecordOutboundAsync(envelope, session, replies, context.RequestAborted);
}
var closeEnvelope = new WebSocketMessageEnvelope
{
ConnectionId = Guid.NewGuid().ToString("N"),
HostName = context.Request.Host.Host,
Path = context.Request.Path.Value ?? "/",
Kind = kind,
Token = token
};
var closeSession = ResolveSession(webSocketService, closeEnvelope);
await telemetrySink.RecordConnectionClosedAsync(closeEnvelope, closeSession, "socket-loop-ended", context.RequestAborted);
});
app.MapGet("/health", () => Results.Json(new { ok = true, service = "OpenJibo Cloud Api" }));
@@ -167,4 +194,29 @@ static string? ResolveToken(HttpRequest request)
return null;
}
static string ReadMessageType(string? text)
{
if (string.IsNullOrWhiteSpace(text))
{
return "BINARY_OR_EMPTY";
}
try
{
using var document = System.Text.Json.JsonDocument.Parse(text);
return document.RootElement.TryGetProperty("type", out var type) && type.ValueKind == System.Text.Json.JsonValueKind.String
? type.GetString() ?? "UNKNOWN"
: "UNKNOWN";
}
catch
{
return "TEXT";
}
}
static CloudSession ResolveSession(JiboWebSocketService webSocketService, WebSocketMessageEnvelope envelope)
{
return webSocketService.GetOrCreateSession(envelope);
}
internal sealed record ReceivedSocketMessage(WebSocketMessageType MessageType, byte[] Buffer);

12
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Application/Abstractions/IWebSocketTelemetrySink.cs Normal file
View File

@@ -0,0 +1,12 @@
using Jibo.Cloud.Domain.Models;
namespace Jibo.Cloud.Application.Abstractions;
public interface IWebSocketTelemetrySink
{
Task RecordConnectionOpenedAsync(WebSocketMessageEnvelope envelope, CloudSession session, CancellationToken cancellationToken = default);
Task RecordInboundAsync(WebSocketMessageEnvelope envelope, CloudSession session, string? messageType, CancellationToken cancellationToken = default);
Task RecordTurnEventAsync(WebSocketMessageEnvelope envelope, CloudSession session, string eventType, IReadOnlyDictionary<string, object?> details, CancellationToken cancellationToken = default);
Task RecordOutboundAsync(WebSocketMessageEnvelope envelope, CloudSession session, IReadOnlyList<WebSocketReply> replies, CancellationToken cancellationToken = default);
Task RecordConnectionClosedAsync(WebSocketMessageEnvelope envelope, CloudSession session, string reason, CancellationToken cancellationToken = default);
}

34
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Application/Services/JiboWebSocketService.cs
View File

@@ -7,17 +7,28 @@ namespace Jibo.Cloud.Application.Services;
public sealed class JiboWebSocketService(
ICloudStateStore stateStore,
IWebSocketTelemetrySink telemetrySink,
WebSocketTurnFinalizationService turnFinalizationService)
{
public CloudSession GetOrCreateSession(WebSocketMessageEnvelope envelope)
{
return stateStore.FindSessionByToken(envelope.Token ?? string.Empty) ??
stateStore.OpenSession(envelope.Kind, null, envelope.Token, envelope.HostName, envelope.Path);
}
public async Task<IReadOnlyList<WebSocketReply>> HandleMessageAsync(WebSocketMessageEnvelope envelope, CancellationToken cancellationToken = default)
{
var session = stateStore.FindSessionByToken(envelope.Token ?? string.Empty) ??
stateStore.OpenSession(envelope.Kind, null, envelope.Token, envelope.HostName, envelope.Path);
var session = GetOrCreateSession(envelope);
session.LastSeenUtc = DateTimeOffset.UtcNow;
if (envelope.IsBinary)
{
return turnFinalizationService.HandleBinaryAudio(session, envelope);
var replies = turnFinalizationService.HandleBinaryAudio(session, envelope);
await telemetrySink.RecordTurnEventAsync(envelope, session, "binary_audio_received", new Dictionary<string, object?>
{
["bytes"] = envelope.Binary?.Length ?? 0
}, cancellationToken);
return replies;
}
var parsedType = ReadMessageType(envelope.Text);
@@ -31,12 +42,25 @@ public sealed class JiboWebSocketService(
if (parsedType == "CONTEXT")
{
return turnFinalizationService.HandleContext(session, envelope.Text);
var replies = turnFinalizationService.HandleContext(session, envelope.Text);
await telemetrySink.RecordTurnEventAsync(envelope, session, "context_received", new Dictionary<string, object?>
{
["transID"] = session.TurnState.TransId
}, cancellationToken);
return replies;
}
if (parsedType is "LISTEN" or "CLIENT_NLU" or "CLIENT_ASR")
{
return await turnFinalizationService.HandleTurnAsync(session, envelope, parsedType, cancellationToken);
var replies = await turnFinalizationService.HandleTurnAsync(session, envelope, parsedType, cancellationToken);
await telemetrySink.RecordTurnEventAsync(envelope, session, "turn_processed", new Dictionary<string, object?>
{
["messageType"] = parsedType,
["replyCount"] = replies.Count,
["transcript"] = session.LastTranscript,
["intent"] = session.LastIntent
}, cancellationToken);
return replies;
}
return

13
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Application/Services/NullWebSocketTelemetrySink.cs Normal file
View File

@@ -0,0 +1,13 @@
using Jibo.Cloud.Application.Abstractions;
using Jibo.Cloud.Domain.Models;
namespace Jibo.Cloud.Application.Services;
public sealed class NullWebSocketTelemetrySink : IWebSocketTelemetrySink
{
public Task RecordConnectionOpenedAsync(WebSocketMessageEnvelope envelope, CloudSession session, CancellationToken cancellationToken = default) => Task.CompletedTask;
public Task RecordInboundAsync(WebSocketMessageEnvelope envelope, CloudSession session, string? messageType, CancellationToken cancellationToken = default) => Task.CompletedTask;
public Task RecordTurnEventAsync(WebSocketMessageEnvelope envelope, CloudSession session, string eventType, IReadOnlyDictionary<string, object?> details, CancellationToken cancellationToken = default) => Task.CompletedTask;
public Task RecordOutboundAsync(WebSocketMessageEnvelope envelope, CloudSession session, IReadOnlyList<WebSocketReply> replies, CancellationToken cancellationToken = default) => Task.CompletedTask;
public Task RecordConnectionClosedAsync(WebSocketMessageEnvelope envelope, CloudSession session, string reason, CancellationToken cancellationToken = default) => Task.CompletedTask;
}

25
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Domain/Models/CapturedWebSocketFixture.cs Normal file
View File

@@ -0,0 +1,25 @@
using System.Text.Json;
namespace Jibo.Cloud.Domain.Models;
public sealed class CapturedWebSocketFixture
{
public string Name { get; init; } = string.Empty;
public CapturedWebSocketFixtureSession Session { get; init; } = new();
public IReadOnlyList<CapturedWebSocketFixtureStep> Steps { get; init; } = [];
}
public sealed class CapturedWebSocketFixtureSession
{
public string HostName { get; init; } = string.Empty;
public string Path { get; init; } = "/";
public string Kind { get; init; } = "unknown";
public string? Token { get; init; }
}
public sealed class CapturedWebSocketFixtureStep
{
public JsonElement? Text { get; init; }
public IReadOnlyList<int>? Binary { get; init; }
public IReadOnlyList<string> ExpectedReplyTypes { get; init; } = [];
}

24
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Domain/Models/WebSocketTelemetryRecord.cs Normal file
View File

@@ -0,0 +1,24 @@
namespace Jibo.Cloud.Domain.Models;
public sealed class WebSocketTelemetryRecord
{
public DateTimeOffset TimestampUtc { get; init; } = DateTimeOffset.UtcNow;
public string EventType { get; init; } = string.Empty;
public string SessionId { get; init; } = string.Empty;
public string ConnectionId { get; init; } = string.Empty;
public string? Token { get; init; }
public string HostName { get; init; } = string.Empty;
public string Path { get; init; } = "/";
public string Kind { get; init; } = "unknown";
public string? TransId { get; init; }
public string? MessageType { get; init; }
public string Direction { get; init; } = "internal";
public string? Text { get; init; }
public int? BinaryLength { get; init; }
public IReadOnlyList<string> ReplyTypes { get; init; } = [];
public int BufferedAudioBytes { get; init; }
public int BufferedAudioChunks { get; init; }
public int FinalizeAttempts { get; init; }
public bool AwaitingTurnCompletion { get; init; }
public IReadOnlyDictionary<string, object?> Details { get; init; } = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
}

10
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Infrastructure/DependencyInjection/ServiceCollectionExtensions.cs
View File

@@ -1,19 +1,27 @@
using Jibo.Cloud.Application.Abstractions;
using Jibo.Cloud.Application.Services;
using Jibo.Cloud.Infrastructure.Persistence;
using Jibo.Cloud.Infrastructure.Telemetry;
using Jibo.Runtime.Abstractions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Configuration;
namespace Jibo.Cloud.Infrastructure.DependencyInjection;
public static class ServiceCollectionExtensions
{
public static IServiceCollection AddOpenJiboCloud(this IServiceCollection services)
public static IServiceCollection AddOpenJiboCloud(this IServiceCollection services, IConfiguration? configuration = null)
{
if (configuration is not null)
{
services.Configure<WebSocketTelemetryOptions>(configuration.GetSection("OpenJibo:Telemetry"));
}
services.AddSingleton<ICloudStateStore, InMemoryCloudStateStore>();
services.AddSingleton<IConversationBroker, DemoConversationBroker>();
services.AddSingleton<ISttStrategy, SyntheticBufferedAudioSttStrategy>();
services.AddSingleton<ISttStrategySelector, DefaultSttStrategySelector>();
services.AddSingleton<IWebSocketTelemetrySink, FileWebSocketTelemetrySink>();
services.AddSingleton<ProtocolToTurnContextMapper>();
services.AddSingleton<ResponsePlanToSocketMessagesMapper>();
services.AddSingleton<WebSocketTurnFinalizationService>();

256
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Infrastructure/Telemetry/FileWebSocketTelemetrySink.cs Normal file
View File

@@ -0,0 +1,256 @@
using System.Collections.Concurrent;
using System.Text.Json;
using Jibo.Cloud.Application.Abstractions;
using Jibo.Cloud.Domain.Models;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace Jibo.Cloud.Infrastructure.Telemetry;
public sealed class FileWebSocketTelemetrySink(
ILogger<FileWebSocketTelemetrySink> logger,
IOptions<WebSocketTelemetryOptions> options) : IWebSocketTelemetrySink
{
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web)
{
WriteIndented = true
};
private readonly ConcurrentDictionary<string, CapturedWebSocketFixtureBuilder> _fixtures = new(StringComparer.OrdinalIgnoreCase);
private readonly SemaphoreSlim _writeLock = new(1, 1);
public async Task RecordConnectionOpenedAsync(WebSocketMessageEnvelope envelope, CloudSession session, CancellationToken cancellationToken = default)
{
if (!options.Value.Enabled)
{
return;
}
_fixtures[session.SessionId] = new CapturedWebSocketFixtureBuilder
{
Session = new CapturedWebSocketFixtureSession
{
HostName = envelope.HostName,
Path = envelope.Path,
Kind = envelope.Kind,
Token = envelope.Token
}
};
await WriteRecordAsync(BuildRecord("connection_opened", envelope, session, null, "internal", null, null), cancellationToken);
}
public Task RecordInboundAsync(WebSocketMessageEnvelope envelope, CloudSession session, string? messageType, CancellationToken cancellationToken = default)
{
if (!options.Value.Enabled)
{
return Task.CompletedTask;
}
return WriteRecordAsync(BuildRecord("message_in", envelope, session, messageType, "in", null, null), cancellationToken);
}
public Task RecordTurnEventAsync(WebSocketMessageEnvelope envelope, CloudSession session, string eventType, IReadOnlyDictionary<string, object?> details, CancellationToken cancellationToken = default)
{
if (!options.Value.Enabled)
{
return Task.CompletedTask;
}
return WriteRecordAsync(BuildRecord(eventType, envelope, session, null, "internal", null, details), cancellationToken);
}
public async Task RecordOutboundAsync(WebSocketMessageEnvelope envelope, CloudSession session, IReadOnlyList<WebSocketReply> replies, CancellationToken cancellationToken = default)
{
if (!options.Value.Enabled)
{
return;
}
var replyTypes = replies
.Select(reply => ReadReplyType(reply.Text))
.Where(type => !string.IsNullOrWhiteSpace(type))
.Select(type => type!)
.ToArray();
await WriteRecordAsync(BuildRecord("message_out", envelope, session, null, "out", replyTypes, null), cancellationToken);
if (_fixtures.TryGetValue(session.SessionId, out var fixture))
{
fixture.Steps.Add(new CapturedWebSocketFixtureStep
{
Text = ParseJsonElement(envelope.Text),
Binary = envelope.Binary?.Select(value => (int)value).ToArray(),
ExpectedReplyTypes = replyTypes
});
}
}
public async Task RecordConnectionClosedAsync(WebSocketMessageEnvelope envelope, CloudSession session, string reason, CancellationToken cancellationToken = default)
{
if (!options.Value.Enabled)
{
return;
}
await WriteRecordAsync(BuildRecord(
"connection_closed",
envelope,
session,
null,
"internal",
null,
new Dictionary<string, object?> { ["reason"] = reason }), cancellationToken);
if (!options.Value.ExportFixtures || !_fixtures.TryRemove(session.SessionId, out var fixture) || fixture.Steps.Count == 0)
{
return;
}
var fixtureName = BuildFixtureName(session, fixture);
var capturedFixture = new CapturedWebSocketFixture
{
Name = fixtureName,
Session = fixture.Session,
Steps = [.. fixture.Steps]
};
var fixtureDirectory = Path.Combine(GetBaseDirectory(), "fixtures");
Directory.CreateDirectory(fixtureDirectory);
var fixturePath = Path.Combine(fixtureDirectory, $"{fixtureName}.flow.json");
await _writeLock.WaitAsync(cancellationToken);
try
{
await File.WriteAllTextAsync(fixturePath, JsonSerializer.Serialize(capturedFixture, JsonOptions), cancellationToken);
}
finally
{
_writeLock.Release();
}
logger.LogInformation("Exported websocket fixture {FixturePath}", fixturePath);
}
private async Task WriteRecordAsync(WebSocketTelemetryRecord record, CancellationToken cancellationToken)
{
var directory = GetBaseDirectory();
Directory.CreateDirectory(directory);
var filePath = Path.Combine(directory, $"{DateTimeOffset.UtcNow:yyyyMMdd}.events.ndjson");
var line = JsonSerializer.Serialize(record) + Environment.NewLine;
await _writeLock.WaitAsync(cancellationToken);
try
{
await File.AppendAllTextAsync(filePath, line, cancellationToken);
}
finally
{
_writeLock.Release();
}
logger.LogInformation(
"WebSocket telemetry {EventType} session={SessionId} transId={TransId} bufferedBytes={BufferedBytes} replyTypes={ReplyTypes}",
record.EventType,
record.SessionId,
record.TransId,
record.BufferedAudioBytes,
string.Join(",", record.ReplyTypes));
}
private static WebSocketTelemetryRecord BuildRecord(
string eventType,
WebSocketMessageEnvelope envelope,
CloudSession session,
string? messageType,
string direction,
IReadOnlyList<string>? replyTypes,
IReadOnlyDictionary<string, object?>? details) => new()
{
EventType = eventType,
SessionId = session.SessionId,
ConnectionId = envelope.ConnectionId,
Token = envelope.Token,
HostName = envelope.HostName,
Path = envelope.Path,
Kind = envelope.Kind,
TransId = session.TurnState.TransId ?? session.LastTransId,
MessageType = messageType,
Direction = direction,
Text = envelope.Text,
BinaryLength = envelope.Binary?.Length,
ReplyTypes = replyTypes ?? [],
BufferedAudioBytes = session.TurnState.BufferedAudioBytes,
BufferedAudioChunks = session.TurnState.BufferedAudioChunkCount,
FinalizeAttempts = session.TurnState.FinalizeAttemptCount,
AwaitingTurnCompletion = session.TurnState.AwaitingTurnCompletion,
Details = details ?? new Dictionary<string, object?>()
};
private static string? ReadReplyType(string? text)
{
if (string.IsNullOrWhiteSpace(text))
{
return null;
}
try
{
using var document = JsonDocument.Parse(text);
return document.RootElement.TryGetProperty("type", out var type) && type.ValueKind == JsonValueKind.String
? type.GetString()
: null;
}
catch
{
return null;
}
}
private static JsonElement? ParseJsonElement(string? text)
{
if (string.IsNullOrWhiteSpace(text))
{
return null;
}
try
{
using var document = JsonDocument.Parse(text);
return document.RootElement.Clone();
}
catch
{
return null;
}
}
private string GetBaseDirectory()
{
return Path.GetFullPath(options.Value.DirectoryPath, AppContext.BaseDirectory);
}
private static string BuildFixtureName(CloudSession session, CapturedWebSocketFixtureBuilder fixture)
{
var host = SanitizeName(fixture.Session.HostName);
var kind = SanitizeName(fixture.Session.Kind);
var transId = SanitizeName(session.TurnState.TransId ?? session.LastTransId ?? session.SessionId);
return $"{host}-{kind}-{transId}";
}
private static string SanitizeName(string value)
{
var chars = value
.ToLowerInvariant()
.Select(character => char.IsLetterOrDigit(character) ? character : '-')
.ToArray();
return string.Join(string.Empty, new string(chars).Split('-', StringSplitOptions.RemoveEmptyEntries));
}
private sealed class CapturedWebSocketFixtureBuilder
{
public CapturedWebSocketFixtureSession Session { get; init; } = new();
public List<CapturedWebSocketFixtureStep> Steps { get; } = [];
}
}

8
OpenJibo/src/Jibo.Cloud/dotnet/src/Jibo.Cloud.Infrastructure/Telemetry/WebSocketTelemetryOptions.cs Normal file
View File

@@ -0,0 +1,8 @@
namespace Jibo.Cloud.Infrastructure.Telemetry;
public sealed class WebSocketTelemetryOptions
{
public bool Enabled { get; set; } = true;
public bool ExportFixtures { get; set; } = true;
public string DirectoryPath { get; set; } = "captures/websocket";
}