using System; using System.Collections.Generic; using System.Linq; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Gott.Extensions; using Gott.Models; using Gott.Services; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Routing; namespace Gott.Endpoints; public static class ChatEndpoint { public static void MapChatEndpoint(this IEndpointRouteBuilder app) { app.MapPost("/api/chat", HandleChat); app.MapPost("/api/chat/stop", HandleStop); app.MapPost("/api/chat/steer", HandleSteer); } private static async Task HandleChat(HttpRequest req, HttpResponse res, ServerSettingsService settings, ChatService chatService, DatabaseService db, StreamingTracker streamingTracker, SteeringQueue steeringQueue, AnthropicService anthropic, OpenAiService openAi, RemoteConnectionManager remoteMgr, EmbeddingBackgroundService embeddingBg, CancellationToken cancellationToken) { var body = await JsonSerializer.DeserializeAsync(req.Body); if (body == null) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "body required" }); return; } var sessionId = body.SessionId; if (string.IsNullOrEmpty(sessionId)) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "sessionId required" }); return; } // Prüfe ob dies eine Choice-Antwort ist (ask_user_choice response) var isChoiceResponse = !string.IsNullOrEmpty(body.ChoiceToolCallId) && !string.IsNullOrEmpty(body.ChoiceResult); if (!isChoiceResponse && body.Content.ValueKind == JsonValueKind.Undefined) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "content required" }); return; } var userId = req.HttpContext.GetUserId(); var maxIterations = settings.GetInt("MaxIterations", 100); var remoteConn = remoteMgr.GetBySessionId(sessionId); var hasRemote = remoteConn != null; var ctx = await chatService.PrepareContext(sessionId, userId, maxIterations, planMode: body.PlanMode, multiModelMode: body.MultiModelMode, hasRemoteConnection: hasRemote, remoteOs: remoteConn?.OsVersion); if (ctx == null) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "Session/Agent nicht gefunden oder kein Agent ausgewählt" }); return; } // API-Key des Users für diesen Agent auflösen (kann null sein) var userApiKey = await db.GetAgentApiKeyRecord(userId, ctx.AgentId); if (isChoiceResponse) { // Choice-Antwort: tool_result in die VORHERIGE Chain speichern (wo der tool_use liegt) // Damit BuildHistoryFromDb die tool_use + tool_result korrekt zusammenbauen kann. var previousChainId = await chatService.FindChainWithToolCall(sessionId, body.ChoiceToolCallId); if (previousChainId != null) { await db.SaveMessage(new MessageRecord { ChainId = previousChainId, Iteration = 999, // Hohe Iteration damit es nach dem tool_use kommt Role = "user", Type = "tool_result", ToolCallId = body.ChoiceToolCallId, Content = body.ChoiceResult }); } else { // Fallback: In aktuelle Chain speichern await db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = 0, Role = "user", Type = "tool_result", ToolCallId = body.ChoiceToolCallId, Content = body.ChoiceResult }); } } else { // Normale User-Message aufbereiten und speichern var userMessage = new ChatMessage { Role = "user", Content = body.Content }; var textContent = userMessage.GetTextContent(); var attachmentBlocks = userMessage.GetAttachmentBlocks(); var attachmentRecords = attachmentBlocks.Select(b => new AttachmentRecord { Type = b.Type, MediaType = b.MediaType, FileName = b.FileName, Data = b.Data }).ToList(); var userMsgId = await db.SaveMessageWithAttachments(new MessageRecord { ChainId = ctx.ChainId, Iteration = 0, Role = "user", Type = "text", Content = textContent }, attachmentRecords); embeddingBg?.Enqueue(userMsgId, ctx.ChainId, ctx.SessionId, ctx.AgentId, "user", textContent ?? ""); } res.Headers["Content-Type"] = "text/event-stream"; res.Headers["Cache-Control"] = "no-cache"; res.Headers["Connection"] = "keep-alive"; // Response-Buffering deaktivieren – wichtig für SSE unter Windows var bufferingFeature = res.HttpContext.Features.Get(); bufferingFeature?.DisableBuffering(); // Kestrel MinResponseDataRate deaktivieren für diesen SSE-Stream. // Ohne dies killt Kestrel die Verbindung wenn bei langen Tool-Ausführungen // keine Daten fließen (Default: 240 bytes/s, 5s Grace → "client_disconnected"). var minRateFeature = res.HttpContext.Features.Get(); if (minRateFeature != null) minRateFeature.MinDataRate = null; // ── Preparer-Agent Phase (vor dem Worker) ── if (ctx.Preparer != null && !isChoiceResponse) { var preparerApiToken = userApiKey?.ApiKey; // Kein Key und kein CLI → Preparer überspringen (main agent wird auch CLI nutzen oder fehlermelden) bool skipPreparer = string.IsNullOrEmpty(ctx.Preparer.Provider) && string.IsNullOrEmpty(preparerApiToken) && !anthropic.IsCliAvailable; if (!skipPreparer) { // Preparer bekommt History MIT seinen eigenen vorherigen Messages (Tag "preparer"), // damit er den Kontext seiner Rückfragen kennt var preparerHistory = await chatService.BuildHistoryFromDb(sessionId, includeTag: "preparer", promptBlocks: ctx.PromptBlocks); var preparerResult = await anthropic.RunPreparer( res, ctx.Preparer, preparerHistory, ctx.ChainId, ctx.Workspace, sessionId, preparerApiToken, cancellationToken); if (!preparerResult.ShouldForward) { // Preparer hat eine Rückfrage → Antwort als Text streamen, Flow beenden // Preparer-Antwort wird schon in RunAgentChat in DB gespeichert await SseHelper.WriteSseEvent(res, "done", new { chainId = ctx.ChainId, stopReason = "preparer_query", planMode = ctx.PlanMode }, cancellationToken); return; } // Preparer ist zufrieden → aufbereiteten Prompt in DB speichern if (!string.IsNullOrWhiteSpace(preparerResult.ForwardedPrompt)) { await db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = 0, Role = "user", Type = "text", Content = $"[Vorbereitet von {ctx.Preparer.Agent?.Name ?? "Vorbereiter-Agent"}]:\n{preparerResult.ForwardedPrompt}" }); } } // !skipPreparer } // History für den Session-Agent: OHNE Sub-Agent-Messages (saubere History). // includeThinking=true für OpenAI-kompatible Provider → Thinking wird als // ...-Prefix dem Assistant-Text vorangestellt (laut MiniMax-Docs essentiell // für Tool-Calling-Reliability / Interleaved Thinking). Anthropic-Pfad braucht das nicht, // weil dort thinking-Blöcke nur mit Signatur akzeptiert werden (die wir nicht persistieren). var includeThinkingForProvider = !string.IsNullOrEmpty(ctx.Provider); var messages = await chatService.BuildHistoryFromDb(sessionId, promptBlocks: ctx.PromptBlocks, includeThinking: includeThinkingForProvider); // Bei Choice-Response: toolChoice auf "auto" setzen, damit Claude frei antworten kann var effectiveToolChoice = isChoiceResponse ? "auto" : ctx.ToolChoice; // Drei separate CancellationTokens: // - workCt: Nur Stop-Button (graceful) → für Anthropic API-Calls, Tool-Ausführungen, DB-Writes // Agent-Arbeit läuft bei Client-Disconnect weiter (Client pollt ja) // - hardCt: Hard-Stop (Doppel-Klick) → sofortiger Abbruch aus while-Loop // - sseCt: Stop-Button + Client-Disconnect → für SSE-Writes an den Browser // Bei Client-Disconnect werden SSE-Writes zu stillen No-Ops var tokens = streamingTracker.Start(sessionId, ctx.ChainId); // Steering-Gate öffnen, damit Einwürfe während dieser Chain angenommen werden. steeringQueue.Open(sessionId); var workCt = tokens.StopToken; var hardCt = tokens.HardToken; using var sseLinkCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, tokens.StopToken); var sseCt = sseLinkCts.Token; // SSE-Heartbeat: Alle 15s ein Kommentar senden, damit die Verbindung // bei langen Tool-Ausführungen nicht von Browser/Netzwerk getrennt wird. var heartbeatTask = Task.Run(async () => { try { while (!sseCt.IsCancellationRequested) { await Task.Delay(15000, sseCt); await res.WriteAsync(": heartbeat\n\n", sseCt); await res.Body.FlushAsync(sseCt); } } catch (OperationCanceledException) { } catch { } }, sseCt); try { // ── Multi-Modell-Panel (Fusion) ── // Vor dem Worker: Panel-Agenten parallel antworten lassen und ihre Antworten dem // Worker als Synthese-Block in die letzte User-Message einmergen. Der Worker streamt // danach als Synthesizer über den unveränderten Dispatch unten. Heartbeat läuft bereits, // Cancellation (workCt/sseCt) ist verfügbar. Nur bei aktivem Fusion-Modus + nicht-leerem // Panel + keiner Choice-Antwort. // // Panel-Token-Kosten (alle Member über alle Runden) werden in panelUsage gesammelt und // unten als seedUsage in die Worker-Streaming-Methode gefaltet → fließen über das // done-Event (Live-Badge) UND FinalizeChain (DB) als Teil der Chain-Kosten und überleben // F5. Bleibt {0} wenn kein Panel läuft → Seeding ist dann ein No-op. var panelUsage = new TokenUsage(); if (!isChoiceResponse && (ctx.MultiModelMode == "fusion" || ctx.MultiModelMode == "moa") && ctx.PanelAgents != null && ctx.PanelAgents.Count > 0) { try { // Fusion = 1 Runde, MoA = MoaLayers Proposer-Runden vor der Synthese. var panelLayers = ctx.MultiModelMode == "moa" ? System.Math.Max(1, ctx.MoaLayers) : 1; await anthropic.RunPanel(res, ctx.PanelAgents, messages, ctx.ChainId, sessionId, userApiKey?.ApiKey, panelLayers, panelUsage, workCt, sseCt); } catch (OperationCanceledException) { throw; } catch (Exception ex) { // Panel-Fehler darf den Worker nicht killen → ohne Synthese normal weiter. Console.WriteLine($"[Panel] Fehler, fahre ohne Synthese fort: {ex.Message}"); } } if (!string.IsNullOrEmpty(ctx.Provider)) { // OpenAI-kompatibler Provider – Key des Users für diesen Agent verwenden string openAiKeyOverride = userApiKey?.ApiKey; // OpenAI-Provider braucht zwingend einen Key if (ctx.Provider == "openai" && string.IsNullOrEmpty(openAiKeyOverride)) { await SseHelper.TryWriteSse(res, "error", new { error = "Kein API-Key für diesen Agent zugewiesen. Bitte einen OpenAI-Key zuweisen." }, sseCt); await SseHelper.TryWriteSse(res, "done", new { chainId = ctx.ChainId }, sseCt); return; } await openAi.StreamChat(res, ctx.ModelId, ctx.SystemPromptText, messages, ctx.ChainId, workspace: ctx.Workspace, sessionId: sessionId, agentId: ctx.AgentId, maxIterations: ctx.MaxIterations, apiKeyOverride: openAiKeyOverride, useAutoContext: ctx.Agent?.UseAutoContext ?? true, seedUsage: panelUsage, workCt: workCt, sseCt: sseCt); } else { // Standard: Anthropic var anthropicToken = userApiKey?.ApiKey; // Kein API-Key → Claude CLI als Fallback verwenden if (string.IsNullOrEmpty(anthropicToken)) { // CLI-Berechtigung prüfen var chatUser = await db.GetUser(userId); if (chatUser != null && !chatUser.CanUseCli && !chatUser.IsAdmin) { await SseHelper.TryWriteSse(res, "error", new { error = "Kein API-Key zugewiesen und keine CLI-Berechtigung. Bitte einen Key zuweisen." }, sseCt); await SseHelper.TryWriteSse(res, "done", new { chainId = ctx.ChainId }, sseCt); return; } if (anthropic.IsCliAvailable) { await anthropic.StreamChatViaCli(res, ctx.ModelId, ctx.SystemPrompt, messages, ctx.ChainId, workspace: ctx.Workspace, allowUserChoice: ctx.Agent?.AllowUserChoice ?? true, planMode: body.PlanMode, ultracode: body.Ultracode, webSearch: !string.IsNullOrEmpty(ctx.WebSearchVersion), sessionId: sessionId, agentId: ctx.AgentId, remoteMgr: remoteMgr, mcpServers: ctx.Session?.McpServers, useAutoContext: ctx.Agent?.UseAutoContext ?? true, effort: ctx.Effort, seedUsage: panelUsage, workCt: workCt, sseCt: sseCt); } else { await SseHelper.TryWriteSse(res, "error", new { error = "Kein API-Key zugewiesen und Claude CLI nicht verfügbar. Bitte einen Key zuweisen oder 'claude login' ausführen." }, sseCt); await SseHelper.TryWriteSse(res, "done", new { chainId = ctx.ChainId }, sseCt); } return; } await anthropic.StreamChat(res, ctx.ModelId, ctx.SystemPrompt, messages, ctx.ChainId, ctx.Workspace, apiToken: anthropicToken, sessionId: sessionId, agentId: ctx.AgentId, userId: userId, useMemory: ctx.UseMemory, webSearchVersion: ctx.WebSearchVersion, maxIterations: ctx.MaxIterations, toolChoice: effectiveToolChoice, review: ctx.Review, finalizer: ctx.Finalizer, allowUserChoice: ctx.Agent?.AllowUserChoice ?? true, planMode: ctx.PlanMode, useAutoContext: ctx.Agent?.UseAutoContext ?? true, useGitHubSearch: ctx.Agent?.UseGitHubSearch ?? false, useSerperSearch: ctx.Agent?.UseSerperSearch ?? false, effort: ctx.Effort, seedUsage: panelUsage, workCt: workCt, hardCt: hardCt, sseCt: sseCt, remoteMgr: remoteMgr); } } finally { streamingTracker.Stop(sessionId); // Steering-Gate schließen. Close() ist atomar gegen TryEnqueue: ein Einwurf, // der dem User als injected:true bestätigt wurde, kann ab hier nicht mehr // verloren gehen. Im Race-Fenster zwischen dem letzten Drain des Loops und // jetzt angekommene Einwürfe werden NICHT verworfen, sondern als Type="steer" // persistiert (sonst Symptom: Einwurf nach F5 weg, da nie gedraint). var leftoverSteers = steeringQueue.Close(sessionId); foreach (var steer in leftoverSteers) { try { await db.SaveMessage(new MessageRecord { ChainId = ctx.ChainId, Iteration = 1, Role = "user", Type = "steer", Content = steer }); } catch { /* Persistenz best-effort – darf das Chain-Ende nicht stören */ } } // Heartbeat-Task beenden (sseLinkCts wird disposed → Token cancelled) } } private static async Task HandleStop(HttpRequest req, HttpResponse res, StreamingTracker streamingTracker) { var body = await JsonSerializer.DeserializeAsync(req.Body); var sessionId = body.TryGetProperty("sessionId", out var sid) ? sid.GetString() : null; if (string.IsNullOrEmpty(sessionId)) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "sessionId required" }); return; } // Erster Klick = "graceful" (laufende Iteration fertig), zweiter = "hard" (sofort) var stopLevel = streamingTracker.Cancel(sessionId); await res.WriteAsJsonAsync(new { stopped = stopLevel != null, level = stopLevel ?? "none" }); } /// /// "Steering": Reiht eine Nachricht out-of-band in die laufende Chain ein. /// Sie wird an der nächsten Iterations-Grenze als User-Message injiziert. /// Anthropic-API-Pfad: Injektion zwischen den messages.create-Calls. /// Claude-CLI-Pfad: Kill-and-Restart am nächsten Tool-Boundary (der Agentic-Loop /// läuft im claude-Child, daher kein echtes Mid-Turn-Inject möglich). /// private static async Task HandleSteer(HttpRequest req, HttpResponse res, StreamingTracker streamingTracker, SteeringQueue steeringQueue) { var body = await JsonSerializer.DeserializeAsync(req.Body); var sessionId = body.TryGetProperty("sessionId", out var sid) ? sid.GetString() : null; var text = body.TryGetProperty("text", out var txt) ? txt.GetString() : null; if (string.IsNullOrEmpty(sessionId) || string.IsNullOrWhiteSpace(text)) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "sessionId and text required" }); return; } // Schneller Vorab-Check: läuft überhaupt eine Chain? if (!streamingTracker.IsStreaming(sessionId)) { await res.WriteAsJsonAsync(new { injected = false, reason = "not_streaming" }); return; } // Maßgeblich ist das Steering-Gate: TryEnqueue ist atomar gegen Close() (Chain-Ende). // Liefert es false, ist das Gate bereits geschlossen (Race am Chain-Ende) → der // Client sendet die Nachricht dann als normale Nachricht statt sie zu verlieren. if (!steeringQueue.TryEnqueue(sessionId, text.Trim())) { await res.WriteAsJsonAsync(new { injected = false, reason = "not_streaming" }); return; } await res.WriteAsJsonAsync(new { injected = true }); } }