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 { /// /// Verpackt einen Live-Einwurf (Steering) des Users mit einem Hinweis, damit das /// Modell ihn aktiv einordnet statt ihn nur passiv anzuhängen: laufende Aufgabe /// ergänzen ODER Kurs ändern. /// private static string WrapSteeringMessage(string text) { return "[Live-Einwurf vom User – kam herein, während du an der laufenden Aufgabe arbeitest]:\n" + text + "\n\nBerücksichtige diesen Einwurf ab sofort. Ergänzt er nur die laufende Aufgabe → arbeite weiter und beziehe ihn ein. Ändert er den Kurs → passe dein Vorgehen an und sag in einem Satz, was du jetzt anders machst."; } /// /// Draint die Steering-Queue der Session und hängt etwaige Live-Einwürfe als /// Text-Blöcke (mit Hinweis-Wrapper) an die übergebene Content-Liste eines /// user-Turns. Persistiert jeden Einwurf RAW (Role=user, Type=steer) für History + /// Re-Chat und meldet ihn via SSE (steer_injected) ans Frontend. /// Gibt die Anzahl injizierter Einwürfe zurück. /// private async Task InjectSteeringAsync(HttpResponse response, List content, string chainId, string sessionId, string agentId, int iteration, CancellationToken sseCt) { if (_steeringQueue == null || string.IsNullOrEmpty(sessionId)) return 0; var steers = _steeringQueue.Drain(sessionId); foreach (var steer in steers) { // API: gewrappter Text-Block (mit Hinweis). DB: RAW-Text (sauber für History/Re-Chat). content.Add(new { type = "text", text = WrapSteeringMessage(steer) }); var msgId = await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = iteration, Role = "user", Type = "steer", Content = steer }); _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "user", steer); await SseHelper.TryWriteSse(response, "steer_injected", new { text = steer }, sseCt); } return steers.Count; } 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, string effort = null, 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) { try { var autoCtx = await BuildAutoContextAsync(_vectorSearch, messages, chainId, agentId); if (autoCtx != null) { systemPrompt.Add(new { type = "text", text = autoCtx }); await SseHelper.TryWriteSse(response, "auto_context", new { context = autoCtx }, sseCt); await _db.SaveMessage(new MessageRecord { ChainId = chainId, Role = "system", Type = "auto_context", Content = autoCtx }); } } 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, ["messages"] = messages, ["stream"] = true }; // Reasoning-Effort (Opus 4.5+, Sonnet 4.6+). "" / "high" / null = Default weglassen. // Werte: low / medium / high / xhigh (Opus 4.7+) / max (Opus 4.6+, Sonnet 4.6). if (!string.IsNullOrEmpty(effort) && effort != "high") payload["output_config"] = new { effort }; if (systemPrompt != null && systemPrompt.Count > 0) { // Cache-Breakpoint auf den letzten System-Prompt-Block (1h TTL). // Idempotent über Iterationen. ApplyCacheControlToLast(systemPrompt); payload["system"] = systemPrompt; } var remoteConn = remoteMgr?.GetBySessionId(sessionId); var hasRemote = remoteConn != null; AddToolsToPayload(payload, new ToolPayloadOptions { Workspace = workspace, AgentId = agentId, UseMemory = useMemory, WebSearchVersion = webSearchVersion, ToolChoice = toolChoice, Iteration = iterations, AllowUserChoice = allowUserChoice, PlanMode = planMode, IncludeRemote = hasRemote, RemoteOs = remoteConn?.OsVersion, 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; } // ── Steering-Injektion (Drain-Punkt C) ── // Das Modell würde jetzt seinen Turn beenden. Hat der User einen Einwurf // geschickt (während des Turns oder der Review/Finalizer-Phase), injizieren // wir ihn als neuen user-Turn und lassen die Chain WEITERlaufen statt zu // finalisieren. NICHT bei awaiting_user_choice – dort wartet die Chain // bewusst auf die Choice-Antwort des Users. if (stopReason != "awaiting_user_choice") { var steerContent = new List(); var injectedCount = await InjectSteeringAsync(response, steerContent, chainId, sessionId, agentId, iterations, sseCt); if (injectedCount > 0) { messages.Add(new { role = "user", content = steerContent }); continue; // → Chain läuft weiter } } 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, effort }, 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 }); } // ── Steering-Injektion (Drain-Punkt B) ── // Live-Einwürfe, die während dieser Tool-Runde eingingen, an DENSELBEN // user-Turn anhängen (tool_result-Blöcke zuerst, dann Text – API-konform). // So sieht das Modell den Einwurf in der nächsten Iteration, mitten in der Arbeit. await InjectSteeringAsync(response, toolResults, chainId, sessionId, agentId, iterations, sseCt); 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, effort }, 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, effort }, 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, new ToolExecutionContext { Workspace = workspace, MemoryService = _memoryService, AgentId = agentId, CancellationToken = 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(); } }