Read workflow-status telemetry from CLI transcript files

Dynamic ultracode workflows run as a detached background task with their
own transcript dir, so their subagent events never flow through the main
-p stream. The previous parent_tool_use_id tap therefore captured nothing
for real workflows and the status stayed "none" the whole run.

Replace the tap with a file-based reader:
- TapWorkflowLaunch parses the "Transcript dir:" line from the Workflow
  tool_result and registers the run.
- WorkflowProgressTracker becomes a per-session registry of transcript
  paths (no longer a stream accumulator).
- WorkflowTelemetryReader reads journal.jsonl for live counts (running)
  and <runId>.json for rich per-agent telemetry (completed); FinalizeChain
  persists the rich JSON into chains.workflow_info.
- Modal renders phases + a per-agent list (label/phase/state/tokens/
  tool-calls/duration) instead of the obsolete tool breakdown.

Sync ARCHITECTURE.md, ARCHITECTURE_QUICK.md and README.md to the new
mechanism.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MelekiandClaude Opus 4.8 committed 2026-06-20 18:14:42 +02:00
1 parent b6e546f618
commit 2b95161596
9 files changed
+453 -175

No files matched your search

+12 -1
View File
@@ -329,7 +329,18 @@ public partial class AnthropicService
if (_workflowTracker != null && !string.IsNullOrEmpty(chain.SessionId)
&& _workflowTracker.ChainIdFor(chain.SessionId) == chain.ChainId)
{
chain.WorkflowInfo = _workflowTracker.GetJson(chain.SessionId);
try
{
// Reiche End-Telemetrie aus der wf_<runId>.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));
if (wp != null)
{
wp.ChainId = chain.ChainId;
chain.WorkflowInfo = JsonSerializer.Serialize(wp);
}
}
catch { /* Persistenz der Telemetrie darf den Finalize nie crashen */ }
_workflowTracker.End(chain.SessionId);
}
return SseHelper.FinalizeChain(_db, chain, usage, iterations, stopReason, sw, streamErrorInfo);
+30 -29
View File
@@ -93,39 +93,35 @@ public class ClaudeCliService
}
/// <summary>
/// Greift aus einem (im Haupt-Stream unterdrückten) Subagent-Turn die Telemetrie ab:
/// die vom Subagenten aufgerufenen Tool-Namen und seine Token-Nutzung. Rein additiv für den
/// Workflow-Status – wirft nie (Telemetrie darf den Stream-Parser nicht stören).
/// Erkennt den tool_result des dynamischen Workflow-Tools ("Workflow launched in background …
/// Transcript dir: …") und registriert den Lauf im WorkflowProgressTracker. Der Workflow läuft
/// als detachter Hintergrund-Task mit eigenem Transcript – seine Live-/End-Telemetrie wird
/// später aus den dort liegenden Dateien (journal.jsonl / wf_&lt;runId&gt;.json) gelesen, nicht aus
/// dem Stream. Rein additiv – wirft nie (Telemetrie darf den Stream-Parser nicht stören).
/// </summary>
private void TapSubagentEvent(string sessionId, string chainId, string parentToolUseId, JsonElement json)
private void TapWorkflowLaunch(string sessionId, string chainId, string resultContent)
{
try
{
if (!json.TryGetProperty("message", out var msg))
if (_workflowTracker == null || string.IsNullOrEmpty(sessionId) || string.IsNullOrEmpty(resultContent))
return;
if (resultContent.IndexOf("Workflow launched", StringComparison.OrdinalIgnoreCase) < 0)
return;
var toolNames = new List<string>();
if (msg.TryGetProperty("content", out var content) && content.ValueKind == JsonValueKind.Array)
// Zeile "Transcript dir: <absoluter Pfad>" extrahieren.
string transcriptDir = null;
foreach (var raw in resultContent.Split('\n'))
{
foreach (var item in content.EnumerateArray())
var line = raw.Trim();
const string marker = "Transcript dir:";
if (line.StartsWith(marker, StringComparison.OrdinalIgnoreCase))
{
if (item.TryGetProperty("type", out var it) && it.GetString() == "tool_use"
&& item.TryGetProperty("name", out var nm))
{
var n = nm.GetString();
if (!string.IsNullOrEmpty(n)) toolNames.Add(n);
}
transcriptDir = line.Substring(marker.Length).Trim();
break;
}
}
long inTok = 0, outTok = 0;
if (msg.TryGetProperty("usage", out var usage))
{
if (usage.TryGetProperty("input_tokens", out var it2) && it2.ValueKind == JsonValueKind.Number) inTok = it2.GetInt64();
if (usage.TryGetProperty("output_tokens", out var ot2) && ot2.ValueKind == JsonValueKind.Number) outTok = ot2.GetInt64();
}
_workflowTracker.RecordSubagent(sessionId, chainId, parentToolUseId, toolNames, inTok, outTok);
if (!string.IsNullOrEmpty(transcriptDir))
_workflowTracker.Register(sessionId, chainId, transcriptDir);
}
catch { /* Telemetrie darf den Stream nie stören */ }
}
@@ -553,12 +549,11 @@ public class ClaudeCliService
&& parentToolUseId.ValueKind == JsonValueKind.String
&& !string.IsNullOrEmpty(parentToolUseId.GetString()))
{
// Subagent-Event: weiterhin aus dem Haupt-Stream AUSBLENDEN (continue) –
// hier wird nur die Telemetrie für den Workflow-Status ABGEGRIFFEN
// (Tool-Aufrufe + Subagent-Tokens), ohne den sichtbaren Verlauf zu verändern.
// Die konsolidierten Subagent-Turns kommen als type:"assistant" (mit usage).
if (_workflowTracker != null && !string.IsNullOrEmpty(sessionId) && type == "assistant")
TapSubagentEvent(sessionId, chainId, parentToolUseId.GetString(), json);
// Subagent-Event aus dem Haupt-Stream AUSBLENDEN (continue). Hinweis: Die
// dynamischen Workflow-Subagenten laufen out-of-band (eigener Hintergrund-
// Task + Transcript) und tauchen hier ohnehin NICHT auf – ihre Telemetrie
// wird aus den Transcript-Dateien gelesen (siehe TapWorkflowLaunch). Dieser
// Guard betrifft nur klassische Task-Subagenten (sichtbarer Verlauf bleibt sauber).
continue;
}
@@ -775,6 +770,12 @@ public class ClaudeCliService
ToolResultContent = resultContent,
ToolResultIsError = resultIsError
});
// Dynamischen Workflow-Start abgreifen (Transcript-Pfad merken),
// damit der /workflow-status-Endpoint die Telemetrie aus den
// Transcript-Dateien lesen kann. Stört den Stream nie.
if (!resultIsError)
TapWorkflowLaunch(sessionId, chainId, resultContent);
}
}
}
+212 -95
View File
@@ -1,131 +1,248 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text.Json;
using System.Text.Json.Serialization;
namespace Gott.Services;
/// <summary>
/// Sammelt Live-Telemetrie laufender dynamischer Workflows ("ultracode"): die Aktivität der
/// Subagenten, die die Claude CLI über das Workflow-/Task-Tool spawnt. Diese Subagent-Events
/// werden im Haupt-Stream bewusst unterdrückt (parent_tool_use_id-Guard in ClaudeCliService) –
/// hier werden sie abgegriffen, damit der Workflow-Fortschritt LIVE (während des Laufs) und
/// danach (persistiert in chains.workflow_info) abrufbar ist. In-Memory pro Session, analog
/// StreamingTracker. Telemetrie darf den Stream nie blockieren → alles defensiv + lock-leicht.
/// Verfolgt dynamische Workflows ("ultracode"), die die Claude CLI über das Workflow-Tool
/// startet. WICHTIG: Diese Workflows laufen als <b>detachter Hintergrund-Task</b> 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:
/// • <c>&lt;transcriptDir&gt;/journal.jsonl</c> – LIVE (append-only: started/result pro Subagent)
/// • <c>&lt;proj&gt;/workflows/&lt;runId&gt;.json</c> – am ENDE geschrieben (reiche Per-Agent-Telemetrie)
/// Beides empirisch verifiziert. Dieser Tracker ist daher nur eine <b>Registry</b>: 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).
/// </summary>
public class WorkflowProgressTracker
{
private readonly ConcurrentDictionary<string, WorkflowProgress> _progress
= new ConcurrentDictionary<string, WorkflowProgress>();
private readonly ConcurrentDictionary<string, WorkflowRunRef> _runs
= new ConcurrentDictionary<string, WorkflowRunRef>();
/// <summary>
/// Verbucht ein Subagent-Event (Tool-Aufrufe + Token-Nutzung eines Subagenten).
/// Lazy: legt den Eintrag bei der ERSTEN Subagent-Aktivität einer Session an, d.h. nur
/// echte Workflow-Läufe (mit gespawnten Subagenten) erzeugen überhaupt einen Eintrag.
/// Registriert einen gestarteten Workflow-Lauf für die Session (aus dem Workflow-tool_result
/// abgegriffene Transcript-Dir). Überschreibt einen früheren Lauf derselben Session.
/// </summary>
public void RecordSubagent(string sessionId, string chainId, string parentToolUseId,
IReadOnlyList<string> toolNames, long inputTokens, long outputTokens)
public void Register(string sessionId, string chainId, string transcriptDir)
{
if (string.IsNullOrEmpty(sessionId)) return;
var wp = _progress.GetOrAdd(sessionId, _ => new WorkflowProgress
if (string.IsNullOrEmpty(sessionId) || string.IsNullOrEmpty(transcriptDir)) return;
try
{
ChainId = chainId,
StartedAt = DateTime.UtcNow
});
lock (wp.Sync)
{
// Stale-Eintrag eines früheren, nicht sauber finalisierten Laufs (gleiche Session,
// andere Chain) → atomar für die neue Chain zurücksetzen, statt fälschlich zu mischen.
if (wp.ChainId != chainId)
var runId = Path.GetFileName(transcriptDir.TrimEnd('/'));
// <transcriptDir> = <proj>/subagents/workflows/<runId>
var workflowsDir = Path.GetDirectoryName(transcriptDir); // <proj>/subagents/workflows
var subagentsDir = Path.GetDirectoryName(workflowsDir); // <proj>/subagents
var projDir = Path.GetDirectoryName(subagentsDir); // <proj>
var journalPath = Path.Combine(transcriptDir, "journal.jsonl");
var richJsonPath = projDir != null ? Path.Combine(projDir, "workflows", runId + ".json") : null;
_runs[sessionId] = new WorkflowRunRef
{
wp.ChainId = chainId;
wp.StartedAt = DateTime.UtcNow;
wp.SubagentIds.Clear();
wp.SubagentCount = 0;
wp.SubagentToolCalls = 0;
wp.SubagentInputTokens = 0;
wp.SubagentOutputTokens = 0;
wp.ToolBreakdown.Clear();
wp.RecentTools.Clear();
}
if (!string.IsNullOrEmpty(parentToolUseId) && wp.SubagentIds.Add(parentToolUseId))
wp.SubagentCount = wp.SubagentIds.Count;
if (toolNames != null)
{
foreach (var tn in toolNames)
{
if (string.IsNullOrEmpty(tn)) continue;
wp.SubagentToolCalls++;
wp.ToolBreakdown[tn] = wp.ToolBreakdown.TryGetValue(tn, out var c) ? c + 1 : 1;
wp.RecentTools.Add(tn);
if (wp.RecentTools.Count > 12) wp.RecentTools.RemoveAt(0);
}
}
wp.SubagentInputTokens += inputTokens;
wp.SubagentOutputTokens += outputTokens;
wp.LastActivityAt = DateTime.UtcNow;
ChainId = chainId,
RunId = runId,
JournalPath = journalPath,
RichJsonPath = richJsonPath,
StartedAt = DateTime.UtcNow
};
}
catch { /* Telemetrie darf den Stream nie stören */ }
}
/// <summary>Threadsicherer Live-Snapshot – null, wenn für die Session kein Workflow läuft.</summary>
public WorkflowProgress GetSnapshot(string sessionId)
=> _progress.TryGetValue(sessionId, out var wp) ? wp.Snapshot() : null;
/// <summary>Registry-Eintrag der Session – null, wenn kein Workflow registriert ist.</summary>
public WorkflowRunRef Get(string sessionId)
=> _runs.TryGetValue(sessionId, out var r) ? r : null;
/// <summary>Aktuellen Stand als JSON serialisieren (threadsicher) – null wenn kein Eintrag.</summary>
public string GetJson(string sessionId)
=> _progress.TryGetValue(sessionId, out var wp) ? wp.ToJson() : null;
/// <summary>ChainId des laufenden Workflow-Eintrags (für die Zuordnung beim Finalize).</summary>
/// <summary>ChainId des registrierten Workflow-Laufs (für die Zuordnung beim Finalize).</summary>
public string ChainIdFor(string sessionId)
=> _progress.TryGetValue(sessionId, out var wp) ? wp.ChainId : null;
=> _runs.TryGetValue(sessionId, out var r) ? r.ChainId : null;
/// <summary>Entfernt den Eintrag (beim Chain-Finalize aufzurufen).</summary>
public void End(string sessionId) => _progress.TryRemove(sessionId, out _);
public void End(string sessionId) => _runs.TryRemove(sessionId, out _);
}
/// <summary>Registrierter Workflow-Lauf einer Session: ChainId + die Pfade seiner Telemetrie-Dateien.</summary>
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; }
}
/// <summary>
/// Telemetrie eines dynamischen Workflows: Subagent-Anzahl, Tool-Aufrufe (+ Aufschlüsselung)
/// und aggregierte Subagent-Tokens. Wird live geführt und beim Chain-Finalize als JSON in
/// chains.workflow_info persistiert.
/// 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 wf_&lt;runId&gt;.json
/// (erst bei Abschluss). Alle Methoden sind defensiv und liefern null statt zu werfen.
/// </summary>
public static class WorkflowTelemetryReader
{
/// <summary>
/// Bevorzugt die reiche, fertige wf_&lt;runId&gt;.json (existiert nur nach Abschluss);
/// fällt sonst auf das live journal.jsonl zurück. Liefert null, wenn nichts lesbar ist.
/// </summary>
public static WorkflowProgress Read(WorkflowRunRef run)
{
if (run == null) return null;
var done = ReadCompleted(run.RichJsonPath);
if (done != null) { done.ChainId = run.ChainId; done.StartedAt = run.StartedAt; return done; }
var live = ReadLive(run.JournalPath, run.StartedAt);
if (live != null) { live.ChainId = run.ChainId; return live; }
return null;
}
/// <summary>Reiche Telemetrie aus der wf_&lt;runId&gt;.json (am Laufende geschrieben).</summary>
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; }
}
/// <summary>Live-Fortschritt aus dem append-only journal.jsonl (started/result pro Subagent).</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 done = new HashSet<string>();
// 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 (t == "started") started.Add(aid);
else if (t == "result") done.Add(aid);
}
if (started.Count == 0 && done.Count == 0) return null;
return new WorkflowProgress
{
Status = "running",
StartedAt = startedAt,
SubagentCount = started.Count,
CompletedCount = done.Count,
DurationMs = (long)Math.Max(0, (DateTime.UtcNow - startedAt).TotalMilliseconds)
};
}
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) + "…";
}
/// <summary>
/// Telemetrie eines dynamischen Workflows. Live (aus journal.jsonl): nur Status + Subagent-Zähler.
/// Abgeschlossen (aus wf_&lt;runId&gt;.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.
/// </summary>
public class WorkflowProgress
{
[JsonIgnore] public object Sync { get; } = new object();
// Dedup-Hilfsstruktur (nicht persistiert; SubagentCount trägt den Wert).
[JsonIgnore] public HashSet<string> SubagentIds { get; set; } = new HashSet<string>();
public string Status { get; set; } // "running" | "completed"
public string ChainId { get; set; }
public string WorkflowName { get; set; }
public string Summary { get; set; }
public DateTime StartedAt { get; set; }
public DateTime LastActivityAt { get; set; }
public int SubagentCount { get; set; }
public int SubagentToolCalls { get; set; }
public long SubagentInputTokens { get; set; }
public long SubagentOutputTokens { get; set; }
public Dictionary<string, int> ToolBreakdown { get; set; } = new Dictionary<string, int>();
public List<string> RecentTools { get; set; } = new List<string>();
public int CompletedCount { get; set; }
public long TotalTokens { get; set; }
public int TotalToolCalls { get; set; }
public long DurationMs { get; set; }
public List<WorkflowPhaseInfo> Phases { get; set; } = new List<WorkflowPhaseInfo>();
public List<WorkflowAgentInfo> Agents { get; set; } = new List<WorkflowAgentInfo>();
}
public WorkflowProgress Snapshot()
{
lock (Sync)
{
return new WorkflowProgress
{
ChainId = ChainId,
StartedAt = StartedAt,
LastActivityAt = LastActivityAt,
SubagentCount = SubagentCount,
SubagentToolCalls = SubagentToolCalls,
SubagentInputTokens = SubagentInputTokens,
SubagentOutputTokens = SubagentOutputTokens,
ToolBreakdown = new Dictionary<string, int>(ToolBreakdown),
RecentTools = new List<string>(RecentTools)
};
}
}
public class WorkflowPhaseInfo
{
public int Index { get; set; }
public string Title { get; set; }
public string Detail { get; set; }
}
// Serialisiert einen entkoppelten Snapshot (Snapshot() kopiert unter Lock) – so läuft die
// Serialisierung außerhalb jeder kritischen Sektion und kann nie auf eine live mutierte
// Collection treffen.
public string ToJson() => JsonSerializer.Serialize(Snapshot());
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; }
}