From 6499864f13b6af0096deab75299372bdc62b15a6 Mon Sep 17 00:00:00 2001 From: Meleki Date: Tue, 10 Mar 2026 11:21:28 +0100 Subject: [PATCH] fix: Agent bricht bei Page-Refresh nicht mehr ab + SSE-Writes crash-sicher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ursache: WriteSseEvent in ProcessStream warf OperationCanceledException bei Client-Disconnect (Page-Refresh) → sseCt war gelinkt mit dem ASP.NET cancellationToken → Exception wurde im catch-Block als 'stopped' behandelt → Agent brach ab obwohl nur der Browser refreshed wurde. Fixes: - ProcessStream: Alle WriteSseEvent durch TryWriteSse ersetzt SSE-Writes bei Client-Disconnect werden jetzt zu stillen No-Ops statt den gesamten Agent-Flow per Exception zu killen - catch-Blöcke: OperationCanceledException nur bei workCt/hardCt als 'stopped' behandeln, neuer catch für Client-Disconnect als 'client_disconnected' (Fallback, sollte nicht mehr auftreten) - done-Event nach Stop: sseCt → CancellationToken.None damit das done-Event trotz gecancelltem sseCt noch gesendet wird - Frontend: done-Handler rendert Chain-Badge direkt im apiGetChain- Callback statt Race-Condition mit finally-Block - Frontend: 'stopped' Badge wieder angezeigt (war nur fälschlich bei Page-Refresh, jetzt korrekt nur bei echtem Stop) --- Services/AnthropicService.cs | 54 ++++++++++++++++++++++++++++++------ wwwroot/js/chat.js | 14 +++++++--- 2 files changed, 55 insertions(+), 13 deletions(-) diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index bb01bd5..3eb23ec 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -272,10 +272,25 @@ public class AnthropicService await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt); await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); } - catch (OperationCanceledException) + catch (OperationCanceledException) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) { // Stop-Button gedrückt (workCt oder hardCt wurde gecancelled) await FinalizeChain(chain, usage, iterations, "stopped", sw); + // sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None + // Der SSE-Stream zum Client ist noch offen – nur der Stop-Button hat ausgelöst + await TryWriteSse(response, "done", new + { + model = modelId, + inputTokens = usage.InputTokens, + outputTokens = usage.OutputTokens, + cacheCreationTokens = usage.CacheCreationTokens, + cacheReadTokens = usage.CacheReadTokens, + iterations, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason = "stopped", + planMode + }, CancellationToken.None); } catch (Exception) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) { @@ -283,6 +298,27 @@ public class AnthropicService // ObjectDisposedException/IOException aus statt OperationCanceledException. // Wenn workCt/hardCt gecancelled ist, ist das ein regulärer Stop. await FinalizeChain(chain, usage, iterations, "stopped", sw); + // sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None + await TryWriteSse(response, "done", new + { + model = modelId, + inputTokens = usage.InputTokens, + outputTokens = usage.OutputTokens, + cacheCreationTokens = usage.CacheCreationTokens, + cacheReadTokens = usage.CacheReadTokens, + iterations, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason = "stopped", + planMode + }, CancellationToken.None); + } + catch (OperationCanceledException) + { + // Client-Disconnect (cancellationToken gecancelled, aber NICHT workCt/hardCt) + // Agent-Arbeit ist leider schon abgebrochen – Chain als "client_disconnected" finalisieren + // (sollte durch TryWriteSse eigentlich nicht mehr passieren) + await FinalizeChain(chain, usage, iterations, "client_disconnected", sw); } catch (Exception ex) { @@ -1136,7 +1172,7 @@ public class AnthropicService if (cbType == "text") { currentTextBlock.Clear(); - await WriteSseEvent(response, "text_start", new { index = idx }, sseCt); + await TryWriteSse(response, "text_start", new { index = idx }, sseCt); } else if (cbType == "tool_use") { @@ -1148,7 +1184,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 }, sseCt); + await TryWriteSse(response, "tool_start", new { name = currentToolName, id = currentToolId }, sseCt); } else if (cbType == "server_tool_use") { @@ -1160,7 +1196,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 }, sseCt); + await TryWriteSse(response, "web_search_start", new { name = currentToolName, id = currentToolId }, sseCt); } else if (cbType == "web_search_tool_result") { @@ -1185,7 +1221,7 @@ public class AnthropicService { var text = delta.GetProperty("text").GetString() ?? ""; currentTextBlock.Append(text); - await WriteSseEvent(response, "text_delta", new { text }, sseCt); + await TryWriteSse(response, "text_delta", new { text }, sseCt); } else if (deltaType == "input_json_delta") { @@ -1206,7 +1242,7 @@ public class AnthropicService if (text.Length > 0) { assistantContent.Add(new { type = "text", text }); - await WriteSseEvent(response, "text_end", new { }, sseCt); + await TryWriteSse(response, "text_end", new { }, sseCt); } currentTextBlock.Clear(); } @@ -1243,7 +1279,7 @@ public class AnthropicService } } - await WriteSseEvent(response, "user_choice", new + await TryWriteSse(response, "user_choice", new { toolCallId = currentToolId, question, @@ -1268,7 +1304,7 @@ public class AnthropicService workCt.ThrowIfCancellationRequested(); executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); - await WriteSseEvent(response, "tool_end", new + await TryWriteSse(response, "tool_end", new { name = currentToolName, input = currentToolInputJson.ToString(), @@ -1296,7 +1332,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 }, sseCt); + await TryWriteSse(response, "web_search_end", new { id = currentToolId, sources }, sseCt); } else if (bt.EndsWith("_tool_result") && serverToolResultBlocks.TryGetValue(idx, out var srvResultBlock)) { diff --git a/wwwroot/js/chat.js b/wwwroot/js/chat.js index fd760d9..c98eda6 100644 --- a/wwwroot/js/chat.js +++ b/wwwroot/js/chat.js @@ -121,11 +121,13 @@ function handleStreamEvent(eventType, data) { if (data.model) { _lastDoneChainData = data; } else if (data.chainId) { + // Fallback: Chain-Daten vom Server nachladen und Badge direkt rendern + // (async – _lastDoneChainData wäre im finally schon gelesen) apiGetChain(data.chainId).then(chain => { - if (chain) _lastDoneChainData = chain; + if (chain) addChainMetaBubble(chain, false); }).catch(() => {}); } - setMeta('Bereit'); + setMeta(data.stopReason === 'stopped' ? 'Gestoppt' : 'Bereit'); break; } } @@ -238,8 +240,7 @@ async function _runStream(fetchPromise) { if (_chatAbandoned) return; endStreamBubble(); if (e.name === 'AbortError') { - setMeta('Abgebrochen'); - await new Promise(r => setTimeout(r, 500)); + setMeta('Gestoppt'); } else { try { const status = await apiIsStreaming(sid); @@ -261,6 +262,11 @@ async function _runStream(fetchPromise) { stopBtn.title = 'Antwort abbrechen'; isStreaming = false; _abortController = null; + // Meta zurücksetzen falls kein done-Event kam (z.B. nach Stop) + var metaEl = $('#meta'); + if (metaEl && metaEl.textContent.includes('Streaming')) { + setMeta(_stopClickCount > 0 ? 'Gestoppt' : 'Bereit'); + } } endStreamBubble(); removeStreamedElements();