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", ReviewContext review = null, CancellationToken cancellationToken = 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) { cancellationToken.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; 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) { var tools = new List(); if (hasTools) tools.AddRange(_toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory)); if (hasWebSearch) tools.Add(new { type = webSearchVersion, name = "web_search", max_uses = 5 }); payload["tools"] = tools; // "any" erzwingt Tool-Nutzung, aber nur in der ersten Iteration, // damit Claude nicht endlos Tools aufruft wenn die Aufgabe erledigt ist. var effectiveChoice = (toolChoice == "any" && iterations == 1) ? "any" : "auto"; payload["tool_choice"] = new { type = effectiveChoice }; } 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"); var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }, HttpCompletionOption.ResponseHeadersRead, cancellationToken ); if (!httpResp.IsSuccessStatusCode) { var errBody = await httpResp.Content.ReadAsStringAsync(cancellationToken); 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 }, cancellationToken); await FinalizeChain(chain, usage, iterations, "error", sw); await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken); return; } var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, cancellationToken); 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() }); } } // 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, cancellationToken); 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 } } 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 }, cancellationToken); return; } var toolResults = new List(); foreach (var tr in executedTools) { // Tool-Results in der History auf 20.000 Zeichen kürzen. // Das vollständige Ergebnis wurde bereits in der DB gespeichert. // 20k Zeichen ≈ ~5k Tokens – verhindert Token-Explosion bei großen // read_file / execute-Outputs die sich über alle weiteren Iterationen schleppen. const int MaxToolResultChars = 20_000; var content = tr.Result.Length > MaxToolResultChars ? tr.Result[..MaxToolResultChars] + $"\n\n[... gekürzt, {tr.Result.Length - MaxToolResultChars} weitere Zeichen ...]" : tr.Result; toolResults.Add(new { type = "tool_result", tool_use_id = tr.Id, content, is_error = tr.IsError }); } messages.Add(new { role = "user", content = toolResults }); } 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); } catch (OperationCanceledException) { // Client hat die Verbindung getrennt (z.B. Browser-Refresh) await FinalizeChain(chain, usage, iterations, "client_disconnected", 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 }, cancellationToken); await TryWriteSse(response, "done", new { chainId = chain.ChainId }, cancellationToken); } } // ── 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 gesamte Konversation /// und kann Tools nutzen (z.B. read_file um die Arbeit zu prüfen). /// Antwortet er mit [APPROVED] → fertig. Sonst → Feedback wird an den Worker zurückgegeben. /// reviewRound = aktuelle Runde (0-basiert), wird vom Caller hochgezählt. /// private async Task RunReview( HttpResponse response, ReviewContext review, List workerMessages, string chainId, string workspace, string sessionId, TokenUsage usage, int baseIterations, int reviewRound, CancellationToken cancellationToken) { cancellationToken.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" }, cancellationToken); // Review-Agent bekommt eigene Message-History var reviewMessages = BuildReviewMessages(workerMessages); // Review-Agent iteriert mit eigenen Tools var (reviewText, reviewIterations) = await RunReviewAgentChat( response, review, reviewMessages, chainId, workspace, sessionId, usage, baseIterations, cancellationToken); var result = new ReviewResult { ReviewIterationCount = reviewIterations }; // Prüfe ob [APPROVED] if (!string.IsNullOrEmpty(reviewText) && reviewText.Contains("[APPROVED]")) { await TryWriteSse(response, "review_approved", new { round = reviewRound + 1, agentName = review.Agent?.Name ?? "Review-Agent" }, cancellationToken); result.HasFeedback = false; return result; } // Feedback – an Worker zurückspielen result.HasFeedback = true; result.Feedback = $"[Review-Feedback von {review.Agent?.Name ?? "Review-Agent"} (Runde {reviewRound + 1}/{review.MaxReviewIterations})]:\n{reviewText}"; await TryWriteSse(response, "review_feedback", new { round = reviewRound + 1, maxRounds = review.MaxReviewIterations, feedback = reviewText, agentName = review.Agent?.Name ?? "Review-Agent" }, cancellationToken); return result; } /// /// 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) { // Worker-Konversation zusammenfassen var summary = new StringBuilder(); summary.AppendLine("Hier ist die bisherige Konversation zwischen User und Worker-Agent. Bitte überprüfe ob die Aufgabe korrekt erledigt wurde:\n"); summary.AppendLine("---"); 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() ?? ""; if (el.TryGetProperty("content", out var content)) { if (content.ValueKind == JsonValueKind.String) { summary.AppendLine($"\n**{roleStr}:** {content.GetString()}"); } 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() ?? "" : ""; summary.AppendLine($"\n**{roleStr}:** {text}"); } else if (blockType == "tool_use") { var toolName = block.TryGetProperty("name", out var tn) ? tn.GetString() ?? "" : ""; var toolInput = block.TryGetProperty("input", out var ti) ? ti.ToString() : "{}"; summary.AppendLine($"\n**{roleStr} [Tool: {toolName}]:** {toolInput}"); } else if (blockType == "tool_result") { var resultContent = block.TryGetProperty("content", out var rc) ? rc.ToString() : ""; // Tool-Results kürzen für Review-Kontext if (resultContent.Length > 2000) resultContent = resultContent[..2000] + "\n[... gekürzt]"; summary.AppendLine($"\n**{roleStr} [Tool-Result]:** {resultContent}"); } } } } } summary.AppendLine("\n---"); summary.AppendLine("\nBitte überprüfe ob die Aufgabe korrekt erledigt wurde. Du kannst dazu auch Tools verwenden (z.B. Dateien lesen). Antworte mit exakt [APPROVED] wenn alles korrekt ist, oder beschreibe was falsch ist und korrigiert werden muss."); return new List { new Dictionary { ["role"] = "user", ["content"] = new List { new { type = "text", text = summary.ToString() } } } }; } /// /// Führt den Review-Agent Chat aus (non-streaming für den Agent, aber SSE-Events für den Client). /// Gibt den finalen Text und die Anzahl der Iterationen zurück. /// private async Task<(string text, int iterations)> RunReviewAgentChat( HttpResponse response, ReviewContext review, List messages, string chainId, string workspace, string sessionId, TokenUsage usage, int baseIterations, CancellationToken cancellationToken) { int iterations = 0; var finalText = new StringBuilder(); int maxToolIterations = 20; // Review-Agent bekommt max 20 Tool-Iterationen while (iterations < maxToolIterations) { cancellationToken.ThrowIfCancellationRequested(); iterations++; var payload = new Dictionary { ["model"] = review.ModelId, ["max_tokens"] = 8192, ["messages"] = messages, ["stream"] = false }; if (review.SystemPrompt != null && review.SystemPrompt.Count > 0) payload["system"] = review.SystemPrompt; var hasWorkspace = !string.IsNullOrEmpty(workspace); var includeMemory = review.UseMemory && _memoryService != null && !string.IsNullOrEmpty(review.AgentId); var hasTools = hasWorkspace || includeMemory; var hasWebSearch = !string.IsNullOrEmpty(review.WebSearchVersion); if (hasTools || hasWebSearch) { var tools = new List(); if (hasTools) tools.AddRange(_toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory)); if (hasWebSearch) tools.Add(new { type = review.WebSearchVersion, name = "web_search", max_uses = 5 }); payload["tools"] = tools; var effectiveChoice = (review.ToolChoice == "any" && iterations == 1) ? "any" : "auto"; payload["tool_choice"] = new { type = effectiveChoice }; } 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"); var httpResp = await client.SendAsync( new HttpRequestMessage(HttpMethod.Post, "https://api.anthropic.com/v1/messages") { Content = httpContent }, cancellationToken: cancellationToken); var respBody = await httpResp.Content.ReadAsStringAsync(cancellationToken); if (!httpResp.IsSuccessStatusCode) { return ($"Review-Agent API Fehler: {(int)httpResp.StatusCode}: {respBody}", iterations); } var resp = JsonSerializer.Deserialize(respBody); // Usage tracken (auf den gemeinsamen Zähler) if (resp.TryGetProperty("usage", out var 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(); } 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); // Review-Text per SSE an den Client streamen await TryWriteSse(response, "review_text", new { text, agentName = review.Agent?.Name ?? "Review-Agent", agentId = review.AgentId }, cancellationToken); // In DB speichern (als Review-Message markiert) await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "assistant", Type = "text", Content = $"[Review: {review.Agent?.Name ?? "Review-Agent"}] {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 }); // Tool-Use per SSE await TryWriteSse(response, "review_tool_use", new { toolName, toolId, input = toolInput.ToString(), agentName = review.Agent?.Name ?? "Review-Agent" }, cancellationToken); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + 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, workspace, _memoryService, review.AgentId, cancellationToken); } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } const int MaxToolResultChars = 20_000; var truncatedResult = toolResult.Length > MaxToolResultChars ? toolResult[..MaxToolResultChars] + $"\n\n[... gekürzt, {toolResult.Length - MaxToolResultChars} weitere Zeichen ...]" : toolResult; toolResults.Add(new { type = "tool_result", tool_use_id = toolId, content = truncatedResult, is_error = isError }); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = baseIterations + iterations, Role = "user", Type = "tool_result", Content = toolResult, ToolName = toolName, ToolCallId = toolId, IsError = isError }); } } 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, ["messages"] = messages, ["stream"] = false }; if (ctx.SystemPrompt != null && ctx.SystemPrompt.Count > 0) payload["system"] = ctx.SystemPrompt; var hasWorkspace = !string.IsNullOrEmpty(ctx.Workspace); var includeMemory = ctx.UseMemory && _memoryService != null && !string.IsNullOrEmpty(ctx.AgentId); var hasTools = hasWorkspace || includeMemory; var hasWebSearch = !string.IsNullOrEmpty(ctx.WebSearchVersion); if (hasTools || hasWebSearch) { var tools = new List(); if (hasTools) tools.AddRange(_toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory)); if (hasWebSearch) tools.Add(new { type = ctx.WebSearchVersion, name = "web_search", max_uses = 5 }); payload["tools"] = tools; var effectiveChoice = (ctx.ToolChoice == "any" && iterations == 1) ? "any" : "auto"; payload["tool_choice"] = new { type = effectiveChoice }; } 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"); 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)) { 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(); } 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 (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } // Tool-Result in der History kürzen (wie bei StreamChat) const int MaxToolResultChars = 20_000; var truncatedResult = toolResult.Length > MaxToolResultChars ? toolResult[..MaxToolResultChars] + $"\n\n[... gekürzt, {toolResult.Length - MaxToolResultChars} weitere Zeichen ...]" : toolResult; toolResults.Add(new { type = "tool_result", tool_use_id = toolId, content = truncatedResult, 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 }); } } 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 cancellationToken = default) { var stream = await httpResp.Content.ReadAsStreamAsync(cancellationToken); 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 executedTools = new List(); var streamErrors = new List(); string line; while ((line = await reader.ReadLineAsync(cancellationToken)) != 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)) { if (u.TryGetProperty("input_tokens", out var it)) usage.InputTokens += it.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(); } 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 }, cancellationToken); } 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 }, cancellationToken); } 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 }, cancellationToken); } else if (cbType == "web_search_tool_result") { // Gesamten Block als JSON speichern für die Conversation-History var rawBlock = JsonSerializer.Deserialize(cb.GetRawText()); webSearchResultBlocks[idx] = rawBlock; } 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 }, cancellationToken); } 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 { }, cancellationToken); } 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 }); // Cancellation prüfen bevor Tool ausgeführt wird cancellationToken.ThrowIfCancellationRequested(); string toolResult; bool isError = false; try { toolResult = await _toolService.ExecuteTool(currentToolName, inputEl, workspace, _memoryService, agentId, cancellationToken); } catch (OperationCanceledException) { throw; } catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } cancellationToken.ThrowIfCancellationRequested(); executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); await WriteSseEvent(response, "tool_end", new { name = currentToolName, input = currentToolInputJson.ToString(), result = toolResult.Length > 500 ? toolResult[..500] + "..." : toolResult, isError }, cancellationToken); } 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]); await WriteSseEvent(response, "web_search_end", new { id = currentToolId }, cancellationToken); } } break; } case "message_delta": { if (evt.TryGetProperty("delta", out var d)) { if (d.TryGetProperty("stop_reason", out var sr)) { 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); } 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; } }