gott/Services/AnthropicService.Panel.cs

using System;
using System.Collections.Generic;
using System.Linq;
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
{
    // ── Multi-Modell-Panel (Fusion / MoA) ──
    //
    // Mehrere Panel-Agenten ("Proposer") beantworten dieselbe Anfrage PARALLEL und
    // streamen live. Anschließend werden ihre Antworten dem Worker-Agent als
    // <panel_responses>-Block vorgelegt, der daraus EINE Gesamtantwort synthetisiert
    // und über die unveränderte Streaming-Pipeline ausgibt.
    //
    // MoA (Mixture-of-Agents): layers>1 → mehrere Proposer-Runden; jede Runde k>1 sieht
    // die Antworten der Vorrunde (<previous_layer_responses>) und verfeinert. Fusion = layers==1.
    // JEDE Runde wird persistiert (mit Layer-Index im JSON-Transkript → F5-sicher, Runden-Divider beim Reload).
    //
    // Member laufen READ-ONLY (Anthropic: PlanMode-Filter, CLI: --permission-mode plan)
    // bzw. tool-los (OpenAI), damit parallele Member keine Dateien gleichzeitig schreiben.
    // Member-Arbeit wird NICHT über den Loop persistiert (Persist=false) – stattdessen
    // wird pro Member genau EINE Proposal-Message mit Tag="panel" + [Panel: Name]-Prefix
    // gespeichert (aus der Worker-History gefiltert, überlebt aber F5 als eigene Bubble).

    private class PanelMemberResult
    {
        public int Index { get; set; }
        public string Name { get; set; }
        public string Model { get; set; }
        public string Text { get; set; }
        public TokenUsage Usage { get; set; } // Token-Verbrauch dieses Members (für die Chain-Kosten)
    }

    /// <summary>
    /// Führt das Panel aus (Fusion = 1 Runde, MoA = <paramref name="layers"/> Runden): pro Runde
    /// alle Member parallel; ab Runde 2 sieht jeder Member die Antworten der Vorrunde. JEDE Runde
    /// wird persistiert (Tag="panel", mit Layer-Index); der Synthese-Block der finalen Runde wird in die letzte User-Message von
    /// <paramref name="workerMessages"/> gemergt. Danach streamt der Worker als Synthesizer ganz
    /// normal weiter. Gibt die Anzahl beigetragener Member der finalen Runde zurück
    /// (0 = nichts injiziert → Worker antwortet regulär).
    /// </summary>
    public async Task<int> RunPanel(
        HttpResponse response,
        List<SubAgentContext> panel,
        List<object> workerMessages,
        string chainId,
        string sessionId,
        string apiToken,
        int layers,
        TokenUsage usageSink,
        CancellationToken workCt,
        CancellationToken sseCt)
    {
        if (panel == null || panel.Count == 0) return 0;
        layers = Math.Clamp(layers, 1, 5); // Kosten-Backstop: max. 5 Proposer-Runden

        // Write-Lock: alle Member streamen LIVE parallel auf DIESELBE HttpResponse →
        // ohne Serialisierung würden sich ihre SSE-Frames auf Byte-Ebene vermischen.
        // Serialisiert nur die µs-Byte-Flushes, nicht die parallele Modell-Arbeit.
        using var sseGate = new SemaphoreSlim(1, 1);

        var mode = layers > 1 ? "moa" : "fusion";
        await SseHelper.TryWriteSse(response, "panel_start", new { members = panel.Count, mode, layers }, sseCt, sseGate);

        // Gesamt-Token aller Member über ALLE Runden in den vom Aufrufer übergebenen Sink
        // akkumulieren. Jeder Member sammelt seine eigene Usage (result.Usage); summiert wird
        // erst NACH der Task.WhenAll-Barrier → kein Lock nötig. Der Aufrufer seedet damit die
        // Worker-Usage → die Panel-Kosten fließen über das done-Event (Live-Badge) UND
        // FinalizeChain (DB) als Teil der Chain-Kosten und überleben so F5.
        var panelUsage = usageSink ?? new TokenUsage();

        // MoA: mehrere Proposer-Runden. Runde k>1 bekommt <previous_layer_responses> und verfeinert.
        // Fusion = layers==1 (keine Vorrunde). prev füttert die nächste Runde bzw. die finale Synthese.
        List<PanelMemberResult> prev = null;
        for (int layer = 1; layer <= layers; layer++)
        {
            await SseHelper.TryWriteSse(response, "panel_layer_start", new { layer, total = layers }, sseCt, sseGate);

            string prevBlock = prev == null ? null : BuildPreviousLayerBlock(prev, layer, layers);

            // Concurrency begrenzen: bei CLI-Membern sind das parallele claude-Children.
            using var sem = new SemaphoreSlim(Math.Min(3, panel.Count));
            var tasks = new List<Task<PanelMemberResult>>();
            for (int i = 0; i < panel.Count; i++)
            {
                var member = panel[i];
                var index = i;
                var lyr = layer;
                tasks.Add(Task.Run(async () =>
                {
                    await sem.WaitAsync(workCt);
                    try { return await RunPanelMember(response, member, workerMessages, index, lyr, layers, prevBlock, chainId, sessionId, apiToken, sseGate, workCt, sseCt); }
                    finally { sem.Release(); }
                }, workCt));
            }

            var results = await Task.WhenAll(tasks);

            // Nach der Barrier (single-threaded): Token ALLER Member dieser Runde aufsummieren –
            // auch von Membern ohne verwertbaren Text (sie haben trotzdem Tokens verbraucht).
            foreach (var r in results)
                if (r?.Usage != null) panelUsage.Add(r.Usage);

            var ordered = results
                .Where(r => r != null && !string.IsNullOrWhiteSpace(r.Text))
                .OrderBy(r => r.Index)
                .ToList();

            await SseHelper.TryWriteSse(response, "panel_layer_done", new { layer, total = layers, members = ordered.Count }, sseCt, sseGate);

            if (ordered.Count == 0)
            {
                // Keine verwertbaren Antworten in dieser Runde → ohne Synthese weiter.
                // Die bisher verbrauchten Member-Tokens stecken bereits im usageSink (oben summiert)
                // und werden vom Aufrufer in die Worker-Usage geseedet → ehrliche Kosten.
                await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = false, members = 0 }, sseCt, sseGate);
                return 0;
            }
            prev = ordered;
        }

        // Synthese-Block aus der LETZTEN Runde bauen und in die letzte User-Message einmergen.
        var sb = new StringBuilder();
        sb.AppendLine();
        sb.AppendLine("---");
        sb.AppendLine("[Multi-Modell-Panel] Mehrere unabhängige KI-Agenten haben parallel auf die obige Anfrage geantwortet. "
            + "Deine Aufgabe: Synthetisiere daraus EINE optimale Gesamtantwort für den User. Nimm die stärksten Argumente, "
            + "korrektesten Fakten und besten Ideen aus allen Beiträgen, korrigiere erkennbare Fehler und vermeide bloßes "
            + "Aneinanderreihen. Widersprechen sich Beiträge, wäge ab und entscheide begründet. Antworte direkt und natürlich "
            + "dem User (du musst das Panel nicht erwähnen). Du kannst deine eigenen Tools nutzen, um zu prüfen oder die Aufgabe abzuschließen.");
        sb.AppendLine();
        sb.AppendLine("<panel_responses>");
        foreach (var r in prev)
        {
            sb.AppendLine();
            sb.AppendLine($"### Antwort von {r.Name} ({r.Model}):");
            sb.AppendLine(r.Text.Trim());
        }
        sb.AppendLine("</panel_responses>");

        MergePanelIntoLastUserText(workerMessages, sb.ToString());

        await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = true, members = prev.Count }, sseCt, sseGate);
        return prev.Count;
    }

    /// <summary>
    /// Führt einen einzelnen Panel-Member aus (Provider-Weiche, read-only/tool-los),
    /// persistiert seine Proposal-Message (Tag="panel") und sendet panel_member_* SSE-Events.
    /// Fängt Member-Fehler ab (ein fehlerhafter Member darf das Panel nicht killen).
    /// </summary>
    private async Task<PanelMemberResult> RunPanelMember(
        HttpResponse response,
        SubAgentContext member,
        List<object> workerMessages,
        int index,
        int layer,
        int totalLayers,
        string previousLayerBlock,
        string chainId,
        string sessionId,
        string apiToken,
        SemaphoreSlim sseGate,
        CancellationToken workCt,
        CancellationToken sseCt)
    {
        var name = member.Agent?.Name ?? $"Panel-Agent {index + 1}";
        await SseHelper.TryWriteSse(response, "panel_member_start", new { index, name, model = member.ModelId, layer, totalLayers }, sseCt, sseGate);

        // Eigene, ISOLIERTE Kopie pro Member. Bei MoA (Runde > 1) wird der Vorrunden-Block in den
        // bestehenden Text-Block der letzten User-Message gemergt – isoliert, damit er nicht in
        // workerMessages oder parallele Member leckt.
        var memberMessages = CloneForMember(workerMessages, previousLayerBlock);
        string text;
        List<object> segments = null; // interleavtes Transkript (Text + Tools) für die Reload-Persistenz
        TokenUsage memberUsage = null; // Token-Verbrauch dieses Members (null = nicht verfügbar, z.B. OpenAI)
        try
        {
            if (!string.IsNullOrEmpty(member.Provider) && _openAiService != null)
            {
                // OpenAI-kompatibler Member: tool-los (CompleteText), Proposer aus Kontext.
                // Kein Tool-Transkript → unten als einzelnes Text-Segment persistiert.
                var systemText = ExtractSystemPromptText(member.SystemPrompt);
                text = await _openAiService.CompleteText(member.ModelId, systemText, memberMessages, apiToken);
            }
            else if (string.IsNullOrEmpty(apiToken) && _cliService != null && _cliService.IsAvailable)
            {
                // Kein API-Key → CLI-Member, read-only (Plan-Modus: nur Read/Glob/Grep).
                // Streamt LIVE (Token + Tool-Uses) über panel_member_*-Events und liefert das Transkript.
                var (t, segs, u) = await RunPanelMemberViaCli(response, member, memberMessages, index, sseGate, workCt, sseCt);
                text = t; segments = segs; memberUsage = u;
            }
            else
            {
                // Anthropic-API-Member: read-only (PlanMode) + NICHT persistieren (Caller persistiert einmal).
                // Live-Streaming über den Loop: pro Tool-Runde panel_member_text / panel_member_tool.
                // TranscriptSink sammelt Text-/Tool-Segmente für die Reload-Persistenz.
                memberUsage = new TokenUsage();
                segments = new List<object>();
                var (t, _) = await RunNonStreamingToolLoop(
                    member.ModelId, member.SystemPrompt,
                    new ToolPayloadOptions
                    {
                        Workspace = member.Workspace, AgentId = member.AgentId,
                        UseMemory = member.UseMemory, WebSearchVersion = member.WebSearchVersion,
                        ToolChoice = "auto", PlanMode = true
                    },
                    memberMessages, chainId, sessionId, apiToken, memberUsage,
                    0, 12, workCt,
                    new NonStreamingLoopCallbacks
                    {
                        Persist = false,
                        SseResponse = response, SseCt = sseCt, SseGate = sseGate,
                        SsePrefix = "panel_member", MemberIndex = index, AgentName = name,
                        TranscriptSink = segments
                    },
                    member.Effort);
                text = t;
            }
        }
        catch (OperationCanceledException) { throw; }
        catch (Exception ex) { text = $"[Fehler bei {name}: {ex.Message}]"; segments = null; }

        text = (text ?? "").Trim();

        // Member-Transkript als EINE strukturierte Message persistieren (Tag="panel" → aus
        // Worker-History gefiltert). Content = JSON {name, model, layer, totalLayers,
        // segments:[{t:"text"|"tool",...}]} → der Reload stellt Text UND Tool-Boxen wieder her
        // (genau wie live) und gruppiert bei MoA nach Runden mit Divider. Fällt auf ein einzelnes
        // Text-Segment zurück (OpenAI-Member, Fehler oder wenn nichts gestreamt wurde). JEDE Runde
        // wird persistiert (Iteration=layer*100+index+1 → stabile, layer-aware Reihenfolge) → MoA-
        // Zwischenrunden überleben F5.
        if (segments == null || segments.Count == 0)
        {
            segments = new List<object>();
            if (!string.IsNullOrEmpty(text))
                segments.Add(new Dictionary<string, object> { ["t"] = "text", ["v"] = text });
        }
        if (segments.Count > 0)
        {
            var content = JsonSerializer.Serialize(new
            {
                name, model = member.ModelId, layer, totalLayers,
                inputTokens = memberUsage?.InputTokens ?? 0,
                outputTokens = memberUsage?.OutputTokens ?? 0,
                cacheReadTokens = memberUsage?.CacheReadTokens ?? 0,
                cacheCreationTokens = memberUsage?.CacheCreationTokens ?? 0,
                segments
            });
            await _db.SaveMessage(new MessageRecord
            {
                ChainId = chainId, Iteration = layer * 100 + index + 1,
                Role = "assistant", Type = "text",
                Content = content, Tag = "panel"
            });
        }

        await SseHelper.TryWriteSse(response, "panel_member_done",
            new
            {
                index, name, model = member.ModelId, text,
                inputTokens = memberUsage?.InputTokens ?? 0,
                outputTokens = memberUsage?.OutputTokens ?? 0,
                cacheReadTokens = memberUsage?.CacheReadTokens ?? 0,
                cacheCreationTokens = memberUsage?.CacheCreationTokens ?? 0
            }, sseCt, sseGate);

        return new PanelMemberResult { Index = index, Name = name, Model = member.ModelId, Text = text, Usage = memberUsage };
    }

    /// <summary>
    /// Führt einen Panel-Member über die Claude CLI aus (read-only via Plan-Modus) und
    /// streamt seine Arbeit LIVE: Text-Deltas (panel_member_text), Tool-Start
    /// (panel_member_tool_use) und Tool-Result (panel_member_tool_end) – jeweils mit
    /// Member-Index, damit das Frontend sie der richtigen Bubble zuordnet. Die SSE-Writes
    /// laufen durch den geteilten <paramref name="sseGate"/> (parallele Member → sonst
    /// vermischen sich die Frames). Persistiert nichts – der Caller speichert eine Proposal-Message.
    /// </summary>
    private async Task<(string text, List<object> segments, TokenUsage usage)> RunPanelMemberViaCli(
        HttpResponse response,
        SubAgentContext member,
        List<object> memberMessages,
        int index,
        SemaphoreSlim sseGate,
        CancellationToken workCt,
        CancellationToken sseCt)
    {
        if (_cliService == null || !_cliService.IsAvailable) return ("", new List<object>(), null);

        var systemText = ExtractSystemPromptText(member.SystemPrompt);
        var prompt = ExtractLastUserText(memberMessages) ?? "";

        // Bild-/Datei-Anhänge der letzten User-Message als Temp-Dateien ablegen, damit der
        // CLI-Member sie per Read-Tool sehen kann (Screenshots, PDFs etc.) – exakt wie der Worker.
        // Ohne das wären die Member bei bildbasierten Fragen blind → read-only Tools nutzlos.
        var tempImagePaths = new List<string>();
        var imageRefs = ExtractLastUserImageRefs(memberMessages, tempImagePaths);
        if (!string.IsNullOrEmpty(imageRefs))
            prompt = imageRefs + "\n" + prompt;

        var historyContext = BuildHistoryContext(memberMessages);
        if (!string.IsNullOrEmpty(historyContext))
            prompt = $"<conversation_history>\n{historyContext}\n</conversation_history>\n\n<current_request>\n{prompt}\n</current_request>";

        var fullText = new StringBuilder();
        // Transkript-Segmente (Text + Tools interleaved) für die Reload-Persistenz.
        var segments = new List<object>();
        var segText = new StringBuilder();
        void FlushText()
        {
            if (segText.Length == 0) return;
            segments.Add(new Dictionary<string, object> { ["t"] = "text", ["v"] = segText.ToString() });
            segText.Clear();
        }
        // Tool-Uses per ID korrelieren: Ein Member kann MEHRERE Tools in einer Assistant-Message
        // aufrufen (ToolStart#1,#2,… ToolResult#1,#2,…). Ein einzelner "currentTool"-Zeiger würde
        // dann Boxen vertauschen/verwaisen lassen. Außerdem ist der Tool-Input erst bei ToolEnd
        // (content_block_stop) vollständig – ToolStart/ToolResult tragen ihn nicht.
        var toolNames = new Dictionary<string, string>();   // toolId -> CLI-Name
        var toolInputs = new Dictionary<string, string>();  // toolId -> Input-JSON
        CliChatResult cliResult = null; // vor dem try, damit die Usage nach dem finally lesbar ist
        try
        {
            cliResult = await _cliService.ChatStreamingAsync(
                prompt,
                async evt =>
                {
                    switch (evt.Type)
                    {
                        case CliEventType.TextDelta:
                            fullText.Append(evt.Text);
                            segText.Append(evt.Text);
                            await SseHelper.TryWriteSse(response, "panel_member_text",
                                new { index, text = evt.Text }, sseCt, sseGate);
                            break;
                        case CliEventType.ToolStart:
                            FlushText(); // Text-Segment vor der Tool-Box schließen (Reihenfolge wie live)
                            if (!string.IsNullOrEmpty(evt.ToolId)) toolNames[evt.ToolId] = evt.ToolName;
                            await SseHelper.TryWriteSse(response, "panel_member_tool_use",
                                new { index, id = evt.ToolId, name = evt.ToolName }, sseCt, sseGate);
                            break;
                        case CliEventType.ToolEnd:
                            // Input streamt erst nach ToolStart ein → hier vollständig. Für ToolResult merken.
                            if (!string.IsNullOrEmpty(evt.ToolId)) toolInputs[evt.ToolId] = evt.ToolInput;
                            break;
                        case CliEventType.ToolResult:
                            var rid = evt.ToolId ?? "";
                            toolNames.TryGetValue(rid, out var rName);
                            toolInputs.TryGetValue(rid, out var rInput);
                            segments.Add(new Dictionary<string, object>
                            {
                                ["t"] = "tool", ["name"] = rName ?? "", ["input"] = rInput ?? "",
                                ["result"] = TruncateToolResult(evt.ToolResultContent), ["error"] = evt.ToolResultIsError
                            });
                            await SseHelper.TryWriteSse(response, "panel_member_tool_end",
                                new { index, id = rid, name = rName, input = rInput, result = evt.ToolResultContent, isError = evt.ToolResultIsError }, sseCt, sseGate);
                            break;
                    }
                },
                modelId: member.ModelId,
                systemPrompt: systemText,
                workspace: member.Workspace,
                hasWorkspace: !string.IsNullOrEmpty(member.Workspace),
                planMode: true,
                effort: member.Effort,
                cancellationToken: workCt
            );
        }
        finally
        {
            CleanupTempImages(tempImagePaths);
        }
        FlushText(); // restliches Text-Segment am Ende sichern
        var usage = cliResult == null ? null : new TokenUsage
        {
            InputTokens = cliResult.InputTokens,
            OutputTokens = cliResult.OutputTokens,
            CacheReadTokens = cliResult.CacheReadTokens,
            CacheCreationTokens = cliResult.CacheCreationTokens
        };
        return (fullText.ToString(), segments, usage);
    }

    /// <summary>Extrahiert den System-Prompt-Text aus den List&lt;object&gt;-Blöcken eines SubAgentContext.</summary>
    private static string ExtractSystemPromptText(List<object> systemPrompt)
    {
        if (systemPrompt == null) return "";
        var parts = new List<string>();
        foreach (var block in systemPrompt)
        {
            var json = JsonSerializer.Serialize(block);
            var el = JsonSerializer.Deserialize<JsonElement>(json);
            if (el.TryGetProperty("text", out var t))
                parts.Add(t.GetString() ?? "");
        }
        return string.Join("\n\n", parts);
    }

    /// <summary>
    /// Hängt den Panel-Synthese-Block an den BESTEHENDEN Text-Block der letzten User-Message an.
    /// Wichtig: kein zweiter Text-Block – sonst sieht der CLI-Worker (ExtractLastUserText liefert
    /// nur den ersten Text-Block) die Synthese-Anweisung nicht. So funktioniert es für API, CLI und OpenAI.
    /// </summary>
    private static void MergePanelIntoLastUserText(List<object> messages, string panelBlock)
    {
        for (int i = messages.Count - 1; i >= 0; i--)
        {
            if (messages[i] is Dictionary<string, object> d &&
                d.TryGetValue("role", out var r) && (r as string) == "user" &&
                d.TryGetValue("content", out var c) && c is List<object> contentList)
            {
                for (int j = 0; j < contentList.Count; j++)
                {
                    var json = JsonSerializer.Serialize(contentList[j]);
                    var el = JsonSerializer.Deserialize<JsonElement>(json);
                    if (el.TryGetProperty("type", out var t) && t.GetString() == "text")
                    {
                        var orig = el.TryGetProperty("text", out var txt) ? txt.GetString() ?? "" : "";
                        contentList[j] = new { type = "text", text = orig + "\n\n" + panelBlock };
                        return;
                    }
                }
                // Kein Text-Block vorhanden → einen anhängen.
                contentList.Add(new { type = "text", text = panelBlock });
                return;
            }
        }
    }

    /// <summary>
    /// Liefert eine pro-Member isolierte Kopie der Worker-Messages. Ist <paramref name="block"/>
    /// gesetzt (MoA, Runde &gt; 1), wird er in den bestehenden Text-Block der letzten User-Message
    /// gemergt – auf einer GEKLONTEN Message + Content-Liste, damit die Mutation NICHT in
    /// workerMessages oder parallele Member leckt. block==null → einfache Shallow-Kopie (wie Fusion).
    /// </summary>
    private static List<object> CloneForMember(List<object> messages, string block)
    {
        var copy = new List<object>(messages);
        if (string.IsNullOrEmpty(block)) return copy;
        for (int i = copy.Count - 1; i >= 0; i--)
        {
            if (copy[i] is Dictionary<string, object> d &&
                d.TryGetValue("role", out var r) && (r as string) == "user" &&
                d.TryGetValue("content", out var c) && c is List<object> contentList)
            {
                var newContent = new List<object>(contentList);
                var newD = new Dictionary<string, object>(d) { ["content"] = newContent };
                bool merged = false;
                for (int j = 0; j < newContent.Count; j++)
                {
                    var json = JsonSerializer.Serialize(newContent[j]);
                    var el = JsonSerializer.Deserialize<JsonElement>(json);
                    if (el.TryGetProperty("type", out var t) && t.GetString() == "text")
                    {
                        var orig = el.TryGetProperty("text", out var txt) ? txt.GetString() ?? "" : "";
                        newContent[j] = new { type = "text", text = orig + "\n\n" + block };
                        merged = true;
                        break;
                    }
                }
                if (!merged) newContent.Add(new { type = "text", text = block });
                copy[i] = newD;
                return copy;
            }
        }
        return copy;
    }

    /// <summary>
    /// Baut den &lt;previous_layer_responses&gt;-Block für MoA: die Antworten der Vorrunde als
    /// Referenz, mit der Anweisung, die eigene Antwort zu VERBESSERN (nicht blind zu kopieren).
    /// </summary>
    private static string BuildPreviousLayerBlock(List<PanelMemberResult> prev, int layer, int total)
    {
        var sb = new StringBuilder();
        sb.AppendLine();
        sb.AppendLine("---");
        sb.AppendLine($"[Multi-Modell-Panel · Runde {layer} von {total}] Andere KI-Agenten haben in der vorherigen Runde "
            + "auf dieselbe Anfrage geantwortet. Nutze ihre Antworten als zusätzliche Referenz: übernimm korrekte Fakten "
            + "und gute Ideen, korrigiere erkennbare Fehler und VERBESSERE deine eigene Antwort. Kopiere nicht blind – "
            + "denke kritisch und liefere die bestmögliche eigenständige Antwort auf die ursprüngliche Anfrage.");
        sb.AppendLine();
        sb.AppendLine("<previous_layer_responses>");
        foreach (var r in prev)
        {
            sb.AppendLine();
            sb.AppendLine($"### Antwort von {r.Name} ({r.Model}):");
            sb.AppendLine(r.Text.Trim());
        }
        sb.AppendLine("</previous_layer_responses>");
        return sb.ToString();
    }
}