gott/Endpoints/ChatEndpoint.cs

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<ChatRequest>(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<Microsoft.AspNetCore.Http.Features.IHttpResponseBodyFeature>();
        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<Microsoft.AspNetCore.Server.Kestrel.Core.Features.IHttpMinResponseDataRateFeature>();
        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
        // <think>...</think>-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: Agent-Key, sonst Server-Setting
                // 'OpenAiApiKey' (analog Gemini/MiniMax – wird in ResolveProviderConfig gezogen).
                if (ctx.Provider == "openai" && string.IsNullOrEmpty(openAiKeyOverride)
                    && string.IsNullOrEmpty(settings.Get("OpenAiApiKey")))
                {
                    await SseHelper.TryWriteSse(res, "error", new { error = "Kein OpenAI-Key vorhanden: weder dem Agent zugewiesen noch als Server-Setting 'OpenAiApiKey' gesetzt." }, 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, maxOutputTokens: ctx.MaxOutputTokens, disableThinking: ctx.DisableThinking, 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, maxOutputTokens: ctx.MaxOutputTokens, 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<JsonElement>(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" });
    }

    /// <summary>
    /// "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).
    /// </summary>
    private static async Task HandleSteer(HttpRequest req, HttpResponse res, StreamingTracker streamingTracker, SteeringQueue steeringQueue)
    {
        var body = await JsonSerializer.DeserializeAsync<JsonElement>(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 });
    }
}