feat(inbound-api): RouteHelper writes an always-on RouteCall audit row, propagates Trace, exposes an inline trace sink

Flow Debugger (design 2026-09-05 §5.1/§5.2/§7.1): RouteTarget.Call emits one ScriptTrace.RouteCall central-direct-write row per routed call (ExecutionId = the inbound parent, Extra.childExecutionId from the reply, best-effort, never throws into the script) whether or not the run is traced. WithTrace(bool) stamps Trace on RouteToCallRequest and RouteToSetAttributesRequest; WithTraceSink hands the row + the reply's inline site events to a callback so a Live run renders without touching the store. DI supplies ICentralAuditWriter / INodeIdentityProvider as optional ctor args; new RouteHelper(locator, router) still compiles. Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP
2026-09-05 23:59:08 -04:00
parent 7e61552d82
commit 3f582a278c
2 changed files with 317 additions and 12 deletions
@@ -1,7 +1,13 @@
+using System.Diagnostics;
+using System.Text.Json;
 using Akka.Actor;
+using ZB.MOM.WW.Audit;
 using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
 using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
 using ZB.MOM.WW.ScadaBridge.Commons.Types;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

 namespace ZB.MOM.WW.ScadaBridge.InboundAPI;

@@ -47,16 +53,27 @@ public class RouteHelper
    private readonly CancellationToken _deadlineToken;
    private readonly CancellationToken _requestAbortedToken;
    private readonly Guid? _parentExecutionId;
+    private readonly ICentralAuditWriter? _auditWriter;
+    private readonly INodeIdentityProvider? _nodeIdentity;
+    private readonly bool _trace;
+    private readonly Action<AuditEvent>? _traceSink;

    /// <summary>
    /// Initializes a new <see cref="RouteHelper"/> with no deadline token and no parent execution id.
+    /// DI (<c>AddScoped&lt;RouteHelper&gt;()</c>) resolves the two optional audit collaborators when the
+    /// host registers them; a bare <c>new RouteHelper(locator, router)</c> writes no audit rows.
    /// </summary>
    /// <param name="instanceLocator">Service to resolve the site id for a given instance code.</param>
    /// <param name="instanceRouter">Service to route cross-site calls to the resolved site.</param>
+    /// <param name="auditWriter">Flow Debugger (design 2026-09-05 §5.1): central-direct-write audit writer for the always-on <c>RouteCall</c> row; null = no row.</param>
+    /// <param name="nodeIdentity">Stamps <c>SourceNode</c> (central-a/central-b) on the <c>RouteCall</c> row; null = NULL.</param>
    public RouteHelper(
        IInstanceLocator instanceLocator,
-        IInstanceRouter instanceRouter)
-        : this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null)
+        IInstanceRouter instanceRouter,
+        ICentralAuditWriter? auditWriter = null,
+        INodeIdentityProvider? nodeIdentity = null)
+        : this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null,
+               auditWriter, nodeIdentity, trace: false, traceSink: null)
    {
    }

@@ -65,13 +82,21 @@ public class RouteHelper
        IInstanceRouter instanceRouter,
        CancellationToken deadlineToken,
        CancellationToken requestAbortedToken,
-        Guid? parentExecutionId)
+        Guid? parentExecutionId,
+        ICentralAuditWriter? auditWriter,
+        INodeIdentityProvider? nodeIdentity,
+        bool trace,
+        Action<AuditEvent>? traceSink)
    {
        _instanceLocator = instanceLocator;
        _instanceRouter = instanceRouter;
        _deadlineToken = deadlineToken;
        _requestAbortedToken = requestAbortedToken;
        _parentExecutionId = parentExecutionId;
+        _auditWriter = auditWriter;
+        _nodeIdentity = nodeIdentity;
+        _trace = trace;
+        _traceSink = traceSink;
    }

    /// <summary>
@@ -84,7 +109,8 @@ public class RouteHelper
    /// <param name="deadlineToken">The executing method's timeout cancellation token to inherit for routed calls.</param>
    /// <returns>A new <see cref="RouteHelper"/> inheriting the given deadline token.</returns>
    public RouteHelper WithDeadline(CancellationToken deadlineToken) =>
