meleki/gott
1
0 Fork 0
Code Activity

Agent-Arbeit vom Client-Disconnect entkoppeln

Browse Source 
Zwei separate CancellationTokens:
- workCt (nur Stop-Button): Anthropic API-Calls, Tool-Ausfuehrungen, DB-Writes
  Agent laeuft bei SSE-Disconnect weiter, Client pollt Ergebnisse
- sseCt (Stop-Button + Client-Disconnect): SSE-Writes an Browser
  Bei Disconnect werden Writes zu stillen No-Ops

Vorher: HttpContext.RequestAborted killte den gesamten Agent-Lauf
wenn die SSE-Verbindung abriss (Browser-Tab idle, Netzwerk-Hickup).
Jetzt arbeitet der Agent im Hintergrund weiter.

OperationCanceledException = 'stopped' (nur Stop-Button)
statt 'client_disconnected' (war falsch verknuepft)
This commit is contained in:
Meleki committed 2026-03-10 06:28:35 +01:00
1 parent dd99216913
commit e9b1e8ef20
2 files changed
+72 -58
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Endpoints/ChatEndpoint.cs
+15 -9
View File
@@ -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)
}
}
Services/AnthropicService.cs
+57 -49
View File
@@ -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<string, object>
@@ -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<string, object>
@@ -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<object>();
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<object>();
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))
{