diff --git a/Endpoints/ChatEndpoint.cs b/Endpoints/ChatEndpoint.cs index 6806b14..8025267 100644 --- a/Endpoints/ChatEndpoint.cs +++ b/Endpoints/ChatEndpoint.cs @@ -175,9 +175,15 @@ public static class ChatEndpoint // Bei Choice-Response: toolChoice auf "auto" setzen, damit Claude frei antworten kann var effectiveToolChoice = isChoiceResponse ? "auto" : ctx.ToolChoice; - // CancellationToken: Verknüpft Stop-Button (über StreamingTracker) + Client-Disconnect + // Zwei separate CancellationTokens: + // - workCt: Nur Stop-Button → für Anthropic API-Calls, Tool-Ausführungen, DB-Writes + // Agent-Arbeit läuft bei Client-Disconnect weiter (Client pollt ja) + // - sseCt: Stop-Button + Client-Disconnect → für SSE-Writes an den Browser + // Bei Client-Disconnect werden SSE-Writes zu stillen No-Ops var stopCts = streamingTracker.Start(sessionId, ctx.ChainId); - using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, stopCts.Token); + var workCt = stopCts.Token; + using var sseLinkCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, stopCts.Token); + var sseCt = sseLinkCts.Token; // SSE-Heartbeat: Alle 15s ein Kommentar senden, damit die Verbindung // bei langen Tool-Ausführungen nicht von Browser/Netzwerk getrennt wird. @@ -185,25 +191,25 @@ public static class ChatEndpoint { try { - while (!linkedCts.Token.IsCancellationRequested) + while (!sseCt.IsCancellationRequested) { - await Task.Delay(15000, linkedCts.Token); - await res.WriteAsync(": heartbeat\n\n", linkedCts.Token); - await res.Body.FlushAsync(linkedCts.Token); + await Task.Delay(15000, sseCt); + await res.WriteAsync(": heartbeat\n\n", sseCt); + await res.Body.FlushAsync(sseCt); } } catch (OperationCanceledException) { } catch { } - }, linkedCts.Token); + }, sseCt); 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); + 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, workCt: workCt, sseCt: sseCt); } finally { streamingTracker.Stop(sessionId); - // Heartbeat-Task beenden (linkedCts wird disposed → Token cancelled) + // Heartbeat-Task beenden (sseLinkCts wird disposed → Token cancelled) } } diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index 06794d7..9766e84 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -51,7 +51,8 @@ public class AnthropicService SubAgentContext finalizer = null, bool allowUserChoice = true, bool planMode = false, - CancellationToken cancellationToken = default) + CancellationToken workCt = default, + CancellationToken sseCt = default) { var sw = System.Diagnostics.Stopwatch.StartNew(); var usage = new TokenUsage(); @@ -70,7 +71,7 @@ public class AnthropicService { while (iterations < maxIterations) { - cancellationToken.ThrowIfCancellationRequested(); + workCt.ThrowIfCancellationRequested(); iterations++; var payload = new Dictionary @@ -95,12 +96,12 @@ public class AnthropicService using var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }, HttpCompletionOption.ResponseHeadersRead, - cancellationToken + workCt ); if (!httpResp.IsSuccessStatusCode) { - var errBody = await httpResp.Content.ReadAsStringAsync(cancellationToken); + var errBody = await httpResp.Content.ReadAsStringAsync(workCt); var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}"; await _db.SaveMessage(new MessageRecord { @@ -108,13 +109,13 @@ public class AnthropicService Role = "system", Type = "error", Content = errMsg, IsError = true }); - await TryWriteSse(response, "error", new { error = errMsg }, cancellationToken); + await TryWriteSse(response, "error", new { error = errMsg }, sseCt); await FinalizeChain(chain, usage, iterations, "error", sw); - await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken); + await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); return; } - var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, cancellationToken); + var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, workCt, sseCt); lastStopReason = stopReason; messages.Add(new { role = "assistant", content = assistantContent }); @@ -198,7 +199,7 @@ public class AnthropicService // ── Review-Agent Phase ── if (review != null && stopReason == "end_turn" && reviewRound < review.MaxReviewIterations) { - var reviewResult = await RunReview(response, review, messages, chainId, workspace, sessionId, usage, iterations, reviewRound, cancellationToken); + var reviewResult = await RunReview(response, review, messages, chainId, workspace, sessionId, usage, iterations, reviewRound, workCt, sseCt); iterations += reviewResult.ReviewIterationCount; reviewRound++; @@ -223,7 +224,7 @@ public class AnthropicService // ── Finalizer-Agent Phase ── if (finalizer != null && stopReason == "end_turn") { - var finalizerIterations = await RunFinalizer(response, finalizer, messages, chainId, workspace, sessionId, usage, iterations, cancellationToken); + var finalizerIterations = await RunFinalizer(response, finalizer, messages, chainId, workspace, sessionId, usage, iterations, workCt, sseCt); iterations += finalizerIterations; } @@ -242,7 +243,7 @@ public class AnthropicService stopReason = finalStopReason, streamErrorInfo, planMode - }, cancellationToken); + }, sseCt); return; } @@ -262,13 +263,13 @@ public class AnthropicService } await FinalizeChain(chain, usage, iterations, "max_iterations", sw); - await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, cancellationToken); - await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken); + await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt); + await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); } catch (OperationCanceledException) { - // Client hat die Verbindung getrennt (z.B. Browser-Refresh) - await FinalizeChain(chain, usage, iterations, "client_disconnected", sw); + // Stop-Button gedrückt (workCt wurde gecancelled) + await FinalizeChain(chain, usage, iterations, "stopped", sw); } catch (Exception ex) { @@ -281,8 +282,8 @@ public class AnthropicService Content = errMsg, IsError = true }); await FinalizeChain(chain, usage, iterations, "error", sw); - await TryWriteSse(response, "error", new { error = errMsg }, cancellationToken); - await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken); + await TryWriteSse(response, "error", new { error = errMsg }, sseCt); + await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); } } @@ -310,6 +311,9 @@ public class AnthropicService string sessionId, CancellationToken cancellationToken) { + // Preparer läuft VOR dem Worker und nutzt den Request-CancellationToken direkt. + // Bei Client-Disconnect wird der Preparer abgebrochen (kein Sinn weiterzumachen, + // da der Worker noch nicht gestartet hat). cancellationToken.ThrowIfCancellationRequested(); await TryWriteSse(response, "preparer_start", new @@ -327,7 +331,7 @@ public class AnthropicService preparer.Workspace, preparer.UseMemory, preparer.WebSearchVersion, preparer.ToolChoice, preparerMessages, chainId, sessionId, usage, 0, "preparer", preparer.Agent?.Name ?? "Vorbereiter-Agent", - cancellationToken); + cancellationToken, cancellationToken); var result = new PreparerResult { IterationCount = iterations, ResponseText = text }; @@ -396,9 +400,10 @@ public class AnthropicService TokenUsage usage, int baseIterations, int reviewRound, - CancellationToken cancellationToken) + CancellationToken workCt, + CancellationToken sseCt) { - cancellationToken.ThrowIfCancellationRequested(); + workCt.ThrowIfCancellationRequested(); // SSE: Review-Phase Start await TryWriteSse(response, "review_start", new @@ -407,7 +412,7 @@ public class AnthropicService maxRounds = review.MaxReviewIterations, agentId = review.AgentId, agentName = review.Agent?.Name ?? "Review-Agent" - }, cancellationToken); + }, sseCt); // Review-Agent bekommt eigene Message-History var reviewMessages = BuildReviewMessages(workerMessages); @@ -418,7 +423,7 @@ public class AnthropicService workspace, review.UseMemory, review.WebSearchVersion, review.ToolChoice, reviewMessages, chainId, sessionId, usage, baseIterations, "review", review.Agent?.Name ?? "Review-Agent", - cancellationToken); + workCt, sseCt); var result = new ReviewResult { ReviewIterationCount = reviewIterations }; @@ -438,7 +443,7 @@ public class AnthropicService maxRounds = review.MaxReviewIterations, feedback = feedbackText, agentName = review.Agent?.Name ?? "Review-Agent" - }, cancellationToken); + }, sseCt); } else { @@ -447,7 +452,7 @@ public class AnthropicService { round = reviewRound + 1, agentName = review.Agent?.Name ?? "Review-Agent" - }, cancellationToken); + }, sseCt); result.HasFeedback = false; } @@ -469,13 +474,14 @@ public class AnthropicService string sessionId, TokenUsage usage, int baseIterations, - CancellationToken cancellationToken) + CancellationToken workCt, + CancellationToken sseCt) { await TryWriteSse(response, "finalizer_start", new { agentId = finalizer.AgentId, agentName = finalizer.Agent?.Name ?? "Finalizer-Agent" - }, cancellationToken); + }, sseCt); // Finalizer bekommt die Konversation als Kontext (gleiche Methode wie Review) var finalizerMessages = BuildReviewMessages(workerMessages); @@ -485,12 +491,12 @@ public class AnthropicService finalizer.Workspace, finalizer.UseMemory, finalizer.WebSearchVersion, finalizer.ToolChoice, finalizerMessages, chainId, sessionId, usage, baseIterations, "finalizer", finalizer.Agent?.Name ?? "Finalizer-Agent", - cancellationToken); + workCt, sseCt); await TryWriteSse(response, "finalizer_done", new { agentName = finalizer.Agent?.Name ?? "Finalizer-Agent" - }, cancellationToken); + }, sseCt); return iterations; } @@ -634,7 +640,8 @@ public class AnthropicService int baseIterations, string agentType, string agentName, - CancellationToken cancellationToken) + CancellationToken workCt, + CancellationToken sseCt = default) { int iterations = 0; var finalText = new StringBuilder(); @@ -643,7 +650,7 @@ public class AnthropicService while (iterations < maxToolIterations) { - cancellationToken.ThrowIfCancellationRequested(); + workCt.ThrowIfCancellationRequested(); iterations++; var payload = new Dictionary @@ -667,9 +674,9 @@ public class AnthropicService using var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }, - cancellationToken: cancellationToken); + cancellationToken: workCt); - var respBody = await httpResp.Content.ReadAsStringAsync(cancellationToken); + var respBody = await httpResp.Content.ReadAsStringAsync(workCt); if (!httpResp.IsSuccessStatusCode) { @@ -702,7 +709,7 @@ public class AnthropicService text, agentName, agentId - }, cancellationToken); + }, sseCt); await _db.SaveMessage(new MessageRecord { @@ -728,7 +735,7 @@ public class AnthropicService toolId, input = toolInput.ToString(), agentName - }, cancellationToken); + }, sseCt); await _db.SaveMessage(new MessageRecord { @@ -744,7 +751,7 @@ public class AnthropicService string toolResult; bool isError = false; - try { toolResult = await _toolService.ExecuteTool(toolName, toolInput, agentWorkspace, _memoryService, agentId, cancellationToken); } + try { toolResult = await _toolService.ExecuteTool(toolName, toolInput, agentWorkspace, _memoryService, agentId, workCt); } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } toolResults.Add(new @@ -763,7 +770,7 @@ public class AnthropicService result = toolResult, isError, agentName - }, cancellationToken); + }, sseCt); await _db.SaveMessage(new MessageRecord { @@ -1056,9 +1063,10 @@ public class AnthropicService TokenUsage usage, string workspace, string agentId = null, - CancellationToken cancellationToken = default) + CancellationToken workCt = default, + CancellationToken sseCt = default) { - var stream = await httpResp.Content.ReadAsStreamAsync(cancellationToken); + var stream = await httpResp.Content.ReadAsStreamAsync(workCt); using var reader = new StreamReader(stream); var stopReason = ""; @@ -1076,7 +1084,7 @@ public class AnthropicService var streamErrors = new List(); string line; - while ((line = await reader.ReadLineAsync(cancellationToken)) != null) + while ((line = await reader.ReadLineAsync(workCt)) != null) { if (!line.StartsWith("data: ")) continue; var data = line["data: ".Length..]; @@ -1108,7 +1116,7 @@ public class AnthropicService if (cbType == "text") { currentTextBlock.Clear(); - await WriteSseEvent(response, "text_start", new { index = idx }, cancellationToken); + await WriteSseEvent(response, "text_start", new { index = idx }, sseCt); } else if (cbType == "tool_use") { @@ -1120,7 +1128,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 }, cancellationToken); + await WriteSseEvent(response, "tool_start", new { name = currentToolName, id = currentToolId }, sseCt); } else if (cbType == "server_tool_use") { @@ -1132,7 +1140,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 }, cancellationToken); + await WriteSseEvent(response, "web_search_start", new { name = currentToolName, id = currentToolId }, sseCt); } else if (cbType == "web_search_tool_result") { @@ -1157,7 +1165,7 @@ public class AnthropicService { var text = delta.GetProperty("text").GetString() ?? ""; currentTextBlock.Append(text); - await WriteSseEvent(response, "text_delta", new { text }, cancellationToken); + await WriteSseEvent(response, "text_delta", new { text }, sseCt); } else if (deltaType == "input_json_delta") { @@ -1178,7 +1186,7 @@ public class AnthropicService if (text.Length > 0) { assistantContent.Add(new { type = "text", text }); - await WriteSseEvent(response, "text_end", new { }, cancellationToken); + await WriteSseEvent(response, "text_end", new { }, sseCt); } currentTextBlock.Clear(); } @@ -1220,7 +1228,7 @@ public class AnthropicService toolCallId = currentToolId, question, choices - }, cancellationToken); + }, sseCt); // Stream restlich konsumieren ohne weitere Verarbeitung // (Claude sendet nach dem tool_use noch message_delta mit stop_reason) @@ -1229,15 +1237,15 @@ public class AnthropicService } // Cancellation prüfen bevor Tool ausgeführt wird - cancellationToken.ThrowIfCancellationRequested(); + workCt.ThrowIfCancellationRequested(); string toolResult; bool isError = false; - try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, cancellationToken); } + try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, workCt); } catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } - cancellationToken.ThrowIfCancellationRequested(); + workCt.ThrowIfCancellationRequested(); executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); await WriteSseEvent(response, "tool_end", new @@ -1246,7 +1254,7 @@ public class AnthropicService input = currentToolInputJson.ToString(), result = toolResult, isError - }, cancellationToken); + }, sseCt); } else if (bt == "server_tool_use") { @@ -1268,7 +1276,7 @@ public class AnthropicService // Web-Search-Ergebnis von Anthropic – als Raw-JSON in assistantContent aufnehmen assistantContent.Add(webSearchResultBlocks[idx]); var sources = webSearchSources.TryGetValue(idx, out var s) ? s : new List(); - await WriteSseEvent(response, "web_search_end", new { id = currentToolId, sources }, cancellationToken); + await WriteSseEvent(response, "web_search_end", new { id = currentToolId, sources }, sseCt); } else if (bt.EndsWith("_tool_result") && serverToolResultBlocks.TryGetValue(idx, out var srvResultBlock)) {