diff --git a/Services/AnthropicService.Agents.cs b/Services/AnthropicService.Agents.cs index 6546556..623904b 100644 --- a/Services/AnthropicService.Agents.cs +++ b/Services/AnthropicService.Agents.cs @@ -416,6 +416,15 @@ public partial class AnthropicService public string SsePrefix { get; set; } public string AgentName { get; set; } + /// Optionaler Write-Lock: serialisiert SSE-Writes, wenn mehrere Agenten + /// (Panel-Member) parallel auf DIESELBE HttpResponse streamen. Null = kein Lock. + public SemaphoreSlim SseGate { get; set; } + + /// Member-Index für Panel-Events (wird in die SSE-Payloads gelegt, + /// damit das Frontend Token/Tool-Events der richtigen Member-Bubble zuordnet). + /// -1 = kein Panel-Member (Review/Finalizer ignorieren das Feld). + public int MemberIndex { get; set; } = -1; + /// Tag für DB-Records (z.B. "review", "finalizer"). public string Tag { get; set; } @@ -522,7 +531,7 @@ public partial class AnthropicService Content = errMsg, IsError = true, Tag = cb.Tag }); if (cb.SseResponse != null) - await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_error", new { error = errMsg }, cb.SseCt); + await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_error", new { index = cb.MemberIndex, error = errMsg }, cb.SseCt, cb.SseGate); break; } @@ -560,7 +569,7 @@ public partial class AnthropicService _embeddingBg?.Enqueue(msgId, chainId, sessionId, toolOpts.AgentId, "assistant", text); } if (cb.SseResponse != null) - await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_text", new { text, agentName = cb.AgentName }, cb.SseCt); + await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_text", new { index = cb.MemberIndex, text, agentName = cb.AgentName }, cb.SseCt, cb.SseGate); } else if (blockType == "tool_use") { @@ -619,7 +628,7 @@ public partial class AnthropicService }); if (cb.SseResponse != null) - await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_tool", new { name = toolName, result = TruncateToolResult(toolResult), isError, agentName = cb.AgentName }, cb.SseCt); + await SseHelper.TryWriteSse(cb.SseResponse, $"{cb.SsePrefix}_tool", new { index = cb.MemberIndex, name = toolName, result = TruncateToolResult(toolResult), isError, agentName = cb.AgentName }, cb.SseCt, cb.SseGate); } } diff --git a/Services/AnthropicService.Panel.cs b/Services/AnthropicService.Panel.cs index d1973a6..500661b 100644 --- a/Services/AnthropicService.Panel.cs +++ b/Services/AnthropicService.Panel.cs @@ -52,7 +52,12 @@ public partial class AnthropicService { if (panel == null || panel.Count == 0) return 0; - await SseHelper.TryWriteSse(response, "panel_start", new { members = panel.Count, mode = "fusion" }, sseCt); + // 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); + + await SseHelper.TryWriteSse(response, "panel_start", new { members = panel.Count, mode = "fusion" }, sseCt, sseGate); // Concurrency begrenzen: bei CLI-Membern sind das parallele claude-Children. using var sem = new SemaphoreSlim(Math.Min(3, panel.Count)); @@ -64,7 +69,7 @@ public partial class AnthropicService tasks.Add(Task.Run(async () => { await sem.WaitAsync(workCt); - try { return await RunPanelMember(response, member, workerMessages, index, chainId, sessionId, apiToken, workCt, sseCt); } + try { return await RunPanelMember(response, member, workerMessages, index, chainId, sessionId, apiToken, sseGate, workCt, sseCt); } finally { sem.Release(); } }, workCt)); } @@ -78,7 +83,7 @@ public partial class AnthropicService if (ordered.Count == 0) { - await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = false, members = 0 }, sseCt); + await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = false, members = 0 }, sseCt, sseGate); return 0; } @@ -103,7 +108,7 @@ public partial class AnthropicService MergePanelIntoLastUserText(workerMessages, sb.ToString()); - await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = true, members = ordered.Count }, sseCt); + await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = true, members = ordered.Count }, sseCt, sseGate); return ordered.Count; } @@ -120,11 +125,12 @@ public partial class AnthropicService 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 }, sseCt); + await SseHelper.TryWriteSse(response, "panel_member_start", new { index, name, model = member.ModelId }, sseCt, sseGate); var memberMessages = new List(workerMessages); // eigene Kopie pro Member string text; @@ -139,11 +145,13 @@ public partial class AnthropicService else if (string.IsNullOrEmpty(apiToken) && _cliService != null && _cliService.IsAvailable) { // Kein API-Key → CLI-Member, read-only (Plan-Modus: nur Read/Glob/Grep). - text = await RunPanelMemberViaCli(member, memberMessages, workCt); + // Streamt LIVE (Token + Tool-Uses) über panel_member_*-Events. + text = await RunPanelMemberViaCli(response, member, memberMessages, index, sseGate, workCt, sseCt); } 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. var memberUsage = new TokenUsage(); var (t, _) = await RunNonStreamingToolLoop( member.ModelId, member.SystemPrompt, @@ -155,7 +163,12 @@ public partial class AnthropicService }, memberMessages, chainId, sessionId, apiToken, memberUsage, 0, 12, workCt, - new NonStreamingLoopCallbacks { Persist = false }, + new NonStreamingLoopCallbacks + { + Persist = false, + SseResponse = response, SseCt = sseCt, SseGate = sseGate, + SsePrefix = "panel_member", MemberIndex = index, AgentName = name + }, member.Effort); text = t; } @@ -178,17 +191,27 @@ public partial class AnthropicService } await SseHelper.TryWriteSse(response, "panel_member_done", - new { index, name, model = member.ModelId, text }, sseCt); + new { index, name, model = member.ModelId, text }, sseCt, sseGate); return new PanelMemberResult { Index = index, Name = name, Model = member.ModelId, Text = text }; } /// - /// Führt einen Panel-Member über die Claude CLI aus (read-only via Plan-Modus). - /// Sammelt nur den Text, persistiert nichts und sendet keine Text-Deltas - /// (parallele Member → Deltas würden interleaven; der volle Text kommt via panel_member_done). + /// 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 (parallele Member → sonst + /// vermischen sich die Frames). Persistiert nichts – der Caller speichert eine Proposal-Message. /// - private async Task RunPanelMemberViaCli(SubAgentContext member, List memberMessages, CancellationToken workCt) + private async Task RunPanelMemberViaCli( + HttpResponse response, + SubAgentContext member, + List memberMessages, + int index, + SemaphoreSlim sseGate, + CancellationToken workCt, + CancellationToken sseCt) { if (_cliService == null || !_cliService.IsAvailable) return ""; @@ -199,13 +222,29 @@ public partial class AnthropicService prompt = $"\n{historyContext}\n\n\n\n{prompt}\n"; var fullText = new StringBuilder(); + string currentToolName = null; // ToolResult trägt keinen Namen → vom ToolStart merken await _cliService.ChatStreamingAsync( prompt, - evt => + async evt => { - if (evt.Type == CliEventType.TextDelta) - fullText.Append(evt.Text); - return Task.CompletedTask; + switch (evt.Type) + { + case CliEventType.TextDelta: + fullText.Append(evt.Text); + await SseHelper.TryWriteSse(response, "panel_member_text", + new { index, text = evt.Text }, sseCt, sseGate); + break; + case CliEventType.ToolStart: + currentToolName = evt.ToolName; + await SseHelper.TryWriteSse(response, "panel_member_tool_use", + new { index, name = evt.ToolName }, sseCt, sseGate); + break; + case CliEventType.ToolResult: + await SseHelper.TryWriteSse(response, "panel_member_tool_end", + new { index, name = currentToolName, result = evt.ToolResultContent, isError = evt.ToolResultIsError }, sseCt, sseGate); + currentToolName = null; + break; + } }, modelId: member.ModelId, systemPrompt: systemText, diff --git a/Services/SseHelper.cs b/Services/SseHelper.cs index a1f25dd..e8831b4 100644 --- a/Services/SseHelper.cs +++ b/Services/SseHelper.cs @@ -25,17 +25,30 @@ public static class SseHelper /// /// Versucht ein SSE-Event zu schreiben, ignoriert Fehler (Client disconnected). + /// Optionaler : serialisiert die Byte-Writes, wenn mehrere + /// Producer (z.B. parallele Panel-Member) auf DIESELBE HttpResponse schreiben – + /// sonst vermischen sich die SSE-Frames auf Byte-Ebene. /// - public static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct) + public static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct, SemaphoreSlim gate = null) { + if (ct.IsCancellationRequested) return; + bool acquired = false; try { - if (!ct.IsCancellationRequested) - await WriteSseEvent(response, eventType, data, ct); + if (gate != null) + { + await gate.WaitAsync(ct); + acquired = true; + } + await WriteSseEvent(response, eventType, data, ct); } catch { - // Client bereits disconnected – ignorieren + // Client bereits disconnected / abgebrochen – ignorieren + } + finally + { + if (acquired) gate.Release(); } } diff --git a/wwwroot/js/chat.js b/wwwroot/js/chat.js index 8bbdac1..9d079e5 100644 --- a/wwwroot/js/chat.js +++ b/wwwroot/js/chat.js @@ -128,7 +128,7 @@ function handleStreamEvent(eventType, data) { finalizeReviewBubble('done', data, 'finalizer'); break; - // Multi-Modell-Panel (Fusion): mehrere Member parallel, non-streaming + // Multi-Modell-Panel (Fusion): mehrere Member streamen parallel LIVE case 'panel_start': startPanel(data); break; @@ -137,6 +137,23 @@ function handleStreamEvent(eventType, data) { addPanelMemberBubble(data); break; + case 'panel_member_text': + appendPanelMemberText(data); + break; + + case 'panel_member_tool_use': + addPanelMemberToolUse(data); + break; + + // tool_end = CLI (offene Box füllen); tool = Anthropic-Loop (kombiniert → fertige Box) + case 'panel_member_tool_end': case 'panel_member_tool': + finishPanelMemberTool(data); + break; + + case 'panel_member_error': + appendPanelMemberText({ index: data.index, text: '\n\n_[Fehler: ' + (data.error || 'unbekannt') + ']_' }); + break; + case 'panel_member_done': finalizePanelMember(data); break; diff --git a/wwwroot/js/stream-ui.js b/wwwroot/js/stream-ui.js index 52ed6da..8d623f5 100644 --- a/wwwroot/js/stream-ui.js +++ b/wwwroot/js/stream-ui.js @@ -206,9 +206,10 @@ function finalizeReviewBubble(type, data, agentType) { scrollDown(); } -// ── Panel Bubbles (Fusion: mehrere Member parallel, non-streaming) ── -// Member liefern ihren Text gesammelt bei panel_member_done → pro Member eine eigene -// Bubble (keyed by index), kein Konflikt mit dem Single-_reviewBubble-State. +// ── Panel Bubbles (Fusion: mehrere Member streamen parallel LIVE) ── +// Pro Member eine eigene Bubble (keyed by index). Innerhalb der Bubble werden Text +// und Tool-Boxen interleaved gerendert (wie beim Session-Agent), sodass parallele +// Member sich NICHT miteinander vermischen – das Demuxing macht der index der Events. let _panelBubbles = {}; @@ -222,21 +223,91 @@ function addPanelMemberBubble(data) { bubble.className = 'msg review-bubble panel-bubble streamed'; var model = (typeof prettyModelName === 'function') ? prettyModelName(data.model) : data.model; bubble.appendChild(_createSubAgentHeader('panel', data.name, { model: model })); - var content = document.createElement('div'); - content.className = 'review-content markdown'; - content.innerHTML = ' denkt nach…'; - bubble.appendChild(content); - bubble._panelContent = content; + var spinner = document.createElement('div'); + spinner.className = 'review-content markdown'; + spinner.innerHTML = ' denkt nach…'; + bubble.appendChild(spinner); + bubble._panelSpinner = spinner; // wird beim ersten Text/Tool entfernt + bubble._segText = null; // aktuell offenes Text-Segment (oder null) + bubble._segTextStr = ''; // dessen akkumulierter Text + bubble._tool = null; // aktuell offene Tool-Box (CLI: Spinner→Füllen) + bubble._fullText = ''; // gesamter gestreamter Text (für Finalize-Fallback) _panelBubbles[data.index] = bubble; chatEl().appendChild(bubble); scrollDown(); } +function _panelClearSpinner(bubble) { + if (bubble._panelSpinner) { bubble._panelSpinner.remove(); bubble._panelSpinner = null; } +} + +function appendPanelMemberText(data) { + var bubble = _panelBubbles[data.index]; + if (!bubble || !data.text) return; + _panelClearSpinner(bubble); + // Neues Text-Segment beginnen, wenn keins offen ist (z.B. direkt nach einer Tool-Box) + if (!bubble._segText) { + bubble._segText = document.createElement('div'); + bubble._segText.className = 'review-content markdown'; + bubble._segTextStr = ''; + bubble.appendChild(bubble._segText); + } + bubble._segTextStr += data.text; + bubble._fullText += data.text; + bubble._segText.innerHTML = renderMarkdown(bubble._segTextStr); + scrollDown(); +} + +function addPanelMemberToolUse(data) { + var bubble = _panelBubbles[data.index]; + if (!bubble) return; + _panelClearSpinner(bubble); + bubble._segText = null; // Text-Segment schließen → Folge-Text kommt UNTER die Tool-Box + if (!isToolVisible(data.name)) { bubble._tool = null; return; } + var toolDiv = document.createElement('div'); + toolDiv.className = 'tool-box review-tool-box streamed'; + toolDiv.dataset.toolName = data.name; + var displayName = TOOL_DISPLAY_NAMES[data.name] || data.name; + toolDiv.innerHTML = '' + esc(displayName) + ' '; + bubble.appendChild(toolDiv); + bubble._tool = toolDiv; + scrollDown(); +} + +function finishPanelMemberTool(data) { + var bubble = _panelBubbles[data.index]; + if (!bubble) return; + _panelClearSpinner(bubble); + bubble._segText = null; // nach Tool-Result neuen Text-Block beginnen + if (bubble._tool) { + // CLI: offene Box (mit Spinner) füllen + renderToolBoxContent(bubble._tool, data.name, null, data.result, data.isError, null); + bubble._tool = null; + } else { + // Anthropic-Loop (kombiniertes panel_member_tool-Event): fertige Box direkt anlegen + if (!isToolVisible(data.name)) return; + var toolDiv = document.createElement('div'); + toolDiv.className = 'tool-box review-tool-box streamed'; + bubble.appendChild(toolDiv); + renderToolBoxContent(toolDiv, data.name, null, data.result, data.isError, null); + } + scrollDown(); +} + function finalizePanelMember(data) { var bubble = _panelBubbles[data.index]; if (!bubble) return; - var text = (data.text && data.text.trim()) ? data.text : '_(keine Antwort)_'; - bubble._panelContent.innerHTML = renderMarkdown(text); + _panelClearSpinner(bubble); + // Wenn LIVE nichts gestreamt wurde (z.B. OpenAI-Member = one-shot): finalen Text rendern. + if (!bubble._fullText || !bubble._fullText.trim()) { + var text = (data.text && data.text.trim()) ? data.text : '_(keine Antwort)_'; + if (!bubble._segText) { + bubble._segText = document.createElement('div'); + bubble._segText.className = 'review-content markdown'; + bubble.appendChild(bubble._segText); + } + bubble._segText.innerHTML = renderMarkdown(text); + } delete _panelBubbles[data.index]; scrollDown(); }