RealtimeSession: idle-timeout watchdog (resets on speech_started)

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<ApiFactory>
+{
+    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<ConversationLog>();
+        var reg = new ToolRegistry(Array.Empty<ITool>());
+        var upstream = new ScriptedRealtimeUpstream();
+        var sink = new RecordingSink();
+        var cfg = new RealtimeSettings("m", "v", "p", IdleTimeoutSeconds: 1, new HashSet<string>());
+        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<ConversationLog>();
+        var reg = new ToolRegistry(Array.Empty<ITool>());
+        var upstream = new ScriptedRealtimeUpstream();
+        var sink = new RecordingSink();
+        var cfg = new RealtimeSettings("m", "v", "p", IdleTimeoutSeconds: 2, new HashSet<string>());
+        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");
+    }
+}

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<byte[]> _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);
+            var evt = await ReceiveOrIdleAsync(ct);
-            if (evt is null) { _endReason = EndReason.Error; return; }
+            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<JsonObject?> 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