-        new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId);
+        new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId,
+            _auditWriter, _nodeIdentity, _trace, _traceSink);

    /// <summary>
    /// Returns a <see cref="RouteHelper"/> carrying the raw request-abort
@@ -97,7 +123,8 @@ public class RouteHelper
    /// <param name="requestAbortedToken">The raw client-disconnect token, independent of the method timeout.</param>
    /// <returns>A new <see cref="RouteHelper"/> carrying the given request-abort token.</returns>
    public RouteHelper WithRequestAborted(CancellationToken requestAbortedToken) =>
-        new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId);
+        new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId,
+            _auditWriter, _nodeIdentity, _trace, _traceSink);

    /// <summary>
    /// Audit Log (ParentExecutionId): returns a <see cref="RouteHelper"/> whose
@@ -111,7 +138,31 @@ public class RouteHelper
    /// <param name="parentExecutionId">The inbound request's execution id to stamp as the spawning parent on routed calls, or null for non-routed runs.</param>
    /// <returns>A new <see cref="RouteHelper"/> carrying the given parent execution id.</returns>
    public RouteHelper WithParentExecutionId(Guid? parentExecutionId) =>
-        new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId);
+        new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId,
+            _auditWriter, _nodeIdentity, _trace, _traceSink);
+
+    /// <summary>
+    /// Flow Debugger (design 2026-09-05 §5.2): returns a <see cref="RouteHelper"/> whose routed
+    /// calls and writes carry <c>Trace = <paramref name="trace"/></c>, so the site traces the
+    /// spawned run. The always-on <c>RouteCall</c> row is written regardless of this flag.
+    /// </summary>
+    /// <param name="trace">True to trace the routed site runs.</param>
+    /// <returns>A new <see cref="RouteHelper"/> carrying the flag.</returns>
+    public RouteHelper WithTrace(bool trace) =>
+        new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
+            _auditWriter, _nodeIdentity, trace, _traceSink);
+
+    /// <summary>
+    /// Flow Debugger (design 2026-09-05 §7.1): returns a <see cref="RouteHelper"/> that also
+    /// hands every <c>RouteCall</c> row AND every inline site trace event on the routed reply
+    /// to <paramref name="traceSink"/>, so a Live run can render without a round-trip to the
+    /// audit store. Null clears the sink.
+    /// </summary>
+    /// <param name="traceSink">Receives the audit events in emission order; null = none.</param>
+    /// <returns>A new <see cref="RouteHelper"/> carrying the sink.</returns>
+    public RouteHelper WithTraceSink(Action<AuditEvent>? traceSink) =>
+        new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
+            _auditWriter, _nodeIdentity, _trace, traceSink);

    /// <summary>
    /// Creates a route target for the specified instance.
@@ -121,7 +172,8 @@ public class RouteHelper
    public RouteTarget To(string instanceCode)
    {
        return new RouteTarget(
-            instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId);
+            instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
+            _auditWriter, _nodeIdentity, _trace, _traceSink);
    }
 }

@@ -136,6 +188,10 @@ public class RouteTarget
    private readonly CancellationToken _deadlineToken;
    private readonly CancellationToken _requestAbortedToken;
    private readonly Guid? _parentExecutionId;
+    private readonly ICentralAuditWriter? _auditWriter;
+    private readonly INodeIdentityProvider? _nodeIdentity;
+    private readonly bool _trace;
+    private readonly Action<AuditEvent>? _traceSink;

    /// <summary>
    /// Initializes a new <see cref="RouteTarget"/> for the specified instance.
@@ -152,7 +208,11 @@ public class RouteTarget
        IInstanceRouter instanceRouter,
        CancellationToken deadlineToken,
        CancellationToken requestAbortedToken,
