fix(commons): Flow graph orders executions by emission sequence, timestamps only within one clock (arch-review F5)

Central rows carry central's clock, site rows the site's, so the global OccurredAtUtc sort let a 1-2 s skew float a whole site execution ahead of the inbound row / RouteCall that started it, and pushed the trigger-cascade spawner search onto its parentChain[0] fallback. FlowGraphBuilder.Build now orders executions by the input index of their earliest-given row and orders steps by timestamp only INSIDE one execution (one clock), breaking ties by kind rank then input index. A folded outbound node keeps its earliest-by-time row's identity, which need not be the row given first, so its index is the minimum over the rows folded into it. The spawner search uses position in the ordered list rather than a raw timestamp compare. FlowDebugRunner.RunMethodAsync keeps the collector's emission order instead of re-sorting; the value-change path keeps its OrderBy (all rows share one site clock). Replay input arrives timestamp-sorted from the repository, so its order is unchanged. The DelmiaShaped test fixture is still shuffled within each execution (the inside-one-execution timestamp rule is unchanged) but now lists the executions in emission order, since that is what fixes their relative order. Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
2026-09-06 14:28:12 -04:00
parent def6bab318
commit 1dfb9cee20
4 changed files with 183 additions and 11 deletions
@@ -16,14 +16,17 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
 ///     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 before outbound), then <c>EventId</c>.</item>
+///   <item><b>Order</b> = emission order between executions, timestamp within one (arch-review F5: central and
+///     site clocks differ). Executions are ordered by the input index of their earliest-given row; inside one
+///     execution — one clock — steps are ordered by <c>OccurredAtUtc</c>, then a kind rank (inbound before route
+///     call before script run before writes/waits before outbound), then the input index. Replay input arrives
+///     timestamp-sorted from the repository, so its order is unchanged; the Live path's collector order wins.</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 — that second case is a <c>Spawn</c> edge (trigger cascade). The edge
+///     <c>childExecutionId</c> matches, else the last step of the parent execution that precedes it in
+///     the order above — 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>
@@ -41,12 +44,47 @@ public static class FlowGraphBuilder
        if (events.Count == 0)
            return new FlowGraph(Array.Empty<FlowStep>(), Array.Empty<FlowEdge>(), Predicted: false, Truncated: truncated);

-        var steps = Fold(events)
-            .OrderBy(s => s.OccurredAtUtc)
+        // Arch-review F5: a step's position in the input is the only cross-node ordering we can trust —
+        // central rows carry central's clock, site rows the site's, and a 1-2 s skew used to sort a whole
+        // site execution ahead of the inbound row that started it.
+        var seq = new Dictionary<Guid, int>(events.Count);
+        for (var i = 0; i < events.Count; i++)
+            seq.TryAdd(events[i].EventId, i);
+
+        // A folded step keeps the identity of its earliest-by-TIME row, which need not be the row given
+        // first — so a folded node's index is the minimum over the rows folded into it.
+        int InputIndex(FlowStep step)
+        {
+            var index = seq.TryGetValue(step.EventId, out var own) ? own : int.MaxValue;
+            if (step.Lifecycle is { Count: > 1 } rows)
+            {
+                foreach (var row in rows)
+                {
+                    if (seq.TryGetValue(row.EventId, out var i) && i < index)
+                        index = i;
+                }
+            }
+
+            return index;
+        }
+
+        var folded = Fold(events).ToList();
+        var firstSeqByExecution = folded
+            .GroupBy(s => s.ExecutionId ?? Guid.Empty)
+            .ToDictionary(g => g.Key, g => g.Min(InputIndex));
+        var steps = folded
+            .OrderBy(s => firstSeqByExecution[s.ExecutionId ?? Guid.Empty])   // executions in emission order
+            .ThenBy(s => s.OccurredAtUtc)                                     // one clock inside an execution
            .ThenBy(s => KindRank(s.Kind))
-            .ThenBy(s => s.EventId)
+            .ThenBy(InputIndex)
            .ToList();

+        // Position in the ordered list — the "before" relation the spawner search uses, so a skewed
+        // clock can no longer push every candidate parent step "after" the child.
+        var position = new Dictionary<Guid, int>(steps.Count);
+        for (var i = 0; i < steps.Count; i++)
+            position[steps[i].EventId] = i;
+
        var edges = new List<FlowEdge>();
        var byExecution = steps
            .Where(s => s.ExecutionId is not null)
@@ -93,8 +131,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 && !IsAsync(p))
+            {
+                // The parent step that precedes the child in DISPLAY order, not by raw timestamp (F5).
+                var childPosition = position[first.EventId];
+                spawner = parentChain.LastOrDefault(p => position[p.EventId] < childPosition && p.Kind != FlowStepKind.Response && !IsAsync(p))
                          ?? parentChain[0];
+            }

            if (spawner is null) continue;
            var spawn = spawner.Kind != FlowStepKind.RouteCall;
