feat: zuverlässiger Stop-Button für Chat-Streaming

- 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
This commit is contained in:
Meleki committed 2026-03-06 15:44:08 +01:00
1 parent 01b3e2c230
commit 40c8fee531
6 files changed
+72 -9

No files matched your search

+22 -2
View File
@@ -18,6 +18,7 @@ public static class ChatEndpoint
public static void MapChatEndpoint(this IEndpointRouteBuilder app) public static void MapChatEndpoint(this IEndpointRouteBuilder app)
{ {
app.MapPost("/api/chat", HandleChat); 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) 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["Cache-Control"] = "no-cache";
res.Headers["Connection"] = "keep-alive"; 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 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 finally
{ {
streamingTracker.Stop(sessionId); streamingTracker.Stop(sessionId);
} }
} }
private static async Task HandleStop(HttpRequest req, HttpResponse res, StreamingTracker streamingTracker)
{
var body = await JsonSerializer.DeserializeAsync<JsonElement>(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 });
}
} }
+1 -1
View File
@@ -612,7 +612,7 @@ public class AnthropicService
string toolResult; string toolResult;
bool isError = false; 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 (OperationCanceledException) { throw; }
catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; }
+31 -4
View File
@@ -1,16 +1,43 @@
using System.Collections.Concurrent; using System.Collections.Concurrent;
using System.Threading;
namespace Gott.Services; namespace Gott.Services;
public class StreamingTracker public class StreamingTracker
{ {
private readonly ConcurrentDictionary<string, string> _streaming = new ConcurrentDictionary<string, string>(); private readonly ConcurrentDictionary<string, StreamingState> _streaming = new ConcurrentDictionary<string, StreamingState>();
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();
}
}
/// <summary>
/// Cancels the running stream for a session. Returns true if a stream was cancelled.
/// </summary>
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 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);
} }
+6 -2
View File
@@ -232,7 +232,7 @@ public class ToolService
}; };
} }
public async Task<string> ExecuteTool(string name, JsonElement input, string workspace, MemoryService memoryService = null, string agentId = null) public async Task<string> ExecuteTool(string name, JsonElement input, string workspace, MemoryService memoryService = null, string agentId = null, CancellationToken cancellationToken = default)
{ {
switch (name) switch (name)
{ {
@@ -322,7 +322,9 @@ public class ToolService
if (proc == null) return "Fehler: Prozess konnte nicht gestartet werden"; if (proc == null) return "Fehler: Prozess konnte nicht gestartet werden";
await proc.StandardInput.WriteAsync(command); await proc.StandardInput.WriteAsync(command);
proc.StandardInput.Close(); 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 try
{ {
await proc.WaitForExitAsync(cts.Token); await proc.WaitForExitAsync(cts.Token);
@@ -330,6 +332,8 @@ public class ToolService
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
proc.Kill(entireProcessTree: true); proc.Kill(entireProcessTree: true);
// Wenn der externe Token abgebrochen wurde, weiterwerfen
cancellationToken.ThrowIfCancellationRequested();
return "Fehler: Timeout nach 180 Sekunden - Prozess abgebrochen"; return "Fehler: Timeout nach 180 Sekunden - Prozess abgebrochen";
} }
var stdout = await proc.StandardOutput.ReadToEndAsync(); var stdout = await proc.StandardOutput.ReadToEndAsync();
+9
View File
@@ -178,6 +178,15 @@ async function apiIsStreaming(sessionId) {
return resp.json(); 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) { async function apiGetChain(chainId) {
const resp = await fetch(`/api/chains/${chainId}`); const resp = await fetch(`/api/chains/${chainId}`);
if (!resp.ok) return null; if (!resp.ok) return null;
+3
View File
@@ -56,6 +56,9 @@ function stopStreaming() {
if (_abortController) { if (_abortController) {
_abortController.abort(); _abortController.abort();
} }
// Backend explizit stoppen (kills laufende Tools/API-Calls)
var sid = getCurrentSessionId();
if (sid) apiStopChat(sid).catch(function() {});
} }
// ── Send Message ── // ── Send Message ──