-        Guid? parentExecutionId)
+        Guid? parentExecutionId,
+        ICentralAuditWriter? auditWriter = null,
+        INodeIdentityProvider? nodeIdentity = null,
+        bool trace = false,
+        Action<AuditEvent>? traceSink = null)
    {
        _instanceCode = instanceCode;
        _instanceLocator = instanceLocator;
@@ -160,6 +220,10 @@ public class RouteTarget
        _deadlineToken = deadlineToken;
        _requestAbortedToken = requestAbortedToken;
        _parentExecutionId = parentExecutionId;
+        _auditWriter = auditWriter;
+        _nodeIdentity = nodeIdentity;
+        _trace = trace;
+        _traceSink = traceSink;
    }

    /// <summary>
@@ -188,11 +252,31 @@ public class RouteTarget
        // (the inbound API request's ExecutionId) so the routed site script
        // records this call's parent. CorrelationId above is a separate concern
        // — the per-operation lifecycle id, freshly minted per routed call.
+        var normalized = ScriptArgs.Normalize(parameters);
        var request = new RouteToCallRequest(
-            correlationId, _instanceCode, scriptName, ScriptArgs.Normalize(parameters),
-            DateTimeOffset.UtcNow, _parentExecutionId);
+            correlationId, _instanceCode, scriptName, normalized,
+            DateTimeOffset.UtcNow, _parentExecutionId,
+            // Flow Debugger (design 2026-09-05 §5.2): the trace flag rides beside the parent id.
+            Trace: _trace);

-        var response = await RouteOrUnreachable(_instanceRouter.RouteToCallAsync(siteId, request, token));
+        // Flow Debugger (design §5.1): the ALWAYS-ON RouteCall row — one per routed call,
+        // written whether or not the run is traced (a writer must be present).
+        var startedUtc = DateTime.UtcNow;
+        var sw = Stopwatch.StartNew();
+        RouteToCallResponse response;
+        try
+        {
+            response = await RouteOrUnreachable(_instanceRouter.RouteToCallAsync(siteId, request, token));
+        }
+        catch (Exception ex)
+        {
+            EmitRouteCall(scriptName, normalized, startedUtc, sw.ElapsedMilliseconds, success: false,
+                returnValue: null, error: ex.Message, childExecutionId: null, traceEvents: null);
+            throw;
+        }
+
+        EmitRouteCall(scriptName, normalized, startedUtc, sw.ElapsedMilliseconds, response.Success,
+            response.ReturnValue, response.ErrorMessage, response.ChildExecutionId, response.TraceEvents);

        if (!response.Success)
        {
@@ -203,6 +287,78 @@ public class RouteTarget
        return response.ReturnValue;
    }

+    /// <summary>
+    /// Flow Debugger (design 2026-09-05 §5.1/§7.1): builds the <c>ScriptTrace.RouteCall</c> row,
+    /// writes it best-effort through the central audit writer (never throws into the script),
+    /// and hands it — followed by the site's inline trace events — to the trace sink.
+    /// </summary>
+    private void EmitRouteCall(
+        string scriptName, IReadOnlyDictionary<string, object?>? args, DateTime startedUtc, long elapsedMs,
+        bool success, object? returnValue, string? error, Guid? childExecutionId,
+        IReadOnlyList<AuditEvent>? traceEvents)
+    {
+        if (_auditWriter is null && _traceSink is null) return;
+        AuditEvent evt;
+        try
+        {
+            evt = ScadaBridgeAuditEventFactory.Create(
+                AuditChannel.ScriptTrace, AuditKind.RouteCall,
+                success ? AuditStatus.Delivered : AuditStatus.Failed,
+                occurredAtUtc: DateTime.SpecifyKind(startedUtc, DateTimeKind.Utc),
+                actor: "inbound-api",
+                target: $"{_instanceCode}.{scriptName}",
+                sourceNode: _nodeIdentity?.NodeName,
+                executionId: _parentExecutionId,
+                parentExecutionId: null,
+                durationMs: (int)Math.Min(elapsedMs, int.MaxValue),
+                errorMessage: error,
+                requestSummary: TraceJson(args),
+                responseSummary: TraceJson(returnValue),
+                extra: childExecutionId is null ? null : new ScriptTraceExtra { ChildExecutionId = childExecutionId }.ToJson());
+        }
+        catch (Exception)
+        {
+            return;
+        }
+
+        if (_auditWriter is not null)
+        {
+            try
+            {
+                var t = _auditWriter.WriteAsync(evt, CancellationToken.None);
+                if (!t.IsCompletedSuccessfully)
+                {
+                    _ = t.ContinueWith(static _ => { }, TaskContinuationOptions.OnlyOnFaulted);
+                }
+            }
+            catch (Exception)
+            {
+                // Best-effort: an audit failure never fails the routed call.
+            }
+        }
+
+        if (_traceSink is null) return;
+        try
+        {
+            _traceSink(evt);
+            if (traceEvents is not null)
+            {
+                foreach (var e in traceEvents) _traceSink(e);
+            }
+        }
+        catch (Exception)
+        {
+            // A misbehaving sink must not fail the script either.
+        }
+    }
+
+    private static string? TraceJson(object? value)
+    {
+        if (value is null) return null;
+        try { return JsonSerializer.Serialize(value, ScriptTraceExtra.JsonOptions); }
+        catch (Exception) { return null; }
+    }
+
    /// <summary>
    /// Gets a single attribute value from the remote instance.
    /// </summary>
