From 40c8fee531039f7ab0845273ea4181010cf40c1c Mon Sep 17 00:00:00 2001 From: Meleki Date: Fri, 6 Mar 2026 15:44:08 +0100 Subject: [PATCH] =?UTF-8?q?feat:=20zuverl=C3=A4ssiger=20Stop-Button=20f?= =?UTF-8?q?=C3=BCr=20Chat-Streaming?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - StreamingTracker: CancellationTokenSource pro Session, Cancel()-Methode - ChatEndpoint: POST /api/chat/stop Endpoint, linkedCts verknüpft Stop-Token mit Client-Disconnect-Token - AnthropicService: cancellationToken an ExecuteTool durchgereicht - ToolService: CancellationToken-Parameter, execute-Tool verknüpft externen Token mit 180s-Timeout, killt Prozess bei Abbruch - Frontend: apiStopChat() + stopStreaming() ruft Backend-Stop auf --- Endpoints/ChatEndpoint.cs | 24 ++++++++++++++++++++++-- Services/AnthropicService.cs | 2 +- Services/StreamingTracker.cs | 35 +++++++++++++++++++++++++++++++---- Services/ToolService.cs | 8 ++++++-- wwwroot/js/api.js | 9 +++++++++ wwwroot/js/chat.js | 3 +++ 6 files changed, 72 insertions(+), 9 deletions(-) diff --git a/Endpoints/ChatEndpoint.cs b/Endpoints/ChatEndpoint.cs index 022f2ec..0cb54af 100644 --- a/Endpoints/ChatEndpoint.cs +++ b/Endpoints/ChatEndpoint.cs @@ -18,6 +18,7 @@ public static class ChatEndpoint public static void MapChatEndpoint(this IEndpointRouteBuilder app) { app.MapPost("/api/chat", HandleChat); + app.MapPost("/api/chat/stop", HandleStop); } private static async Task HandleChat(HttpRequest req, HttpResponse res, IConfiguration config, ChatService chatService, DatabaseService db, StreamingTracker streamingTracker, AnthropicService anthropic, CancellationToken cancellationToken) @@ -78,14 +79,33 @@ public static class ChatEndpoint res.Headers["Cache-Control"] = "no-cache"; res.Headers["Connection"] = "keep-alive"; - streamingTracker.Start(sessionId, ctx.ChainId); + // CancellationToken: Verknüpft Stop-Button (über StreamingTracker) + Client-Disconnect + var stopCts = streamingTracker.Start(sessionId, ctx.ChainId); + using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, stopCts.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: ctx.ToolChoice, cancellationToken: cancellationToken); + 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: ctx.ToolChoice, cancellationToken: linkedCts.Token); } finally { streamingTracker.Stop(sessionId); } } + + private static async Task HandleStop(HttpRequest req, HttpResponse res, StreamingTracker streamingTracker) + { + var body = await JsonSerializer.DeserializeAsync(req.Body); + var sessionId = body.TryGetProperty("sessionId", out var sid) ? sid.GetString() : null; + + if (string.IsNullOrEmpty(sessionId)) + { + res.StatusCode = 400; + await res.WriteAsJsonAsync(new { error = "sessionId required" }); + return; + } + + var cancelled = streamingTracker.Cancel(sessionId); + await res.WriteAsJsonAsync(new { stopped = cancelled }); + } } diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index cda6604..cef42f0 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -612,7 +612,7 @@ public class AnthropicService string toolResult; bool isError = false; - try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId); } + try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, cancellationToken); } catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } diff --git a/Services/StreamingTracker.cs b/Services/StreamingTracker.cs index 420d278..7373170 100644 --- a/Services/StreamingTracker.cs +++ b/Services/StreamingTracker.cs @@ -1,16 +1,43 @@ using System.Collections.Concurrent; +using System.Threading; namespace Gott.Services; public class StreamingTracker { - private readonly ConcurrentDictionary _streaming = new ConcurrentDictionary(); + private readonly ConcurrentDictionary _streaming = new ConcurrentDictionary(); - public void Start(string sessionId, string chainId) => _streaming[sessionId] = chainId; + public CancellationTokenSource Start(string sessionId, string chainId) + { + var cts = new CancellationTokenSource(); + _streaming[sessionId] = new StreamingState(chainId, cts); + return cts; + } - public void Stop(string sessionId) => _streaming.TryRemove(sessionId, out _); + public void Stop(string sessionId) + { + if (_streaming.TryRemove(sessionId, out var state)) + { + state.Cts.Dispose(); + } + } + + /// + /// Cancels the running stream for a session. Returns true if a stream was cancelled. + /// + public bool Cancel(string sessionId) + { + if (_streaming.TryGetValue(sessionId, out var state)) + { + try { state.Cts.Cancel(); } catch { } + return true; + } + return false; + } public bool IsStreaming(string sessionId) => _streaming.ContainsKey(sessionId); - public string GetChainId(string sessionId) => _streaming.TryGetValue(sessionId, out var chainId) ? chainId : null; + public string GetChainId(string sessionId) => _streaming.TryGetValue(sessionId, out var state) ? state.ChainId : null; + + private record StreamingState(string ChainId, CancellationTokenSource Cts); } diff --git a/Services/ToolService.cs b/Services/ToolService.cs index 1195ff7..12ff7f0 100644 --- a/Services/ToolService.cs +++ b/Services/ToolService.cs @@ -232,7 +232,7 @@ public class ToolService }; } - public async Task ExecuteTool(string name, JsonElement input, string workspace, MemoryService memoryService = null, string agentId = null) + public async Task ExecuteTool(string name, JsonElement input, string workspace, MemoryService memoryService = null, string agentId = null, CancellationToken cancellationToken = default) { switch (name) { @@ -322,7 +322,9 @@ public class ToolService if (proc == null) return "Fehler: Prozess konnte nicht gestartet werden"; await proc.StandardInput.WriteAsync(command); proc.StandardInput.Close(); - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(180)); + // Timeout (180s) UND externen CancellationToken verknüpfen + using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + cts.CancelAfter(TimeSpan.FromSeconds(180)); try { await proc.WaitForExitAsync(cts.Token); @@ -330,6 +332,8 @@ public class ToolService catch (OperationCanceledException) { proc.Kill(entireProcessTree: true); + // Wenn der externe Token abgebrochen wurde, weiterwerfen + cancellationToken.ThrowIfCancellationRequested(); return "Fehler: Timeout nach 180 Sekunden - Prozess abgebrochen"; } var stdout = await proc.StandardOutput.ReadToEndAsync(); diff --git a/wwwroot/js/api.js b/wwwroot/js/api.js index b39c069..b7cb497 100644 --- a/wwwroot/js/api.js +++ b/wwwroot/js/api.js @@ -178,6 +178,15 @@ async function apiIsStreaming(sessionId) { return resp.json(); } +async function apiStopChat(sessionId) { + const resp = await fetch('/api/chat/stop', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ sessionId }) + }); + return resp.json(); +} + async function apiGetChain(chainId) { const resp = await fetch(`/api/chains/${chainId}`); if (!resp.ok) return null; diff --git a/wwwroot/js/chat.js b/wwwroot/js/chat.js index b069db9..ce13d34 100644 --- a/wwwroot/js/chat.js +++ b/wwwroot/js/chat.js @@ -56,6 +56,9 @@ function stopStreaming() { if (_abortController) { _abortController.abort(); } + // Backend explizit stoppen (kills laufende Tools/API-Calls) + var sid = getCurrentSessionId(); + if (sid) apiStopChat(sid).catch(function() {}); } // ── Send Message ──