RealtimeSession: execute server-side tools at response.done

backend.tests/RealtimeSessionToolsTests.cs

@@ -0,0 +1,67 @@
+using System.Text.Json.Nodes;
+using backend.Conversations;
+using backend.Data;
+using backend.Realtime;
+using backend.Tools;
+using FluentAssertions;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+
+namespace backend.tests;
+
+public class RealtimeSessionToolsTests(ApiFactory factory) : IClassFixture<ApiFactory>
+{
+    private readonly ApiFactory _factory = factory;
+
+    [Fact]
+    public async Task Get_current_time_call_returns_function_call_output_and_persists_tool_turn()
+    {
+        using var scope = _factory.Services.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
+        var log = scope.ServiceProvider.GetRequiredService<ConversationLog>();
+        var reg = new ToolRegistry(new ITool[] { new GetCurrentTimeTool() });
+        var upstream = new ScriptedRealtimeUpstream();
+        var sink = new RecordingSink();
+        var cfg = new RealtimeSettings("m", "v", "p", 30, new HashSet<string> { "get_current_time" });
+        var session = new RealtimeSession(Guid.NewGuid(), upstream, sink, log, reg, cfg,
+            new FakeDeviceChannel(), NullLogger.Instance);
+
+        upstream.Push(new JsonObject { ["type"] = "session.updated" });
+        var cts = new CancellationTokenSource();
+        var run = session.RunAsync(cts.Token);
+        await sink.WaitForAsync("session_started", TimeSpan.FromSeconds(2));
+
+        upstream.Push(new JsonObject
+        {
+            ["type"] = "response.function_call_arguments.done",
+            ["call_id"] = "call_abc",
+            ["name"] = "get_current_time",
+            ["arguments"] = "{}",
+        });
+        upstream.Push(new JsonObject { ["type"] = "response.done" });
+
+        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2);
+        while (DateTime.UtcNow < deadline && !upstream.Sent.Any(e =>
+            (string?)e["type"] == "conversation.item.create"))
+        {
+            await Task.Delay(20);
+        }
+
+        var item = upstream.Sent.First(e => (string?)e["type"] == "conversation.item.create");
+        ((string?)item["item"]!["type"]).Should().Be("function_call_output");
+        ((string?)item["item"]!["call_id"]).Should().Be("call_abc");
+        ((string?)item["item"]!["output"]).Should().Contain("\"now\"");
+
+        upstream.Sent.Should().Contain(e => (string?)e["type"] == "response.create");
+
+        var convoId = session.ConversationId!.Value;
+        var toolTurn = await db.Turns.FirstAsync(t =>
+            t.ConversationId == convoId && t.Role == TurnRole.Tool);
+        toolTurn.ToolName.Should().Be("get_current_time");
+        toolTurn.ToolResultJson.Should().Contain("\"now\"");
+
+        cts.Cancel();
+        await run;
+    }
+}

backend/Realtime/RealtimeSession.cs

@@ -25,6 +25,7 @@ public class RealtimeSession
     private Conversation? _conversation;
     private EndReason _endReason = EndReason.Unknown;
     private bool _endRequested;
+    private readonly List<(string CallId, string Name, string ArgsJson)> _pendingCalls = new();
 
     private readonly System.Threading.Channels.Channel<byte[]> _uplink =
         System.Threading.Channels.Channel.CreateBounded<byte[]>(
@@ -135,14 +136,77 @@ public class RealtimeSession
                         await _log.AppendAssistantAsync(_conversation.Id, text!, ct);
                     break;
                 }
+                case "response.function_call_arguments.done":
+                {
+                    var callId = (string?)evt["call_id"] ?? "";
+                    var name = (string?)evt["name"] ?? "";
+                    var args = (string?)evt["arguments"] ?? "{}";
+                    _pendingCalls.Add((callId, name, args));
+                    break;
+                }
                 case "response.done":
+                {
                     await _output.WriteEnvelopeAsync(
                         new JsonObject { ["type"] = "assistant_done" }, ct);
+
+                    if (_pendingCalls.Count == 0) break;
+                    var calls = _pendingCalls.ToList();
+                    _pendingCalls.Clear();
+
+                    if (calls.Any(c => c.Name == "end_session"))
+                    {
+                        _endReason = EndReason.Tool;
+                        _endRequested = true;
+                        break;
+                    }
+
+                    foreach (var c in calls)
+                    {
+                        await ExecuteToolAsync(c.CallId, c.Name, c.ArgsJson, ct);
+                    }
                     break;
+                }
             }
         }
     }
 
+    private async Task ExecuteToolAsync(string callId, string name, string argsJson, CancellationToken ct)
+    {
+        var tool = _registry.Get(name);
+        string outputJson;
+        if (tool is null)
+        {
+            outputJson = $$"""{"ok":false,"error":"unknown tool: {{name}}"}""";
+        }
+        else
+        {
+            try
+            {
+                using var argsDoc = System.Text.Json.JsonDocument.Parse(argsJson);
+                var ctx = new DeviceContext(_deviceId, _conversation!.Id, _channel);
+                var res = await tool.ExecuteAsync(
+                    new ToolInvocation(callId, argsDoc.RootElement.Clone()), ctx, ct);
+                outputJson = res.Output.GetRawText();
+            }
+            catch (Exception ex)
+            {
+                outputJson = System.Text.Json.JsonSerializer.Serialize(new
+                {
+                    ok = false,
+                    error = ex.Message,
+                });
+            }
+        }
+
+        if (_conversation is not null)
+        {
+            await _log.AppendToolAsync(_conversation.Id, name, argsJson, outputJson, ct);
+        }
+
+        await _upstream.SendJsonAsync(RealtimeEvents.FunctionCallOutput(callId, outputJson), ct);
+        await _upstream.SendJsonAsync(RealtimeEvents.ResponseCreate(), ct);
+    }
+
     private async Task ForwardAudioDeltaAsync(JsonObject evt, CancellationToken ct)
     {
         var b64 = (string?)evt["delta"];