@@ -353,7 +509,9 @@ public class RouteTarget
        // across Call / GetAttributes / SetAttributes.
        var request = new RouteToSetAttributesRequest(
            correlationId, _instanceCode, attributeValues, DateTimeOffset.UtcNow,
-            _parentExecutionId);
+            _parentExecutionId,
+            // Flow Debugger (design §5.3): a traced routed write arms the site's echo stamp.
+            Trace: _trace);

        var response = await RouteOrUnreachable(_instanceRouter.RouteToSetAttributesAsync(siteId, request, token));

@@ -0,0 +1,147 @@
+using NSubstitute;
+using ZB.MOM.WW.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests;
+
+/// <summary>
+/// Flow Debugger (design 2026-09-05 §5.1/§5.2/§7.1): <see cref="RouteTarget.Call"/> writes an
+/// ALWAYS-ON <c>ScriptTrace.RouteCall</c> row (traced or not), propagates <c>Trace</c> on both
+/// route requests, and feeds the row plus the reply's inline site events to the trace sink.
+/// </summary>
+public class RouteHelperTraceTests
+{
+    private readonly IInstanceLocator _locator = Substitute.For<IInstanceLocator>();
+    private readonly IInstanceRouter _router = Substitute.For<IInstanceRouter>();
+    private readonly List<AuditEvent> _written = new();
+    private readonly ICentralAuditWriter _writer = Substitute.For<ICentralAuditWriter>();
+    private readonly INodeIdentityProvider _node = Substitute.For<INodeIdentityProvider>();
+
+    public RouteHelperTraceTests()
+    {
+        _locator.GetSiteIdForInstanceAsync("inst-1", Arg.Any<CancellationToken>()).Returns("SiteA");
+        _writer.WriteAsync(Arg.Do<AuditEvent>(e => _written.Add(e)), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
+        _node.NodeName.Returns("central-a");
+    }
+
+    private RouteHelper Helper() => new(_locator, _router, _writer, _node);
+
+    private static AuditEvent SiteEvent(Guid exec) => ScadaBridgeAuditEventFactory.Create(
+        AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, executionId: exec, target: "doWork");
+
+    [Fact]
+    public async Task Call_WritesRouteCallRow_WithParentAsExecutionId_AndChildFromReply()
+    {
+        var parent = Guid.NewGuid();
+        var child = Guid.NewGuid();
+        _router.RouteToCallAsync("SiteA", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 99, null,
+                DateTimeOffset.UtcNow, child, null));
+
+        var result = await Helper().WithParentExecutionId(parent).To("inst-1").Call("doWork", new { x = 1 });
+
+        Assert.Equal(99, result);
+        var evt = Assert.Single(_written);
+        var row = evt.AsRow();
+        Assert.Equal("ScriptTrace.RouteCall", evt.Action);
+        Assert.Equal("inbound-api", evt.Actor);
+        Assert.Equal("inst-1.doWork", evt.Target);
+        Assert.Equal("central-a", evt.SourceNode);
+        Assert.Equal(parent, row.ExecutionId);
+        Assert.Null(row.ParentExecutionId);
+        Assert.Equal(AuditStatus.Delivered, row.Status);
+        Assert.Equal(child, ScriptTraceExtra.TryParse(row.Extra)!.ChildExecutionId);
+        Assert.Contains("\"x\":1", row.RequestSummary);
+        Assert.Equal("99", row.ResponseSummary);
+    }
+
+    [Fact]
+    public async Task Call_IsAlwaysOn_EvenWhenNotTraced_AndTraceFalseOnRequest()
+    {
+        RouteToCallRequest? captured = null;
+        _router.RouteToCallAsync("SiteA", Arg.Do<RouteToCallRequest>(r => captured = r), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, null, null, DateTimeOffset.UtcNow));
+
+        await Helper().To("inst-1").Call("doWork");
+
+        Assert.False(captured!.Trace);
+        Assert.Single(_written);
+    }
+
+    [Fact]
+    public async Task WithTrace_PropagatesOnCallAndSetAttributes()
+    {
+        RouteToCallRequest? call = null;
+        RouteToSetAttributesRequest? set = null;
+        _router.RouteToCallAsync("SiteA", Arg.Do<RouteToCallRequest>(r => call = r), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, null, null, DateTimeOffset.UtcNow));
+        _router.RouteToSetAttributesAsync("SiteA", Arg.Do<RouteToSetAttributesRequest>(r => set = r), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
+
+        var traced = Helper().WithTrace(true).WithDeadline(CancellationToken.None).To("inst-1");
+        await traced.Call("doWork");
+        await traced.SetAttributes(new Dictionary<string, string> { ["A"] = "1" });
+
+        Assert.True(call!.Trace);
+        Assert.True(set!.Trace);
+    }
+
+    [Fact]
+    public async Task Sink_ReceivesRouteCallRow_ThenTheReplysTraceEvents_InOrder()
+    {
+        var child = Guid.NewGuid();
+        var siteEvents = new[] { SiteEvent(child), SiteEvent(child) };
+        _router.RouteToCallAsync("SiteA", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, null, null,
+                DateTimeOffset.UtcNow, child, siteEvents));
+        var sunk = new List<AuditEvent>();
+
+        await Helper().WithTrace(true).WithTraceSink(sunk.Add).To("inst-1").Call("doWork");
+
+        Assert.Equal(3, sunk.Count);
+        Assert.Equal("ScriptTrace.RouteCall", sunk[0].Action);
+        Assert.Same(siteEvents[0], sunk[1]);
+        Assert.Same(siteEvents[1], sunk[2]);
+    }
+
+    [Fact]
+    public async Task Failure_WritesFailedRow_AndStillThrows()
+    {
+        _router.RouteToCallAsync("SiteA", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, false, null, "script faulted", DateTimeOffset.UtcNow));
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => Helper().To("inst-1").Call("doWork"));
+
+        var row = Assert.Single(_written).AsRow();
+        Assert.Equal(AuditStatus.Failed, row.Status);
+        Assert.Equal("script faulted", row.ErrorMessage);
+    }
+
+    [Fact]
+    public async Task NoWriterAndNoSink_WritesNothing_AndNeverThrows()
+    {
+        _router.RouteToCallAsync("SiteA", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 1, null, DateTimeOffset.UtcNow));
+
+        var result = await new RouteHelper(_locator, _router).WithTrace(true).To("inst-1").Call("doWork");
+
+        Assert.Equal(1, result);
+        Assert.Empty(_written);
+    }
+
+    [Fact]
+    public async Task FaultingWriter_DoesNotFailTheCall()
+    {
+        var badWriter = Substitute.For<ICentralAuditWriter>();
+        badWriter.WriteAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>()).Returns(Task.FromException(new IOException("disk")));
+        _router.RouteToCallAsync("SiteA", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
+            .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 5, null, DateTimeOffset.UtcNow));
+
+        var result = await new RouteHelper(_locator, _router, badWriter, null).To("inst-1").Call("doWork");
+        Assert.Equal(5, result);
+    }
+}