Stream panel member responses live with a parallel SSE write-lock

Panel members now stream token-by-token and show tool uses in real time
(like the session agent) instead of delivering text only at completion.
The key constraint is byte-level safety: N members flushing SSE frames
concurrently on the shared HttpResponse would interleave and corrupt the
stream, so a SemaphoreSlim gate serializes the writes (not the model work).

- SseHelper.TryWriteSse takes an optional write-lock (default null, backward
  compatible)
- NonStreamingLoopCallbacks carries SseGate + MemberIndex; _text/_tool/_error
  events are tagged with the member index (review/finalizer stay at -1, no gate)
- CLI members forward TextDelta/ToolStart/ToolResult live instead of
  discarding everything but the final text
- Anthropic-API members stream per tool-round via the loop callbacks; OpenAI
  members fall back to collected text at panel_member_done
- Frontend demuxes by member index into per-bubble interleaved text/tool
  rendering

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MelekiandClaude Opus 4.8 committed 2026-06-18 13:18:13 +02:00
1 parent 268f11fa6d
commit f21ed9ff9f
5 files changed
+183 -34

No files matched your search

+55 -16
View File
@@ -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<object>(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 };
}
/// <summary>
/// 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 <paramref name="sseGate"/> (parallele Member → sonst
/// vermischen sich die Frames). Persistiert nichts – der Caller speichert eine Proposal-Message.
/// </summary>
private async Task<string> RunPanelMemberViaCli(SubAgentContext member, List<object> memberMessages, CancellationToken workCt)
private async Task<string> RunPanelMemberViaCli(
HttpResponse response,
SubAgentContext member,
List<object> memberMessages,
int index,
SemaphoreSlim sseGate,
CancellationToken workCt,
CancellationToken sseCt)
{
if (_cliService == null || !_cliService.IsAvailable) return "";
@@ -199,13 +222,29 @@ public partial class AnthropicService
prompt = $"<conversation_history>\n{historyContext}\n</conversation_history>\n\n<current_request>\n{prompt}\n</current_request>";
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,