Workflow-Totalverlust beheben: inkrementelle Persistenz + Token-Fix + Resume

Bisher wurden Workflow-Subagent-Outputs nur beim sauberen FinalizeChain in
die DB gerettet - bei Abbruch (529/Timeout/Stop/Hard-Kill) war alles weg,
Tokens standen auf 0.

- A) Inkrementelle Persistenz: MaybeFlushWorkflowProgress() (throttled 6s)
  schreibt workflow_info + Token + runId waehrend des Laufs -> ueberlebt
  Hard-Kill (OOM/Redeploy), wo kein catch/finally laeuft.
- B) Output-Rettung: ReadLive liest jetzt vollstaendige Subagent-Outputs aus
  journal.jsonl (nicht nur Counts), inkl. State.
- C) Token-Fix: neuer assistant-Branch -> CliEventType.Usage; EffectiveUsage()
  rettet Orchestrator-Token bei Abbruch statt 0.
- D) Resume: Feld/Migration workflow_run_id; FinalizeChain persistiert die
  runId; naechster Chat injiziert einen resumeFromRunId-Hinweis.
- UI: Workflow-Status-Modal zeigt geretteten Output + runId/"fortsetzbar"-Banner.

Build 0/0, ReadLive gegen echtes journal.jsonl verifiziert, adversarial 2x geprueft.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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; }

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>();
         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<workflow_resume>\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</workflow_resume>");
+                        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,

Services/AnthropicService.cs

@@ -332,13 +332,17 @@ public partial class AnthropicService
             try
             {
                 // Reiche End-Telemetrie aus der <runId>.json des Laufs lesen und persistieren
-                // (Fallback: Live-Counts aus dem journal.jsonl, falls die json noch nicht da ist).
+                // (Fallback: Live-Counts + gerettete Subagent-Outputs aus dem journal.jsonl, falls
-                var wp = WorkflowTelemetryReader.Read(_workflowTracker.Get(chain.SessionId));
+                // 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);

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

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);
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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<ChainRecord> GetChain(string chainId)
     {
         using var conn = CreateConnection();

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");

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; }
     }
 
-    /// <summary>Live-Fortschritt aus dem append-only journal.jsonl (started/result pro Subagent).</summary>
+    /// <summary>
+    /// Live-Fortschritt aus dem append-only journal.jsonl (started/result pro Subagent).
+    /// WICHTIG: Die <c>result</c>-Events enthalten den VOLLSTÄNDIGEN Subagent-Output (Feld
+    /// <c>result</c>, 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
+    /// &lt;runId&gt;.json mit Tokens/Labels wird erst bei Completion geschrieben). Tokens/Labels
+    /// sind hier daher noch 0/agentId – die kennt nur die abgeschlossene &lt;runId&gt;.json.
+    /// </summary>
     public static WorkflowProgress ReadLive(string journalPath, DateTime startedAt)
     {
         if (string.IsNullOrEmpty(journalPath) || !File.Exists(journalPath)) return null;
         try
         {
-            var started = new HashSet<string>();
+            var order = new List<string>();                    // Reihenfolge der agentIds (started/result)
-            var done = new HashSet<string>();
+            var seen = new HashSet<string>();
+            var outputs = new Dictionary<string, string>();     // 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);
+                if (seen.Add(aid)) order.Add(aid);
-                else if (t == "result") done.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;
+            if (order.Count == 0) return null;
-            return new WorkflowProgress
+            var wp = new WorkflowProgress
             {
                 Status = "running",
                 StartedAt = startedAt,
-                SubagentCount = started.Count,
+                SubagentCount = order.Count,
-                CompletedCount = done.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; }

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;

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 = '<span title="Workflow-runId">runId: <code>' + esc(wf.runId) + '</code></span>' +
+      (status !== 'completed' ? ' <span class="wf-resume-hint">· fortsetzbar via resumeFromRunId</span>' : '');
+    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 <runId>.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) {
           '<span class="wf-agent-label">' + esc(a.label || ('Agent ' + (a.index || ''))) + '</span>' +
           (a.phaseTitle ? '<span class="wf-agent-phase">' + esc(a.phaseTitle) + '</span>' : '') +
         '</div>' +
+        (hasMetrics ?
         '<div class="wf-agent-metrics">' +
           '<span title="Tokens"><i class="fa fa-database"></i> ' + _wfNum(a.tokens) + '</span>' +
           '<span title="Tool-Aufrufe"><i class="fa fa-wrench"></i> ' + _wfNum(a.toolCalls) + '</span>' +
           '<span title="Dauer"><i class="fa fa-clock-o"></i> ' + _wfDur(a.durationMs) + '</span>' +
-        '</div>';
+        '</div>' : '') +
+        (a.resultPreview ? '<div class="wf-agent-output">' + esc(a.resultPreview) + '</div>' : '');
       alist.appendChild(row);
     });
     asec.appendChild(alist);