diff --git a/Models/ChainRecord.cs b/Models/ChainRecord.cs index f7e98ce..ebe4389 100644 --- a/Models/ChainRecord.cs +++ b/Models/ChainRecord.cs @@ -20,6 +20,7 @@ public class ChainRecord public string ArchiveId { get; set; } public string StreamErrorInfo { get; set; } public string WorkflowInfo { get; set; } // JSON: Telemetrie eines dynamischen Workflows (ultracode), null = kein Workflow + public string WorkflowRunId { get; set; } // runId (wf_…) des zuletzt gestarteten Workflows – für Resume nach Abbruch public bool PlanMode { get; set; } public string Effort { get; set; } public string CreatedAt { get; set; } diff --git a/Services/AnthropicService.Cli.cs b/Services/AnthropicService.Cli.cs index 3f92e99..d5f4669 100644 --- a/Services/AnthropicService.Cli.cs +++ b/Services/AnthropicService.Cli.cs @@ -54,6 +54,19 @@ public partial class AnthropicService // 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); + // Token-Fix bei Abbruch: Die Token stammen aus dem finalen result-Event – das kommt bei einem + // Abbruch NIE, also bliebe usage 0. Wir akkumulieren die Orchestrator-Usage live aus den + // assistant-Turns (CliEventType.Usage) in partialUsage und nehmen bei Abbruch/Zwischen-Persistenz + // das jeweils Größere pro Feld (kein Doppelzählen mit dem normalen result-Pfad). + var partialUsage = new TokenUsage(); + TokenUsage EffectiveUsage() => new TokenUsage + { + InputTokens = Math.Max(usage.InputTokens, partialUsage.InputTokens), + OutputTokens = Math.Max(usage.OutputTokens, partialUsage.OutputTokens), + CacheCreationTokens = Math.Max(usage.CacheCreationTokens, partialUsage.CacheCreationTokens), + CacheReadTokens = Math.Max(usage.CacheReadTokens, partialUsage.CacheReadTokens) + }; + var lastWfFlush = DateTime.MinValue; // Throttle für die inkrementelle Workflow-Persistenz var fullText = new StringBuilder(); var tempImagePaths = new List(); string mcpConfigPath = null; @@ -119,6 +132,32 @@ public partial class AnthropicService } } + // ── Resume-Hinweis: Lief in dieser Session zuletzt ein Workflow, der ABGEBROCHEN wurde, + // injizieren wir seine runId in den System-Prompt. Der Agent kann den Workflow dann mit + // resumeFromRunId erneut starten → bereits fertige Subagenten werden aus dem Journal + // wiederverwendet statt neu ausgeführt. Best-effort (hängt an LLM-Kooperation). Nur bei + // ultracode relevant (nur dann kann der Agent überhaupt das Workflow-Tool nutzen). + if (ultracode && !string.IsNullOrEmpty(sessionId)) + { + try + { + var priorChains = await _db.GetActiveChains(sessionId); // ORDER BY id DESC + // Nur der ZULETZT gelaufene Workflow zählt: ist er abgeschlossen → kein Resume; + // wurde er abgebrochen → Hinweis anbieten. + var lastWf = priorChains.FirstOrDefault(c => + c.ChainId != chainId && !string.IsNullOrEmpty(c.WorkflowRunId)); + if (lastWf != null && lastWf.StopReason != "end_turn" + && !(lastWf.WorkflowInfo != null && lastWf.WorkflowInfo.Contains("\"Status\":\"completed\""))) + { + systemText.AppendLine( + $"\n\nIn dieser Session wurde zuletzt ein dynamischer Workflow (runId: {lastWf.WorkflowRunId}) vorzeitig beendet (Grund: {lastWf.StopReason ?? "unbekannt"}) und ist unvollständig. Falls die aktuelle Aufgabe die Fortsetzung genau dieses Workflows ist: starte ihn erneut mit DEMSELBEN Skript und übergib dem Workflow-Tool resumeFromRunId: \"{lastWf.WorkflowRunId}\" – bereits abgeschlossene Subagenten werden dann aus dem Journal wiederverwendet statt neu ausgeführt (spart Token und Zeit). Ist die aktuelle Aufgabe NICHT die Fortsetzung, ignoriere diesen Hinweis vollständig.\n"); + await SseHelper.TryWriteSse(response, "workflow_resume_available", + new { runId = lastWf.WorkflowRunId, stopReason = lastWf.StopReason }, sseCt); + } + } + catch (Exception ex) { Console.WriteLine($"[Wf-Resume/CLI] Error: {ex.Message}"); } + } + // Letzte User-Message als Prompt extrahieren var prompt = ExtractLastUserText(messages) ?? ""; @@ -282,7 +321,38 @@ public partial class AnthropicService } break; } + case CliEventType.Usage: + // Orchestrator-Usage live akkumulieren (Token-Fix-Fallback für Abbruch): + // Output/Cache summieren (pro Turn additiv), Input = größter Turn-Wert + // (Kontextgröße wächst pro Turn – Summieren würde massiv doppelt zählen). + partialUsage.OutputTokens += evt.OutputTokens; + partialUsage.CacheReadTokens += evt.CacheReadTokens; + partialUsage.CacheCreationTokens += evt.CacheCreationTokens; + if (evt.InputTokens > partialUsage.InputTokens) partialUsage.InputTokens = evt.InputTokens; + break; } + // Inkrementelle Zwischen-Persistenz (throttled) – so überlebt der letzte Workflow-Stand + // (Subagent-Outputs + Token) auch einen Hard-Kill (OOM/Redeploy), bei dem KEIN catch läuft. + await MaybeFlushWorkflowProgress(); + } + + // Persistiert – höchstens alle ~6 s – die aktuelle Workflow-Telemetrie dieser Chain in die DB. + async Task MaybeFlushWorkflowProgress() + { + if (_workflowTracker == null || string.IsNullOrEmpty(sessionId)) return; + if (_workflowTracker.ChainIdFor(sessionId) != chainId) return; // nur bei Workflow FÜR diese Chain + var now = DateTime.UtcNow; + if ((now - lastWfFlush).TotalSeconds < 6) return; // Throttle + lastWfFlush = now; + try + { + var run = _workflowTracker.Get(sessionId); + var wp = WorkflowTelemetryReader.Read(run); + if (wp == null) return; + wp.ChainId = chainId; + await _db.UpdateChainWorkflowProgress(chainId, JsonSerializer.Serialize(wp), run?.RunId, EffectiveUsage()); + } + catch { /* inkrementelle Persistenz darf den Stream nie stören */ } } do @@ -440,7 +510,7 @@ public partial class AnthropicService // Wenn ask_user_choice aktiv war: als awaiting_user_choice markieren, // damit die nächste Chain den Kontext (Frage + Optionen) behält. var stoppedReason = isUserChoice ? "awaiting_user_choice" : "stopped"; - await FinalizeChain(chain, usage, 1, stoppedReason, sw); + await FinalizeChain(chain, EffectiveUsage(), 1, stoppedReason, sw); await SseHelper.TryWriteSse(response, "done", new { model = modelId, @@ -456,7 +526,7 @@ public partial class AnthropicService // Client-Disconnect (sseCt gecancelled, aber NICHT workCt) // CLI-Prozess läuft weiter im Hintergrund – hier nur Chain finalisieren // (sollte nicht mehr passieren, da CLI jetzt workCt nutzt) - await FinalizeChain(chain, usage, 1, "client_disconnected", sw); + await FinalizeChain(chain, EffectiveUsage(), 1, "client_disconnected", sw); } catch (Exception ex) { @@ -470,7 +540,7 @@ public partial class AnthropicService Content = errorMsg }); await SseHelper.TryWriteSse(response, "error", new { error = errorMsg }, sseCt); - await FinalizeChain(chain, usage, 1, "error", sw); + await FinalizeChain(chain, EffectiveUsage(), 1, "error", sw); await SseHelper.TryWriteSse(response, "done", new { model = modelId, diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index 1654b82..5e162a8 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -332,13 +332,17 @@ public partial class AnthropicService try { // Reiche End-Telemetrie aus der .json des Laufs lesen und persistieren - // (Fallback: Live-Counts aus dem journal.jsonl, falls die json noch nicht da ist). - var wp = WorkflowTelemetryReader.Read(_workflowTracker.Get(chain.SessionId)); + // (Fallback: Live-Counts + gerettete Subagent-Outputs aus dem journal.jsonl, falls + // die json noch nicht da ist – z.B. bei Abbruch). + var run = _workflowTracker.Get(chain.SessionId); + var wp = WorkflowTelemetryReader.Read(run); if (wp != null) { wp.ChainId = chain.ChainId; chain.WorkflowInfo = JsonSerializer.Serialize(wp); } + // runId als Resume-Anker festhalten (auch wenn der Lauf abgebrochen wurde). + if (run != null) chain.WorkflowRunId = run.RunId; } catch { /* Persistenz der Telemetrie darf den Finalize nie crashen */ } _workflowTracker.End(chain.SessionId); diff --git a/Services/ClaudeCliService.cs b/Services/ClaudeCliService.cs index 543ce27..c1f8b44 100644 --- a/Services/ClaudeCliService.cs +++ b/Services/ClaudeCliService.cs @@ -803,6 +803,24 @@ public class ClaudeCliService if (isError) result.Error = json.TryGetProperty("result", out var errMsg) ? errMsg.GetString() : "CLI error"; } + else if (type == "assistant" && json.TryGetProperty("message", out var asMsg) + && asMsg.TryGetProperty("usage", out var asUsage)) + { + // Konsolidierter assistant-Turn des HAUPT-Streams (Subagenten sind durch den + // parent_tool_use_id-Guard oben bereits ausgefiltert). message.usage liefert die + // Live-Token dieses Turns – die einzige Token-Quelle VOR dem finalen result-Event. + // Wir emittieren sie, damit der Aufrufer bei einem Abbruch (kein result-Event) + // nicht 0 Token persistiert. Der Stream selbst bleibt unberührt. + long AInt(string n) => asUsage.TryGetProperty(n, out var v) && v.ValueKind == JsonValueKind.Number ? v.GetInt64() : 0L; + await onEvent(new CliStreamEvent + { + Type = CliEventType.Usage, + InputTokens = AInt("input_tokens"), + OutputTokens = AInt("output_tokens"), + CacheReadTokens = AInt("cache_read_input_tokens"), + CacheCreationTokens = AInt("cache_creation_input_tokens") + }); + } else if (type == "user" && json.TryGetProperty("message", out var userMsg)) { // CLI sendet tool_result als type:"user" message @@ -1120,7 +1138,8 @@ public enum CliEventType TextEnd, ToolStart, ToolEnd, - ToolResult + ToolResult, + Usage // Orchestrator-Usage eines assistant-Turns (Live-Akkumulation, Token-Fix bei Abbruch) } public class CliStreamEvent @@ -1132,6 +1151,11 @@ public class CliStreamEvent public string ToolInput { get; set; } public string ToolResultContent { get; set; } public bool ToolResultIsError { get; set; } + // Usage-Event: per-Turn-Usage des Haupt-Streams (kein Subagent – parent_tool_use_id ist null) + public long InputTokens { get; set; } + public long OutputTokens { get; set; } + public long CacheReadTokens { get; set; } + public long CacheCreationTokens { get; set; } } public class CliChatResult diff --git a/Services/DatabaseService.Sessions.cs b/Services/DatabaseService.Sessions.cs index 7c0d6ee..8b7419f 100644 --- a/Services/DatabaseService.Sessions.cs +++ b/Services/DatabaseService.Sessions.cs @@ -255,11 +255,42 @@ public partial class DatabaseService stop_reason = @StopReason, stream_error_info = @StreamErrorInfo, workflow_info = @WorkflowInfo, + workflow_run_id = @WorkflowRunId, duration_ms = @DurationMs WHERE chain_id = @ChainId ", chain); } + /// + /// Inkrementelle Zwischen-Persistenz eines laufenden Workflows: schreibt NUR die Workflow-Telemetrie + /// und die (partiellen) Token, ohne stop_reason/duration/iterations zu berühren. Wird während des + /// Laufs periodisch aufgerufen, damit bei einem Hard-Kill (OOM/Redeploy) der letzte Zwischenstand + /// – inkl. bereits fertiger Subagent-Outputs und verbrauchter Token – erhalten bleibt. + /// + public async Task UpdateChainWorkflowProgress(string chainId, string workflowInfo, string workflowRunId, TokenUsage usage) + { + using var conn = CreateConnection(); + await conn.ExecuteAsync(@" + UPDATE chains SET + workflow_info = @workflowInfo, + workflow_run_id = @workflowRunId, + input_tokens = @InputTokens, + output_tokens = @OutputTokens, + cache_creation_tokens = @CacheCreationTokens, + cache_read_tokens = @CacheReadTokens + WHERE chain_id = @chainId + ", new + { + chainId, + workflowInfo, + workflowRunId, + usage.InputTokens, + usage.OutputTokens, + usage.CacheCreationTokens, + usage.CacheReadTokens + }); + } + public async Task GetChain(string chainId) { using var conn = CreateConnection(); diff --git a/Services/DatabaseService.cs b/Services/DatabaseService.cs index b4f37c0..2fb1cbc 100644 --- a/Services/DatabaseService.cs +++ b/Services/DatabaseService.cs @@ -149,6 +149,7 @@ public partial class DatabaseService catch { /* already migrated */ } TryAddColumn(conn, "chains", "stream_error_info", "TEXT"); TryAddColumn(conn, "chains", "workflow_info", "TEXT"); + TryAddColumn(conn, "chains", "workflow_run_id", "TEXT"); TryAddColumn(conn, "agents", "use_soul", "INTEGER NOT NULL DEFAULT 1"); TryAddColumn(conn, "agents", "use_identity", "INTEGER NOT NULL DEFAULT 1"); TryAddColumn(conn, "agents", "use_user_info", "INTEGER NOT NULL DEFAULT 1"); diff --git a/Services/WorkflowProgressTracker.cs b/Services/WorkflowProgressTracker.cs index 886f5e1..ada91af 100644 --- a/Services/WorkflowProgressTracker.cs +++ b/Services/WorkflowProgressTracker.cs @@ -90,9 +90,9 @@ public static class WorkflowTelemetryReader { if (run == null) return null; var done = ReadCompleted(run.RichJsonPath); - if (done != null) { done.ChainId = run.ChainId; done.StartedAt = run.StartedAt; return done; } + if (done != null) { done.ChainId = run.ChainId; done.RunId = run.RunId; done.StartedAt = run.StartedAt; return done; } var live = ReadLive(run.JournalPath, run.StartedAt); - if (live != null) { live.ChainId = run.ChainId; return live; } + if (live != null) { live.ChainId = run.ChainId; live.RunId = run.RunId; return live; } return null; } @@ -158,14 +158,23 @@ public static class WorkflowTelemetryReader catch { return null; } } - /// Live-Fortschritt aus dem append-only journal.jsonl (started/result pro Subagent). + /// + /// Live-Fortschritt aus dem append-only journal.jsonl (started/result pro Subagent). + /// WICHTIG: Die result-Events enthalten den VOLLSTÄNDIGEN Subagent-Output (Feld + /// result, Objekt oder String) – die einzige Quelle, die schon WÄHREND des Laufs + /// existiert. Wir bauen daraus die Per-Agent-Liste inkl. Output-Preview, damit bei einem + /// Abbruch die bereits geleistete Arbeit fertiger Subagenten NICHT verloren geht (die reiche + /// <runId>.json mit Tokens/Labels wird erst bei Completion geschrieben). Tokens/Labels + /// sind hier daher noch 0/agentId – die kennt nur die abgeschlossene <runId>.json. + /// public static WorkflowProgress ReadLive(string journalPath, DateTime startedAt) { if (string.IsNullOrEmpty(journalPath) || !File.Exists(journalPath)) return null; try { - var started = new HashSet(); - var done = new HashSet(); + var order = new List(); // Reihenfolge der agentIds (started/result) + var seen = new HashSet(); + var outputs = new Dictionary(); // agentId -> voller Output (nur fertige) // ReadAllText statt ReadAllLines: robust gegen halb geschriebene letzte Zeile. foreach (var raw in File.ReadAllText(journalPath).Split('\n')) { @@ -177,18 +186,39 @@ public static class WorkflowTelemetryReader var t = GetStr(ev, "type"); var aid = GetStr(ev, "agentId"); if (string.IsNullOrEmpty(aid)) continue; - if (t == "started") started.Add(aid); - else if (t == "result") done.Add(aid); + if (seen.Add(aid)) order.Add(aid); + if (t == "result") + { + string output = null; + if (ev.TryGetProperty("result", out var rv)) + output = rv.ValueKind == JsonValueKind.String ? rv.GetString() : rv.GetRawText(); + outputs[aid] = output ?? ""; + } } - if (started.Count == 0 && done.Count == 0) return null; - return new WorkflowProgress + if (order.Count == 0) return null; + var wp = new WorkflowProgress { Status = "running", StartedAt = startedAt, - SubagentCount = started.Count, - CompletedCount = done.Count, + SubagentCount = order.Count, + CompletedCount = outputs.Count, DurationMs = (long)Math.Max(0, (DateTime.UtcNow - startedAt).TotalMilliseconds) }; + int idx = 0; + foreach (var aid in order) + { + var isDone = outputs.TryGetValue(aid, out var outp); + wp.Agents.Add(new WorkflowAgentInfo + { + Index = ++idx, + Label = aid, + State = isDone ? "done" : "running", + // Voller Output fertiger Subagenten (großzügig gekappt), damit die Arbeit + // bei Abbruch rekonstruierbar bleibt statt verloren zu gehen. + ResultPreview = isDone ? Trim(outp, 1500) : null + }); + } + return wp; } catch { return null; } } @@ -215,6 +245,7 @@ public class WorkflowProgress { public string Status { get; set; } // "running" | "completed" public string ChainId { get; set; } + public string RunId { get; set; } // wf_… – für Resume nach Abbruch public string WorkflowName { get; set; } public string Summary { get; set; } public DateTime StartedAt { get; set; } diff --git a/wwwroot/css/chat.css b/wwwroot/css/chat.css index 98ef4e8..f33c3ad 100644 --- a/wwwroot/css/chat.css +++ b/wwwroot/css/chat.css @@ -1114,12 +1114,33 @@ display: flex; align-items: center; justify-content: space-between; + flex-wrap: wrap; gap: var(--badge-spacing); padding: .4em var(--badge-spacing); background: var(--f3-bg); border-radius: var(--border-radius); } +/* Geretteter Subagent-Output (aus journal.jsonl) – volle Breite, unter Label/Metriken. */ +.wf-agent-output { + flex-basis: 100%; + margin-top: .35em; + padding: .4em .55em; + background: var(--f2-bg); + border-radius: var(--border-radius); + font-family: var(--mono); + font-size: .8em; + white-space: pre-wrap; + word-break: break-word; + max-height: 12em; + overflow-y: auto; + opacity: var(--semi-transparent); +} + +.wf-resume-hint { + color: var(--info); +} + .wf-agent-main { display: flex; align-items: center; diff --git a/wwwroot/js/modals.js b/wwwroot/js/modals.js index f362b2d..9f47d1e 100644 --- a/wwwroot/js/modals.js +++ b/wwwroot/js/modals.js @@ -144,6 +144,16 @@ function _renderWorkflowStatus(body, data) { body.appendChild(title); } + // runId + Resume-Hinweis: Bei einem abgebrochenen Lauf kann der Agent den Workflow mit dieser + // runId (resumeFromRunId) fortsetzen – fertige Subagenten werden aus dem Journal wiederverwendet. + if (wf.runId) { + var rid = document.createElement('div'); + rid.className = 'wf-meta'; + rid.innerHTML = 'runId: ' + esc(wf.runId) + '' + + (status !== 'completed' ? ' · fortsetzbar via resumeFromRunId' : ''); + body.appendChild(rid); + } + // Kennzahlen. var stats = [ { label: 'Subagenten', value: _wfNum(wf.subagentCount) }, @@ -182,7 +192,9 @@ function _renderWorkflowStatus(body, data) { body.appendChild(psec); } - // Subagenten-Details (nur nach Abschluss; live liegen noch keine Labels vor). + // Subagenten-Details. Nach Abschluss: Labels + Tokens/Tool-Calls/Dauer (aus .json). + // Live/bei Abbruch: agentId als Label + State + geretteter Output (aus journal.jsonl) – so ist + // die Arbeit fertiger Subagenten auch dann sichtbar, wenn der Lauf nie sauber endete. var agents = wf.agents || []; if (agents.length) { var asec = document.createElement('div'); @@ -193,6 +205,7 @@ function _renderWorkflowStatus(body, data) { agents.forEach(function (a) { var st = a.state || ''; var stCls = st === 'done' ? 'wf-ag-done' : (st === 'running' ? 'wf-ag-running' : 'wf-ag-other'); + var hasMetrics = (a.tokens || a.toolCalls || a.durationMs); var row = document.createElement('div'); row.className = 'wf-agent-row'; row.innerHTML = @@ -201,11 +214,13 @@ function _renderWorkflowStatus(body, data) { '' + esc(a.label || ('Agent ' + (a.index || ''))) + '' + (a.phaseTitle ? '' + esc(a.phaseTitle) + '' : '') + '' + + (hasMetrics ? '
' + ' ' + _wfNum(a.tokens) + '' + ' ' + _wfNum(a.toolCalls) + '' + ' ' + _wfDur(a.durationMs) + '' + - '
'; + '' : '') + + (a.resultPreview ? '
' + esc(a.resultPreview) + '
' : ''); alist.appendChild(row); }); asec.appendChild(alist);