Persist panel member token costs to DB via usage seed
Replace the separate panel_usage SSE event + frontend accumulator with server-side seeding of the worker's usage object. Both the done-event and FinalizeChain read from the same usage, so panel (Fusion/MoA) member tokens now flow into the live badge AND the DB as a single source of truth — the badge no longer drops to the worker-only value after F5. - RunPanel takes a TokenUsage usageSink; accumulates member costs after the Task.WhenAll barrier (single-threaded, no lock) - ChatEndpoint passes panelUsage as seedUsage to all three worker paths - StreamChat / StreamChatViaCli / OpenAI StreamChat fold seedUsage into usage at creation time, so every exit path persists it - OpenAI done-events extended with cacheCreation/cacheRead fields for live consistency when seeded by Anthropic members - chat.js: remove redundant _panelUsage / _addPanelUsage / panel_usage - TODO.md: mark the live-only v1 boundary closed, note the new mechanism Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
1 parent
1b203707ec
commit
5c89adb6df
7 files changed
+41
-46
No files matched your search
@@ -236,13 +236,19 @@ public static class ChatEndpoint
|
||||
// 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, workCt, sseCt);
|
||||
await anthropic.RunPanel(res, ctx.PanelAgents, messages, ctx.ChainId, sessionId, userApiKey?.ApiKey, panelLayers, panelUsage, workCt, sseCt);
|
||||
}
|
||||
catch (OperationCanceledException) { throw; }
|
||||
catch (Exception ex)
|
||||
@@ -265,7 +271,7 @@ public static class ChatEndpoint
|
||||
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, workCt: workCt, sseCt: sseCt);
|
||||
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
|
||||
{
|
||||
@@ -286,7 +292,7 @@ public static class ChatEndpoint
|
||||
|
||||
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, webSearch: !string.IsNullOrEmpty(ctx.WebSearchVersion), sessionId: sessionId, agentId: ctx.AgentId, remoteMgr: remoteMgr, unityMcpUrl: ctx.Session?.UnityMcpUrl, unityMcpToken: ctx.Session?.UnityMcpToken, useAutoContext: ctx.Agent?.UseAutoContext ?? true, effort: ctx.Effort, workCt: workCt, sseCt: sseCt);
|
||||
await anthropic.StreamChatViaCli(res, ctx.ModelId, ctx.SystemPrompt, messages, ctx.ChainId, workspace: ctx.Workspace, allowUserChoice: ctx.Agent?.AllowUserChoice ?? true, planMode: body.PlanMode, webSearch: !string.IsNullOrEmpty(ctx.WebSearchVersion), sessionId: sessionId, agentId: ctx.AgentId, remoteMgr: remoteMgr, unityMcpUrl: ctx.Session?.UnityMcpUrl, unityMcpToken: ctx.Session?.UnityMcpToken, useAutoContext: ctx.Agent?.UseAutoContext ?? true, effort: ctx.Effort, seedUsage: panelUsage, workCt: workCt, sseCt: sseCt);
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -296,7 +302,7 @@ public static class ChatEndpoint
|
||||
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, workCt: workCt, hardCt: hardCt, sseCt: sseCt, remoteMgr: remoteMgr);
|
||||
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, seedUsage: panelUsage, workCt: workCt, hardCt: hardCt, sseCt: sseCt, remoteMgr: remoteMgr);
|
||||
}
|
||||
}
|
||||
finally
|
||||
|
||||
@@ -38,6 +38,7 @@ public partial class AnthropicService
|
||||
string unityMcpToken = null,
|
||||
bool useAutoContext = true,
|
||||
string effort = null,
|
||||
TokenUsage seedUsage = null,
|
||||
CancellationToken workCt = default,
|
||||
CancellationToken sseCt = default)
|
||||
{
|
||||
@@ -50,6 +51,9 @@ public partial class AnthropicService
|
||||
|
||||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||
var usage = new TokenUsage();
|
||||
// Panel-Member-Tokens (Fusion/MoA) vorab in die Chain-Usage falten → erscheinen im
|
||||
// done-Event (Live-Badge) UND in FinalizeChain (DB, überlebt F5). Null/leer = No-op.
|
||||
if (seedUsage != null) usage.Add(seedUsage);
|
||||
var fullText = new StringBuilder();
|
||||
var tempImagePaths = new List<string>();
|
||||
string mcpConfigPath = null;
|
||||
|
||||
@@ -54,6 +54,7 @@ public partial class AnthropicService
|
||||
string sessionId,
|
||||
string apiToken,
|
||||
int layers,
|
||||
TokenUsage usageSink,
|
||||
CancellationToken workCt,
|
||||
CancellationToken sseCt)
|
||||
{
|
||||
@@ -68,22 +69,12 @@ public partial class AnthropicService
|
||||
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. Jeder Member sammelt seine eigene Usage
|
||||
// (result.Usage); summiert wird erst NACH der Task.WhenAll-Barrier → kein Lock nötig.
|
||||
// Vor jedem panel_done als panel_usage-Event ans Frontend → addiert auf die Worker-Tokens.
|
||||
var panelUsage = new TokenUsage();
|
||||
async Task SendPanelUsage()
|
||||
{
|
||||
if (panelUsage.InputTokens == 0 && panelUsage.OutputTokens == 0
|
||||
&& panelUsage.CacheReadTokens == 0 && panelUsage.CacheCreationTokens == 0) return;
|
||||
await SseHelper.TryWriteSse(response, "panel_usage", new
|
||||
{
|
||||
inputTokens = panelUsage.InputTokens,
|
||||
outputTokens = panelUsage.OutputTokens,
|
||||
cacheCreationTokens = panelUsage.CacheCreationTokens,
|
||||
cacheReadTokens = panelUsage.CacheReadTokens
|
||||
}, 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.
|
||||
@@ -127,8 +118,8 @@ public partial class AnthropicService
|
||||
if (ordered.Count == 0)
|
||||
{
|
||||
// Keine verwertbaren Antworten in dieser Runde → ohne Synthese weiter.
|
||||
// Trotzdem die bisher verbrauchten Member-Tokens melden (ehrliche Kosten).
|
||||
await SendPanelUsage();
|
||||
// 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;
|
||||
}
|
||||
@@ -156,7 +147,6 @@ public partial class AnthropicService
|
||||
|
||||
MergePanelIntoLastUserText(workerMessages, sb.ToString());
|
||||
|
||||
await SendPanelUsage();
|
||||
await SseHelper.TryWriteSse(response, "panel_done", new { synthesizing = true, members = prev.Count }, sseCt, sseGate);
|
||||
return prev.Count;
|
||||
}
|
||||
|
||||
@@ -79,10 +79,14 @@ public partial class AnthropicService
|
||||
CancellationToken workCt = default,
|
||||
CancellationToken hardCt = default,
|
||||
CancellationToken sseCt = default,
|
||||
RemoteConnectionManager remoteMgr = null)
|
||||
RemoteConnectionManager remoteMgr = null,
|
||||
TokenUsage seedUsage = null)
|
||||
{
|
||||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||
var usage = new TokenUsage();
|
||||
// Panel-Member-Tokens (Fusion/MoA) vorab in die Chain-Usage falten → erscheinen im
|
||||
// done-Event (Live-Badge) UND in FinalizeChain (DB, überlebt F5). Null/leer = No-op.
|
||||
if (seedUsage != null) usage.Add(seedUsage);
|
||||
int iterations = 0;
|
||||
int reviewRound = 0;
|
||||
string lastStopReason = "";
|
||||
|
||||
@@ -298,11 +298,15 @@ public class OpenAiService
|
||||
int maxIterations = 10,
|
||||
string apiKeyOverride = null,
|
||||
bool useAutoContext = true,
|
||||
TokenUsage seedUsage = null,
|
||||
CancellationToken workCt = default,
|
||||
CancellationToken sseCt = default)
|
||||
{
|
||||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||
var usage = new TokenUsage();
|
||||
// Panel-Member-Tokens (Fusion/MoA) vorab in die Chain-Usage falten → erscheinen im
|
||||
// done-Event (Live-Badge) UND in FinalizeChain (DB, überlebt F5). Null/leer = No-op.
|
||||
if (seedUsage != null) usage.Add(seedUsage);
|
||||
var iteration = 0;
|
||||
var autoContinueCount = 0;
|
||||
const int MaxAutoContinues = 3;
|
||||
@@ -884,6 +888,8 @@ public class OpenAiService
|
||||
model = modelString,
|
||||
inputTokens = usage.InputTokens,
|
||||
outputTokens = usage.OutputTokens,
|
||||
cacheCreationTokens = usage.CacheCreationTokens,
|
||||
cacheReadTokens = usage.CacheReadTokens,
|
||||
iterations = iteration,
|
||||
chainId,
|
||||
durationMs = sw.ElapsedMilliseconds,
|
||||
@@ -1038,6 +1044,8 @@ public class OpenAiService
|
||||
model = modelString,
|
||||
inputTokens = usage.InputTokens,
|
||||
outputTokens = usage.OutputTokens,
|
||||
cacheCreationTokens = usage.CacheCreationTokens,
|
||||
cacheReadTokens = usage.CacheReadTokens,
|
||||
iterations = iteration,
|
||||
chainId,
|
||||
durationMs = sw.ElapsedMilliseconds,
|
||||
|
||||
@@ -56,4 +56,4 @@
|
||||
|
||||
## Multi-Modell-Panel (Fusion/MoA)
|
||||
- [x] 🔄 MoA-Zwischenrunden persistieren (F5 überleben) - `RunPanelMember` (`Services/AnthropicService.Panel.cs`) persistiert jetzt JEDE Proposer-Runde (statt nur der finalen) mit `layer`/`totalLayers` im JSON-Transkript; `wwwroot/js/history.js` gruppiert die Panel-Bubbles nach Layer und rendert die "Runde k/N"-Divider beim Reload.
|
||||
- [x] 💰 Member-Token in die Chain-Kosten - Jeder Panel-Member sammelt seine eigene `TokenUsage` (Anthropic-Loop via `memberUsage`, CLI via `CliChatResult`; OpenAI=null/nachrüstbar); `RunPanel` summiert sie NACH der `Task.WhenAll`-Barrier (single-threaded → kein Lock) über alle Runden und sendet ein `panel_usage`-SSE-Event. Das Frontend (`wwwroot/js/chat.js`) addiert die Panel-Tokens im `done`-Event auf die Worker-Tokens → Badge = Gesamtkosten. v1-Grenze: nur Live-Badge (DB/Reload zeigt weiter nur Worker-Tokens, da `FinalizeChain` unverändert).
|
||||
- [x] 💰 Member-Token in die Chain-Kosten - Jeder Panel-Member sammelt seine eigene `TokenUsage` (Anthropic-Loop via `memberUsage`, CLI via `CliChatResult`; OpenAI=null/nachrüstbar); `RunPanel` summiert sie NACH der `Task.WhenAll`-Barrier (single-threaded → kein Lock) über alle Runden in einen `usageSink`. Der Dispatch (`Endpoints/ChatEndpoint.cs`) reicht diesen als `seedUsage` an die Worker-Streaming-Methode weiter, die ihn in ihre `usage` faltet → Panel-Kosten fließen über das `done`-Event (Live-Badge) UND `FinalizeChain` (DB) als Teil der Chain-Kosten. **Single Source of Truth**, überlebt F5; das frühere `panel_usage`-SSE + Frontend-Addition wurde dadurch entfernt.
|
||||
+4
-21
@@ -170,11 +170,6 @@ function handleStreamEvent(eventType, data) {
|
||||
endPanel(data);
|
||||
break;
|
||||
|
||||
case 'panel_usage':
|
||||
// Panel-Member-Tokens merken → im done-Event auf die Worker-Tokens addieren.
|
||||
_panelUsage = data;
|
||||
break;
|
||||
|
||||
case 'auto_context':
|
||||
addAutoContextBubble(data.context);
|
||||
break;
|
||||
@@ -204,18 +199,17 @@ function handleStreamEvent(eventType, data) {
|
||||
if (data.stopReason === 'max_tokens') {
|
||||
addWarningBubble('Antwort wurde abgeschnitten (max_tokens erreicht). Bitte frage nach dem Rest.', true);
|
||||
}
|
||||
// Panel-Member-Tokens (live gesammelt) auf die Worker-Tokens addieren → Badge = Gesamtkosten.
|
||||
var _pu = _panelUsage;
|
||||
_panelUsage = null;
|
||||
// Panel-Member-Tokens (Fusion/MoA) sind server-seitig bereits in die Worker-Usage gefaltet
|
||||
// (seedUsage) → das done-Event trägt die Gesamtkosten, und FinalizeChain persistiert sie
|
||||
// (überlebt F5). Hier nichts mehr zu addieren.
|
||||
_lastDoneChainData = null;
|
||||
if (data.model) {
|
||||
_addPanelUsage(data, _pu);
|
||||
_lastDoneChainData = data;
|
||||
} else if (data.chainId) {
|
||||
// Fallback: Chain-Daten vom Server nachladen und Badge direkt rendern
|
||||
// (async – _lastDoneChainData wäre im finally schon gelesen)
|
||||
apiGetChain(data.chainId).then(chain => {
|
||||
if (chain) { _addPanelUsage(chain, _pu); addChainMetaBubble(chain, false); }
|
||||
if (chain) { addChainMetaBubble(chain, false); }
|
||||
}).catch(() => {});
|
||||
}
|
||||
setMeta(data.stopReason === 'stopped' ? 'Gestoppt' : 'Bereit');
|
||||
@@ -318,17 +312,7 @@ async function cycleCompositeMode() {
|
||||
|
||||
let isStreaming = false;
|
||||
let _lastDoneChainData = null; // Chain-Daten vom letzten done-Event
|
||||
let _panelUsage = null; // Panel-Member-Tokens (panel_usage-Event) → werden auf die Worker-Tokens des done-Events addiert
|
||||
|
||||
// Addiert die live gesammelten Panel-Member-Tokens auf ein Chain-/done-Objekt, sodass das
|
||||
// Token-Badge die Gesamtkosten (Worker + alle Panel-Member aller Runden) zeigt.
|
||||
function _addPanelUsage(target, pu) {
|
||||
if (!target || !pu) return;
|
||||
target.inputTokens = (target.inputTokens || 0) + (pu.inputTokens || 0);
|
||||
target.outputTokens = (target.outputTokens || 0) + (pu.outputTokens || 0);
|
||||
target.cacheCreationTokens = (target.cacheCreationTokens || 0) + (pu.cacheCreationTokens || 0);
|
||||
target.cacheReadTokens = (target.cacheReadTokens || 0) + (pu.cacheReadTokens || 0);
|
||||
}
|
||||
let _currentToolBox = null; // Aktuell laufende Tool-Box (für Stream-Events)
|
||||
let _lastFinishedToolBox = null; // Letzte fertige Tool-Box (für tool_result Zuordnung)
|
||||
let _chatAbandoned = false; // true wenn User weg-navigiert während Stream läuft
|
||||
@@ -414,7 +398,6 @@ async function _runStream(fetchPromise) {
|
||||
sendBtn.hidden = true;
|
||||
stopBtn.hidden = false;
|
||||
isStreaming = true;
|
||||
_panelUsage = null; // frische Chain → alte Panel-Tokens verwerfen
|
||||
_abortController = new AbortController();
|
||||
// Eine Live-SSE-Antwort rendert ALLEIN. Einen evtl. noch laufenden Recovery-Poll
|
||||
// (_streamingPollInterval – gestartet beim Betreten des Chats WAEHREND eine Chain
|
||||
|
||||
Reference in new issue
Block a user