feat(central-ui): FlowDebugService — management-command bridge + predicted graph
IFlowDebugService/FlowDebugService (scoped, registered beside IDataConnectionMoveService). Run/replay go to the per-node ManagementActor through ManagementActorHolder with the Blazor principal projected to AuthenticatedUser, exactly as DataConnectionMoveService does; the FlowDebugRunResult payload is deserialised (camelCase) and fed to FlowGraphBuilder, while ManagementError/Unauthorized/holder-null/transport faults collapse into FlowDebugOutcome with the server's error code verbatim (STANDBY_NODE reaches the page untouched). Prediction is DB-only: PredictMethodAsync hands FlowShapePredictor a memoised resolver that maps a route target → Instance.UniqueName → template parent chain (nearest definition wins, cycle guarded) → TemplateScript.Code; PredictValueChangeAsync roots at the write and appends each script whose value-change trigger names the attribute (same attributeName/attribute keys the site's TriggerRouting parser reads). Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP
This commit is contained in:
Joseph Doherty
@@ -140,6 +140,7 @@ public static class ServiceCollectionExtensions
|
||||
// name collision, no instance binding, no native-alarm-source name reference);
|
||||
// the move dialog only SUBMITS the command and renders the returned outcome.
|
||||
services.AddScoped<IDataConnectionMoveService, DataConnectionMoveService>();
|
||||
services.AddScoped<IFlowDebugService, FlowDebugService>();
|
||||
|
||||
// Schema Library: authoring + read accessors for the reusable named
|
||||
// JSON-Schema library (SharedSchema + ISharedSchemaRepository).
|
||||
|
||||
@@ -0,0 +1,301 @@
|
||||
using System.Security.Claims;
|
||||
using System.Text.Json;
|
||||
using Akka.Actor;
|
||||
using Microsoft.AspNetCore.Components.Authorization;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||
using ZB.MOM.WW.ScadaBridge.ManagementService;
|
||||
using ZB.MOM.WW.ScadaBridge.ScriptAnalysis;
|
||||
using ZB.MOM.WW.ScadaBridge.Security;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Default <see cref="IFlowDebugService"/>: prediction runs in-process over the repositories
|
||||
/// (method script + instance → template chain → script bodies, DB-only, nothing executes);
|
||||
/// the run/replay commands go to the per-node <c>ManagementActor</c> through
|
||||
/// <see cref="ManagementActorHolder"/>, exactly as <see cref="DataConnectionMoveService"/>
|
||||
/// does, with the current Blazor principal projected to an <see cref="AuthenticatedUser"/>
|
||||
/// so the server's Deployer gate runs against the real identity. The Live-run Ask waits up
|
||||
/// to <see cref="RunAskTimeout"/> (the server bounds a run by the method timeout + grace and
|
||||
/// a value-change by its wait window; both fit inside two minutes).
|
||||
/// </summary>
|
||||
public sealed class FlowDebugService : IFlowDebugService
|
||||
{
|
||||
/// <summary>Ask timeout for the two Live-run commands.</summary>
|
||||
public static readonly TimeSpan RunAskTimeout = TimeSpan.FromMinutes(2);
|
||||
|
||||
/// <summary>Ask timeout for Replay (a central DB read).</summary>
|
||||
public static readonly TimeSpan ReplayAskTimeout = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>Upper bound on the template parent chain the script resolver walks (cycle guard).</summary>
|
||||
private const int MaxTemplateChain = 16;
|
||||
|
||||
private static readonly JsonSerializerOptions ResultJsonOptions = new()
|
||||
{
|
||||
PropertyNameCaseInsensitive = true,
|
||||
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
||||
};
|
||||
|
||||
private readonly ManagementActorHolder _holder;
|
||||
private readonly AuthenticationStateProvider _auth;
|
||||
private readonly IInboundApiRepository _inboundApi;
|
||||
private readonly ITemplateEngineRepository _templates;
|
||||
private readonly ILogger<FlowDebugService> _logger;
|
||||
|
||||
/// <summary>Initializes the service.</summary>
|
||||
public FlowDebugService(
|
||||
ManagementActorHolder holder,
|
||||
AuthenticationStateProvider auth,
|
||||
IInboundApiRepository inboundApi,
|
||||
ITemplateEngineRepository templates,
|
||||
ILogger<FlowDebugService> logger)
|
||||
{
|
||||
_holder = holder ?? throw new ArgumentNullException(nameof(holder));
|
||||
_auth = auth ?? throw new ArgumentNullException(nameof(auth));
|
||||
_inboundApi = inboundApi ?? throw new ArgumentNullException(nameof(inboundApi));
|
||||
_templates = templates ?? throw new ArgumentNullException(nameof(templates));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<PredictedFlow> PredictMethodAsync(int apiMethodId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var method = await _inboundApi.GetApiMethodByIdAsync(apiMethodId, cancellationToken);
|
||||
if (method == null)
|
||||
return PredictedFlow.Empty;
|
||||
|
||||
// The predictor is synchronous and takes a synchronous resolver, so the instance →
|
||||
// template → script lookups are pre-fetched lazily through a small memo that blocks
|
||||
// on the repository — acceptable for a handful of constant route targets, and the
|
||||
// predictor never asks for anything else.
|
||||
var memo = new Dictionary<(string, string), string?>();
|
||||
string? Resolve(string instanceCode, string scriptName)
|
||||
{
|
||||
var key = (instanceCode, scriptName);
|
||||
if (memo.TryGetValue(key, out var cached))
|
||||
return cached;
|
||||
string? body;
|
||||
try
|
||||
{
|
||||
body = ResolveSiteScriptAsync(instanceCode, scriptName, cancellationToken).GetAwaiter().GetResult();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogDebug(ex, "Flow prediction could not resolve {Instance}.{Script}", instanceCode, scriptName);
|
||||
body = null;
|
||||
}
|
||||
|
||||
memo[key] = body;
|
||||
return body;
|
||||
}
|
||||
|
||||
return FlowShapePredictor.PredictInboundMethod(method.Name, method.Script ?? string.Empty, Resolve);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<PredictedFlow> PredictValueChangeAsync(int instanceId, string attributeName, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var instance = await _templates.GetInstanceByIdAsync(instanceId, cancellationToken);
|
||||
if (instance == null)
|
||||
return PredictedFlow.Empty;
|
||||
|
||||
var scripts = await CollectEffectiveScriptsAsync(instance.TemplateId, cancellationToken);
|
||||
var triggered = scripts.Values
|
||||
.Where(s => IsValueChangeOn(s, attributeName))
|
||||
.OrderBy(s => s.Name, StringComparer.Ordinal)
|
||||
.ToList();
|
||||
|
||||
// Root: the write itself (Tags lane), then each triggered script predicted as a site script.
|
||||
var nodes = new List<PredictedNode>
|
||||
{
|
||||
new("n1", FlowStepKind.AttributeWrite, FlowLane.Tags, $"{instance.UniqueName}.{attributeName}", false,
|
||||
triggered.Count == 0 ? "no value-change script on this attribute" : null),
|
||||
};
|
||||
var edges = new List<PredictedEdge>();
|
||||
foreach (var script in triggered)
|
||||
{
|
||||
var sub = FlowShapePredictor.PredictSiteScript($"{instance.UniqueName}.{script.Name}", script.Code ?? string.Empty);
|
||||
var offset = nodes.Count;
|
||||
var map = new Dictionary<string, string>(StringComparer.Ordinal);
|
||||
foreach (var n in sub.Nodes)
|
||||
{
|
||||
var id = "n" + (nodes.Count + 1);
|
||||
map[n.Id] = id;
|
||||
nodes.Add(n with { Id = id });
|
||||
}
|
||||
|
||||
if (sub.Nodes.Count > 0)
|
||||
edges.Add(new PredictedEdge("n1", nodes[offset].Id, false));
|
||||
foreach (var e in sub.Edges)
|
||||
edges.Add(e with { From = map[e.From], To = map[e.To] });
|
||||
}
|
||||
|
||||
return new PredictedFlow(nodes, edges);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public Task<FlowDebugOutcome> RunMethodAsync(int apiMethodId, JsonElement? parameters, CancellationToken cancellationToken = default)
|
||||
=> SendAndBuildAsync(new FlowDebugRunMethodCommand(apiMethodId, parameters), RunAskTimeout, cancellationToken);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public Task<FlowDebugOutcome> RunValueChangeAsync(int instanceId, string attributeName, string value, CancellationToken cancellationToken = default)
|
||||
=> SendAndBuildAsync(new FlowDebugRunValueChangeCommand(instanceId, attributeName, value), RunAskTimeout, cancellationToken);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public Task<FlowDebugOutcome> ReplayAsync(Guid executionId, CancellationToken cancellationToken = default)
|
||||
=> SendAndBuildAsync(new FlowDebugReplayCommand(executionId), ReplayAskTimeout, cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Resolves <c>instanceCode.scriptName</c> to a script body: instance by unique name →
|
||||
/// its template chain (nearest definition wins) → the named script's code.
|
||||
/// </summary>
|
||||
private async Task<string?> ResolveSiteScriptAsync(string instanceCode, string scriptName, CancellationToken ct)
|
||||
{
|
||||
var instance = await _templates.GetInstanceByUniqueNameAsync(instanceCode, ct);
|
||||
if (instance == null)
|
||||
return null;
|
||||
|
||||
var scripts = await CollectEffectiveScriptsAsync(instance.TemplateId, ct);
|
||||
return scripts.TryGetValue(scriptName, out var script) ? script.Code : null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Walks the template parent chain from <paramref name="templateId"/> upward, collecting
|
||||
/// scripts by name with the nearest (most-derived) definition winning — so an override in
|
||||
/// a derived template is what gets predicted, and a base-only script is still found.
|
||||
/// </summary>
|
||||
private async Task<Dictionary<string, TemplateScript>> CollectEffectiveScriptsAsync(int templateId, CancellationToken ct)
|
||||
{
|
||||
var result = new Dictionary<string, TemplateScript>(StringComparer.OrdinalIgnoreCase);
|
||||
int? current = templateId;
|
||||
var hops = 0;
|
||||
var seen = new HashSet<int>();
|
||||
while (current is int id && hops++ < MaxTemplateChain && seen.Add(id))
|
||||
{
|
||||
var scripts = await _templates.GetScriptsByTemplateIdAsync(id, ct);
|
||||
foreach (var s in scripts)
|
||||
{
|
||||
if (!result.ContainsKey(s.Name))
|
||||
result[s.Name] = s;
|
||||
}
|
||||
|
||||
var template = await _templates.GetTemplateByIdAsync(id, ct);
|
||||
current = template?.ParentTemplateId;
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>True when the script's trigger is a value-change on <paramref name="attributeName"/> (case-insensitive).</summary>
|
||||
private static bool IsValueChangeOn(TemplateScript script, string attributeName)
|
||||
{
|
||||
if (!string.Equals(script.TriggerType, "valuechange", StringComparison.OrdinalIgnoreCase)
|
||||
&& !string.Equals(script.TriggerType, "ValueChange", StringComparison.OrdinalIgnoreCase))
|
||||
return false;
|
||||
|
||||
var cfg = script.TriggerConfiguration;
|
||||
if (string.IsNullOrWhiteSpace(cfg))
|
||||
return false;
|
||||
|
||||
try
|
||||
{
|
||||
using var doc = JsonDocument.Parse(cfg);
|
||||
if (doc.RootElement.ValueKind != JsonValueKind.Object)
|
||||
return false;
|
||||
foreach (var prop in doc.RootElement.EnumerateObject())
|
||||
{
|
||||
if (prop.NameEquals("attribute") || prop.NameEquals("Attribute")
|
||||
|| prop.NameEquals("attributeName") || prop.NameEquals("AttributeName"))
|
||||
{
|
||||
return prop.Value.ValueKind == JsonValueKind.String
|
||||
&& string.Equals(prop.Value.GetString(), attributeName, StringComparison.OrdinalIgnoreCase);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
// Malformed trigger config — not something prediction should fail over.
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
private async Task<FlowDebugOutcome> SendAndBuildAsync(object command, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
var response = await SendAsync(command, timeout, ct);
|
||||
switch (response)
|
||||
{
|
||||
case ManagementSuccess success:
|
||||
{
|
||||
FlowDebugRunResult? result;
|
||||
try
|
||||
{
|
||||
result = JsonSerializer.Deserialize<FlowDebugRunResult>(success.JsonData, ResultJsonOptions);
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.LogError(ex, "Flow Debugger result could not be parsed for {Command}", command.GetType().Name);
|
||||
return FlowDebugOutcome.Fail("The server returned an unreadable result.", "BAD_RESULT");
|
||||
}
|
||||
|
||||
if (result == null)
|
||||
return FlowDebugOutcome.Fail("The server returned no result.", "BAD_RESULT");
|
||||
|
||||
var graph = FlowGraphBuilder.Build(result.Events ?? [], result.Truncated);
|
||||
return new FlowDebugOutcome(graph, result, null, null);
|
||||
}
|
||||
case ManagementUnauthorized unauthorized:
|
||||
return FlowDebugOutcome.Fail(unauthorized.Message, "UNAUTHORIZED");
|
||||
case ManagementError error:
|
||||
return FlowDebugOutcome.Fail(error.Error, error.ErrorCode);
|
||||
default:
|
||||
return FlowDebugOutcome.Fail("Unexpected response from the management service.", "INTERNAL_ERROR");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Wraps <paramref name="command"/> in a <see cref="ManagementEnvelope"/> for the current
|
||||
/// principal and Asks the <c>ManagementActor</c>. Transport faults become a synthetic
|
||||
/// <see cref="ManagementError"/> so callers handle one response shape.
|
||||
/// </summary>
|
||||
private async Task<object> SendAsync(object command, TimeSpan timeout, CancellationToken cancellationToken)
|
||||
{
|
||||
var actor = _holder.ActorRef;
|
||||
if (actor is null)
|
||||
return new ManagementError(string.Empty, "Management service is not ready.", "SERVICE_UNAVAILABLE");
|
||||
|
||||
var user = await BuildAuthenticatedUserAsync();
|
||||
var envelope = new ManagementEnvelope(user, command, Guid.NewGuid().ToString("N"));
|
||||
|
||||
try
|
||||
{
|
||||
return await actor.Ask<object>(envelope, timeout, cancellationToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "ManagementActor Ask failed for {Command}", command.GetType().Name);
|
||||
return new ManagementError(string.Empty, ex.Message, "TRANSPORT_ERROR");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<AuthenticatedUser> BuildAuthenticatedUserAsync()
|
||||
{
|
||||
var state = await _auth.GetAuthenticationStateAsync();
|
||||
var principal = state.User;
|
||||
|
||||
var username = principal.FindFirst(JwtTokenService.UsernameClaimType)?.Value ?? "unknown";
|
||||
var displayName = principal.FindFirst(JwtTokenService.DisplayNameClaimType)?.Value ?? username;
|
||||
var roles = principal.FindAll(JwtTokenService.RoleClaimType).Select(c => c.Value).ToArray();
|
||||
var permittedSiteIds = principal.FindAll(JwtTokenService.SiteIdClaimType).Select(c => c.Value).ToArray();
|
||||
|
||||
return new AuthenticatedUser(username, displayName, roles, permittedSiteIds);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
using System.Text.Json;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||
using ZB.MOM.WW.ScadaBridge.ScriptAnalysis;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Outcome of a Flow Debugger run or replay as the page consumes it: the built
|
||||
/// <see cref="FlowGraph"/> (null on failure), the raw server result (null when the command
|
||||
/// itself was refused), and a classified error.
|
||||
/// </summary>
|
||||
/// <param name="Graph">The flow graph built from the returned audit events; null when the command failed outright.</param>
|
||||
/// <param name="Raw">The server's <see cref="FlowDebugRunResult"/>; null when no result came back.</param>
|
||||
/// <param name="Error">Human-readable failure, null on success.</param>
|
||||
/// <param name="ErrorCode">Machine-readable failure code (<c>STANDBY_NODE</c>, <c>NO_TRIGGER_WITHIN_WINDOW</c>, <c>UNAUTHORIZED</c>, <c>SERVICE_UNAVAILABLE</c>, …), null on success.</param>
|
||||
public sealed record FlowDebugOutcome(FlowGraph? Graph, FlowDebugRunResult? Raw, string? Error, string? ErrorCode)
|
||||
{
|
||||
/// <summary>True when a graph was built (the run may still have finished incomplete — see <see cref="Raw"/>).</summary>
|
||||
public bool Success => Graph != null && Error == null;
|
||||
|
||||
/// <summary>Creates a failed outcome.</summary>
|
||||
public static FlowDebugOutcome Fail(string error, string errorCode) => new(null, null, error, errorCode);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// CentralUI facade over the Flow Debugger (design 2026-09-05 §5.5/§7.1): static shape
|
||||
/// prediction from script syntax, and the three management commands (Live run of a method,
|
||||
/// Live run of a value change, Replay) dispatched to the per-node <c>ManagementActor</c>
|
||||
/// through the in-process <c>ManagementActorHolder</c> — the SAME Ask path the HTTP
|
||||
/// <c>/management</c> endpoint uses, so the server stays the single enforcer of the Deployer
|
||||
/// gate and the active-node refusal (<c>STANDBY_NODE</c>).
|
||||
/// </summary>
|
||||
public interface IFlowDebugService
|
||||
{
|
||||
/// <summary>Predicts the flow shape of an inbound API method from its script and the site scripts its constant routes resolve to.</summary>
|
||||
/// <param name="apiMethodId">The method id.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
/// <returns>The predicted flow; <see cref="PredictedFlow.Empty"/> when the method does not exist.</returns>
|
||||
Task<PredictedFlow> PredictMethodAsync(int apiMethodId, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Predicts the flow a value change on <paramref name="attributeName"/> would trigger: the
|
||||
/// instance's template scripts whose value-change trigger names that attribute, each predicted
|
||||
/// as a site script.
|
||||
/// </summary>
|
||||
/// <param name="instanceId">The instance id.</param>
|
||||
/// <param name="attributeName">Canonical attribute name.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
Task<PredictedFlow> PredictValueChangeAsync(int instanceId, string attributeName, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>Runs an inbound API method with tracing on the active central node and returns the flow.</summary>
|
||||
/// <param name="apiMethodId">The method id.</param>
|
||||
/// <param name="parameters">Flat JSON object of parameters, or null.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
Task<FlowDebugOutcome> RunMethodAsync(int apiMethodId, JsonElement? parameters, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>Writes an attribute through the traced path and returns the flow the change triggered.</summary>
|
||||
/// <param name="instanceId">The instance id.</param>
|
||||
/// <param name="attributeName">Canonical attribute name.</param>
|
||||
/// <param name="value">Value in canonical string form.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
Task<FlowDebugOutcome> RunValueChangeAsync(int instanceId, string attributeName, string value, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>Rebuilds a past flow from the central audit store.</summary>
|
||||
/// <param name="executionId">Any execution id in the chain.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
Task<FlowDebugOutcome> ReplayAsync(Guid executionId, CancellationToken cancellationToken = default);
|
||||
}
|
||||
Reference in New Issue
Block a user