using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; using System.Linq; using System.Text.Json; namespace Gott.Services; /// /// Verfolgt dynamische Workflows ("ultracode"), die die Claude CLI über das Workflow-Tool /// startet. WICHTIG: Diese Workflows laufen als detachter Hintergrund-Task mit eigenem /// Transcript-Verzeichnis – ihre Subagent-Events fließen NICHT durch den Haupt-Stream (anders /// als klassische Task-Subagenten mit parent_tool_use_id). Die echte Telemetrie liegt in zwei /// Dateien des CLI-Transcripts: /// • <transcriptDir>/journal.jsonl – LIVE (append-only: started/result pro Subagent) /// • <proj>/workflows/<runId>.json – am ENDE geschrieben (reiche Per-Agent-Telemetrie) /// Beides empirisch verifiziert. Dieser Tracker ist daher nur eine Registry: er merkt sich /// pro Session die aus dem Workflow-tool_result abgegriffenen Pfade; gelesen wird erst bei Abruf /// (Endpoint) bzw. beim Finalize. In-Memory pro Session, defensiv (nie werfend). /// public class WorkflowProgressTracker { private readonly ConcurrentDictionary _runs = new ConcurrentDictionary(); /// /// Registriert einen gestarteten Workflow-Lauf für die Session (aus dem Workflow-tool_result /// abgegriffene Transcript-Dir). Überschreibt einen früheren Lauf derselben Session. /// public void Register(string sessionId, string chainId, string transcriptDir) { if (string.IsNullOrEmpty(sessionId) || string.IsNullOrEmpty(transcriptDir)) return; try { var runId = Path.GetFileName(transcriptDir.TrimEnd('/')); // = /subagents/workflows/ var workflowsDir = Path.GetDirectoryName(transcriptDir); // /subagents/workflows var subagentsDir = Path.GetDirectoryName(workflowsDir); // /subagents var projDir = Path.GetDirectoryName(subagentsDir); // var journalPath = Path.Combine(transcriptDir, "journal.jsonl"); var richJsonPath = projDir != null ? Path.Combine(projDir, "workflows", runId + ".json") : null; _runs[sessionId] = new WorkflowRunRef { ChainId = chainId, RunId = runId, JournalPath = journalPath, RichJsonPath = richJsonPath, StartedAt = DateTime.UtcNow }; } catch { /* Telemetrie darf den Stream nie stören */ } } /// Registry-Eintrag der Session – null, wenn kein Workflow registriert ist. public WorkflowRunRef Get(string sessionId) => _runs.TryGetValue(sessionId, out var r) ? r : null; /// ChainId des registrierten Workflow-Laufs (für die Zuordnung beim Finalize). public string ChainIdFor(string sessionId) => _runs.TryGetValue(sessionId, out var r) ? r.ChainId : null; /// Entfernt den Eintrag (beim Chain-Finalize aufzurufen). public void End(string sessionId) => _runs.TryRemove(sessionId, out _); } /// Registrierter Workflow-Lauf einer Session: ChainId + die Pfade seiner Telemetrie-Dateien. public class WorkflowRunRef { public string ChainId { get; set; } public string RunId { get; set; } public string JournalPath { get; set; } public string RichJsonPath { get; set; } public DateTime StartedAt { get; set; } } /// /// Liest die Workflow-Telemetrie aus den CLI-Transcript-Dateien. Zwei Quellen, je nach Phase: /// das append-only journal.jsonl (live, während des Laufs) und die reiche <runId>.json /// (erst bei Abschluss). Alle Methoden sind defensiv und liefern null statt zu werfen. /// public static class WorkflowTelemetryReader { /// /// Bevorzugt die reiche, fertige <runId>.json (existiert nur nach Abschluss); /// fällt sonst auf das live journal.jsonl zurück. Liefert null, wenn nichts lesbar ist. /// public static WorkflowProgress Read(WorkflowRunRef run) { if (run == null) return null; var done = ReadCompleted(run.RichJsonPath); 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; live.RunId = run.RunId; return live; } return null; } /// Reiche Telemetrie aus der <runId>.json (am Laufende geschrieben). public static WorkflowProgress ReadCompleted(string richJsonPath) { if (string.IsNullOrEmpty(richJsonPath) || !File.Exists(richJsonPath)) return null; try { using var doc = JsonDocument.Parse(File.ReadAllText(richJsonPath)); var root = doc.RootElement; var wp = new WorkflowProgress { Status = "completed", WorkflowName = GetStr(root, "workflowName"), Summary = GetStr(root, "summary"), DurationMs = GetLong(root, "durationMs"), StartedAt = DateTime.UtcNow }; // Phasen if (root.TryGetProperty("phases", out var phases) && phases.ValueKind == JsonValueKind.Array) { int pi = 0; foreach (var p in phases.EnumerateArray()) wp.Phases.Add(new WorkflowPhaseInfo { Index = ++pi, Title = GetStr(p, "title"), Detail = GetStr(p, "detail") }); } // Per-Agent aus workflowProgress (type == "workflow_agent") if (root.TryGetProperty("workflowProgress", out var prog) && prog.ValueKind == JsonValueKind.Array) { foreach (var ev in prog.EnumerateArray()) { if (GetStr(ev, "type") != "workflow_agent") continue; var a = new WorkflowAgentInfo { Index = (int)GetLong(ev, "index"), Label = GetStr(ev, "label"), PhaseTitle = GetStr(ev, "phaseTitle"), State = GetStr(ev, "state"), Model = GetStr(ev, "model"), Tokens = GetLong(ev, "tokens"), ToolCalls = (int)GetLong(ev, "toolCalls"), DurationMs = GetLong(ev, "durationMs"), ResultPreview = Trim(GetStr(ev, "resultPreview"), 240) }; wp.Agents.Add(a); wp.TotalTokens += a.Tokens; wp.TotalToolCalls += a.ToolCalls; } } wp.SubagentCount = wp.Agents.Count; wp.CompletedCount = wp.Agents.Count(x => x.State == "done"); if (wp.SubagentCount == 0) wp.SubagentCount = (int)GetLong(root, "agentCount"); return wp; } catch { return null; } } /// /// 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 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')) { var line = raw.Trim(); if (line.Length == 0) continue; JsonElement ev; try { using var d = JsonDocument.Parse(line); ev = d.RootElement.Clone(); } catch { continue; } var t = GetStr(ev, "type"); var aid = GetStr(ev, "agentId"); if (string.IsNullOrEmpty(aid)) continue; 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 (order.Count == 0) return null; var wp = new WorkflowProgress { Status = "running", StartedAt = startedAt, 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; } } private static string GetStr(JsonElement e, string name) => e.ValueKind == JsonValueKind.Object && e.TryGetProperty(name, out var v) && v.ValueKind == JsonValueKind.String ? v.GetString() : null; private static long GetLong(JsonElement e, string name) => e.ValueKind == JsonValueKind.Object && e.TryGetProperty(name, out var v) && v.ValueKind == JsonValueKind.Number ? v.GetInt64() : 0L; private static string Trim(string s, int max) => string.IsNullOrEmpty(s) || s.Length <= max ? s : s.Substring(0, max) + "…"; } /// /// Telemetrie eines dynamischen Workflows. Live (aus journal.jsonl): nur Status + Subagent-Zähler. /// Abgeschlossen (aus <runId>.json): zusätzlich Phasen, Per-Agent-Details (Tokens, Tool-Calls, /// Dauer, Label, Phase) und Summen. Wird beim Chain-Finalize als JSON in chains.workflow_info /// persistiert und vom /workflow-status-Endpoint ausgeliefert. /// 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; } public int SubagentCount { get; set; } public int CompletedCount { get; set; } public long TotalTokens { get; set; } public int TotalToolCalls { get; set; } public long DurationMs { get; set; } public List Phases { get; set; } = new List(); public List Agents { get; set; } = new List(); } public class WorkflowPhaseInfo { public int Index { get; set; } public string Title { get; set; } public string Detail { get; set; } } public class WorkflowAgentInfo { public int Index { get; set; } public string Label { get; set; } public string PhaseTitle { get; set; } public string State { get; set; } // "done" | "running" | "queued" | … public string Model { get; set; } public long Tokens { get; set; } public int ToolCalls { get; set; } public long DurationMs { get; set; } public string ResultPreview { get; set; } }