@@ -173,8 +173,12 @@ public sealed class FlowDebugRunner
                result.ErrorMessage ?? "Access denied: routed call outside your site scope.");
        }

+        // Arch-review F5: the collector's emission order is authoritative — central rows carry central's
+        // clock and site rows the site's, so re-sorting by OccurredAtUtc lets a skewed site clock float
+        // its steps ahead of the RouteCall that started them. (The value-change path below keeps its
+        // OrderBy: every row it merges was stamped by the same site.)
        var ordered = new List<AuditEvent>(events.Count + 1) { inboundRow };
-        ordered.AddRange(events.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId));
+        ordered.AddRange(events);

        return new FlowDebugRunResult(
            root, result.Success, result.ErrorMessage, result.Success ? null : result.ErrorCode,
@@ -46,8 +46,10 @@ public class FlowGraphBuilderTests
            executionId: Cascade, parentExecutionId: SiteRun, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 10,
            extra: new ScriptTraceExtra { TriggerKind = "valuechange", TriggerAttribute = "DelmiaReceiver.Trigger", TriggerValue = "false" }.ToJson());

-        // Deliberately shuffled: the builder must order by time.
-        return [wait, cascade, inbound, write2, run, route, write1];
+        // Deliberately shuffled WITHIN each execution: the builder must order by time inside one execution.
+        // The executions themselves are given in emission order (root, site run, cascade) — since arch-review
+        // F5 that is what fixes their relative order, because they do not share a clock.
+        return [inbound, route, wait, run, write2, write1, cascade];
    }

    [Fact]
@@ -365,6 +367,105 @@ public class FlowGraphBuilderTests
        Assert.Equal(graph.Steps[1].EventId, edge.ToEventId);
    }

