From 1c622abb3780b2415cbf099db2bc685bc52c3460 Mon Sep 17 00:00:00 2001 From: Meleki Date: Wed, 18 Mar 2026 14:54:39 +0100 Subject: [PATCH] refactor: AnthropicService partial-class Aufteilung + ToolPayloadOptions + Non-Streaming Deduplizierung MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AnthropicService.cs (337 Zeilen): Felder, Konstruktor, Setters, Utilities - AnthropicService.Streaming.cs (750 Zeilen): StreamChat, ProcessStream, SSE-Parsing - AnthropicService.Agents.cs (680 Zeilen): RunAgentChat, ChatNonStreaming, Review/Preparer/Finalizer - AnthropicService.Cli.cs (530 Zeilen): StreamChatViaCli, MCP-Config, CLI-Helpers - AddToolsToPayload: 12 Bool-Parameter → ToolPayloadOptions Record - RunAgentChat + ChatNonStreaming → gemeinsame RunNonStreamingToolLoop-Methode Co-Authored-By: Claude Opus 4.6 --- ...XECUTE.md => [DONE] MCP_REMOTE_EXECUTE.md} | 0 Services/AnthropicService.Agents.cs | 680 ++++++ Services/AnthropicService.Cli.cs | 530 ++++ Services/AnthropicService.Streaming.cs | 750 ++++++ Services/AnthropicService.cs | 2171 +---------------- 5 files changed, 2020 insertions(+), 2111 deletions(-) rename Plans/{MCP_REMOTE_EXECUTE.md => [DONE] MCP_REMOTE_EXECUTE.md} (100%) create mode 100644 Services/AnthropicService.Agents.cs create mode 100644 Services/AnthropicService.Cli.cs create mode 100644 Services/AnthropicService.Streaming.cs diff --git a/Plans/MCP_REMOTE_EXECUTE.md b/Plans/[DONE] MCP_REMOTE_EXECUTE.md similarity index 100% rename from Plans/MCP_REMOTE_EXECUTE.md rename to Plans/[DONE] MCP_REMOTE_EXECUTE.md diff --git a/Services/AnthropicService.Agents.cs b/Services/AnthropicService.Agents.cs new file mode 100644 index 0000000..2cb2be3 --- /dev/null +++ b/Services/AnthropicService.Agents.cs @@ -0,0 +1,680 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Net.Http; +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 partial class AnthropicService +{ + // ── 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, + string apiToken, + 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 SseHelper.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(); + string text; + int iterations; + + if (!string.IsNullOrEmpty(preparer.Provider) && _openAiService != null) + { + // OpenAI-kompatibler Preparer + var subCtx = BuildSubAgentChatContext(preparer, chainId, sessionId, apiToken); + text = await _openAiService.ChatNonStreaming(subCtx, preparerMessages); + iterations = 1; + } + else + { + (text, iterations) = await RunAgentChat( + response, preparer.ModelId, preparer.SystemPrompt, preparer.AgentId, + preparer.Workspace, preparer.UseMemory, preparer.WebSearchVersion, + preparer.ToolChoice, preparerMessages, chainId, sessionId, apiToken, 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 SseHelper.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 SseHelper.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, + string apiToken, + TokenUsage usage, + int baseIterations, + int reviewRound, + CancellationToken workCt, + CancellationToken sseCt) + { + await SseHelper.TryWriteSse(response, "review_start", new + { + agentId = review.AgentId, + agentName = review.Agent?.Name ?? "Review-Agent", + round = reviewRound + }, sseCt); + + // Review-Agent bekommt die Worker-Konversation als Kontext + var reviewMessages = BuildReviewMessages(workerMessages); + + string text; + int iterations; + + if (!string.IsNullOrEmpty(review.Provider) && _openAiService != null) + { + // OpenAI-kompatibler Reviewer + var subCtx = BuildSubAgentChatContext(review, chainId, sessionId, apiToken); + text = await _openAiService.ChatNonStreaming(subCtx, reviewMessages); + iterations = 1; + } + else + { + (text, iterations) = await RunAgentChat( + response, review.ModelId, review.SystemPrompt, review.AgentId, + review.Workspace, review.UseMemory, review.WebSearchVersion, + review.ToolChoice, reviewMessages, chainId, sessionId, apiToken, usage, + baseIterations, "review", review.Agent?.Name ?? "Review-Agent", + workCt, sseCt); + } + + var result = new ReviewResult { ReviewIterationCount = iterations }; + + if (!string.IsNullOrEmpty(text) && text.Contains("[FEEDBACK]")) + { + var feedbackText = text.Substring(text.IndexOf("[FEEDBACK]") + "[FEEDBACK]".Length).Trim(); + result.HasFeedback = true; + result.Feedback = $"[Review-Feedback, Runde {reviewRound + 1}]: {feedbackText}"; + } + + await SseHelper.TryWriteSse(response, "review_done", new + { + agentName = review.Agent?.Name ?? "Review-Agent", + round = reviewRound, + hasFeedback = result.HasFeedback + }, sseCt); + + return result; + } + + /// + /// 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, + string apiToken, + TokenUsage usage, + int baseIterations, + CancellationToken workCt, + CancellationToken sseCt) + { + await SseHelper.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); + + string text; + int iterations; + + if (!string.IsNullOrEmpty(finalizer.Provider) && _openAiService != null) + { + var subCtx = BuildSubAgentChatContext(finalizer, chainId, sessionId, apiToken); + text = await _openAiService.ChatNonStreaming(subCtx, finalizerMessages); + iterations = 1; + } + else + { + (text, iterations) = await RunAgentChat( + response, finalizer.ModelId, finalizer.SystemPrompt, finalizer.AgentId, + finalizer.Workspace, finalizer.UseMemory, finalizer.WebSearchVersion, + finalizer.ToolChoice, finalizerMessages, chainId, sessionId, apiToken, usage, + baseIterations, "finalizer", finalizer.Agent?.Name ?? "Finalizer-Agent", + workCt, sseCt); + } + + await SseHelper.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() ?? ""; + + if (!el.TryGetProperty("content", out var content)) continue; + + string textContent = null; + + if (content.ValueKind == JsonValueKind.String) + { + textContent = content.GetString(); + } + else if (content.ValueKind == JsonValueKind.Array) + { + var parts = new List(); + foreach (var block in content.EnumerateArray()) + { + var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : ""; + if (blockType == "text" && block.TryGetProperty("text", out var txt)) + parts.Add(txt.GetString() ?? ""); + } + if (parts.Count > 0) + textContent = string.Join("\n", parts); + } + + if (string.IsNullOrEmpty(textContent)) continue; + + if (roleStr == lastRole) + { + // Gleiche Rolle → Text anhängen + currentParts.Add(textContent); + } + else + { + // Rolle wechselt → vorherige Message abschließen + if (lastRole != null && currentParts.Count > 0) + { + messages.Add(new + { + role = lastRole, + content = string.Join("\n\n", currentParts) + }); + } + lastRole = roleStr; + currentParts = new List { textContent }; + } + } + + // Letzte Message abschließen + if (lastRole != null && currentParts.Count > 0) + { + messages.Add(new + { + role = lastRole, + content = string.Join("\n\n", currentParts) + }); + } + + // Sicherstellen, dass die Messages mit "user" beginnen + if (messages.Count > 0) + { + var firstJson = JsonSerializer.Serialize(messages[0]); + var firstEl = JsonSerializer.Deserialize(firstJson); + if (firstEl.GetProperty("role").GetString() != "user") + { + messages.Insert(0, new { role = "user", content = "Bitte überprüfe die folgende Konversation und die Arbeit des Agents." }); + } + } + else + { + messages.Add(new { role = "user", content = "Es gibt keine Konversation zum Überprüfen." }); + } + + return messages; + } + + /// + /// Baut einen ChatContext für Sub-Agents (Preparer/Review/Finalizer) auf, + /// wenn diese einen OpenAI-kompatiblen Provider nutzen. + /// + private ChatContext BuildSubAgentChatContext(SubAgentContext sub, string chainId, string sessionId, string apiToken) + { + // SystemPrompt als Text extrahieren (SubAgent hat List mit { type, text } Blöcken) + var systemText = ""; + if (sub.SystemPrompt != null) + { + var parts = new List(); + foreach (var block in sub.SystemPrompt) + { + var json = JsonSerializer.Serialize(block); + var el = JsonSerializer.Deserialize(json); + if (el.TryGetProperty("text", out var t)) + parts.Add(t.GetString() ?? ""); + } + systemText = string.Join("\n\n", parts); + } + + return new ChatContext + { + ChainId = chainId, + SessionId = sessionId, + ModelId = sub.ModelId, + Provider = sub.Provider, + Workspace = sub.Workspace, + AgentId = sub.AgentId, + SystemPromptText = systemText, + UseMemory = sub.UseMemory, + WebSearchVersion = sub.WebSearchVersion, + MaxIterations = 20, + ApiToken = apiToken + }; + } + + // ── Shared Non-Streaming Tool Loop ── + + /// + /// Optionale Callbacks für RunNonStreamingToolLoop. + /// + private class NonStreamingLoopCallbacks + { + /// Wird zu Beginn jeder Iteration aufgerufen (z.B. Typing-Indicator). + public Func OnIteration { get; set; } + + /// SSE-Response für Client-Events. Null = keine SSE-Events. + public HttpResponse SseResponse { get; set; } + public CancellationToken SseCt { get; set; } + + /// SSE-Event-Prefix (z.B. "review" → "review_text", "review_tool"). + public string SsePrefix { get; set; } + public string AgentName { get; set; } + + /// Tag für DB-Records (z.B. "review", "finalizer"). + public string Tag { get; set; } + + /// Ob Text-Embeddings in die Queue geschrieben werden. + public bool EnqueueEmbeddings { get; set; } + + /// + /// Optionaler Tool-Executor. Parameter: (toolName, input, ct) → result. + /// Bei Fehler Exception werfen. Wenn null, wird Standard-Executor verwendet. + /// + public Func> ExecuteTool { get; set; } + } + + /// + /// Gemeinsamer Non-Streaming Tool-Loop für RunAgentChat und ChatNonStreaming. + /// Baut Payload, sendet API-Request, verarbeitet Content-Blöcke, führt Tools aus. + /// systemPrompt kann List<object> (Agent) oder string (ChatContext) sein. + /// + private async Task<(string text, int iterations)> RunNonStreamingToolLoop( + string modelId, + object systemPrompt, + ToolPayloadOptions toolOpts, + List messages, + string chainId, + string sessionId, + string apiToken, + TokenUsage usage, + int baseIteration, + int maxIterations, + CancellationToken ct, + NonStreamingLoopCallbacks cb = null) + { + cb ??= new NonStreamingLoopCallbacks(); + int iterations = 0; + var finalText = new StringBuilder(); + + while (iterations < maxIterations) + { + ct.ThrowIfCancellationRequested(); + iterations++; + if (cb.OnIteration != null) await cb.OnIteration(iterations); + + var payload = new Dictionary + { + ["model"] = modelId, + ["max_tokens"] = 16384, + ["messages"] = messages + }; + + if (systemPrompt is List spBlocks && spBlocks.Count > 0) + payload["system"] = spBlocks; + else if (systemPrompt is string spText && !string.IsNullOrEmpty(spText)) + payload["system"] = spText; + + AddToolsToPayload(payload, toolOpts with { Iteration = iterations }); + + var client = CreateClient(apiToken); + var json = JsonSerializer.Serialize(payload); + LogPayload(json, sessionId, chainId, baseIteration + 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, + ct + ); + + TryUpdateKeyMeta(httpResp, apiToken); + + if (!httpResp.IsSuccessStatusCode) + { + var errBody = await httpResp.Content.ReadAsStringAsync(ct); + var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}"; + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = baseIteration + iterations, + Role = "system", Type = "error", + Content = errMsg, IsError = true, Tag = cb.Tag + }); + if (cb.SseResponse != null) + await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_error", new { error = errMsg }, cb.SseCt); + break; + } + + var respJson = await httpResp.Content.ReadAsStringAsync(ct); + var respDoc = JsonSerializer.Deserialize(respJson); + + if (respDoc.TryGetProperty("usage", out var u)) + TrackUsage(usage, u); + + var stopReason = respDoc.TryGetProperty("stop_reason", out var sr) ? sr.GetString() ?? "" : ""; + var contentArr = respDoc.TryGetProperty("content", out var c) ? c : default; + if (contentArr.ValueKind != JsonValueKind.Array) break; + + var assistantContent = new List(); + var toolResults = new List(); + + foreach (var block in contentArr.EnumerateArray()) + { + var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : ""; + if (blockType == "text") + { + var text = block.GetProperty("text").GetString() ?? ""; + finalText.Append(text); + assistantContent.Add(new { type = "text", text }); + + var msgId = await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = baseIteration + iterations, + Role = "assistant", Type = "text", + Content = text, Tag = cb.Tag + }); + if (cb.EnqueueEmbeddings) + _embeddingBg?.Enqueue(msgId, chainId, sessionId, toolOpts.AgentId, "assistant", text); + if (cb.SseResponse != null) + await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_text", new { text, agentName = cb.AgentName }, cb.SseCt); + } + else if (blockType == "tool_use") + { + var toolId = block.GetProperty("id").GetString() ?? ""; + var toolName = block.GetProperty("name").GetString() ?? ""; + JsonElement inputEl; + try { inputEl = block.GetProperty("input"); } + catch { inputEl = JsonSerializer.Deserialize("{}"); } + + assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = inputEl }); + + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = baseIteration + iterations, + Role = "assistant", Type = "tool_use", + ToolName = toolName, ToolCallId = toolId, + Content = inputEl.ToString(), Tag = cb.Tag + }); + + ct.ThrowIfCancellationRequested(); + + string toolResult; + bool isError = false; + try + { + toolResult = cb.ExecuteTool != null + ? await cb.ExecuteTool(toolName, inputEl, ct) + : await _toolService.ExecuteTool(toolName, inputEl, toolOpts.Workspace, _memoryService, toolOpts.AgentId, ct, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg); + } + catch (OperationCanceledException) { throw; } + catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } + + ct.ThrowIfCancellationRequested(); + + toolResults.Add(new + { + type = "tool_result", + tool_use_id = toolId, + content = TruncateToolResult(toolResult), + is_error = isError + }); + + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = baseIteration + iterations, + Role = "user", Type = "tool_result", + ToolCallId = toolId, + Content = toolResult, IsError = isError, Tag = cb.Tag + }); + + if (cb.SseResponse != null) + await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_tool", new { name = toolName, result = TruncateToolResult(toolResult), isError, agentName = cb.AgentName }, cb.SseCt); + } + } + + messages.Add(new { role = "assistant", content = assistantContent }); + + if (stopReason != "tool_use" || toolResults.Count == 0) + break; + + messages.Add(new { role = "user", content = toolResults }); + } + + return (finalText.ToString(), iterations); + } + + /// + /// Agent-Chat (non-streaming API, SSE-Events für Client). + /// Wrapper um RunNonStreamingToolLoop für Review-/Finalizer-Agents. + /// + private 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, + string apiToken, + TokenUsage usage, + int baseIterations, + string agentType, + string agentName, + CancellationToken workCt, + CancellationToken sseCt) + { + return RunNonStreamingToolLoop( + modelId, systemPrompt, + new ToolPayloadOptions + { + Workspace = agentWorkspace, AgentId = agentId, + UseMemory = useMemory, WebSearchVersion = webSearchVersion, + ToolChoice = agentToolChoice + }, + messages, chainId, sessionId, apiToken, usage, + baseIterations, 20, workCt, + new NonStreamingLoopCallbacks + { + SseResponse = response, SseCt = sseCt, + SsePrefix = agentType, AgentName = agentName, + Tag = agentType + }); + } + + /// + /// Non-Streaming Chat für Telegram. Gibt den finalen Antworttext zurück. + /// Wrapper um RunNonStreamingToolLoop mit Chain-Management und Embedding-Enqueue. + /// 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(); + + var chain = new ChainRecord + { + ChainId = ctx.ChainId, + SessionId = ctx.SessionId, + Model = ctx.ModelId + }; + + try + { + var (text, iterations) = await RunNonStreamingToolLoop( + ctx.ModelId, ctx.SystemPromptText, + new ToolPayloadOptions + { + Workspace = ctx.Workspace, AgentId = ctx.AgentId, + UseMemory = ctx.UseMemory, WebSearchVersion = ctx.WebSearchVersion, + ToolChoice = "auto" + }, + messages, ctx.ChainId, ctx.SessionId, ctx.ApiToken, usage, + 0, ctx.MaxIterations, CancellationToken.None, + new NonStreamingLoopCallbacks + { + OnIteration = onIteration, + EnqueueEmbeddings = true, + ExecuteTool = (name, input, ct) => + _toolService.ExecuteTool(name, input, ctx.Workspace, _memoryService, ctx.AgentId, ct, + vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db, + chatService: _chatService, anthropicService: this, openAiService: _openAiService) + }); + + var finalStopReason = text.Length > 0 ? "end_turn" : "error"; + await FinalizeChain(chain, usage, iterations, finalStopReason, sw); + return text; + } + catch (Exception ex) + { + await _db.SaveMessage(new MessageRecord + { + ChainId = ctx.ChainId, Iteration = 0, + Role = "system", Type = "error", + Content = $"Fehler: {ex.Message}", IsError = true + }); + await FinalizeChain(chain, usage, 0, "error", sw); + return ""; + } + } +} diff --git a/Services/AnthropicService.Cli.cs b/Services/AnthropicService.Cli.cs new file mode 100644 index 0000000..3faad3c --- /dev/null +++ b/Services/AnthropicService.Cli.cs @@ -0,0 +1,530 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Gott.Endpoints; +using Gott.Models; +using Microsoft.AspNetCore.Http; + +namespace Gott.Services; + +public partial class AnthropicService +{ + /// + /// Streaming-Chat über die Claude CLI. + /// Die CLI verwendet das lokale Token aus ~/.claude/.credentials.json. + /// + public async Task StreamChatViaCli( + HttpResponse response, + string modelId, + List systemPrompt, + List messages, + string chainId, + string workspace = null, + bool allowUserChoice = false, + bool planMode = false, + bool webSearch = false, + string sessionId = null, + string agentId = null, + RemoteConnectionManager remoteMgr = null, + CancellationToken workCt = default, + CancellationToken sseCt = default) + { + if (_cliService == null || !_cliService.IsAvailable) + { + await SseHelper.TryWriteSse(response, "error", new { error = "Claude CLI nicht verfügbar" }, sseCt); + await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt); + return; + } + + var sw = System.Diagnostics.Stopwatch.StartNew(); + var usage = new TokenUsage(); + var fullText = new StringBuilder(); + var tempImagePaths = new List(); + string mcpConfigPath = null; + bool isUserChoice = false; + + var chain = new ChainRecord + { + ChainId = chainId, + SessionId = sessionId, + Model = modelId + }; + + try + { + // MCP-Config generieren (immer – für ask_user_choice, optional Memory-Tools + remote_execute) + var hasRemote = remoteMgr != null && !string.IsNullOrEmpty(sessionId) && remoteMgr.GetBySessionId(sessionId) != null; + mcpConfigPath = GenerateMcpConfig(sessionId, agentId, hasRemote); + + // System-Prompt als Text extrahieren + var systemText = new StringBuilder(); + if (systemPrompt != null) + { + foreach (var block in systemPrompt) + { + var json = JsonSerializer.Serialize(block); + var el = JsonSerializer.Deserialize(json); + if (el.TryGetProperty("text", out var t)) + systemText.AppendLine(t.GetString()); + } + } + + + // Letzte User-Message als Prompt extrahieren + var prompt = ExtractLastUserText(messages) ?? ""; + + // Bild-Attachments aus letzter User-Message als Temp-Dateien speichern + // (CLI kann Bilder nicht inline empfangen, aber per Read-Tool lesen) + var imageRefs = ExtractLastUserImageRefs(messages, tempImagePaths); + if (!string.IsNullOrEmpty(imageRefs)) + { + prompt = imageRefs + "\n" + prompt; + } + + // Konversations-History als Kontext zum Prompt hinzufügen + var historyContext = BuildHistoryContext(messages); + if (!string.IsNullOrEmpty(historyContext)) + { + prompt = $"\n{historyContext}\n\n\n\n{prompt}\n"; + } + + string choiceToolCallId = null; + + var result = await _cliService.ChatStreamingAsync( + prompt, + async evt => + { + switch (evt.Type) + { + case CliEventType.TextStart: + await SseHelper.TryWriteSse(response, "text_start", new { index = 0 }, sseCt); + break; + case CliEventType.TextDelta: + fullText.Append(evt.Text); + await SseHelper.TryWriteSse(response, "text_delta", new { text = evt.Text }, sseCt); + break; + case CliEventType.TextEnd: + await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt); + break; + case CliEventType.ToolStart: + { + // Text-Puffer flushen bevor Tool-Use gespeichert wird (korrekte Reihenfolge in DB) + if (fullText.Length > 0) + { + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "assistant", + Type = "text", + Content = fullText.ToString() + }); + fullText.Clear(); + } + var mappedName = MapCliToolName(evt.ToolName); + await SseHelper.TryWriteSse(response, "tool_start", new { name = mappedName, id = evt.ToolId }, sseCt); + break; + } + case CliEventType.ToolEnd: + { + var mappedName = MapCliToolName(evt.ToolName); + var mappedInput = MapCliToolInput(evt.ToolName, evt.ToolInput); + // Tool-Use in DB speichern (damit Poll die Tools anzeigt) + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "assistant", + Type = "tool_use", + ToolName = mappedName, + ToolCallId = evt.ToolId, + Content = mappedInput + }); + await SseHelper.TryWriteSse(response, "tool_end", new { name = mappedName, id = evt.ToolId, input = mappedInput }, sseCt); + + // ask_user_choice via MCP: TCS erstellen + SSE user_choice Event + if (mappedName == "ask_user_choice") + { + try + { + var choiceInput = JsonSerializer.Deserialize(mappedInput); + var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : ""; + var cChoices = new List(); + if (choiceInput.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array) + foreach (var cOpt in cc.EnumerateArray()) + cChoices.Add(cOpt.GetString() ?? ""); + + // Alte Antwort entfernen (falls vorhanden) + InternalEndpoint.ChoiceAnswers.TryRemove(sessionId, out _); + + // SSE Event mit viaCli-Flag → Frontend zeigt Buttons + nutzt choice-respond Endpoint + choiceToolCallId = evt.ToolId; + await SseHelper.TryWriteSse(response, "user_choice", new + { + toolCallId = evt.ToolId, + question = cQuestion, + choices = cChoices, + viaCli = true + }, sseCt); + + isUserChoice = true; + } + catch (JsonException) { /* Input nicht parsbar – MCP-Server gibt Fehler */ } + } + break; + } + case CliEventType.ToolResult: + { + // Tool-Result in DB speichern (damit es beim Neuladen sichtbar ist) + var resultContent = SseHelper.TruncateToolResult(evt.ToolResultContent ?? ""); + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "user", + Type = "tool_result", + ToolCallId = evt.ToolId, + Content = resultContent, + IsError = evt.ToolResultIsError + }); + // SSE Event an Client senden + await SseHelper.TryWriteSse(response, "tool_result", new + { + id = evt.ToolId, + result = resultContent, + isError = evt.ToolResultIsError + }, sseCt); + break; + } + } + }, + modelId: modelId, + systemPrompt: systemText.Length > 0 ? systemText.ToString() : null, + workspace: workspace, + hasWorkspace: !string.IsNullOrEmpty(workspace), + planMode: planMode, + webSearch: webSearch, + mcpConfigPath: mcpConfigPath, + cancellationToken: workCt + ); + + // Bei CLI-Fehler → Fehlermeldung an Client senden und in DB speichern + if (!result.Success && !isUserChoice) + { + var errorMsg = !string.IsNullOrEmpty(result.Error) ? result.Error : "CLI hat keine Antwort geliefert"; + // Error in DB speichern (damit er beim Neuladen sichtbar ist) + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "system", + Type = "error", + Content = $"CLI-Fehler: {errorMsg}" + }); + await SseHelper.TryWriteSse(response, "error", new { error = $"CLI-Fehler: {errorMsg}" }, sseCt); + } + + // Antwort in DB speichern + if (fullText.Length > 0) + { + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "assistant", + Type = "text", + Content = fullText.ToString() + }); + } + + // Token-Usage vom CLI-Result übernehmen + usage.InputTokens = result.InputTokens; + usage.OutputTokens = result.OutputTokens; + usage.CacheReadTokens = result.CacheReadTokens; + usage.CacheCreationTokens = result.CacheCreationTokens; + + var stopReason = isUserChoice ? "awaiting_user_choice" : (result.Success ? "end_turn" : "error"); + await FinalizeChain(chain, usage, result.NumTurns > 0 ? result.NumTurns : 1, stopReason, sw); + + await SseHelper.TryWriteSse(response, "done", new + { + model = modelId, + inputTokens = result.InputTokens, + outputTokens = result.OutputTokens, + cacheCreationTokens = result.CacheCreationTokens, + cacheReadTokens = result.CacheReadTokens, + costUsd = result.CostUsd, + iterations = result.NumTurns > 0 ? result.NumTurns : 1, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason, + viaCli = true + }, sseCt); + } + catch (OperationCanceledException) when (workCt.IsCancellationRequested) + { + // Stop-Button gedrückt → CLI-Prozess wurde gestoppt + // Wenn ask_user_choice aktiv war: als awaiting_user_choice markieren, + // damit die nächste Chain den Kontext (Frage + Optionen) behält. + var stoppedReason = isUserChoice ? "awaiting_user_choice" : "stopped"; + await FinalizeChain(chain, usage, 1, stoppedReason, sw); + await SseHelper.TryWriteSse(response, "done", new + { + model = modelId, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason = stoppedReason, + viaCli = true + }, CancellationToken.None); + } + catch (OperationCanceledException) + { + // Client-Disconnect (sseCt gecancelled, aber NICHT workCt) + // CLI-Prozess läuft weiter im Hintergrund – hier nur Chain finalisieren + // (sollte nicht mehr passieren, da CLI jetzt workCt nutzt) + await FinalizeChain(chain, usage, 1, "client_disconnected", sw); + } + catch (Exception ex) + { + var errorMsg = $"CLI-Fehler: {ex.Message}"; + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "system", + Type = "error", + Content = errorMsg + }); + await SseHelper.TryWriteSse(response, "error", new { error = errorMsg }, sseCt); + await FinalizeChain(chain, usage, 1, "error", sw); + await SseHelper.TryWriteSse(response, "done", new + { + model = modelId, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason = "error", + viaCli = true + }, sseCt); + } + finally + { + // Temp-Bilddateien aufräumen + CleanupTempImages(tempImagePaths); + + // MCP-Config Temp-Datei aufräumen + if (mcpConfigPath != null) + { + try { File.Delete(mcpConfigPath); } catch { } + } + } + } + + /// + /// Mappt CLI-Tool-Namen (Edit, Write, Bash, etc.) auf gott-Tool-Namen (edit_file, write_file, etc.). + /// + private static string MapCliToolName(string cliName) + { + return cliName switch + { + "Edit" => "edit_file", + "Write" => "write_file", + "Read" => "read_file", + "Bash" => "execute", + "Glob" => "list_files", + "Grep" => "search_files", + "WebSearch" => "web_search", + "WebFetch" => "web_fetch", + "NotebookEdit" => "notebook_edit", + "mcp__gott-remote__remote_execute" => "remote_execute", + _ => cliName?.StartsWith("mcp__") == true ? cliName.Split("__").LastOrDefault() ?? cliName : cliName + }; + } + + /// + /// Mappt CLI-Tool-Input-Felder auf gott-Konventionen (z.B. file_path→path, old_string→old_text). + /// + private static string MapCliToolInput(string cliToolName, string inputJson) + { + if (string.IsNullOrEmpty(inputJson)) return inputJson; + + try + { + var input = JsonSerializer.Deserialize(inputJson); + if (input.ValueKind != JsonValueKind.Object) return inputJson; + + switch (cliToolName) + { + case "Edit": + { + // file_path → path, old_string → old_text, new_string → new_text + var mapped = new Dictionary(); + foreach (var prop in input.EnumerateObject()) + { + var key = prop.Name switch + { + "file_path" => "path", + "old_string" => "old_text", + "new_string" => "new_text", + _ => prop.Name + }; + mapped[key] = prop.Value; + } + return JsonSerializer.Serialize(mapped); + } + case "Write": + { + // file_path → path + var mapped = new Dictionary(); + foreach (var prop in input.EnumerateObject()) + { + var key = prop.Name == "file_path" ? "path" : prop.Name; + mapped[key] = prop.Value; + } + return JsonSerializer.Serialize(mapped); + } + case "Read": + { + // file_path → path + var mapped = new Dictionary(); + foreach (var prop in input.EnumerateObject()) + { + var key = prop.Name == "file_path" ? "path" : prop.Name; + mapped[key] = prop.Value; + } + return JsonSerializer.Serialize(mapped); + } + case "Bash": + { + // command bleibt gleich + return inputJson; + } + default: + return inputJson; + } + } + catch + { + return inputJson; + } + } + + /// + /// Baut einen Kontext-String aus der Konversations-History. + /// + private static string BuildHistoryContext(List messages) + { + var sb = new StringBuilder(); + // Letzte Message überspringen wenn es ein Text-Prompt ist (wird separat als current_request gesendet). + // Wenn die letzte Message ein tool_result ist (z.B. Choice-Antwort), NICHT überspringen. + int takeCount = messages.Count; + if (messages.Count > 0) + { + var lastJson = JsonSerializer.Serialize(messages[messages.Count - 1]); + var lastEl = JsonSerializer.Deserialize(lastJson); + bool lastIsToolResult = false; + if (lastEl.TryGetProperty("content", out var lc) && lc.ValueKind == JsonValueKind.Array) + { + foreach (var b in lc.EnumerateArray()) + { + var bType = b.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : ""; + if (bType == "tool_result") { lastIsToolResult = true; break; } + } + } + if (!lastIsToolResult) + takeCount = messages.Count - 1; + } + foreach (var msg in messages.Take(takeCount)) + { + 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) + { + sb.AppendLine($"[{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" && block.TryGetProperty("text", out var txt)) + { + sb.AppendLine($"[{roleStr}]: {txt.GetString()}"); + } + else if (blockType == "tool_use" && block.TryGetProperty("name", out var tn) && tn.GetString() == "ask_user_choice") + { + if (block.TryGetProperty("input", out var inp)) + { + var q = inp.TryGetProperty("question", out var qq) ? qq.GetString() ?? "" : ""; + var opts = new List(); + if (inp.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array) + foreach (var o in cc.EnumerateArray()) opts.Add(o.GetString() ?? ""); + sb.AppendLine($"[assistant]: Ich habe gefragt: \"{q}\" mit Optionen: {string.Join(", ", opts)}"); + } + } + else if (blockType == "tool_result" && block.TryGetProperty("content", out var trContent)) + { + sb.AppendLine($"[user]: Du hast gewählt: \"{trContent.GetString()}\""); + } + } + } + } + } + return sb.ToString().Trim(); + } + + /// + /// Generiert eine temporäre MCP-Config-Datei für die Claude CLI. + /// Enthält den MCP-Server für remote_execute. + /// + private string GenerateMcpConfig(string sessionId, string agentId = null, bool hasRemote = false) + { + // Port aus Konfiguration oder Default + var port = 19000; + var scriptPath = Path.Combine(AppContext.BaseDirectory, "scripts", "mcp-remote-execute.py"); + + // Fallback: Quell-Verzeichnis prüfen (für Dev-Modus) + if (!File.Exists(scriptPath)) + { + var projectRoot = Path.GetDirectoryName(AppPaths.WwwRoot) ?? AppContext.BaseDirectory; + var srcPath = Path.Combine(projectRoot, "scripts", "mcp-remote-execute.py"); + if (File.Exists(srcPath)) scriptPath = srcPath; + } + + var env = new Dictionary + { + ["GOTT_API_URL"] = $"http://localhost:{port}", + ["GOTT_SESSION_ID"] = sessionId ?? "", + ["GOTT_AUTH_TOKEN"] = Endpoints.InternalEndpoint.InternalToken, + ["GOTT_AGENT_ID"] = agentId ?? "", + ["GOTT_HAS_REMOTE"] = hasRemote ? "true" : "false" + }; + + var config = new + { + mcpServers = new Dictionary + { + ["gott-remote"] = new + { + command = "python3", + args = new[] { scriptPath }, + env + } + } + }; + + var json = System.Text.Json.JsonSerializer.Serialize(config, new System.Text.Json.JsonSerializerOptions { WriteIndented = true }); + var tempPath = Path.Combine(Path.GetTempPath(), $"gott-mcp-{sessionId}.json"); + File.WriteAllText(tempPath, json); + return tempPath; + } +} diff --git a/Services/AnthropicService.Streaming.cs b/Services/AnthropicService.Streaming.cs new file mode 100644 index 0000000..3064bdb --- /dev/null +++ b/Services/AnthropicService.Streaming.cs @@ -0,0 +1,750 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Net.Http; +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 partial class AnthropicService +{ + public async Task StreamChat( + HttpResponse response, + string modelId, + List systemPrompt, + List messages, + string chainId, + string workspace, + string apiToken, + string sessionId = null, + string agentId = null, + string userId = 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, + bool useAutoContext = true, + bool useGitHubSearch = false, + bool useSerperSearch = false, + CancellationToken workCt = default, + CancellationToken hardCt = default, + CancellationToken sseCt = default, + RemoteConnectionManager remoteMgr = null) + { + var sw = System.Diagnostics.Stopwatch.StartNew(); + var usage = new TokenUsage(); + int iterations = 0; + int reviewRound = 0; + string lastStopReason = ""; + int promptTrimRetries = 0; + + var chain = new ChainRecord + { + ChainId = chainId, + SessionId = sessionId, + Model = modelId + }; + + try + { + // ── Automatische Kontext-Injektion via Embedding-Suche ── + if (useAutoContext && _vectorSearch != null && systemPrompt != null && messages.Count > 0) + { + try + { + // Letzte User-Message als Query extrahieren + var lastUserText = ExtractLastUserText(messages); + if (!string.IsNullOrWhiteSpace(lastUserText)) + { + var contextResults = await _vectorSearch.SearchAsync(lastUserText, agentId, topN: 5, minScore: 0.45f); + // Nur Ergebnisse aus anderen Sessions (nicht aktuelle Chain) und nicht zu kurze + contextResults.RemoveAll(r => r.ChainId == chainId || (r.TextPreview?.Length ?? 0) < 20); + // Memory-Dateien durchsuchen + var memoryResults = await _vectorSearch.SearchMemoryFilesAsync(lastUserText, agentId, topN: 3, minScore: 0.45f); + + if (contextResults.Count > 0 || memoryResults.Count > 0) + { + var ctxSb = new StringBuilder(); + ctxSb.AppendLine("\n"); + ctxSb.AppendLine("Die folgende Information wurde automatisch abgerufen und könnte relevant sein:"); + foreach (var r in contextResults) + { + var date = DateTime.TryParse(r.CreatedAt, out var dt) ? dt.ToString("yyyy-MM-dd HH:mm") : r.CreatedAt; + ctxSb.AppendLine($"- [Gespräch, {r.Role}, {date}]: {r.TextPreview}"); + } + foreach (var r in memoryResults) + { + ctxSb.AppendLine($"- [Memory, {r.FilePath}]: {r.TextPreview}"); + } + ctxSb.AppendLine(""); + systemPrompt.Add(new { type = "text", text = ctxSb.ToString() }); + } + } + } + catch (Exception ex) + { + Console.WriteLine($"[AutoContext] Error: {ex.Message}"); + } + } + + 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; + + var hasRemote = remoteMgr?.GetBySessionId(sessionId) != null; + AddToolsToPayload(payload, new ToolPayloadOptions + { + Workspace = workspace, + AgentId = agentId, + UseMemory = useMemory, + WebSearchVersion = webSearchVersion, + ToolChoice = toolChoice, + Iteration = iterations, + AllowUserChoice = allowUserChoice, + PlanMode = planMode, + IncludeRemote = hasRemote, + IncludeGitHubSearch = useGitHubSearch && _toolService.HasGitHubToken, + IncludeSerperSearch = useSerperSearch && _toolService.HasSerperKey + }); + + var client = CreateClient(apiToken); + 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 + ); + + TryUpdateKeyMeta(httpResp, apiToken); + + if (!httpResp.IsSuccessStatusCode) + { + var errBody = await httpResp.Content.ReadAsStringAsync(workCt); + var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}"; + + // Auto-Retry bei "prompt too long" / "exceed context limit": History kürzen und erneut versuchen + if ((int)httpResp.StatusCode == 400 && + (errBody.Contains("prompt is too long") || errBody.Contains("exceed context limit")) && + messages.Count > 4 && promptTrimRetries < 5) + { + promptTrimRetries++; + int trimCount = Math.Max(2, messages.Count / 5); // 20% der Messages entfernen + trimCount = TrimMessagesCleanly(messages, trimCount); + if (trimCount > 0) + { + await SseHelper.TryWriteSse(response, "info", new { message = $"History zu lang – {trimCount} ältere Messages entfernt, Retry ({promptTrimRetries}/5)..." }, sseCt); + iterations--; // Iteration nicht zählen, war ja kein echter Versuch + continue; // → nächste while-Iteration mit kürzerer History + } + } + + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = iterations, + Role = "system", Type = "error", + Content = errMsg, IsError = true + }); + await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt); + await FinalizeChain(chain, usage, iterations, "error", sw); + await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); + return; + } + + var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, workCt, sseCt, remoteMgr, sessionId, userId, apiToken); + lastStopReason = stopReason; + promptTrimRetries = 0; // Erfolgreicher API-Call → Retry-Counter resetten + + 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") + { + var textContent = el.GetProperty("text").GetString(); + var msgId = await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = iterations, + Role = "assistant", Type = "text", + Content = textContent + }); + _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "assistant", textContent ?? ""); + } + 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") + { + // Raw-JSON speichern für History-Rebuild (API braucht den vollständigen Block) + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = iterations, + Role = "assistant", Type = "web_search_result", + Content = el.GetRawText() + }); + } + 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, apiToken, 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, apiToken, usage, iterations, workCt, sseCt); + iterations += finalizerIterations; + } + + var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason; + await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo); + await SseHelper.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 SseHelper.TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt); + await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); + } + catch (OperationCanceledException) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) + { + // Stop-Button gedrückt (workCt oder hardCt wurde gecancelled) + await FinalizeChain(chain, usage, iterations, "stopped", sw); + // sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None + // Der SSE-Stream zum Client ist noch offen – nur der Stop-Button hat ausgelöst + await SseHelper.TryWriteSse(response, "done", new + { + model = modelId, + inputTokens = usage.InputTokens, + outputTokens = usage.OutputTokens, + cacheCreationTokens = usage.CacheCreationTokens, + cacheReadTokens = usage.CacheReadTokens, + iterations, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason = "stopped", + planMode + }, CancellationToken.None); + } + catch (Exception) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) + { + // 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); + // sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None + await SseHelper.TryWriteSse(response, "done", new + { + model = modelId, + inputTokens = usage.InputTokens, + outputTokens = usage.OutputTokens, + cacheCreationTokens = usage.CacheCreationTokens, + cacheReadTokens = usage.CacheReadTokens, + iterations, + chainId, + durationMs = sw.ElapsedMilliseconds, + stopReason = "stopped", + planMode + }, CancellationToken.None); + } + catch (OperationCanceledException) + { + // Client-Disconnect (cancellationToken gecancelled, aber NICHT workCt/hardCt) + // Agent-Arbeit ist leider schon abgebrochen – Chain als "client_disconnected" finalisieren + // (sollte durch TryWriteSse eigentlich nicht mehr passieren) + await FinalizeChain(chain, usage, iterations, "client_disconnected", sw); + } + catch (Exception ex) + { + // 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 SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt); + await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); + } + } + + 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, + RemoteConnectionManager remoteMgr = null, + string sessionId = null, + string userId = null, + string apiToken = null) + { + 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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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, remoteMgr: remoteMgr, sessionId: sessionId, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db, chatService: _chatService, anthropicService: this, openAiService: _openAiService, userId: userId, telegramService: _telegramService, apiToken: apiToken); } + catch (OperationCanceledException) { throw; } + catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } + + workCt.ThrowIfCancellationRequested(); + executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); + + await SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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); + } + + /// + /// 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(); + } +} diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index a56ec8c..61f1c62 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -15,8 +15,26 @@ using Microsoft.AspNetCore.Http; namespace Gott.Services; -public class AnthropicService +public partial class AnthropicService { + /// + /// Optionen für AddToolsToPayload – ersetzt die 12 bool-Parameter. + /// + public record ToolPayloadOptions + { + public string Workspace { get; init; } + public string AgentId { get; init; } + public bool UseMemory { get; init; } + public string WebSearchVersion { get; init; } + public string ToolChoice { get; init; } + public int Iteration { get; init; } + public bool AllowUserChoice { get; init; } + public bool PlanMode { get; init; } + public bool IncludeRemote { get; init; } + public bool IncludeGitHubSearch { get; init; } + public bool IncludeSerperSearch { get; init; } + } + private readonly IHttpClientFactory _httpFactory; private readonly ToolService _toolService; private readonly DatabaseService _db; @@ -84,1545 +102,6 @@ public class AnthropicService catch { /* silent */ } } - public async Task StreamChat( - HttpResponse response, - string modelId, - List systemPrompt, - List messages, - string chainId, - string workspace, - string apiToken, - string sessionId = null, - string agentId = null, - string userId = 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, - bool useAutoContext = true, - bool useGitHubSearch = false, - bool useSerperSearch = false, - CancellationToken workCt = default, - CancellationToken hardCt = default, - CancellationToken sseCt = default, - RemoteConnectionManager remoteMgr = null) - { - var sw = System.Diagnostics.Stopwatch.StartNew(); - var usage = new TokenUsage(); - int iterations = 0; - int reviewRound = 0; - string lastStopReason = ""; - int promptTrimRetries = 0; - - var chain = new ChainRecord - { - ChainId = chainId, - SessionId = sessionId, - Model = modelId - }; - - try - { - // ── Automatische Kontext-Injektion via Embedding-Suche ── - if (useAutoContext && _vectorSearch != null && systemPrompt != null && messages.Count > 0) - { - try - { - // Letzte User-Message als Query extrahieren - var lastUserText = ExtractLastUserText(messages); - if (!string.IsNullOrWhiteSpace(lastUserText)) - { - var contextResults = await _vectorSearch.SearchAsync(lastUserText, agentId, topN: 5, minScore: 0.45f); - // Nur Ergebnisse aus anderen Sessions (nicht aktuelle Chain) und nicht zu kurze - contextResults.RemoveAll(r => r.ChainId == chainId || (r.TextPreview?.Length ?? 0) < 20); - // Memory-Dateien durchsuchen - var memoryResults = await _vectorSearch.SearchMemoryFilesAsync(lastUserText, agentId, topN: 3, minScore: 0.45f); - - if (contextResults.Count > 0 || memoryResults.Count > 0) - { - var ctxSb = new StringBuilder(); - ctxSb.AppendLine("\n"); - ctxSb.AppendLine("Die folgende Information wurde automatisch abgerufen und könnte relevant sein:"); - foreach (var r in contextResults) - { - var date = DateTime.TryParse(r.CreatedAt, out var dt) ? dt.ToString("yyyy-MM-dd HH:mm") : r.CreatedAt; - ctxSb.AppendLine($"- [Gespräch, {r.Role}, {date}]: {r.TextPreview}"); - } - foreach (var r in memoryResults) - { - ctxSb.AppendLine($"- [Memory, {r.FilePath}]: {r.TextPreview}"); - } - ctxSb.AppendLine(""); - systemPrompt.Add(new { type = "text", text = ctxSb.ToString() }); - } - } - } - catch (Exception ex) - { - Console.WriteLine($"[AutoContext] Error: {ex.Message}"); - } - } - - 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; - - var hasRemote = remoteMgr?.GetBySessionId(sessionId) != null; - AddToolsToPayload(payload, workspace, agentId, useMemory, webSearchVersion, toolChoice, iterations, allowUserChoice, planMode, includeRemote: hasRemote, includeGitHubSearch: useGitHubSearch && _toolService.HasGitHubToken, includeSerperSearch: useSerperSearch && _toolService.HasSerperKey); - - var client = CreateClient(apiToken); - 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 - ); - - TryUpdateKeyMeta(httpResp, apiToken); - - if (!httpResp.IsSuccessStatusCode) - { - var errBody = await httpResp.Content.ReadAsStringAsync(workCt); - var errMsg = $"API {(int)httpResp.StatusCode}: {errBody}"; - - // Auto-Retry bei "prompt too long" / "exceed context limit": History kürzen und erneut versuchen - if ((int)httpResp.StatusCode == 400 && - (errBody.Contains("prompt is too long") || errBody.Contains("exceed context limit")) && - messages.Count > 4 && promptTrimRetries < 5) - { - promptTrimRetries++; - int trimCount = Math.Max(2, messages.Count / 5); // 20% der Messages entfernen - trimCount = TrimMessagesCleanly(messages, trimCount); - if (trimCount > 0) - { - await SseHelper.TryWriteSse(response, "info", new { message = $"History zu lang – {trimCount} ältere Messages entfernt, Retry ({promptTrimRetries}/5)..." }, sseCt); - iterations--; // Iteration nicht zählen, war ja kein echter Versuch - continue; // → nächste while-Iteration mit kürzerer History - } - } - - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, Iteration = iterations, - Role = "system", Type = "error", - Content = errMsg, IsError = true - }); - await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt); - await FinalizeChain(chain, usage, iterations, "error", sw); - await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); - return; - } - - var (stopReason, assistantContent, executedTools, streamErrorInfo) = await ProcessStream(response, httpResp, usage, workspace, agentId, workCt, sseCt, remoteMgr, sessionId, userId, apiToken); - lastStopReason = stopReason; - promptTrimRetries = 0; // Erfolgreicher API-Call → Retry-Counter resetten - - 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") - { - var textContent = el.GetProperty("text").GetString(); - var msgId = await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, Iteration = iterations, - Role = "assistant", Type = "text", - Content = textContent - }); - _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "assistant", textContent ?? ""); - } - 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") - { - // Raw-JSON speichern für History-Rebuild (API braucht den vollständigen Block) - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, Iteration = iterations, - Role = "assistant", Type = "web_search_result", - Content = el.GetRawText() - }); - } - 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, apiToken, 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, apiToken, usage, iterations, workCt, sseCt); - iterations += finalizerIterations; - } - - var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason; - await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo); - await SseHelper.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 SseHelper.TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt); - await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt); - } - catch (OperationCanceledException) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) - { - // Stop-Button gedrückt (workCt oder hardCt wurde gecancelled) - await FinalizeChain(chain, usage, iterations, "stopped", sw); - // sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None - // Der SSE-Stream zum Client ist noch offen – nur der Stop-Button hat ausgelöst - await SseHelper.TryWriteSse(response, "done", new - { - model = modelId, - inputTokens = usage.InputTokens, - outputTokens = usage.OutputTokens, - cacheCreationTokens = usage.CacheCreationTokens, - cacheReadTokens = usage.CacheReadTokens, - iterations, - chainId, - durationMs = sw.ElapsedMilliseconds, - stopReason = "stopped", - planMode - }, CancellationToken.None); - } - catch (Exception) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested) - { - // 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); - // sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None - await SseHelper.TryWriteSse(response, "done", new - { - model = modelId, - inputTokens = usage.InputTokens, - outputTokens = usage.OutputTokens, - cacheCreationTokens = usage.CacheCreationTokens, - cacheReadTokens = usage.CacheReadTokens, - iterations, - chainId, - durationMs = sw.ElapsedMilliseconds, - stopReason = "stopped", - planMode - }, CancellationToken.None); - } - catch (OperationCanceledException) - { - // Client-Disconnect (cancellationToken gecancelled, aber NICHT workCt/hardCt) - // Agent-Arbeit ist leider schon abgebrochen – Chain als "client_disconnected" finalisieren - // (sollte durch TryWriteSse eigentlich nicht mehr passieren) - await FinalizeChain(chain, usage, iterations, "client_disconnected", sw); - } - catch (Exception ex) - { - // 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 SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt); - await SseHelper.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, - string apiToken, - 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 SseHelper.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(); - string text; - int iterations; - - if (!string.IsNullOrEmpty(preparer.Provider) && _openAiService != null) - { - // OpenAI-kompatibler Preparer - var subCtx = BuildSubAgentChatContext(preparer, chainId, sessionId, apiToken); - text = await _openAiService.ChatNonStreaming(subCtx, preparerMessages); - iterations = 1; - } - else - { - (text, iterations) = await RunAgentChat( - response, preparer.ModelId, preparer.SystemPrompt, preparer.AgentId, - preparer.Workspace, preparer.UseMemory, preparer.WebSearchVersion, - preparer.ToolChoice, preparerMessages, chainId, sessionId, apiToken, 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 SseHelper.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 SseHelper.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, - string apiToken, - TokenUsage usage, - int baseIterations, - int reviewRound, - CancellationToken workCt, - CancellationToken sseCt) - { - workCt.ThrowIfCancellationRequested(); - - // SSE: Review-Phase Start - await SseHelper.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); - - string reviewText; - int reviewIterations; - - if (!string.IsNullOrEmpty(review.Provider) && _openAiService != null) - { - var subCtx = BuildSubAgentChatContext(review, chainId, sessionId, apiToken); - reviewText = await _openAiService.ChatNonStreaming(subCtx, reviewMessages); - reviewIterations = 1; - } - else - { - (reviewText, reviewIterations) = await RunAgentChat( - response, review.ModelId, review.SystemPrompt, review.AgentId, - workspace, review.UseMemory, review.WebSearchVersion, - review.ToolChoice, reviewMessages, chainId, sessionId, apiToken, 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 SseHelper.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 SseHelper.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, - string apiToken, - TokenUsage usage, - int baseIterations, - CancellationToken workCt, - CancellationToken sseCt) - { - await SseHelper.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); - - string text; - int iterations; - - if (!string.IsNullOrEmpty(finalizer.Provider) && _openAiService != null) - { - var subCtx = BuildSubAgentChatContext(finalizer, chainId, sessionId, apiToken); - text = await _openAiService.ChatNonStreaming(subCtx, finalizerMessages); - iterations = 1; - } - else - { - (text, iterations) = await RunAgentChat( - response, finalizer.ModelId, finalizer.SystemPrompt, finalizer.AgentId, - finalizer.Workspace, finalizer.UseMemory, finalizer.WebSearchVersion, - finalizer.ToolChoice, finalizerMessages, chainId, sessionId, apiToken, usage, - baseIterations, "finalizer", finalizer.Agent?.Name ?? "Finalizer-Agent", - workCt, sseCt); - } - - await SseHelper.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; - } - - /// - /// Baut einen ChatContext für Sub-Agents (Preparer/Review/Finalizer) auf, - /// wenn diese einen OpenAI-kompatiblen Provider nutzen. - /// - private ChatContext BuildSubAgentChatContext(SubAgentContext sub, string chainId, string sessionId, string apiToken) - { - // SystemPrompt als Text extrahieren (SubAgent hat List mit { type, text } Blöcken) - var systemText = ""; - if (sub.SystemPrompt != null) - { - var parts = new List(); - foreach (var block in sub.SystemPrompt) - { - var json = JsonSerializer.Serialize(block); - var el = JsonSerializer.Deserialize(json); - if (el.TryGetProperty("text", out var t)) - parts.Add(t.GetString() ?? ""); - } - systemText = string.Join("\n\n", parts); - } - - return new ChatContext - { - ChainId = chainId, - SessionId = sessionId, - ModelId = sub.ModelId, - Provider = sub.Provider, - Workspace = sub.Workspace, - AgentId = sub.AgentId, - SystemPromptText = systemText, - UseMemory = sub.UseMemory, - WebSearchVersion = sub.WebSearchVersion, - MaxIterations = 20, - ApiToken = apiToken - }; - } - - /// - /// 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, - string apiToken, - 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(apiToken); - 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); - - TryUpdateKeyMeta(httpResp, apiToken); - - var respBody = await httpResp.Content.ReadAsStringAsync(workCt); - - if (!httpResp.IsSuccessStatusCode) - { - // Auto-Retry bei "prompt too long" in Sub-Agents - if ((int)httpResp.StatusCode == 400 && - (respBody.Contains("prompt is too long") || respBody.Contains("exceed context limit")) && - messages.Count > 4) - { - int trimCount = Math.Max(2, messages.Count / 5); - trimCount = TrimMessagesCleanly(messages, trimCount); - if (trimCount > 0) - { - await SseHelper.TryWriteSse(response, "info", new { message = $"{tag}: History zu lang – {trimCount} Messages entfernt, Retry..." }, sseCt); - iterations--; - continue; - } - } - 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 SseHelper.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 SseHelper.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, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db, chatService: _chatService, anthropicService: this, openAiService: _openAiService, telegramService: _telegramService, apiToken: apiToken); } - 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 SseHelper.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())); - // Raw-JSON speichern für History-Rebuild (API braucht den vollständigen Block) - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = baseIterations + iterations, - Role = "assistant", - Type = "web_search_result", - Content = block.GetRawText(), - 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(ctx.ApiToken); - 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 }); - - TryUpdateKeyMeta(httpResp, ctx.ApiToken); - - var respBody = await httpResp.Content.ReadAsStringAsync(); - - if (!httpResp.IsSuccessStatusCode) - { - // Auto-Retry bei "prompt too long" in Non-Streaming - if ((int)httpResp.StatusCode == 400 && - (respBody.Contains("prompt is too long") || respBody.Contains("exceed context limit")) && - messages.Count > 4) - { - int trimCount = Math.Max(2, messages.Count / 5); - trimCount = TrimMessagesCleanly(messages, trimCount); - if (trimCount > 0) - { - iterations--; - continue; - } - } - - 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, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db, chatService: _chatService, anthropicService: this, openAiService: _openAiService, userId: ctx.Session?.UserId, telegramService: _telegramService, apiToken: ctx.ApiToken); } - 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())); - // Raw-JSON speichern für History-Rebuild (API braucht den vollständigen Block) - await _db.SaveMessage(new MessageRecord - { - ChainId = ctx.ChainId, Iteration = iterations, - Role = "assistant", Type = "web_search_result", - Content = block.GetRawText() - }); - } - 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 (_settings == null || !_settings.GetBool("PayloadLogging", false)) 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 (_settings?.GetBool("PayloadLoggingGzip", false) ?? false) - { - 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 Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw, string streamErrorInfo = null) - => SseHelper.FinalizeChain(_db, chain, usage, iterations, stopReason, sw, streamErrorInfo); - - 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, - RemoteConnectionManager remoteMgr = null, - string sessionId = null, - string userId = null, - string apiToken = null) - { - 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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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, remoteMgr: remoteMgr, sessionId: sessionId, vectorSearch: _vectorSearch, embeddingBg: _embeddingBg, db: _db, chatService: _chatService, anthropicService: this, openAiService: _openAiService, userId: userId, telegramService: _telegramService, apiToken: apiToken); } - catch (OperationCanceledException) { throw; } - catch (Exception ex) { toolResult = $"Fehler: {ex.Message}"; isError = true; } - - workCt.ThrowIfCancellationRequested(); - executedTools.Add(new ToolResult(currentToolId, toolResult, isError)); - - await SseHelper.TryWriteSse(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 SseHelper.TryWriteSse(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 ── // TruncateToolResult → SseHelper.cs @@ -1667,21 +146,28 @@ public class AnthropicService { var json = JsonSerializer.Serialize(messages[i]); var el = JsonSerializer.Deserialize(json); + if (!el.TryGetProperty("role", out var role) || role.GetString() != "user") continue; - if (!el.TryGetProperty("content", out var content) || content.ValueKind != JsonValueKind.Array) break; + if (!el.TryGetProperty("content", out var content) || content.ValueKind != JsonValueKind.Array) continue; var refs = new List(); + var tempDir = Path.Combine(Path.GetTempPath(), "gott-cli-images"); + Directory.CreateDirectory(tempDir); + foreach (var block in content.EnumerateArray()) { - var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() : ""; - if (blockType != "image" && blockType != "document") continue; + var blockType = block.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : ""; + if (blockType != "image" && blockType != "file") continue; + + var source = block.TryGetProperty("source", out var src) ? src : default; + if (source.ValueKind == JsonValueKind.Undefined) continue; + + var sourceType = source.TryGetProperty("type", out var st) ? st.GetString() ?? "" : ""; + if (sourceType != "base64") continue; - if (!block.TryGetProperty("source", out var source)) continue; - var mediaType = source.TryGetProperty("media_type", out var mt) ? mt.GetString() ?? "" : ""; var data = source.TryGetProperty("data", out var d) ? d.GetString() ?? "" : ""; - if (string.IsNullOrEmpty(data)) continue; + var mediaType = source.TryGetProperty("media_type", out var mt) ? mt.GetString() ?? "" : ""; - // Dateiendung aus media_type ableiten var ext = mediaType switch { "image/png" => ".png", @@ -1691,11 +177,6 @@ public class AnthropicService "application/pdf" => ".pdf", _ => ".bin" }; - - // Temp-Verzeichnis sicherstellen - var tempDir = Path.Combine(Path.GetTempPath(), "gott-attachments"); - Directory.CreateDirectory(tempDir); - var fileName = $"{Guid.NewGuid():N}{ext}"; var filePath = Path.Combine(tempDir, fileName); @@ -1734,128 +215,64 @@ public class AnthropicService } } - /// - /// Generiert eine temporäre MCP-Config-Datei für die Claude CLI. - /// Enthält den MCP-Server für remote_execute. - /// - private string GenerateMcpConfig(string sessionId, string agentId = null, bool hasRemote = false) + private void LogPayload(string json, string sessionId, string chainId, int iteration) { - // Port aus Konfiguration oder Default - var port = 19000; - var scriptPath = Path.Combine(AppContext.BaseDirectory, "scripts", "mcp-remote-execute.py"); - - // Fallback: Quell-Verzeichnis prüfen (für Dev-Modus) - if (!File.Exists(scriptPath)) + if (_settings == null || !_settings.GetBool("PayloadLogging", false)) return; + try { - var projectRoot = Path.GetDirectoryName(AppPaths.WwwRoot) ?? AppContext.BaseDirectory; - var srcPath = Path.Combine(projectRoot, "scripts", "mcp-remote-execute.py"); - if (File.Exists(srcPath)) scriptPath = srcPath; - } + var logsDir = Path.Combine(AppContext.BaseDirectory, "data", "logs"); + Directory.CreateDirectory(logsDir); + var timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds(); + var baseName = $"{timestamp}-{sessionId}-{chainId}-iter{iteration}-payload"; - var env = new Dictionary - { - ["GOTT_API_URL"] = $"http://localhost:{port}", - ["GOTT_SESSION_ID"] = sessionId ?? "", - ["GOTT_AUTH_TOKEN"] = Endpoints.InternalEndpoint.InternalToken, - ["GOTT_AGENT_ID"] = agentId ?? "", - ["GOTT_HAS_REMOTE"] = hasRemote ? "true" : "false" - }; - - var config = new - { - mcpServers = new Dictionary + if (_settings?.GetBool("PayloadLoggingGzip", false) ?? false) { - ["gott-remote"] = new - { - command = "python3", - args = new[] { scriptPath }, - env - } + 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); } - }; - - var json = System.Text.Json.JsonSerializer.Serialize(config, new System.Text.Json.JsonSerializerOptions { WriteIndented = true }); - var tempPath = Path.Combine(Path.GetTempPath(), $"gott-mcp-{sessionId}.json"); - File.WriteAllText(tempPath, json); - return tempPath; - } - - /// - /// 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()) + else { - 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 }); - } + File.WriteAllText(Path.Combine(logsDir, baseName + ".json"), json); } } - return sources; + catch { /* Logging darf nie die App crashen */ } } - /// - /// 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(); - } + private Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw, string streamErrorInfo = null) + => SseHelper.FinalizeChain(_db, chain, usage, iterations, stopReason, sw, streamErrorInfo); /// /// 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, bool includeRemote = false, bool includeGitHubSearch = false, bool includeSerperSearch = false) + private void AddToolsToPayload(Dictionary payload, ToolPayloadOptions opts) { - var hasWorkspace = !string.IsNullOrEmpty(workspace); - var includeMemory = useMemory && _memoryService != null && !string.IsNullOrEmpty(agentId); + var hasWorkspace = !string.IsNullOrEmpty(opts.Workspace); + var includeMemory = opts.UseMemory && _memoryService != null && !string.IsNullOrEmpty(opts.AgentId); var includeMemorySearch = _vectorSearch != null && _vectorSearch.GetEmbeddingCount() > 0; var includeSendToSession = _db != null && _chatService != null; - var hasTools = hasWorkspace || includeMemory || includeRemote || includeMemorySearch || includeSendToSession || includeGitHubSearch || includeSerperSearch; - var hasWebSearch = !string.IsNullOrEmpty(webSearchVersion); + var hasTools = hasWorkspace || includeMemory || opts.IncludeRemote || includeMemorySearch || includeSendToSession || opts.IncludeGitHubSearch || opts.IncludeSerperSearch; + var hasWebSearch = !string.IsNullOrEmpty(opts.WebSearchVersion); if (!hasTools && !hasWebSearch) return; var tools = new List(); if (hasTools) { - var allTools = _toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory, includeUserChoice: allowUserChoice, includeRemote: includeRemote, includeMemorySearch: includeMemorySearch, includeSendToSession: includeSendToSession, includeGitHubSearch: includeGitHubSearch, includeSerperSearch: includeSerperSearch); - if (planMode) + var allTools = _toolService.GetToolDefinitions(includeFileTools: hasWorkspace, includeMemory: includeMemory, includeUserChoice: opts.AllowUserChoice, includeRemote: opts.IncludeRemote, includeMemorySearch: includeMemorySearch, includeSendToSession: includeSendToSession, includeGitHubSearch: opts.IncludeGitHubSearch, includeSerperSearch: opts.IncludeSerperSearch); + if (opts.PlanMode) allTools = _toolService.FilterReadOnlyTools(allTools); tools.AddRange(allTools); } if (hasWebSearch) { - tools.Add(new { type = webSearchVersion, name = "web_search", max_uses = 5 }); + tools.Add(new { type = opts.WebSearchVersion, name = "web_search", max_uses = 5 }); // web_search_20260209: Code Execution wird von der API automatisch injected – NICHT manuell hinzufügen! } payload["tools"] = tools; - var effectiveChoice = (!planMode && toolChoice == "any" && iteration == 1) ? "any" : "auto"; + var effectiveChoice = (!opts.PlanMode && opts.ToolChoice == "any" && opts.Iteration == 1) ? "any" : "auto"; payload["tool_choice"] = new { type = effectiveChoice }; } @@ -1917,472 +334,4 @@ public class AnthropicService } // WriteSseEvent, TryWriteSse, TokenUsage, ToolResult → SseHelper.cs - - /// - /// Streaming-Chat über die Claude CLI. - /// Die CLI verwendet das lokale Token aus ~/.claude/.credentials.json. - /// - public async Task StreamChatViaCli( - HttpResponse response, - string modelId, - List systemPrompt, - List messages, - string chainId, - string workspace = null, - bool allowUserChoice = false, - bool planMode = false, - bool webSearch = false, - string sessionId = null, - string agentId = null, - RemoteConnectionManager remoteMgr = null, - CancellationToken workCt = default, - CancellationToken sseCt = default) - { - if (_cliService == null || !_cliService.IsAvailable) - { - await SseHelper.TryWriteSse(response, "error", new { error = "Claude CLI nicht verfügbar" }, sseCt); - await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt); - return; - } - - var sw = System.Diagnostics.Stopwatch.StartNew(); - var usage = new TokenUsage(); - var fullText = new StringBuilder(); - var tempImagePaths = new List(); - string mcpConfigPath = null; - bool isUserChoice = false; - - var chain = new ChainRecord - { - ChainId = chainId, - SessionId = sessionId, - Model = modelId - }; - - try - { - // MCP-Config generieren (immer – für ask_user_choice, optional Memory-Tools + remote_execute) - var hasRemote = remoteMgr != null && !string.IsNullOrEmpty(sessionId) && remoteMgr.GetBySessionId(sessionId) != null; - mcpConfigPath = GenerateMcpConfig(sessionId, agentId, hasRemote); - - // System-Prompt als Text extrahieren - var systemText = new StringBuilder(); - if (systemPrompt != null) - { - foreach (var block in systemPrompt) - { - var json = JsonSerializer.Serialize(block); - var el = JsonSerializer.Deserialize(json); - if (el.TryGetProperty("text", out var t)) - systemText.AppendLine(t.GetString()); - } - } - - - // Letzte User-Message als Prompt extrahieren - var prompt = ExtractLastUserText(messages) ?? ""; - - // Bild-Attachments aus letzter User-Message als Temp-Dateien speichern - // (CLI kann Bilder nicht inline empfangen, aber per Read-Tool lesen) - var imageRefs = ExtractLastUserImageRefs(messages, tempImagePaths); - if (!string.IsNullOrEmpty(imageRefs)) - { - prompt = imageRefs + "\n" + prompt; - } - - // Konversations-History als Kontext zum Prompt hinzufügen - var historyContext = BuildHistoryContext(messages); - if (!string.IsNullOrEmpty(historyContext)) - { - prompt = $"\n{historyContext}\n\n\n\n{prompt}\n"; - } - - string choiceToolCallId = null; - - var result = await _cliService.ChatStreamingAsync( - prompt, - async evt => - { - switch (evt.Type) - { - case CliEventType.TextStart: - await SseHelper.TryWriteSse(response, "text_start", new { index = 0 }, sseCt); - break; - case CliEventType.TextDelta: - fullText.Append(evt.Text); - await SseHelper.TryWriteSse(response, "text_delta", new { text = evt.Text }, sseCt); - break; - case CliEventType.TextEnd: - await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt); - break; - case CliEventType.ToolStart: - { - // Text-Puffer flushen bevor Tool-Use gespeichert wird (korrekte Reihenfolge in DB) - if (fullText.Length > 0) - { - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "assistant", - Type = "text", - Content = fullText.ToString() - }); - fullText.Clear(); - } - var mappedName = MapCliToolName(evt.ToolName); - await SseHelper.TryWriteSse(response, "tool_start", new { name = mappedName, id = evt.ToolId }, sseCt); - break; - } - case CliEventType.ToolEnd: - { - var mappedName = MapCliToolName(evt.ToolName); - var mappedInput = MapCliToolInput(evt.ToolName, evt.ToolInput); - // Tool-Use in DB speichern (damit Poll die Tools anzeigt) - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "assistant", - Type = "tool_use", - ToolName = mappedName, - ToolCallId = evt.ToolId, - Content = mappedInput - }); - await SseHelper.TryWriteSse(response, "tool_end", new { name = mappedName, id = evt.ToolId, input = mappedInput }, sseCt); - - // ask_user_choice via MCP: TCS erstellen + SSE user_choice Event - if (mappedName == "ask_user_choice") - { - try - { - var choiceInput = JsonSerializer.Deserialize(mappedInput); - var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : ""; - var cChoices = new List(); - if (choiceInput.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array) - foreach (var cOpt in cc.EnumerateArray()) - cChoices.Add(cOpt.GetString() ?? ""); - - // Alte Antwort entfernen (falls vorhanden) - InternalEndpoint.ChoiceAnswers.TryRemove(sessionId, out _); - - // SSE Event mit viaCli-Flag → Frontend zeigt Buttons + nutzt choice-respond Endpoint - choiceToolCallId = evt.ToolId; - await SseHelper.TryWriteSse(response, "user_choice", new - { - toolCallId = evt.ToolId, - question = cQuestion, - choices = cChoices, - viaCli = true - }, sseCt); - - isUserChoice = true; - } - catch (JsonException) { /* Input nicht parsbar – MCP-Server gibt Fehler */ } - } - break; - } - case CliEventType.ToolResult: - { - // Tool-Result in DB speichern (damit es beim Neuladen sichtbar ist) - var resultContent = SseHelper.TruncateToolResult(evt.ToolResultContent ?? ""); - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "user", - Type = "tool_result", - ToolCallId = evt.ToolId, - Content = resultContent, - IsError = evt.ToolResultIsError - }); - // SSE Event an Client senden - await SseHelper.TryWriteSse(response, "tool_result", new - { - id = evt.ToolId, - result = resultContent, - isError = evt.ToolResultIsError - }, sseCt); - break; - } - } - }, - modelId: modelId, - systemPrompt: systemText.Length > 0 ? systemText.ToString() : null, - workspace: workspace, - hasWorkspace: !string.IsNullOrEmpty(workspace), - planMode: planMode, - webSearch: webSearch, - mcpConfigPath: mcpConfigPath, - cancellationToken: workCt - ); - - // Bei CLI-Fehler → Fehlermeldung an Client senden und in DB speichern - if (!result.Success && !isUserChoice) - { - var errorMsg = !string.IsNullOrEmpty(result.Error) ? result.Error : "CLI hat keine Antwort geliefert"; - // Error in DB speichern (damit er beim Neuladen sichtbar ist) - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "system", - Type = "error", - Content = $"CLI-Fehler: {errorMsg}" - }); - await SseHelper.TryWriteSse(response, "error", new { error = $"CLI-Fehler: {errorMsg}" }, sseCt); - } - - // Antwort in DB speichern - if (fullText.Length > 0) - { - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "assistant", - Type = "text", - Content = fullText.ToString() - }); - } - - // Token-Usage vom CLI-Result übernehmen - usage.InputTokens = result.InputTokens; - usage.OutputTokens = result.OutputTokens; - usage.CacheReadTokens = result.CacheReadTokens; - usage.CacheCreationTokens = result.CacheCreationTokens; - - var stopReason = isUserChoice ? "awaiting_user_choice" : (result.Success ? "end_turn" : "error"); - await FinalizeChain(chain, usage, result.NumTurns > 0 ? result.NumTurns : 1, stopReason, sw); - - await SseHelper.TryWriteSse(response, "done", new - { - model = modelId, - inputTokens = result.InputTokens, - outputTokens = result.OutputTokens, - cacheCreationTokens = result.CacheCreationTokens, - cacheReadTokens = result.CacheReadTokens, - costUsd = result.CostUsd, - iterations = result.NumTurns > 0 ? result.NumTurns : 1, - chainId, - durationMs = sw.ElapsedMilliseconds, - stopReason, - viaCli = true - }, sseCt); - } - catch (OperationCanceledException) when (workCt.IsCancellationRequested) - { - // Stop-Button gedrückt → CLI-Prozess wurde gestoppt - // Wenn ask_user_choice aktiv war: als awaiting_user_choice markieren, - // damit die nächste Chain den Kontext (Frage + Optionen) behält. - var stoppedReason = isUserChoice ? "awaiting_user_choice" : "stopped"; - await FinalizeChain(chain, usage, 1, stoppedReason, sw); - await SseHelper.TryWriteSse(response, "done", new - { - model = modelId, - chainId, - durationMs = sw.ElapsedMilliseconds, - stopReason = stoppedReason, - viaCli = true - }, CancellationToken.None); - } - catch (OperationCanceledException) - { - // Client-Disconnect (sseCt gecancelled, aber NICHT workCt) - // CLI-Prozess läuft weiter im Hintergrund – hier nur Chain finalisieren - // (sollte nicht mehr passieren, da CLI jetzt workCt nutzt) - await FinalizeChain(chain, usage, 1, "client_disconnected", sw); - } - catch (Exception ex) - { - var errorMsg = $"CLI-Fehler: {ex.Message}"; - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "system", - Type = "error", - Content = errorMsg - }); - await SseHelper.TryWriteSse(response, "error", new { error = errorMsg }, sseCt); - await FinalizeChain(chain, usage, 1, "error", sw); - await SseHelper.TryWriteSse(response, "done", new - { - model = modelId, - chainId, - durationMs = sw.ElapsedMilliseconds, - stopReason = "error", - viaCli = true - }, sseCt); - } - finally - { - // Temp-Bilddateien aufräumen - CleanupTempImages(tempImagePaths); - - // MCP-Config Temp-Datei aufräumen - if (mcpConfigPath != null) - { - try { File.Delete(mcpConfigPath); } catch { } - } - } - } - - /// - /// Mappt CLI-Tool-Namen (Edit, Write, Bash, etc.) auf gott-Tool-Namen (edit_file, write_file, etc.). - /// - private static string MapCliToolName(string cliName) - { - return cliName switch - { - "Edit" => "edit_file", - "Write" => "write_file", - "Read" => "read_file", - "Bash" => "execute", - "Glob" => "list_files", - "Grep" => "search_files", - "WebSearch" => "web_search", - "WebFetch" => "web_fetch", - "NotebookEdit" => "notebook_edit", - "mcp__gott-remote__remote_execute" => "remote_execute", - _ => cliName?.StartsWith("mcp__") == true ? cliName.Split("__").LastOrDefault() ?? cliName : cliName - }; - } - - /// - /// Mappt CLI-Tool-Input-Felder auf gott-Konventionen (z.B. file_path→path, old_string→old_text). - /// - private static string MapCliToolInput(string cliToolName, string inputJson) - { - if (string.IsNullOrEmpty(inputJson)) return inputJson; - - try - { - var input = JsonSerializer.Deserialize(inputJson); - if (input.ValueKind != JsonValueKind.Object) return inputJson; - - switch (cliToolName) - { - case "Edit": - { - // file_path → path, old_string → old_text, new_string → new_text - var mapped = new Dictionary(); - foreach (var prop in input.EnumerateObject()) - { - var key = prop.Name switch - { - "file_path" => "path", - "old_string" => "old_text", - "new_string" => "new_text", - _ => prop.Name - }; - mapped[key] = prop.Value; - } - return JsonSerializer.Serialize(mapped); - } - case "Write": - { - // file_path → path, content bleibt - var mapped = new Dictionary(); - foreach (var prop in input.EnumerateObject()) - { - var key = prop.Name == "file_path" ? "path" : prop.Name; - mapped[key] = prop.Value; - } - return JsonSerializer.Serialize(mapped); - } - case "Read": - { - // file_path → path - var mapped = new Dictionary(); - foreach (var prop in input.EnumerateObject()) - { - var key = prop.Name == "file_path" ? "path" : prop.Name; - mapped[key] = prop.Value; - } - return JsonSerializer.Serialize(mapped); - } - case "Bash": - { - // command bleibt gleich - return inputJson; - } - default: - return inputJson; - } - } - catch - { - return inputJson; - } - } - - /// - /// Baut einen Kontext-String aus der Konversations-History. - /// - private static string BuildHistoryContext(List messages) - { - var sb = new StringBuilder(); - // Letzte Message überspringen wenn es ein Text-Prompt ist (wird separat als current_request gesendet). - // Wenn die letzte Message ein tool_result ist (z.B. Choice-Antwort), NICHT überspringen. - int takeCount = messages.Count; - if (messages.Count > 0) - { - var lastJson = JsonSerializer.Serialize(messages[messages.Count - 1]); - var lastEl = JsonSerializer.Deserialize(lastJson); - bool lastIsToolResult = false; - if (lastEl.TryGetProperty("content", out var lc) && lc.ValueKind == JsonValueKind.Array) - { - foreach (var b in lc.EnumerateArray()) - { - var bType = b.TryGetProperty("type", out var bt) ? bt.GetString() ?? "" : ""; - if (bType == "tool_result") { lastIsToolResult = true; break; } - } - } - if (!lastIsToolResult) - takeCount = messages.Count - 1; - } - foreach (var msg in messages.Take(takeCount)) - { - 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) - { - sb.AppendLine($"[{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" && block.TryGetProperty("text", out var txt)) - { - sb.AppendLine($"[{roleStr}]: {txt.GetString()}"); - } - else if (blockType == "tool_use" && block.TryGetProperty("name", out var tn) && tn.GetString() == "ask_user_choice") - { - if (block.TryGetProperty("input", out var inp)) - { - var q = inp.TryGetProperty("question", out var qq) ? qq.GetString() ?? "" : ""; - var opts = new List(); - if (inp.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array) - foreach (var o in cc.EnumerateArray()) opts.Add(o.GetString() ?? ""); - sb.AppendLine($"[assistant]: Ich habe gefragt: \"{q}\" mit Optionen: {string.Join(", ", opts)}"); - } - } - else if (blockType == "tool_result" && block.TryGetProperty("content", out var trContent)) - { - sb.AppendLine($"[user]: Du hast gewählt: \"{trContent.GetString()}\""); - } - } - } - } - } - return sb.ToString().Trim(); - } }