using System; using System.Collections.Generic; using System.IO; using System.IO.Compression; using System.Net.Http; using System.Net.Http.Headers; using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Gott.Models; using Microsoft.AspNetCore.Http; namespace Gott.Services; public class AnthropicService { private readonly IHttpClientFactory _httpFactory; private readonly string _token; private readonly ToolService _toolService; private readonly DatabaseService _db; private readonly MemoryService _memoryService; private readonly bool _payloadLogging; private readonly bool _payloadLoggingGzip; public AnthropicService(IHttpClientFactory httpFactory, string token, ToolService toolService, DatabaseService db, MemoryService memoryService = null, bool payloadLogging = false, bool payloadLoggingGzip = false) { _httpFactory = httpFactory; _token = token; _toolService = toolService; _db = db; _memoryService = memoryService; _payloadLogging = payloadLogging; _payloadLoggingGzip = payloadLoggingGzip; } public async Task StreamChat( HttpResponse response, string modelId, List systemPrompt, List messages, string chainId, string workspace, string sessionId = null, string agentId = null, bool useMemory = true, string webSearchVersion = null, int maxIterations = 100, string toolChoice = "any", SubAgentContext review = null, SubAgentContext finalizer = null, bool allowUserChoice = true, bool planMode = false, CancellationToken workCt = default, CancellationToken hardCt = default, CancellationToken sseCt = default) { var sw = System.Diagnostics.Stopwatch.StartNew(); var usage = new TokenUsage(); int iterations = 0; int reviewRound = 0; string lastStopReason = ""; var chain = new ChainRecord { ChainId = chainId, SessionId = sessionId, Model = modelId }; try { while (iterations < maxIterations) { // Hard-Stop: Sofortiger Abbruch aus while-Loop (Doppel-Klick Stop) hardCt.ThrowIfCancellationRequested(); workCt.ThrowIfCancellationRequested(); iterations++; var payload = new Dictionary { ["model"] = modelId, ["max_tokens"] = 16384, ["cache_control"] = new { type = "ephemeral" }, ["messages"] = messages, ["stream"] = true }; if (systemPrompt != null && systemPrompt.Count > 0) payload["system"] = systemPrompt; AddToolsToPayload(payload, workspace, agentId, useMemory, webSearchVersion, toolChoice, iterations, allowUserChoice, planMode); var client = CreateClient(); var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true }); LogPayload(json, sessionId, chainId, iterations); var httpContent = new StringContent(json, Encoding.UTF8, "application/json"); using var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }, HttpCompletionOption.ResponseHeadersRead, workCt ); if (!httpResp.IsSuccessStatusCode) { var errBody = await httpResp.Content.ReadAsStringAsync(workCt); var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}"; await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "system", Type = "error", Content = errMsg, IsError = true }); await TryWriteSse(response, "error", new { error = errMsg }, sseCt); await FinalizeChain(chain, usage, iterations, "error", sw); await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); return; } var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, workCt, sseCt); lastStopReason = stopReason; messages.Add(new { role = "assistant", content = assistantContent }); // Assistant-Blöcke persistieren foreach (var block in assistantContent) { var blockJson = JsonSerializer.Serialize(block); var el = JsonSerializer.Deserialize(blockJson); var blockType = el.GetProperty("type").GetString() ?? ""; if (blockType == "text") { await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "assistant", Type = "text", Content = el.GetProperty("text").GetString() }); } else if (blockType == "tool_use") { await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "assistant", Type = "tool_use", ToolName = el.GetProperty("name").GetString(), ToolCallId = el.GetProperty("id").GetString(), Content = el.GetProperty("input").ToString() }); } else if (blockType == "server_tool_use") { await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "assistant", Type = "server_tool_use", Content = el.GetRawText() }); } else if (blockType == "web_search_tool_result") { var sources = ExtractWebSearchSources(el); if (sources.Count > 0) { await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "assistant", Type = "web_search_result", Content = JsonSerializer.Serialize(sources) }); } } else if (blockType.EndsWith("_tool_result")) { // Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.) await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "assistant", Type = "server_tool_result", Content = el.GetRawText() }); } } // Tool Results persistieren foreach (var tr in executedTools) { await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "user", Type = "tool_result", ToolCallId = tr.Id, Content = tr.Result, IsError = tr.IsError }); } if (stopReason != "tool_use") { // ── 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, workCt, sseCt); iterations += reviewResult.ReviewIterationCount; reviewRound++; if (reviewResult.HasFeedback) { // Feedback als User-Message an Worker zurückgeben → nächste Iteration messages.Add(new Dictionary { ["role"] = "user", ["content"] = new List { new { type = "text", text = reviewResult.Feedback } } }); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "user", Type = "text", Content = reviewResult.Feedback }); continue; // → nächste Worker-Iteration } } // ── Finalizer-Agent Phase ── if (finalizer != null && stopReason == "end_turn") { var finalizerIterations = await RunFinalizer(response, finalizer, messages, chainId, workspace, sessionId, usage, iterations, workCt, sseCt); iterations += finalizerIterations; } var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason; await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo); 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 = finalStopReason, streamErrorInfo, planMode }, sseCt); return; } var toolResults = new List(); foreach (var tr in executedTools) { toolResults.Add(new { type = "tool_result", tool_use_id = tr.Id, content = TruncateToolResult(tr.Result), is_error = tr.IsError }); } 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); await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt); await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); } catch (OperationCanceledException) { // 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) { // Unerwarteter Fehler – in DB speichern und versuchen an Client zu senden var errMsg = $"Stream-Fehler: {ex.Message}"; await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iterations, Role = "system", Type = "error", Content = errMsg, IsError = true }); await FinalizeChain(chain, usage, iterations, "error", sw); await TryWriteSse(response, "error", new { error = errMsg }, sseCt); await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); } } // ── Preparer-Agent Logik ── public class PreparerResult { public bool ShouldForward { get; set; } public string ForwardedPrompt { get; set; } public string ResponseText { get; set; } public int IterationCount { get; set; } } /// /// Führt den Preparer-Agent aus. Der Preparer bekommt die User-Nachricht und die bisherige Konversation. /// Antwortet er mit [FORWARD] → der Text nach [FORWARD] wird an den Session-Agent weitergegeben. /// Sonst → seine Antwort wird als Rückfrage an den User gestreamt. /// public async Task RunPreparer( HttpResponse response, SubAgentContext preparer, List historyMessages, string chainId, string workspace, 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 { agentId = preparer.AgentId, agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent" }, cancellationToken); // Preparer bekommt die History als Kontext var preparerMessages = new List(historyMessages); var usage = new TokenUsage(); var (text, iterations) = await RunAgentChat( response, preparer.ModelId, preparer.SystemPrompt, preparer.AgentId, preparer.Workspace, preparer.UseMemory, preparer.WebSearchVersion, preparer.ToolChoice, preparerMessages, chainId, sessionId, usage, 0, "preparer", preparer.Agent?.Name ?? "Vorbereiter-Agent", cancellationToken, cancellationToken); var result = new PreparerResult { IterationCount = iterations, ResponseText = text }; // Logik: Standardmäßig wird IMMER weitergeleitet. // Nur wenn der Preparer explizit [QUESTION] nutzt, wird gestoppt (= Rückfrage an User). // [FORWARD] wird weiterhin erkannt und entfernt, ist aber nicht mehr zwingend nötig. if (!string.IsNullOrEmpty(text) && text.Contains("[QUESTION]")) { // Explizite Rückfrage an User var questionText = text.Replace("[QUESTION]", "").Trim(); result.ShouldForward = false; result.ResponseText = questionText; await TryWriteSse(response, "preparer_done", new { agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent", forwarded = false, responseText = questionText }, cancellationToken); } else { // Weiterleitung (Standard): [FORWARD]-Tag entfernen falls vorhanden var forwardedText = text ?? ""; if (forwardedText.Contains("[FORWARD]")) { var idx = forwardedText.IndexOf("[FORWARD]"); forwardedText = forwardedText.Substring(idx + "[FORWARD]".Length).Trim(); } result.ShouldForward = true; result.ForwardedPrompt = forwardedText; await TryWriteSse(response, "preparer_done", new { agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent", forwarded = true }, cancellationToken); } return result; } // ── Review-Agent Logik ── private class ReviewResult { public bool HasFeedback { get; set; } public string Feedback { get; set; } public int ReviewIterationCount { get; set; } } /// /// Führt den Review-Agent aus. Der Review-Agent bekommt die Text-Konversation /// und kann Tools nutzen (z.B. read_file um die Arbeit zu prüfen). /// Default = Approved. Nur bei explizitem [FEEDBACK]-Tag wird Feedback an den Worker zurückgegeben. /// reviewRound = aktuelle Runde (0-basiert), wird vom Caller hochgezählt. /// private async Task RunReview( HttpResponse response, SubAgentContext review, List workerMessages, string chainId, string workspace, string sessionId, TokenUsage usage, int baseIterations, int reviewRound, CancellationToken workCt, CancellationToken sseCt) { workCt.ThrowIfCancellationRequested(); // SSE: Review-Phase Start await TryWriteSse(response, "review_start", new { round = reviewRound + 1, maxRounds = review.MaxReviewIterations, agentId = review.AgentId, agentName = review.Agent?.Name ?? "Review-Agent" }, sseCt); // Review-Agent bekommt eigene Message-History var reviewMessages = BuildReviewMessages(workerMessages); // Review-Agent iteriert mit eigenen Tools var (reviewText, reviewIterations) = await RunAgentChat( response, review.ModelId, review.SystemPrompt, review.AgentId, workspace, review.UseMemory, review.WebSearchVersion, review.ToolChoice, reviewMessages, chainId, sessionId, usage, baseIterations, "review", review.Agent?.Name ?? "Review-Agent", workCt, sseCt); var result = new ReviewResult { ReviewIterationCount = reviewIterations }; // Default = Approved (analog zum Preparer, wo Default = Forward ist). // Nur bei explizitem [FEEDBACK]-Tag wird Feedback an den Worker zurückgespielt. // [APPROVED] wird weiterhin erkannt, ist aber nicht mehr zwingend nötig. if (!string.IsNullOrEmpty(reviewText) && reviewText.Contains("[FEEDBACK]")) { // Explizites Feedback – an Worker zurückspielen var feedbackText = reviewText.Replace("[FEEDBACK]", "").Replace("[APPROVED]", "").Trim(); result.HasFeedback = true; result.Feedback = $"[Review-Feedback von {review.Agent?.Name ?? "Review-Agent"} (Runde {reviewRound + 1}/{review.MaxReviewIterations})]:\n{feedbackText}"; await TryWriteSse(response, "review_feedback", new { round = reviewRound + 1, maxRounds = review.MaxReviewIterations, feedback = feedbackText, agentName = review.Agent?.Name ?? "Review-Agent" }, sseCt); } else { // Default: Approved (kein [FEEDBACK]-Tag → alles OK) await TryWriteSse(response, "review_approved", new { round = reviewRound + 1, agentName = review.Agent?.Name ?? "Review-Agent" }, sseCt); result.HasFeedback = false; } return result; } // ── Finalizer-Agent Logik ── /// /// Führt den Finalizer-Agent aus. Wird einmal nach der Review-Phase (oder direkt nach dem Worker) aufgerufen. /// Der Finalizer kann Tools nutzen, gibt aber kein Feedback an den Worker zurück. /// private async Task RunFinalizer( HttpResponse response, SubAgentContext finalizer, List workerMessages, string chainId, string workspace, string sessionId, TokenUsage usage, int baseIterations, CancellationToken workCt, CancellationToken sseCt) { await TryWriteSse(response, "finalizer_start", new { agentId = finalizer.AgentId, agentName = finalizer.Agent?.Name ?? "Finalizer-Agent" }, sseCt); // Finalizer bekommt die Konversation als Kontext (gleiche Methode wie Review) var finalizerMessages = BuildReviewMessages(workerMessages); var (text, iterations) = await RunAgentChat( response, finalizer.ModelId, finalizer.SystemPrompt, finalizer.AgentId, finalizer.Workspace, finalizer.UseMemory, finalizer.WebSearchVersion, finalizer.ToolChoice, finalizerMessages, chainId, sessionId, usage, baseIterations, "finalizer", finalizer.Agent?.Name ?? "Finalizer-Agent", workCt, sseCt); await TryWriteSse(response, "finalizer_done", new { agentName = finalizer.Agent?.Name ?? "Finalizer-Agent" }, sseCt); return iterations; } /// /// Baut die Messages für den Review-Agent auf. /// Der Review-Agent bekommt die komplette Worker-Konversation als eine User-Message. /// private List BuildReviewMessages(List workerMessages) { // Nur die Text-Konversation extrahieren – Tool-Calls und Tool-Results weglassen. // Der Review/Finalizer-Agent hat eigene Tools und kann bei Bedarf selbst Dateien lesen. // Aufeinanderfolgende Messages gleicher Rolle werden zusammengeführt, // da die Anthropic API abwechselnde user/assistant-Rollen erfordert. var messages = new List(); string lastRole = null; var currentParts = new List(); foreach (var msg in workerMessages) { var json = JsonSerializer.Serialize(msg); var el = JsonSerializer.Deserialize(json); if (!el.TryGetProperty("role", out var role)) continue; var roleStr = role.GetString() ?? ""; var textParts = new List(); if (el.TryGetProperty("content", out var content)) { if (content.ValueKind == JsonValueKind.String) { var s = content.GetString() ?? ""; if (!string.IsNullOrWhiteSpace(s)) textParts.Add(s); } else if (content.ValueKind == JsonValueKind.Array) { foreach (var block in content.EnumerateArray()) { var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : ""; if (blockType == "text") { var text = block.TryGetProperty("text", out var t) ? t.GetString() ?? "" : ""; if (!string.IsNullOrWhiteSpace(text)) textParts.Add(text); } } } } if (textParts.Count == 0) continue; if (roleStr == lastRole) { // Gleiche Rolle wie vorher → zusammenführen currentParts.AddRange(textParts); } else { // Neue Rolle → vorherige Message abschließen if (lastRole != null && currentParts.Count > 0) { messages.Add(new Dictionary { ["role"] = lastRole, ["content"] = new List { new { type = "text", text = string.Join("\n\n", currentParts) } } }); } lastRole = roleStr; currentParts = new List(textParts); } } // Letzte angesammelte Message abschließen if (lastRole != null && currentParts.Count > 0) { messages.Add(new Dictionary { ["role"] = lastRole, ["content"] = new List { new { type = "text", text = string.Join("\n\n", currentParts) } } }); } // Anthropic erfordert: erste Message = user, letzte Message = user, // und abwechselnde Rollen (durch die Zusammenführung oben bereits sichergestellt). if (messages.Count > 0) { var first = (Dictionary)messages[0]; if ((string)first["role"] == "assistant") { messages.Insert(0, new Dictionary { ["role"] = "user", ["content"] = new List { new { type = "text", text = "Hier ist die bisherige Konversation. Bitte überprüfe den aktuellen Zustand." } } }); } var last = (Dictionary)messages[messages.Count - 1]; if ((string)last["role"] == "assistant") { messages.Add(new Dictionary { ["role"] = "user", ["content"] = new List { new { type = "text", text = "Bitte überprüfe ob die Aufgabe korrekt erledigt wurde." } } }); } } // Falls keine Text-Messages vorhanden, Fallback mit Hinweis if (messages.Count == 0) { messages.Add(new Dictionary { ["role"] = "user", ["content"] = new List { new { type = "text", text = "Die bisherige Konversation enthielt nur Tool-Aufrufe ohne Text. Bitte überprüfe den aktuellen Zustand des Workspaces mit deinen Tools." } } }); } return messages; } /// /// Generischer Agent-Chat (non-streaming für den Agent, aber SSE-Events für den Client). /// Wird für Review-Agent und Finalizer-Agent verwendet. /// agentType steuert SSE-Event-Prefixe und DB-Message-Tags ("review" oder "finalizer"). /// private async Task<(string text, int iterations)> RunAgentChat( HttpResponse response, string modelId, List systemPrompt, string agentId, string agentWorkspace, bool useMemory, string webSearchVersion, string agentToolChoice, List messages, string chainId, string sessionId, TokenUsage usage, int baseIterations, string agentType, string agentName, CancellationToken workCt, CancellationToken sseCt = default) { int iterations = 0; var finalText = new StringBuilder(); int maxToolIterations = 20; var tag = agentType == "preparer" ? "Preparer" : agentType == "finalizer" ? "Finalizer" : "Review"; while (iterations < maxToolIterations) { workCt.ThrowIfCancellationRequested(); iterations++; var payload = new Dictionary { ["model"] = modelId, ["max_tokens"] = 8192, ["cache_control"] = new { type = "ephemeral" }, ["messages"] = messages, ["stream"] = false }; if (systemPrompt != null && systemPrompt.Count > 0) payload["system"] = systemPrompt; AddToolsToPayload(payload, agentWorkspace, agentId, useMemory, webSearchVersion, agentToolChoice, iterations); var client = CreateClient(); var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true }); LogPayload(json, sessionId, chainId, baseIterations + iterations); var httpContent = new StringContent(json, Encoding.UTF8, "application/json"); using var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }, cancellationToken: workCt); var respBody = await httpResp.Content.ReadAsStringAsync(workCt); if (!httpResp.IsSuccessStatusCode) { return ($"{tag}-Agent API Fehler: {(int)httpResp.StatusCode}: {respBody}", iterations); } var resp = JsonSerializer.Deserialize(respBody); if (resp.TryGetProperty("usage", out var u)) TrackUsage(usage, u); var stopReason = resp.TryGetProperty("stop_reason", out var sr) ? sr.GetString() ?? "" : ""; var content = resp.GetProperty("content"); var assistantContent = new List(); var toolResults = new List(); foreach (var block in content.EnumerateArray()) { var blockType = block.GetProperty("type").GetString() ?? ""; if (blockType == "text") { var text = block.GetProperty("text").GetString() ?? ""; assistantContent.Add(new { type = "text", text }); finalText.Append(text); await TryWriteSse(response, $"{agentType}_text", new { text, agentName, agentId }, sseCt); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "assistant", Type = "text", Content = $"[{tag}: {agentName}] {text}", Tag = agentType }); } else if (blockType == "tool_use") { var toolId = block.GetProperty("id").GetString() ?? ""; var toolName = block.GetProperty("name").GetString() ?? ""; var toolInput = block.GetProperty("input"); assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = toolInput }); await TryWriteSse(response, $"{agentType}_tool_use", new { toolName, toolId, input = toolInput.ToString(), agentName }, sseCt); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "assistant", Type = "tool_use", Content = toolInput.GetRawText(), ToolName = toolName, ToolCallId = toolId, Tag = agentType }); 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 { type = "tool_result", tool_use_id = toolId, content = TruncateToolResult(toolResult), is_error = isError }); await TryWriteSse(response, $"{agentType}_tool_end", new { toolName, toolId, input = toolInput.ToString(), result = toolResult, isError, agentName }, sseCt); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "user", Type = "tool_result", Content = toolResult, ToolName = toolName, ToolCallId = toolId, IsError = isError, Tag = agentType }); } else if (blockType == "web_search_tool_result") { assistantContent.Add(JsonSerializer.Deserialize(block.GetRawText())); var sources = ExtractWebSearchSources(block); if (sources.Count > 0) { await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "assistant", Type = "web_search_result", Content = JsonSerializer.Serialize(sources), Tag = agentType }); } } else if (blockType == "server_tool_use") { var (toolId, toolName, inputEl) = ParseServerToolUse(block); assistantContent.Add(new { type = "server_tool_use", id = toolId, name = toolName, input = inputEl }); } else if (blockType.EndsWith("_tool_result")) { // Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.) assistantContent.Add(JsonSerializer.Deserialize(block.GetRawText())); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "assistant", Type = "server_tool_result", Content = block.GetRawText(), Tag = agentType }); } } messages.Add(new Dictionary { ["role"] = "assistant", ["content"] = assistantContent }); if (stopReason == "end_turn" || stopReason == "stop_sequence" || toolResults.Count == 0) break; messages.Add(new Dictionary { ["role"] = "user", ["content"] = toolResults }); } return (finalText.ToString(), iterations); } /// /// Non-Streaming Chat für Telegram. Gibt den finalen Antworttext zurück. /// onIteration wird nach jeder Iteration aufgerufen (z.B. für Typing-Indicator). /// public async Task ChatNonStreaming( ChatContext ctx, List messages, Func onIteration = null) { var sw = System.Diagnostics.Stopwatch.StartNew(); var usage = new TokenUsage(); int iterations = 0; var finalText = new StringBuilder(); var chain = new ChainRecord { ChainId = ctx.ChainId, SessionId = ctx.SessionId, Model = ctx.ModelId }; try { while (iterations < ctx.MaxIterations) { iterations++; onIteration?.Invoke(iterations); var payload = new Dictionary { ["model"] = ctx.ModelId, ["max_tokens"] = 16384, ["cache_control"] = new { type = "ephemeral" }, ["messages"] = messages, ["stream"] = false }; if (ctx.SystemPrompt != null && ctx.SystemPrompt.Count > 0) payload["system"] = ctx.SystemPrompt; AddToolsToPayload(payload, ctx.Workspace, ctx.AgentId, ctx.UseMemory, ctx.WebSearchVersion, ctx.ToolChoice, iterations); var client = CreateClient(); var json = JsonSerializer.Serialize(payload, new JsonSerializerOptions { WriteIndented = true }); LogPayload(json, ctx.SessionId, ctx.ChainId, iterations); var httpContent = new StringContent(json, Encoding.UTF8, "application/json"); using var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }); var respBody = await httpResp.Content.ReadAsStringAsync(); if (!httpResp.IsSuccessStatusCode) { var errMsg = $"API {(int)httpResp.StatusCode}: {respBody}"; await _db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = iterations, Role = "system", Type = "error", Content = errMsg, IsError = true }); await FinalizeChain(chain, usage, iterations, "error", sw); throw new Exception($"Claude API Fehler: {(int)httpResp.StatusCode}"); } var resp = JsonSerializer.Deserialize(respBody); // Usage tracken if (resp.TryGetProperty("usage", out var u)) TrackUsage(usage, u); var stopReason = resp.TryGetProperty("stop_reason", out var sr) ? sr.GetString() ?? "" : ""; var content = resp.GetProperty("content"); var assistantContent = new List(); var toolResults = new List(); foreach (var block in content.EnumerateArray()) { var blockType = block.GetProperty("type").GetString() ?? ""; if (blockType == "text") { var text = block.GetProperty("text").GetString() ?? ""; assistantContent.Add(new { type = "text", text }); finalText.Append(text); await _db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = iterations, Role = "assistant", Type = "text", Content = text }); } else if (blockType == "tool_use") { var toolId = block.GetProperty("id").GetString() ?? ""; var toolName = block.GetProperty("name").GetString() ?? ""; var toolInput = block.GetProperty("input"); assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = toolInput }); await _db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = iterations, Role = "assistant", Type = "tool_use", Content = toolInput.GetRawText(), ToolName = toolName, ToolCallId = toolId }); 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 { type = "tool_result", tool_use_id = toolId, content = TruncateToolResult(toolResult), is_error = isError }); await _db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = iterations, Role = "user", Type = "tool_result", Content = toolResult, ToolName = toolName, ToolCallId = toolId, IsError = isError }); } else if (blockType == "web_search_tool_result") { assistantContent.Add(JsonSerializer.Deserialize(block.GetRawText())); var sources = ExtractWebSearchSources(block); if (sources.Count > 0) { await _db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = iterations, Role = "assistant", Type = "web_search_result", Content = JsonSerializer.Serialize(sources) }); } } else if (blockType == "server_tool_use") { var (toolId, toolName, inputEl) = ParseServerToolUse(block); assistantContent.Add(new { type = "server_tool_use", id = toolId, name = toolName, input = inputEl }); } else if (blockType.EndsWith("_tool_result")) { // Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.) assistantContent.Add(JsonSerializer.Deserialize(block.GetRawText())); await _db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = iterations, Role = "assistant", Type = "server_tool_result", Content = block.GetRawText() }); } } messages.Add(new Dictionary { ["role"] = "assistant", ["content"] = assistantContent }); if (stopReason == "end_turn" || stopReason == "stop_sequence" || toolResults.Count == 0) { await FinalizeChain(chain, usage, iterations, stopReason, sw); break; } messages.Add(new Dictionary { ["role"] = "user", ["content"] = toolResults }); } } catch { if (iterations > 0) await FinalizeChain(chain, usage, iterations, "error", sw); throw; } return finalText.Length > 0 ? finalText.ToString() : "(Keine Textantwort)"; } private void LogPayload(string json, string sessionId, string chainId, int iteration) { if (!_payloadLogging) return; try { var logsDir = Path.Combine(AppContext.BaseDirectory, "data", "logs"); Directory.CreateDirectory(logsDir); var timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds(); var baseName = $"{timestamp}-{sessionId}-{chainId}-iter{iteration}-payload"; if (_payloadLoggingGzip) { var filePath = Path.Combine(logsDir, baseName + ".json.gz"); using var fs = File.Create(filePath); using var gz = new GZipStream(fs, CompressionLevel.Fastest); using var writer = new StreamWriter(gz, Encoding.UTF8); writer.Write(json); } else { File.WriteAllText(Path.Combine(logsDir, baseName + ".json"), json); } } catch { /* Logging darf nie die App crashen */ } } private async Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw, string streamErrorInfo = null) { chain.InputTokens = usage.InputTokens; chain.OutputTokens = usage.OutputTokens; chain.CacheCreationTokens = usage.CacheCreationTokens; chain.CacheReadTokens = usage.CacheReadTokens; chain.Iterations = iterations; chain.StopReason = stopReason; chain.StreamErrorInfo = streamErrorInfo; chain.DurationMs = sw.ElapsedMilliseconds; await _db.UpdateChain(chain); } private async Task<(string StopReason, List Content, List Tools, string StreamErrorInfo)> ProcessStream( HttpResponse response, HttpResponseMessage httpResp, TokenUsage usage, string workspace, string agentId = null, CancellationToken workCt = default, 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 = ""; var assistantContent = new List(); var currentToolId = ""; var currentToolName = ""; var currentToolInputJson = new StringBuilder(); var currentTextBlock = new StringBuilder(); var blockTypes = new Dictionary(); var webSearchResultBlocks = new Dictionary(); var webSearchSources = new Dictionary>(); var serverToolResultBlocks = new Dictionary(); // bash_code_execution_tool_result, code_execution_tool_result etc. var executedTools = new List(); var streamErrors = new List(); string line; while ((line = await reader.ReadLineAsync(workCt)) != null) { if (!line.StartsWith("data: ")) continue; var data = line["data: ".Length..]; if (string.IsNullOrWhiteSpace(data)) continue; JsonElement evt; try { evt = JsonSerializer.Deserialize(data); } catch { continue; } var evtType = evt.TryGetProperty("type", out var t) ? t.GetString() ?? "" : ""; switch (evtType) { case "message_start": { if (evt.TryGetProperty("message", out var msg) && msg.TryGetProperty("usage", out var u)) TrackUsage(usage, u); break; } case "content_block_start": { var idx = evt.GetProperty("index").GetInt32(); var cb = evt.GetProperty("content_block"); var cbType = cb.GetProperty("type").GetString() ?? ""; blockTypes[idx] = cbType; if (cbType == "text") { currentTextBlock.Clear(); await WriteSseEvent(response, "text_start", new { index = idx }, sseCt); } else if (cbType == "tool_use") { if (currentTextBlock.Length > 0) { assistantContent.Add(new { type = "text", text = currentTextBlock.ToString() }); currentTextBlock.Clear(); } currentToolId = cb.GetProperty("id").GetString() ?? ""; currentToolName = cb.GetProperty("name").GetString() ?? ""; currentToolInputJson.Clear(); await WriteSseEvent(response, "tool_start", new { name = currentToolName, id = currentToolId }, sseCt); } else if (cbType == "server_tool_use") { if (currentTextBlock.Length > 0) { assistantContent.Add(new { type = "text", text = currentTextBlock.ToString() }); currentTextBlock.Clear(); } currentToolId = cb.GetProperty("id").GetString() ?? ""; currentToolName = cb.GetProperty("name").GetString() ?? ""; currentToolInputJson.Clear(); await WriteSseEvent(response, "web_search_start", new { name = currentToolName, id = currentToolId }, sseCt); } else if (cbType == "web_search_tool_result") { var rawBlock = JsonSerializer.Deserialize(cb.GetRawText()); webSearchResultBlocks[idx] = rawBlock; webSearchSources[idx] = ExtractWebSearchSources(cb); } else if (cbType.EndsWith("_tool_result")) { // Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.) serverToolResultBlocks[idx] = JsonSerializer.Deserialize(cb.GetRawText()); } break; } case "content_block_delta": { var delta = evt.GetProperty("delta"); var deltaType = delta.GetProperty("type").GetString() ?? ""; if (deltaType == "text_delta") { var text = delta.GetProperty("text").GetString() ?? ""; currentTextBlock.Append(text); await WriteSseEvent(response, "text_delta", new { text }, sseCt); } else if (deltaType == "input_json_delta") { var partial = delta.GetProperty("partial_json").GetString() ?? ""; currentToolInputJson.Append(partial); } break; } case "content_block_stop": { var idx = evt.GetProperty("index").GetInt32(); if (blockTypes.TryGetValue(idx, out var bt)) { if (bt == "text") { var text = currentTextBlock.ToString(); if (text.Length > 0) { assistantContent.Add(new { type = "text", text }); await WriteSseEvent(response, "text_end", new { }, sseCt); } currentTextBlock.Clear(); } else if (bt == "tool_use") { JsonElement inputEl; try { inputEl = JsonSerializer.Deserialize(currentToolInputJson.ToString()); } catch { inputEl = JsonSerializer.Deserialize("{}"); } assistantContent.Add(new { type = "tool_use", id = currentToolId, name = currentToolName, input = inputEl }); // ask_user_choice: NICHT ausführen, sondern als SSE-Event an Client senden if (currentToolName == ToolService.AskUserChoiceToolName) { var question = inputEl.TryGetProperty("question", out var q) ? q.GetString() ?? "" : ""; var choices = new List(); if (inputEl.TryGetProperty("choices", out var c)) { if (c.ValueKind == JsonValueKind.Array) { foreach (var ch in c.EnumerateArray()) choices.Add(ch.GetString() ?? ""); } else if (c.ValueKind == JsonValueKind.String) { // Fallback: Claude hat einen einzelnen String statt Array gesendet choices.Add(c.GetString() ?? ""); } } await WriteSseEvent(response, "user_choice", new { toolCallId = currentToolId, question, choices }, sseCt); // Stream restlich konsumieren ohne weitere Verarbeitung // (Claude sendet nach dem tool_use noch message_delta mit stop_reason) stopReason = "awaiting_user_choice"; continue; } // Cancellation prüfen bevor Tool ausgeführt wird workCt.ThrowIfCancellationRequested(); string toolResult; bool isError = false; try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, workCt); } catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } workCt.ThrowIfCancellationRequested(); executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); await WriteSseEvent(response, "tool_end", new { name = currentToolName, input = currentToolInputJson.ToString(), result = toolResult, isError }, sseCt); } else if (bt == "server_tool_use") { // Server-side tool (Web Search) – Anthropic führt es aus, wir speichern nur die Anfrage JsonElement inputEl; try { inputEl = JsonSerializer.Deserialize(currentToolInputJson.ToString()); } catch { inputEl = JsonSerializer.Deserialize("{}"); } assistantContent.Add(new { type = "server_tool_use", id = currentToolId, name = currentToolName, input = inputEl }); } else if (bt == "web_search_tool_result") { // 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); } else if (bt.EndsWith("_tool_result") && serverToolResultBlocks.TryGetValue(idx, out var srvResultBlock)) { // Generische Server-Tool-Results (bash_code_execution_tool_result, code_execution_tool_result etc.) assistantContent.Add(srvResultBlock); } } break; } case "message_delta": { if (evt.TryGetProperty("delta", out var d)) { if (d.TryGetProperty("stop_reason", out var sr)) { // Nicht überschreiben wenn bereits auf awaiting_user_choice gesetzt if (stopReason != "awaiting_user_choice") stopReason = sr.GetString() ?? ""; // Bei problematischen stop_reasons Debug-Infos sammeln if (stopReason is "unexpected_end" or "max_tokens") { var debugInfo = new Dictionary { ["event"] = "message_delta", ["stop_reason"] = stopReason, ["output_tokens"] = usage.OutputTokens, ["text_length"] = currentTextBlock.Length, ["timestamp"] = DateTime.UtcNow.ToString("o") }; if (d.TryGetProperty("stop_sequence", out var ss) && ss.ValueKind != JsonValueKind.Null) debugInfo["stop_sequence"] = ss.GetString(); streamErrors.Add(debugInfo); } } } if (evt.TryGetProperty("usage", out var u2) && u2.TryGetProperty("output_tokens", out var ot)) usage.OutputTokens += ot.GetInt64(); break; } case "error": { // API Error-Event (z.B. overloaded_error) var errorInfo = new Dictionary { ["event"] = "error", ["timestamp"] = DateTime.UtcNow.ToString("o"), ["raw"] = data }; if (evt.TryGetProperty("error", out var err)) { if (err.TryGetProperty("type", out var errType)) errorInfo["error_type"] = errType.GetString(); if (err.TryGetProperty("message", out var errMsg)) errorInfo["error_message"] = errMsg.GetString(); } streamErrors.Add(errorInfo); break; } } } if (currentTextBlock.Length > 0) assistantContent.Add(new { type = "text", text = currentTextBlock.ToString() }); // Stream-Fehler als JSON serialisieren (null wenn keine Fehler) string streamErrorInfo = streamErrors.Count > 0 ? JsonSerializer.Serialize(streamErrors) : null; return (stopReason, assistantContent, executedTools, streamErrorInfo); } // ── Shared Helpers ── internal const int MaxToolResultChars = 20_000; internal static string TruncateToolResult(string result) { return result.Length > MaxToolResultChars ? result[..MaxToolResultChars] + $"\n\n[... gekürzt, {result.Length - MaxToolResultChars} weitere Zeichen ...]" : result; } /// /// Extrahiert Web-Search-Sources aus einem web_search_tool_result-Block. /// private static List ExtractWebSearchSources(JsonElement block) { var sources = new List(); if (block.TryGetProperty("content", out var searchContent) && searchContent.ValueKind == JsonValueKind.Array) { foreach (var item in searchContent.EnumerateArray()) { var itemType = item.TryGetProperty("type", out var it) ? it.GetString() : ""; if (itemType == "web_search_result") { var url = item.TryGetProperty("url", out var u) ? u.GetString() ?? "" : ""; var title = item.TryGetProperty("title", out var tt) ? tt.GetString() ?? "" : ""; if (!string.IsNullOrEmpty(url)) sources.Add(new { url, title }); } } } return sources; } /// /// Parst einen server_tool_use-Block und gibt (id, name, input) zurück. /// private static (string id, string name, JsonElement input) ParseServerToolUse(JsonElement block) { var toolId = block.GetProperty("id").GetString() ?? ""; var toolName = block.GetProperty("name").GetString() ?? ""; JsonElement inputEl; try { inputEl = block.GetProperty("input"); } catch { inputEl = JsonSerializer.Deserialize("{}"); } return (toolId, toolName, inputEl); } /// /// Aktualisiert den TokenUsage-Tracker aus einem usage-JSON-Element. /// private static void TrackUsage(TokenUsage usage, JsonElement u) { if (u.TryGetProperty("input_tokens", out var it)) usage.InputTokens += it.GetInt64(); if (u.TryGetProperty("output_tokens", out var ot)) usage.OutputTokens += ot.GetInt64(); if (u.TryGetProperty("cache_creation_input_tokens", out var cc)) usage.CacheCreationTokens += cc.GetInt64(); if (u.TryGetProperty("cache_read_input_tokens", out var cr)) usage.CacheReadTokens += cr.GetInt64(); } /// /// Fügt Tool-Definitionen und WebSearch zum Payload hinzu. /// private void AddToolsToPayload(Dictionary payload, string workspace, string agentId, bool useMemory, string webSearchVersion, string toolChoice, int iteration, bool allowUserChoice = false, bool planMode = false) { var hasWorkspace = !string.IsNullOrEmpty(workspace); var includeMemory = useMemory && _memoryService != null && !string.IsNullOrEmpty(agentId); var hasTools = hasWorkspace || includeMemory; var hasWebSearch = !string.IsNullOrEmpty(webSearchVersion); if (!hasTools && !hasWebSearch) return; var tools = new List(); if (hasTools) { var allTools = _toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory, includeUserChoice: allowUserChoice); if (planMode) allTools = _toolService.FilterReadOnlyTools(allTools); tools.AddRange(allTools); } if (hasWebSearch) tools.Add(new { type = webSearchVersion, name = "web_search", max_uses = 5 }); payload["tools"] = tools; var effectiveChoice = (!planMode && toolChoice == "any" && iteration == 1) ? "any" : "auto"; payload["tool_choice"] = new { type = effectiveChoice }; } private HttpClient CreateClient() { var client = _httpFactory.CreateClient(); client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token); client.DefaultRequestHeaders.Add("anthropic-version", "2023-06-01"); client.DefaultRequestHeaders.Add("anthropic-beta", "claude-code-20250219,oauth-2025-04-20"); client.DefaultRequestHeaders.Add("User-Agent", "claude-cli/2.1.2 (external, cli)"); client.DefaultRequestHeaders.Add("x-app", "cli"); client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json")); client.DefaultRequestHeaders.Add("anthropic-dangerous-direct-browser-access", "true"); return client; } public static async Task WriteSseEvent(HttpResponse response, string eventType, object data, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); var json = JsonSerializer.Serialize(data); await response.WriteAsync($"event: {eventType}\ndata: {json}\n\n", ct); await response.Body.FlushAsync(ct); } private static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct) { try { if (!ct.IsCancellationRequested) await WriteSseEvent(response, eventType, data, ct); } catch { // Client bereits disconnected – ignorieren } } } public record ToolResult(string Id, string Result, bool IsError); public class TokenUsage { public long InputTokens { get; set; } public long OutputTokens { get; set; } public long CacheCreationTokens { get; set; } public long CacheReadTokens { get; set; } }