From dd99216913653bc8aa42256e5c516e8117fa2e35 Mon Sep 17 00:00:00 2001 From: Meleki Date: Mon, 9 Mar 2026 20:13:22 +0100 Subject: [PATCH] =?UTF-8?q?fix:=20SSE-Stream-Stabilit=C3=A4t=20und=20Histo?= =?UTF-8?q?ry-Trimming=20tool=5Fuse/tool=5Fresult-Schutz?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Kestrel MinResponseDataRate per-Request deaktivieren für SSE-Streams (Default 240 bytes/s + 5s Grace → client_disconnected bei langen Tools) - SSE-Heartbeat alle 15s während Stream (schützt gegen Browser/Netzwerk-Idle) - History-Trimming: RemoveRange schneidet nicht mehr in tool_use/tool_result-Paare - Reminder-Insertion: kein Einfügen zwischen tool_use und tool_result --- Endpoints/ChatEndpoint.cs | 26 ++++++++++++++ Program.cs | 1 + Services/ChatService.cs | 76 +++++++++++++++++++++++++++++++++++++-- 3 files changed, 101 insertions(+), 2 deletions(-) diff --git a/Endpoints/ChatEndpoint.cs b/Endpoints/ChatEndpoint.cs index 4c9df0a..6806b14 100644 --- a/Endpoints/ChatEndpoint.cs +++ b/Endpoints/ChatEndpoint.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using System.Linq; using System.Text.Json; @@ -124,6 +125,13 @@ public static class ChatEndpoint var bufferingFeature = res.HttpContext.Features.Get(); bufferingFeature?.DisableBuffering(); + // Kestrel MinResponseDataRate deaktivieren für diesen SSE-Stream. + // Ohne dies killt Kestrel die Verbindung wenn bei langen Tool-Ausführungen + // keine Daten fließen (Default: 240 bytes/s, 5s Grace → "client_disconnected"). + var minRateFeature = res.HttpContext.Features.Get(); + if (minRateFeature != null) + minRateFeature.MinDataRate = null; + // ── Preparer-Agent Phase (vor dem Worker) ── if (ctx.Preparer != null && !isChoiceResponse) { @@ -171,6 +179,23 @@ public static class ChatEndpoint var stopCts = streamingTracker.Start(sessionId, ctx.ChainId); using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, stopCts.Token); + // SSE-Heartbeat: Alle 15s ein Kommentar senden, damit die Verbindung + // bei langen Tool-Ausführungen nicht von Browser/Netzwerk getrennt wird. + var heartbeatTask = Task.Run(async () => + { + try + { + while (!linkedCts.Token.IsCancellationRequested) + { + await Task.Delay(15000, linkedCts.Token); + await res.WriteAsync(": heartbeat\n\n", linkedCts.Token); + await res.Body.FlushAsync(linkedCts.Token); + } + } + catch (OperationCanceledException) { } + catch { } + }, linkedCts.Token); + try { await anthropic.StreamChat(res, ctx.ModelId, ctx.SystemPrompt, messages, ctx.ChainId, ctx.Workspace, sessionId, agentId: ctx.AgentId, useMemory: ctx.UseMemory, webSearchVersion: ctx.WebSearchVersion, maxIterations: ctx.MaxIterations, toolChoice: effectiveToolChoice, review: ctx.Review, finalizer: ctx.Finalizer, allowUserChoice: ctx.Agent?.AllowUserChoice ?? true, planMode: ctx.PlanMode, cancellationToken: linkedCts.Token); @@ -178,6 +203,7 @@ public static class ChatEndpoint finally { streamingTracker.Stop(sessionId); + // Heartbeat-Task beenden (linkedCts wird disposed → Token cancelled) } } diff --git a/Program.cs b/Program.cs index d9550c2..a84dd5b 100644 --- a/Program.cs +++ b/Program.cs @@ -9,6 +9,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; var builder = WebApplication.CreateBuilder(args); + builder.Services.AddHttpClient(); var dataDir = Path.Combine(AppContext.BaseDirectory, "data"); diff --git a/Services/ChatService.cs b/Services/ChatService.cs index 66dcbfc..e741ed3 100644 --- a/Services/ChatService.cs +++ b/Services/ChatService.cs @@ -445,8 +445,67 @@ Für welche Plattform soll die App entwickelt werden? Web, Mobile oder Desktop? } // History auf maximal N Messages begrenzen (älteste entfernen) + // WICHTIG: Nicht mitten in einem tool_use/tool_result-Paar abschneiden! if (history.Count > _maxHistory) - history.RemoveRange(0, history.Count - _maxHistory); + { + int removeCount = history.Count - _maxHistory; + // Nach vorne korrigieren bis wir an einer sicheren Stelle sind: + // Nicht abschneiden wenn die erste verbleibende Message ein tool_result ist + // oder die letzte entfernte Message ein assistant mit tool_use ist + while (removeCount < history.Count - 1) + { + var firstRemaining = history[removeCount]; + bool isToolResult = false; + if (firstRemaining is Dictionary frDict && + frDict.TryGetValue("role", out var frRole) && frRole.ToString() == "user" && + frDict.TryGetValue("content", out var frContent) && frContent is List frList) + { + isToolResult = frList.Any(item => + { + var json = JsonSerializer.Serialize(item); + return json.Contains("\"tool_result\""); + }); + } + + if (isToolResult) + { + // Die erste verbleibende Message ist ein tool_result → muss den vorherigen + // assistant[tool_use] auch behalten → ein Element weniger entfernen + removeCount--; + // Jetzt prüfen ob der neue Start auch sicher ist (könnte auch tool_result sein) + continue; + } + + // Prüfen ob die letzte entfernte Message ein assistant mit tool_use war + bool lastRemovedHasToolUse = false; + if (removeCount > 0) + { + var lastRemoved = history[removeCount - 1]; + if (lastRemoved is Dictionary lrDict && + lrDict.TryGetValue("role", out var lrRole) && lrRole.ToString() == "assistant" && + lrDict.TryGetValue("content", out var lrContent) && lrContent is List lrList) + { + lastRemovedHasToolUse = lrList.Any(item => + { + var json = JsonSerializer.Serialize(item); + return json.Contains("\"tool_use\""); + }); + } + } + + if (lastRemovedHasToolUse) + { + // Letztes entferntes Element enthält tool_use → auch das tool_result entfernen + removeCount++; + continue; + } + + break; + } + + if (removeCount > 0 && removeCount < history.Count) + history.RemoveRange(0, removeCount); + } // Reminder einfügen: alle ~30 user/assistant-Paare ein Reminder-Paar einfügen var reminder = SystemPromptBuilder.BuildReminder(promptBlocks); @@ -473,7 +532,20 @@ Für welche Plattform soll die App entwickelt werden? Web, Mobile oder Desktop? { // Position nach dem aktuellen Element – aber nur einfügen wenn es eine // assistant-Message ist (damit user/assistant-Alternierung nicht bricht) - if (history[i] is Dictionary d && + // UND die nächste Message kein tool_result ist (sonst bricht tool_use/tool_result-Paar) + bool nextIsToolResult = false; + if (i + 1 < history.Count && history[i + 1] is Dictionary nextDict && + nextDict.TryGetValue("role", out var nextRole) && nextRole.ToString() == "user" && + nextDict.TryGetValue("content", out var nextContent) && nextContent is List nextList) + { + nextIsToolResult = nextList.Any(item => + { + var json = JsonSerializer.Serialize(item); + return json.Contains("\"tool_result\""); + }); + } + + if (!nextIsToolResult && history[i] is Dictionary d && d.TryGetValue("role", out var r) && r.ToString() == "assistant") { insertPositions.Add(i + 1);