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; using Microsoft.Extensions.Configuration; 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); } private static async Task HandleChat(HttpRequest req, HttpResponse res, IConfiguration config, ChatService chatService, DatabaseService db, StreamingTracker streamingTracker, AnthropicService anthropic, 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 = config.GetValue("MaxIterations", 100); var hasRemote = remoteMgr.GetBySessionId(sessionId) != null; var ctx = await chatService.PrepareContext(sessionId, userId, maxIterations, planMode: body.PlanMode, hasRemoteConnection: hasRemote); if (ctx == null) { res.StatusCode = 400; await res.WriteAsJsonAsync(new { error = "Session/Agent nicht gefunden oder kein Agent ausgewählt" }); return; } 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) { // 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, 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 AnthropicService.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}" }); } } // History für den Session-Agent: OHNE Sub-Agent-Messages (saubere History) var messages = await chatService.BuildHistoryFromDb(sessionId, promptBlocks: ctx.PromptBlocks); // 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); 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 { await anthropic.StreamChat(res, ctx.ModelId, ctx.SystemPrompt, messages, ctx.ChainId, ctx.Workspace, sessionId, agentId: ctx.AgentId, 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, workCt: workCt, hardCt: hardCt, sseCt: sseCt, remoteMgr: remoteMgr); } finally { streamingTracker.Stop(sessionId); // 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" }); } }