From 7dada5fdb9dd7962f69c82e1cc95368171927051 Mon Sep 17 00:00:00 2001 From: Assistant builder Date: Thu, 11 Jun 2026 21:38:31 +0000 Subject: [PATCH] RealtimeSession: execute server-side tools at response.done --- backend.tests/RealtimeSessionToolsTests.cs | 67 ++++++++++++++++++++++ backend/Realtime/RealtimeSession.cs | 64 +++++++++++++++++++++ 2 files changed, 131 insertions(+) create mode 100644 backend.tests/RealtimeSessionToolsTests.cs diff --git a/backend.tests/RealtimeSessionToolsTests.cs b/backend.tests/RealtimeSessionToolsTests.cs new file mode 100644 index 0000000..ea10c0a --- /dev/null +++ b/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 +{ + 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(); + var log = scope.ServiceProvider.GetRequiredService(); + 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 { "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; + } +} diff --git a/backend/Realtime/RealtimeSession.cs b/backend/Realtime/RealtimeSession.cs index c49c2a9..3f74d9a 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 readonly List<(string CallId, string Name, string ArgsJson)> _pendingCalls = new(); private readonly System.Threading.Channels.Channel _uplink = System.Threading.Channels.Channel.CreateBounded( @@ -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"];