+    // ---- Arch-review F5: emission order between executions, timestamps within one ------------
+
+    /// <summary>
+    /// Central and site clocks differ. The four rows are given in EMISSION order (the Live path's
+    /// collector order); the graph must keep the executions in that order whichever way the skew runs.
+    /// </summary>
+    private static List<AuditEvent> SkewedRouteShape(int siteSkewMs)
+    {
+        var inbound = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered,
+            occurredAtUtc: T0, actor: "k", target: "M", sourceNode: "central-a", executionId: Root, durationMs: 900,
+            requestSummary: "{}", responseSummary: "{\"ok\":true}");
+        var route = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.RouteCall, AuditStatus.Delivered,
+            occurredAtUtc: T0.AddMilliseconds(50), target: "M1.S", sourceNode: "central-a", executionId: Root,
+            extra: new ScriptTraceExtra { ChildExecutionId = SiteRun }.ToJson());
+        var run = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
+            occurredAtUtc: T0.AddMilliseconds(siteSkewMs), actor: "S", target: "S", sourceNode: "node-a",
+            executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 20,
+            extra: new ScriptTraceExtra { TriggerKind = "call" }.ToJson());
+        var write = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.AttributeWrite, AuditStatus.Delivered,
+            occurredAtUtc: T0.AddMilliseconds(siteSkewMs + 10), actor: "S", target: "A", sourceNode: "node-a",
+            executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 2,
+            extra: new ScriptTraceExtra { Writes = [new AttributeWriteDetail("A", "1", true, null)] }.ToJson());
+        return [inbound, route, run, write];
+    }
+
+    [Fact]
+    public void SiteClockAhead_SiteStepsStillFollowTheirRouteCall_AndPrecedeResponse()
+    {
+        var events = SkewedRouteShape(siteSkewMs: 2000); // site 2 s ahead of central
+        var graph = FlowGraphBuilder.Build(events);
+
+        Assert.Equal(
+            new[]
+            {
+                FlowStepKind.InboundRequest, FlowStepKind.RouteCall, FlowStepKind.ScriptRun,
+                FlowStepKind.AttributeWrite, FlowStepKind.Response,
+            },
+            graph.Steps.Select(s => s.Kind));
+        Assert.Equal(events.Select(e => e.EventId), graph.Steps.Take(4).Select(s => s.EventId));
+        // The Response time still comes off the inbound row's own clock (inbound + DurationMs).
+        Assert.Equal(T0.AddMilliseconds(900), graph.Steps[4].OccurredAtUtc.UtcDateTime);
+    }
+
+    [Fact]
+    public void SiteClockBehind_SiteStepsDoNotSortAheadOfTheInboundRequest()
+    {
+        var events = SkewedRouteShape(siteSkewMs: -2000); // site 2 s behind central
+        var graph = FlowGraphBuilder.Build(events);
+
+        Assert.Equal(
+            new[]
+            {
+                FlowStepKind.InboundRequest, FlowStepKind.RouteCall, FlowStepKind.ScriptRun,
+                FlowStepKind.AttributeWrite, FlowStepKind.Response,
+            },
+            graph.Steps.Select(s => s.Kind));
+        Assert.Equal(events.Select(e => e.EventId), graph.Steps.Take(4).Select(s => s.EventId));
+    }
+
+    [Fact]
+    public void SiteClockBehind_TriggerCascade_SpawnsFromTheLastParentStep_NotTheFirst()
+    {
+        // Mixed-version shape: no RouteCall row, so the spawner is found by position, and a site
+        // clock behind central used to push every parent step "after" the child → parentChain[0].
+        var inbound = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered,
+            occurredAtUtc: T0, actor: "k", target: "M", sourceNode: "central-a", executionId: Root, durationMs: 900);
+        var centralDb = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
+            occurredAtUtc: T0.AddMilliseconds(10), actor: "api:M", target: "CfgDb", sourceNode: "central-a", executionId: Root);
+        var run = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
+            occurredAtUtc: T0.AddMilliseconds(-2000), actor: "S", target: "S", sourceNode: "node-a",
+            executionId: SiteRun, parentExecutionId: Root, sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 20,
+            extra: new ScriptTraceExtra { TriggerKind = "call" }.ToJson());
+
+        var graph = FlowGraphBuilder.Build([inbound, centralDb, run]);
+
+        var spawn = Assert.Single(graph.Edges, e => e.Spawn);
+        Assert.Equal(centralDb.EventId, spawn.FromEventId); // the LAST parent step, not the inbound row
+        Assert.Equal(run.EventId, spawn.ToEventId);
+    }
+
+    [Fact]
+    public void FoldedNode_TakesTheEmissionIndexOfItsEarliestEmittedRow()
+    {
+        // The folded step keeps the earliest-by-TIME row's id (the resolve here is emitted first),
+        // so its sequence must be the minimum input index over the rows folded into it.
+        var corr = Guid.NewGuid();
+        var resolve = Outbound(AuditChannel.ApiOutbound, AuditKind.CachedResolve, AuditStatus.Delivered, 4000, corr, "Delmia.Ack", response: "res");
+        var submit = Outbound(AuditChannel.ApiOutbound, AuditKind.CachedSubmit, AuditStatus.Submitted, 35, corr, "Delmia.Ack", request: "req");
+        var run = SiteRunRow(20, Cascade);
+
+        // Emission order interleaves the executions: the folded node's earliest row is index 0, so
+        // its execution leads — even though the id the folded step keeps (the submit) is index 2.
+        var graph = FlowGraphBuilder.Build([resolve, run, submit]);
+
+        var api = Assert.Single(graph.Steps, s => s.Kind == FlowStepKind.ApiCall);
+        Assert.Equal(submit.EventId, api.EventId);
+        Assert.Equal(new[] { FlowStepKind.ApiCall, FlowStepKind.ScriptRun }, graph.Steps.Select(s => s.Kind));
+    }
+
    // ---- Review findings on Task 2 (2026-09-06) --------------------------------------------

    [Fact]
@@ -124,6 +124,31 @@ public class FlowDebugRunnerTests
        Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.RouteCall);
    }

+    [Fact]
+    public async Task RunMethod_SiteClockBehindCentral_StillReturnsTheSinkEventsInEmissionOrder()
+    {
+        // Arch-review F5: the site's rows carry the SITE clock. A site 2 s behind central used to
+        // re-sort ahead of the always-on central RouteCall row.
+        var method = new ApiMethod("M", "x") { Id = 6, TimeoutSeconds = 5 };
+        _apiRepo.GetApiMethodByIdAsync(6, Arg.Any<CancellationToken>()).Returns(method);
+        _locator.GetSiteIdForInstanceAsync("inst", Arg.Any<CancellationToken>()).Returns("site-a");
+        var child = Guid.NewGuid();
+        var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
+            occurredAtUtc: DateTime.UtcNow.AddSeconds(-2), executionId: child, target: "work");
+        _router.RouteToCallAsync("site-a", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow, child, [siteRun]));
+        _executor.RegisterHandler("M", ctx => ctx.Route.To("inst").Call("work"));
+
+        var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
+            new FlowDebugRunMethodCommand(6, null), Deployer, CancellationToken.None);
+
+        Assert.True(result.Completed, result.Error);
+        Assert.Equal(3, result.Events.Count);
+        Assert.Equal(AuditKind.InboundRequest, result.Events[0].AsRow().Kind);
+        Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind);
+        Assert.Same(siteRun, result.Events[2]);
+    }
+
    [Fact]
    public async Task RunMethod_UnknownMethod_IsNotFound()
    {