From 8f53f9daa9f4d9a28dc953f5b461a63eaa70e985 Mon Sep 17 00:00:00 2001 From: Assistant builder Date: Thu, 11 Jun 2026 21:46:16 +0000 Subject: [PATCH] RealtimeSession: idle-timeout watchdog (resets on speech_started) --- .../RealtimeSessionIdleTimeoutTests.cs | 63 +++++++++++++++++++ backend/Realtime/RealtimeSession.cs | 44 ++++++++++++- 2 files changed, 105 insertions(+), 2 deletions(-) create mode 100644 backend.tests/RealtimeSessionIdleTimeoutTests.cs diff --git a/backend.tests/RealtimeSessionIdleTimeoutTests.cs b/backend.tests/RealtimeSessionIdleTimeoutTests.cs new file mode 100644 index 0000000..8692a9b --- /dev/null +++ b/backend.tests/RealtimeSessionIdleTimeoutTests.cs @@ -0,0 +1,63 @@ +using System.Text.Json.Nodes; +using backend.Conversations; +using backend.Realtime; +using backend.Tools; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace backend.tests; + +public class RealtimeSessionIdleTimeoutTests(ApiFactory factory) : IClassFixture +{ + private readonly ApiFactory _factory = factory; + + [Fact] + public async Task Session_ends_with_reason_idle_when_no_speech_for_timeout() + { + using var scope = _factory.Services.CreateScope(); + var log = scope.ServiceProvider.GetRequiredService(); + var reg = new ToolRegistry(Array.Empty()); + var upstream = new ScriptedRealtimeUpstream(); + var sink = new RecordingSink(); + var cfg = new RealtimeSettings("m", "v", "p", IdleTimeoutSeconds: 1, new HashSet()); + var session = new RealtimeSession(Guid.NewGuid(), upstream, sink, log, reg, cfg, + new FakeDeviceChannel(), NullLogger.Instance); + + upstream.Push(new JsonObject { ["type"] = "session.updated" }); + var run = session.RunAsync(default); + await sink.WaitForAsync("session_started", TimeSpan.FromSeconds(2)); + + await run.WaitAsync(TimeSpan.FromSeconds(4)); + + var ended = sink.Envelopes.Last(e => (string?)e["type"] == "session_ended"); + ((string?)ended["reason"]).Should().Be("idle"); + } + + [Fact] + public async Task Speech_started_resets_the_idle_timer() + { + using var scope = _factory.Services.CreateScope(); + var log = scope.ServiceProvider.GetRequiredService(); + var reg = new ToolRegistry(Array.Empty()); + var upstream = new ScriptedRealtimeUpstream(); + var sink = new RecordingSink(); + var cfg = new RealtimeSettings("m", "v", "p", IdleTimeoutSeconds: 2, new HashSet()); + var session = new RealtimeSession(Guid.NewGuid(), upstream, sink, log, reg, cfg, + new FakeDeviceChannel(), NullLogger.Instance); + + upstream.Push(new JsonObject { ["type"] = "session.updated" }); + var run = session.RunAsync(default); + await sink.WaitForAsync("session_started", TimeSpan.FromSeconds(2)); + + await Task.Delay(1000); + upstream.Push(new JsonObject { ["type"] = "input_audio_buffer.speech_started" }); + await Task.Delay(1500); + + sink.Envelopes.Any(e => (string?)e["type"] == "session_ended").Should().BeFalse(); + + await run.WaitAsync(TimeSpan.FromSeconds(4)); + var ended = sink.Envelopes.Last(e => (string?)e["type"] == "session_ended"); + ((string?)ended["reason"]).Should().Be("idle"); + } +} diff --git a/backend/Realtime/RealtimeSession.cs b/backend/Realtime/RealtimeSession.cs index 3f74d9a..a1fca32 100644 --- a/backend/Realtime/RealtimeSession.cs +++ b/backend/Realtime/RealtimeSession.cs @@ -25,6 +25,7 @@ public class RealtimeSession private Conversation? _conversation; private EndReason _endReason = EndReason.Unknown; private bool _endRequested; + private DateTime _lastSpeechAt = DateTime.UtcNow; private readonly List<(string CallId, string Name, string ArgsJson)> _pendingCalls = new(); private readonly System.Threading.Channels.Channel _uplink = @@ -79,11 +80,14 @@ public class RealtimeSession ["type"] = "session_started", ["conversation_id"] = _conversation.Id.ToString(), }, ct); + _lastSpeechAt = DateTime.UtcNow; var uplinkTask = Task.Run(() => PumpUplinkAsync(ct), ct); + var idleTask = Task.Run(() => WatchdogAsync(ct), ct); await PumpAsync(ct); _uplink.Writer.TryComplete(); await uplinkTask; + await idleTask; } catch (OperationCanceledException) { @@ -114,11 +118,16 @@ public class RealtimeSession { while (!ct.IsCancellationRequested && !_endRequested) { - var evt = await _upstream.ReceiveJsonAsync(ct); - if (evt is null) { _endReason = EndReason.Error; return; } + var evt = await ReceiveOrIdleAsync(ct); + if (evt is null && _endRequested) return; + if (evt is null && ct.IsCancellationRequested) return; + if (evt is null) continue; var type = (string?)evt["type"]; switch (type) { + case "input_audio_buffer.speech_started": + _lastSpeechAt = DateTime.UtcNow; + break; case "response.audio.delta": await ForwardAudioDeltaAsync(evt, ct); break; @@ -215,6 +224,37 @@ public class RealtimeSession await _output.WriteBinaryAsync(bytes, ct); } + private async Task WatchdogAsync(CancellationToken ct) + { + try + { + var timeout = TimeSpan.FromSeconds(_cfg.IdleTimeoutSeconds); + while (!ct.IsCancellationRequested && !_endRequested) + { + await Task.Delay(250, ct); + if (DateTime.UtcNow - _lastSpeechAt > timeout) + { + _endReason = EndReason.Idle; + _endRequested = true; + return; + } + } + } + catch (OperationCanceledException) { } + } + + private async Task ReceiveOrIdleAsync(CancellationToken ct) + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + var receive = _upstream.ReceiveJsonAsync(timeout.Token); + var delay = Task.Delay(250, timeout.Token); + var done = await Task.WhenAny(receive, delay); + if (done == receive) return await receive; + timeout.Cancel(); + try { await receive; } catch { /* ignore */ } + return null; + } + private async Task PumpUplinkAsync(CancellationToken ct) { try