diff --git a/Endpoints/ChatEndpoint.cs b/Endpoints/ChatEndpoint.cs index 8025267..e1d25e8 100644 --- a/Endpoints/ChatEndpoint.cs +++ b/Endpoints/ChatEndpoint.cs @@ -175,14 +175,16 @@ public static class ChatEndpoint // Bei Choice-Response: toolChoice auf "auto" setzen, damit Claude frei antworten kann var effectiveToolChoice = isChoiceResponse ? "auto" : ctx.ToolChoice; - // Zwei separate CancellationTokens: - // - workCt: Nur Stop-Button → für Anthropic API-Calls, Tool-Ausführungen, DB-Writes + // Drei separate CancellationTokens: + // - workCt: Nur Stop-Button (graceful) → für Anthropic API-Calls, Tool-Ausführungen, DB-Writes // Agent-Arbeit läuft bei Client-Disconnect weiter (Client pollt ja) + // - hardCt: Hard-Stop (Doppel-Klick) → sofortiger Abbruch aus while-Loop // - 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); - var workCt = stopCts.Token; - using var sseLinkCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, stopCts.Token); + var tokens = streamingTracker.Start(sessionId, ctx.ChainId); + var workCt = tokens.StopToken; + var hardCt = tokens.HardToken; + using var sseLinkCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, tokens.StopToken); var sseCt = sseLinkCts.Token; // SSE-Heartbeat: Alle 15s ein Kommentar senden, damit die Verbindung @@ -204,7 +206,7 @@ public static class ChatEndpoint 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, workCt: workCt, sseCt: sseCt); + 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, hardCt: hardCt, sseCt: sseCt); } finally { @@ -225,7 +227,8 @@ public static class ChatEndpoint return; } - var cancelled = streamingTracker.Cancel(sessionId); - await res.WriteAsJsonAsync(new { stopped = cancelled }); + // Erster Klick = "graceful" (laufende Iteration fertig), zweiter = "hard" (sofort) + var stopLevel = streamingTracker.Cancel(sessionId); + await res.WriteAsJsonAsync(new { stopped = stopLevel != null, level = stopLevel ?? "none" }); } } diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index 9766e84..bb01bd5 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -52,6 +52,7 @@ public class AnthropicService bool allowUserChoice = true, bool planMode = false, CancellationToken workCt = default, + CancellationToken hardCt = default, CancellationToken sseCt = default) { var sw = System.Diagnostics.Stopwatch.StartNew(); @@ -71,6 +72,8 @@ public class AnthropicService { while (iterations < maxIterations) { + // Hard-Stop: Sofortiger Abbruch aus while-Loop (Doppel-Klick Stop) + hardCt.ThrowIfCancellationRequested(); workCt.ThrowIfCancellationRequested(); iterations++; @@ -260,6 +263,9 @@ public class AnthropicService } messages.Add(new { role = "user", content = toolResults }); + + // Hard-Stop Check nach Tool-Ausführung: Sofort raus statt nächste Iteration + hardCt.ThrowIfCancellationRequested(); } await FinalizeChain(chain, usage, iterations, "max_iterations", sw); @@ -268,7 +274,14 @@ public class AnthropicService } catch (OperationCanceledException) { - // Stop-Button gedrückt (workCt wurde gecancelled) + // Stop-Button gedrückt (workCt oder hardCt wurde gecancelled) + await FinalizeChain(chain, usage, iterations, "stopped", sw); + } + catch (Exception) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) + { + // Stream-Dispose durch CancellationToken-Registration löst + // ObjectDisposedException/IOException aus statt OperationCanceledException. + // Wenn workCt/hardCt gecancelled ist, ist das ein regulärer Stop. await FinalizeChain(chain, usage, iterations, "stopped", sw); } catch (Exception ex) @@ -752,6 +765,7 @@ public class AnthropicService string toolResult; bool isError = false; try { toolResult = await _toolService.ExecuteTool(toolName, toolInput, agentWorkspace, _memoryService, agentId, workCt); } + catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } toolResults.Add(new @@ -946,6 +960,7 @@ public class AnthropicService string toolResult; bool isError = false; try { toolResult = await _toolService.ExecuteTool(toolName, toolInput, ctx.Workspace, _memoryService, ctx.AgentId); } + catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } toolResults.Add(new @@ -1067,6 +1082,11 @@ public class AnthropicService CancellationToken sseCt = default) { var stream = await httpResp.Content.ReadAsStreamAsync(workCt); + // Bei Cancellation den Stream disposen, damit ReadLineAsync sofort abbricht. + // StreamReader.ReadLineAsync propagiert den CancellationToken nicht zuverlässig + // an den underlying NetworkStream – der Read blockiert sonst bis Anthropic + // das nächste Chunk sendet (kann bei extended thinking Minuten dauern). + using var ctReg = workCt.Register(() => { try { stream.Dispose(); } catch { } }); using var reader = new StreamReader(stream); var stopReason = ""; diff --git a/Services/StreamingTracker.cs b/Services/StreamingTracker.cs index 7373170..4572495 100644 --- a/Services/StreamingTracker.cs +++ b/Services/StreamingTracker.cs @@ -7,37 +7,56 @@ public class StreamingTracker { private readonly ConcurrentDictionary _streaming = new ConcurrentDictionary(); - public CancellationTokenSource Start(string sessionId, string chainId) + public StreamingTokens Start(string sessionId, string chainId) { - var cts = new CancellationTokenSource(); - _streaming[sessionId] = new StreamingState(chainId, cts); - return cts; + var stopCts = new CancellationTokenSource(); + var hardCts = new CancellationTokenSource(); + var state = new StreamingState(chainId, stopCts, hardCts); + _streaming[sessionId] = state; + return new StreamingTokens(stopCts.Token, hardCts.Token); } public void Stop(string sessionId) { if (_streaming.TryRemove(sessionId, out var state)) { - state.Cts.Dispose(); + state.StopCts.Dispose(); + state.HardCts.Dispose(); } } /// - /// Cancels the running stream for a session. Returns true if a stream was cancelled. + /// Cancels the running stream for a session. + /// First call = graceful (laufende Iteration wird fertig). + /// Second call = hard (sofortiger Abbruch aus while-Loop). + /// Returns "graceful", "hard" or null if not streaming. /// - public bool Cancel(string sessionId) + public string Cancel(string sessionId) { if (_streaming.TryGetValue(sessionId, out var state)) { - try { state.Cts.Cancel(); } catch { } - return true; + if (!state.StopCts.IsCancellationRequested) + { + // Erster Klick → graceful stop + try { state.StopCts.Cancel(); } catch { } + return "graceful"; + } + else if (!state.HardCts.IsCancellationRequested) + { + // Zweiter Klick → hard stop + try { state.HardCts.Cancel(); } catch { } + return "hard"; + } + return "hard"; // Bereits beides gecancelled } - return false; + return null; } public bool IsStreaming(string sessionId) => _streaming.ContainsKey(sessionId); public string GetChainId(string sessionId) => _streaming.TryGetValue(sessionId, out var state) ? state.ChainId : null; - private record StreamingState(string ChainId, CancellationTokenSource Cts); + public record StreamingTokens(CancellationToken StopToken, CancellationToken HardToken); + + private record StreamingState(string ChainId, CancellationTokenSource StopCts, CancellationTokenSource HardCts); } diff --git a/wwwroot/css/chat.css b/wwwroot/css/chat.css index 9efa3ae..f841fef 100644 --- a/wwwroot/css/chat.css +++ b/wwwroot/css/chat.css @@ -440,6 +440,18 @@ background: rgb(from var(--dangerous) r g b / 0.15); } +.stop-btn.stop-hard { + background: var(--dangerous); + color: var(--primary-text); + animation: pulse-hard 0.6s ease-in-out; +} + +@keyframes pulse-hard { + 0% { transform: scale(1); } + 50% { transform: scale(1.15); } + 100% { transform: scale(1); } +} + .input-area textarea { flex: 1; min-height: 0; diff --git a/wwwroot/js/chat.js b/wwwroot/js/chat.js index 5d2f132..fd760d9 100644 --- a/wwwroot/js/chat.js +++ b/wwwroot/js/chat.js @@ -170,11 +170,23 @@ async function clearChat() { showArchiveLoadButton(); } +let _stopClickCount = 0; + function stopStreaming() { - if (_abortController) { - _abortController.abort(); + _stopClickCount++; + + if (_stopClickCount >= 2) { + // Hard-Stop: Sofortiger Abbruch (Doppel-Klick) + var btn = $('#stop'); + if (btn) { + btn.classList.add('stop-hard'); + btn.title = 'Hard-Stop gesendet!'; + } + // SSE-Stream clientseitig sofort abbrechen + if (_abortController) _abortController.abort(); } - // Backend explizit stoppen (kills laufende Tools/API-Calls) + + // Backend explizit stoppen (graceful beim 1. Klick, hard beim 2.) var sid = getCurrentSessionId(); if (sid) apiStopChat(sid).catch(function() {}); } @@ -193,6 +205,7 @@ async function _runStream(fetchPromise) { _abortController = new AbortController(); _currentToolBox = null; _chatAbandoned = false; + _stopClickCount = 0; setMetaHtml(' Streaming...'); try { @@ -244,6 +257,8 @@ async function _runStream(fetchPromise) { sendBtn.disabled = false; sendBtn.hidden = false; stopBtn.hidden = true; + stopBtn.classList.remove('stop-hard'); + stopBtn.title = 'Antwort abbrechen'; isStreaming = false; _abortController = null; }