From 4ae894304fe7995b5956a0198b224818617e673c Mon Sep 17 00:00:00 2001 From: Meleki Date: Thu, 5 Mar 2026 15:28:48 +0100 Subject: [PATCH] =?UTF-8?q?fix:=20CancellationToken=20durchg=C3=A4ngig=20i?= =?UTF-8?q?n=20AnthropicService=20propagieren?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ThrowIfCancellationRequested() am Anfang jeder while-Loop-Iteration - WriteSseEvent() erhält CancellationToken-Parameter, prüft vor Schreiben - WriteAsync/FlushAsync mit CancellationToken aufgerufen - Cancellation-Check vor und nach Tool-Execution - OperationCanceledException in ExecuteTool wird re-thrown statt geschluckt - Behebt: Stop-Button brach nur Frontend-Stream ab, Backend lief weiter --- Services/AnthropicService.cs | 29 ++++++++++++++++++----------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index e78e39d..cda6604 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -65,6 +65,7 @@ public class AnthropicService { while (iterations < maxIterations) { + cancellationToken.ThrowIfCancellationRequested(); iterations++; var payload = new Dictionary @@ -523,7 +524,7 @@ public class AnthropicService if (cbType == "text") { currentTextBlock.Clear(); - await WriteSseEvent(response, "text_start", new { index = idx }); + await WriteSseEvent(response, "text_start", new { index = idx }, cancellationToken); } else if (cbType == "tool_use") { @@ -535,7 +536,7 @@ public class AnthropicService currentToolId = cb.GetProperty("id").GetString() ?? ""; currentToolName = cb.GetProperty("name").GetString() ?? ""; currentToolInputJson.Clear(); - await WriteSseEvent(response, "tool_start", new { name = currentToolName, id = currentToolId }); + await WriteSseEvent(response, "tool_start", new { name = currentToolName, id = currentToolId }, cancellationToken); } else if (cbType == "server_tool_use") { @@ -547,7 +548,7 @@ public class AnthropicService currentToolId = cb.GetProperty("id").GetString() ?? ""; currentToolName = cb.GetProperty("name").GetString() ?? ""; currentToolInputJson.Clear(); - await WriteSseEvent(response, "web_search_start", new { name = currentToolName, id = currentToolId }); + await WriteSseEvent(response, "web_search_start", new { name = currentToolName, id = currentToolId }, cancellationToken); } else if (cbType == "web_search_tool_result") { @@ -567,7 +568,7 @@ public class AnthropicService { var text = delta.GetProperty("text").GetString() ?? ""; currentTextBlock.Append(text); - await WriteSseEvent(response, "text_delta", new { text }); + await WriteSseEvent(response, "text_delta", new { text }, cancellationToken); } else if (deltaType == "input_json_delta") { @@ -588,7 +589,7 @@ public class AnthropicService if (text.Length > 0) { assistantContent.Add(new { type = "text", text }); - await WriteSseEvent(response, "text_end", new { }); + await WriteSseEvent(response, "text_end", new { }, cancellationToken); } currentTextBlock.Clear(); } @@ -606,11 +607,16 @@ public class AnthropicService input = inputEl }); + // Cancellation prüfen bevor Tool ausgeführt wird + cancellationToken.ThrowIfCancellationRequested(); + string toolResult; bool isError = false; try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId); } + catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } + cancellationToken.ThrowIfCancellationRequested(); executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); await WriteSseEvent(response, "tool_end", new @@ -619,7 +625,7 @@ public class AnthropicService input = currentToolInputJson.ToString(), result = toolResult.Length > 500 ? toolResult[..500] + "..." : toolResult, isError - }); + }, cancellationToken); } else if (bt == "server_tool_use") { @@ -640,7 +646,7 @@ public class AnthropicService { // Web-Search-Ergebnis von Anthropic – als Raw-JSON in assistantContent aufnehmen assistantContent.Add(webSearchResultBlocks[idx]); - await WriteSseEvent(response, "web_search_end", new { id = currentToolId }); + await WriteSseEvent(response, "web_search_end", new { id = currentToolId }, cancellationToken); } } break; @@ -723,11 +729,12 @@ public class AnthropicService return client; } - public static async Task WriteSseEvent(HttpResponse response, string eventType, object data) + public static async Task WriteSseEvent(HttpResponse response, string eventType, object data, CancellationToken ct = default) { + ct.ThrowIfCancellationRequested(); var json = JsonSerializer.Serialize(data); - await response.WriteAsync($"event: {eventType}\ndata: {json}\n\n"); - await response.Body.FlushAsync(); + await response.WriteAsync($"event: {eventType}\ndata: {json}\n\n", ct); + await response.Body.FlushAsync(ct); } private static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct) @@ -735,7 +742,7 @@ public class AnthropicService try { if (!ct.IsCancellationRequested) - await WriteSseEvent(response, eventType, data); + await WriteSseEvent(response, eventType, data, ct); } catch {