feat(commons): FlowGraphBuilder folds outbound lifecycles into one node; async/spawn edges; Response from the inbound execution

Fold rule: outbound rows sharing (kind, CorrelationId) become one FlowStep with
a Lifecycle list — first row's identity/time/input, latest row's outcome/
subtitle/output/error, DurationMs = sync row's own or submit→terminal wall time
(null while pending). Chain skips async steps (Notify, or first row
CachedSubmit) and links them with Async edges from the preceding chain step.
Spawner edges from a non-RouteCall are Spawn: true. The Response links from the
last non-async step of the inbound execution (the RouteCall), not the globally
last cascaded step (D9). FlowGraphView places a Response at max column + 1 so
the D9 origin does not pull it left (pulled forward from Task 7 to keep the
layout test green).

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 01:42:48 -04:00
parent e26616bebc
commit 29129ef17a
5 changed files with 332 additions and 23 deletions
@@ -396,6 +396,14 @@ public partial class FlowGraphView : ComponentBase
}
}
// The Response is the flow's final node: it sits right of everything real regardless of
// its predecessor's column (the builder links it from the route call — D9 — which would
// otherwise drag it back to column 2 while the site work sits to its right).
var isResponse = (stepByKey.TryGetValue(k, out var sk) && sk.Kind == FlowStepKind.Response)
|| (predByKey.TryGetValue(k, out var pk) && pk.Kind == FlowStepKind.Response);
if (isResponse && column.Count > 0)
col = Math.Max(col, column.Values.Max() + 1);
while (!occupied.Add((lane[k], col)))
col++;
column[k] = col;
@@ -1,5 +1,6 @@
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
@@ -8,18 +9,27 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
/// (Flow Debugger design 2026-09-05 §7.2). Pure: same events in, same graph out.
/// </summary>
/// <remarks>
/// Rules:
/// Rules (v1.1, 2026-09-06 outbound-calls plan):
/// <list type="bullet">
/// <item><b>Fold.</b> Outbound rows (<c>ApiOutbound</c> / <c>DbOutbound</c> / <c>Notification</c>) sharing a kind and
/// <c>CorrelationId</c> fold into ONE step (a sync row with a null correlation is a group of one). The folded step
/// takes the first row's identity, time, input and title, the latest row's outcome, subtitle, output and error;
/// <c>DurationMs</c> is the sync row's own, or submit→terminal wall time for a cached operation (null while pending);
/// <c>Lifecycle</c> lists every row in time order.</item>
/// <item>Steps are ordered by <c>OccurredAtUtc</c>, then a kind rank (inbound before route call
/// before script run before writes/waits), then <c>EventId</c>.</item>
/// <item>Within one <c>ExecutionId</c> consecutive steps are chained in that order.</item>
/// before script run before writes/waits before outbound), then <c>EventId</c>.</item>
/// <item>Within one <c>ExecutionId</c> consecutive steps are chained in that order, <b>skipping async steps</b> — a
/// <c>Notify</c>, or a cached operation whose first row is <c>CachedSubmit</c>. Each skipped step instead gets an
/// <c>Async</c> (dashed) edge from the chain step immediately before it in time.</item>
/// <item>The first step of an execution is linked from its spawner: a <c>RouteCall</c> whose
/// <c>childExecutionId</c> matches, else the latest step of the parent execution that occurred
/// no later than it. The edge <c>CrossesBoundary</c> when exactly one end is in the Central lane.</item>
/// no later than it — that second case is a <c>Spawn</c> edge (trigger cascade). The edge
/// <c>CrossesBoundary</c> when exactly one end is in the Central lane.</item>
/// <item>An <c>AttributeWait</c> that was satisfied gets an <c>External</c> (dashed) edge from the
/// preceding <c>AttributeWrite</c> in the same execution — the device echo the system inferred.</item>
/// <item>When an <c>InboundRequest</c> row carries a response summary, a <c>Response</c> step is
/// synthesized as the final Central node, linked from the last real step.</item>
/// synthesized as the final Central node, linked from the last non-async step of the <i>inbound</i>
/// execution (the route call, or the inbound row itself), never from a cascaded site step.</item>
/// </list>
/// </remarks>
public static class FlowGraphBuilder
@@ -31,8 +41,7 @@ public static class FlowGraphBuilder
if (events.Count == 0)
return new FlowGraph(Array.Empty<FlowStep>(), Array.Empty<FlowEdge>(), Predicted: false, Truncated: truncated);
var steps = events
.Select(FlowStepProjector.Project)
var steps = Fold(events)
.OrderBy(s => s.OccurredAtUtc)
.ThenBy(s => KindRank(s.Kind))
.ThenBy(s => s.EventId)
@@ -44,11 +53,23 @@ public static class FlowGraphBuilder
.GroupBy(s => s.ExecutionId!.Value)
.ToDictionary(g => g.Key, g => g.ToList());
// 1. Chain consecutive steps inside each execution.
// 1. Chain consecutive non-async steps inside each execution; async steps branch off the step before them.
foreach (var chain in byExecution.Values)
{
for (var i = 1; i < chain.Count; i++)
edges.Add(new FlowEdge(chain[i - 1].EventId, chain[i].EventId, CrossesBoundary(chain[i - 1], chain[i]), External: false));
FlowStep? prev = null;
foreach (var step in chain)
{
if (IsAsync(step))
{
if (prev is not null)
edges.Add(new FlowEdge(prev.EventId, step.EventId, CrossesBoundary(prev, step), External: false, Async: true));
continue;
}
if (prev is not null)
edges.Add(new FlowEdge(prev.EventId, step.EventId, CrossesBoundary(prev, step), External: false));
prev = step;
}
}
// 2. Link each child execution's first step from its spawner.
@@ -66,11 +87,12 @@ public static class FlowGraphBuilder
if (routeCallsByChild.TryGetValue(executionId, out var routeCall))
spawner = routeCall;
else if (byExecution.TryGetValue(parentId, out var parentChain))
spawner = parentChain.LastOrDefault(p => p.OccurredAtUtc <= first.OccurredAtUtc && p.Kind != FlowStepKind.Response)
spawner = parentChain.LastOrDefault(p => p.OccurredAtUtc <= first.OccurredAtUtc && p.Kind != FlowStepKind.Response && !IsAsync(p))
?? parentChain[0];
if (spawner is not null && spawner.EventId != first.EventId)
edges.Add(new FlowEdge(spawner.EventId, first.EventId, CrossesBoundary(spawner, first), External: false));
edges.Add(new FlowEdge(spawner.EventId, first.EventId, CrossesBoundary(spawner, first), External: false,
Spawn: spawner.Kind != FlowStepKind.RouteCall));
}
// 3. Satisfied wait ⇐ preceding write in the same execution (inferred device echo).
@@ -95,6 +117,10 @@ public static class FlowGraphBuilder
if (inbound is not null && inbound.OutputJson is not null)
{
var last = steps[^1];
var origin = (inbound.ExecutionId is { } inboundExecution && byExecution.TryGetValue(inboundExecution, out var inboundChain)
? inboundChain.LastOrDefault(s => s.Kind != FlowStepKind.Response && !IsAsync(s))
: null)
?? last;
var response = new FlowStep(
EventId: DeriveResponseId(inbound.EventId),
ExecutionId: inbound.ExecutionId,
@@ -115,12 +141,88 @@ public static class FlowGraphBuilder
Target: inbound.Target,
Extra: null);
steps.Add(response);
edges.Add(new FlowEdge(last.EventId, response.EventId, CrossesBoundary(last, response), External: false));
edges.Add(new FlowEdge(origin.EventId, response.EventId, CrossesBoundary(origin, response), External: false));
}
return new FlowGraph(steps, edges, Predicted: false, Truncated: truncated);
}
/// <summary>
/// A step the script did not wait for: a notification, or a cached operation (first lifecycle row
/// <c>CachedSubmit</c>). Such steps hang off the chain on an <see cref="FlowEdge.Async"/> edge (D4).
/// </summary>
public static bool IsAsync(FlowStep step)
=> step.Kind == FlowStepKind.Notify
|| (step.Kind is FlowStepKind.ApiCall or FlowStepKind.DbCall
&& string.Equals(step.Lifecycle?.FirstOrDefault()?.Kind, nameof(AuditKind.CachedSubmit), StringComparison.Ordinal));
/// <summary>
/// Projects every event and folds the outbound rows of one operation (same kind + <c>CorrelationId</c>) into one step
/// carrying the whole lifecycle. Non-outbound rows and null-correlation outbound rows pass through one-to-one.
/// </summary>
private static IEnumerable<FlowStep> Fold(IReadOnlyList<AuditEvent> events)
{
var groups = new Dictionary<(FlowStepKind Kind, Guid Correlation), List<(FlowStep Step, AuditEvent Evt)>>();
var order = new List<object>(events.Count); // FlowStep, or the group key — preserves first-seen order
foreach (var evt in events)
{
var step = FlowStepProjector.Project(evt);
if (!FlowStepProjector.IsOutbound(step.Kind))
{
order.Add(step);
continue;
}
if (evt.CorrelationId is not { } correlation)
{
order.Add(step with { Lifecycle = [FlowStepProjector.ToLifecycleRow(evt)] });
continue;
}
var key = (step.Kind, correlation);
if (!groups.TryGetValue(key, out var group))
{
group = [];
groups[key] = group;
order.Add(key);
}
group.Add((step, evt));
}
foreach (var item in order)
{
if (item is FlowStep single)
{
yield return single;
continue;
}
var key = ((FlowStepKind, Guid))item;
var rows = groups[key]
.OrderBy(r => r.Step.OccurredAtUtc)
.ThenBy(r => r.Step.EventId)
.ToList();
var first = rows[0].Step;
var latest = rows[^1].Step;
var lifecycle = rows.Select(r => FlowStepProjector.ToLifecycleRow(r.Evt)).ToList();
var terminal = latest.Outcome != FlowStepOutcome.Pending;
int? duration = rows.Count == 1
? first.DurationMs
: terminal ? (int?)Math.Max(0, (int)Math.Round((latest.OccurredAtUtc - first.OccurredAtUtc).TotalMilliseconds)) : null;
yield return first with
{
Outcome = latest.Outcome,
Subtitle = latest.Subtitle,
OutputJson = latest.OutputJson,
ErrorMessage = latest.ErrorMessage,
DurationMs = rows.Count == 1 && !terminal ? null : duration,
Lifecycle = lifecycle,
};
}
}
/// <summary>Display rank used to break timestamp ties: spawners before what they spawn.</summary>
public static int KindRank(FlowStepKind kind) => kind switch
{
@@ -45,6 +45,21 @@ public static class FlowStepProjector
Extra: extra);
}
/// <summary>Projects one outbound audit row onto its lifecycle entry (<see cref="FlowStep.Lifecycle"/>).</summary>
public static FlowLifecycleRow ToLifecycleRow(AuditEvent evt)
{
ArgumentNullException.ThrowIfNull(evt);
var d = AuditDetailsCodec.Deserialize(evt.DetailsJson);
return new FlowLifecycleRow(
EventId: evt.EventId,
Kind: d.Kind ?? string.Empty,
Status: d.Status ?? string.Empty,
OccurredAtUtc: evt.OccurredAtUtc,
Node: evt.SourceNode,
ErrorMessage: d.ErrorMessage,
DurationMs: d.DurationMs);
}
/// <summary>Maps a channel/kind string pair onto a step kind.</summary>
public static FlowStepKind ResolveKind(string? channel, string? kind)
{