diff --git a/docs/plans/2026-09-05-flow-debugger.md.tasks.json b/docs/plans/2026-09-05-flow-debugger.md.tasks.json index 6596d66c..3cdb7c95 100644 --- a/docs/plans/2026-09-05-flow-debugger.md.tasks.json +++ b/docs/plans/2026-09-05-flow-debugger.md.tasks.json @@ -4,12 +4,12 @@ { "id": 0, "subject": "Task 0: Amend the design doc with the four planning corrections", - "status": "pending" + "status": "completed" }, { "id": 1, - "subject": "Task 1: Audit enums \u2014 ScriptTrace channel and four kinds", - "status": "pending", + "subject": "Task 1: Audit enums — ScriptTrace channel and four kinds", + "status": "completed", "blockedBy": [ 0 ] @@ -17,15 +17,15 @@ { "id": 2, "subject": "Task 2: Additive Trace / trace-payload fields on the message records", - "status": "pending", + "status": "completed", "blockedBy": [ 0 ] }, { "id": 3, - "subject": "Task 3: ApiMethod.TraceAllCalls \u2014 entity, EF migration, management command, CLI flag", - "status": "pending", + "subject": "Task 3: ApiMethod.TraceAllCalls — entity, EF migration, management command, CLI flag", + "status": "completed", "blockedBy": [ 0 ] @@ -33,7 +33,7 @@ { "id": 4, "subject": "Task 4: FlowStep projection + FlowGraph model + FlowGraphBuilder (pure)", - "status": "pending", + "status": "completed", "blockedBy": [ 1, 2 @@ -42,15 +42,15 @@ { "id": 5, "subject": "Task 5: FlowTraceCollector (site, per run)", - "status": "pending", + "status": "completed", "blockedBy": [ 0 ] }, { "id": 6, - "subject": "Task 6: ScriptRuntimeContext \u2014 traced flag, trace emission for writes and waits, flag inheritance", - "status": "pending", + "subject": "Task 6: ScriptRuntimeContext — traced flag, trace emission for writes and waits, flag inheritance", + "status": "completed", "blockedBy": [ 4, 5 @@ -58,24 +58,24 @@ }, { "id": 7, - "subject": "Task 7: ScriptRunLauncher + ScriptActor \u2014 ScriptRun rows, ExecutionId + events, value-change inheritance", - "status": "pending", + "subject": "Task 7: ScriptRunLauncher + ScriptActor — ScriptRun rows, ExecutionId + events, value-change inheritance", + "status": "completed", "blockedBy": [ 6 ] }, { "id": 8, - "subject": "Task 8: InstanceActor \u2014 trace arm for data-sourced writes; Trace on published changes", - "status": "pending", + "subject": "Task 8: InstanceActor — trace arm for data-sourced writes; Trace on published changes", + "status": "completed", "blockedBy": [ 2 ] }, { "id": 9, - "subject": "Task 9: DeploymentManagerActor routing \u2014 pass Trace, return child id + events", - "status": "pending", + "subject": "Task 9: DeploymentManagerActor routing — pass Trace, return child id + events", + "status": "completed", "blockedBy": [ 7, 8 @@ -83,8 +83,8 @@ }, { "id": 10, - "subject": "Task 10: Wire \u2014 site_command.proto trace fields, mapper, golden samples", - "status": "pending", + "subject": "Task 10: Wire — site_command.proto trace fields, mapper, golden samples", + "status": "completed", "blockedBy": [ 2, 11 @@ -92,16 +92,16 @@ }, { "id": 11, - "subject": "Task 11: Wire \u2014 sitestream.proto source_execution_id/trace + PullTraceEvents RPC; site SQLite read by execution ids", - "status": "pending", + "subject": "Task 11: Wire — sitestream.proto source_execution_id/trace + PullTraceEvents RPC; site SQLite read by execution ids", + "status": "completed", "blockedBy": [ 2 ] }, { "id": 12, - "subject": "Task 12: RouteHelper \u2014 always-on RouteCall audit row, WithTrace, inline trace sink", - "status": "pending", + "subject": "Task 12: RouteHelper — always-on RouteCall audit row, WithTrace, inline trace sink", + "status": "completed", "blockedBy": [ 1, 2, @@ -111,7 +111,7 @@ { "id": 13, "subject": "Task 13: InboundScriptExecutor.ExecuteAsync(trace) + endpoint honours TraceAllCalls", - "status": "pending", + "status": "completed", "blockedBy": [ 12 ] @@ -119,14 +119,14 @@ { "id": 14, "subject": "Task 14: IAuditLogRepository.GetTraceEventsAsync(rootExecutionId)", - "status": "pending", + "status": "completed", "blockedBy": [ 1 ] }, { "id": 15, - "subject": "Task 15: Management commands \u2014 FlowDebugRunMethod / RunValueChange / Replay", + "subject": "Task 15: Management commands — FlowDebugRunMethod / RunValueChange / Replay", "status": "pending", "blockedBy": [ 9, @@ -155,7 +155,7 @@ }, { "id": 18, - "subject": "Task 18: Central UI service \u2014 IFlowDebugService / FlowDebugService", + "subject": "Task 18: Central UI service — IFlowDebugService / FlowDebugService", "status": "pending", "blockedBy": [ 15, @@ -164,7 +164,7 @@ }, { "id": 19, - "subject": "Task 19: FlowGraphView \u2014 custom SVG swimlane renderer", + "subject": "Task 19: FlowGraphView — custom SVG swimlane renderer", "status": "pending", "blockedBy": [ 4 @@ -172,7 +172,7 @@ }, { "id": 20, - "subject": "Task 20: Flow Debugger page \u2014 toolbar, inputs, predicted graph, run, detail panel", + "subject": "Task 20: Flow Debugger page — toolbar, inputs, predicted graph, run, detail panel", "status": "pending", "blockedBy": [ 18, @@ -190,7 +190,7 @@ }, { "id": 22, - "subject": "Task 22: Rig \u2014 retention override, redeploy, verification step H", + "subject": "Task 22: Rig — retention override, redeploy, verification step H", "status": "pending", "blockedBy": [ 16, @@ -207,5 +207,5 @@ ] } ], - "lastUpdated": "2026-09-06T03:22:55.422160+00:00" + "lastUpdated": "2026-09-06T03:58:13.917546+00:00" } \ No newline at end of file diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs index a7613bcf..6434d474 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs @@ -4,6 +4,7 @@ using Grpc.Core; using Grpc.Net.Client; using Microsoft.Extensions.Logging; using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Communication; using ZB.MOM.WW.ScadaBridge.Communication.Grpc; diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IPullTraceEventsClient.cs similarity index 96% rename from src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs rename to src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IPullTraceEventsClient.cs index d78f227a..1beb7149 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullTraceEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IPullTraceEventsClient.cs @@ -1,6 +1,6 @@ using ZB.MOM.WW.Audit; -namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; +namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; /// /// Flow Debugger (design 2026-09-05 §5.4): central-side client for the site-hosted diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/FlowDebugCommands.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/FlowDebugCommands.cs new file mode 100644 index 00000000..7231b1e9 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Management/FlowDebugCommands.cs @@ -0,0 +1,51 @@ +using System.Text.Json; +using ZB.MOM.WW.Audit; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; + +/// +/// Flow Debugger Live run of an inbound API method (design 2026-09-05 §7.1): executes the +/// method script on the ACTIVE central node with as the request +/// body, tracing every routed site run, and returns the whole flow's audit events inline. +/// Deployer role; refuses on a standby with error code STANDBY_NODE. +/// +/// The method to run. +/// Flat JSON object of parameters (validated against the method's schema); null = none. +public record FlowDebugRunMethodCommand(int ApiMethodId, JsonElement? Parameters); + +/// +/// Flow Debugger Live run of a value-change trigger (design 2026-09-05 §5.3): writes +/// to on the instance through the +/// traced routed-write path, then polls the site for the traced run(s) the change spawned. +/// +/// The instance to write to. +/// Canonical attribute name. +/// Canonical string value to write. +/// How long to wait for a spawned run before reporting NO_TRIGGER_WITHIN_WINDOW. +public record FlowDebugRunValueChangeCommand(int InstanceId, string AttributeName, string Value, int WaitSeconds = 45); + +/// +/// Flow Debugger Replay (design 2026-09-05 §7.1): returns the audit events of the execution +/// tree that belongs to, from the central store. +/// +/// Any execution id in the chain. +public record FlowDebugReplayCommand(Guid ExecutionId); + +/// The result of a Flow Debugger run or replay: the flow's audit events plus outcome metadata. +/// The root execution id (the inbound request / the value-change write). +/// True when the run finished (or the replay found rows); false on timeout / no trigger. +/// Human-readable failure, null on success. +/// Machine-readable failure code, null on success. +/// The flow's audit events, oldest first — feed FlowGraphBuilder.Build. +/// True when the inline event list was capped (the audit rows remain complete). +/// When the run started. +/// When the run finished (or the replay was served). +public record FlowDebugRunResult( + Guid RootExecutionId, + bool Completed, + string? Error, + string? ErrorCode, + IReadOnlyList Events, + bool Truncated, + DateTimeOffset StartedAtUtc, + DateTimeOffset FinishedAtUtc); diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/SchemaRefResolver.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/SchemaRefResolver.cs index 903ba381..07b9e596 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/SchemaRefResolver.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/SchemaRefResolver.cs @@ -27,7 +27,7 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI; /// today's behavior. /// /// -internal static class SchemaRefResolver +public static class SchemaRefResolver { /// /// Builds a lib:Name → schema-JSON resolution seam for the supplied diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs new file mode 100644 index 00000000..90cf04fe --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs @@ -0,0 +1,317 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Text.Json; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.InboundAPI; +using ZB.MOM.WW.ScadaBridge.Security; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.FlowDebug; + +/// +/// Flow Debugger execution path (design 2026-09-05 §7.1): the single place a Live run or a +/// Replay happens, called by the per-node ManagementActor's three +/// FlowDebug* handlers so the Central UI (via ManagementActorHolder) and the +/// CLI (via /management) share one path and one audit story. Resolves every +/// collaborator from the scoped the actor hands it, so it is +/// unit-testable with substitutes and needs no actor. +/// +public sealed class FlowDebugRunner +{ + /// Error code when a Live run is attempted on a standby central node. + public const string StandbyNodeCode = "STANDBY_NODE"; + + /// Error code when a value-change run spawns nothing inside its wait window. + public const string NoTriggerCode = "NO_TRIGGER_WITHIN_WINDOW"; + + /// Error code when the request parameters fail the method's schema. + public const string ValidationFailedCode = "VALIDATION_FAILED"; + + /// Error code when the method / instance / site does not exist. + public const string NotFoundCode = "NOT_FOUND"; + + /// The Extra.source value stamped on rows this runner synthesizes. + public const string SourceTag = "flow-debugger"; + + /// Grace added to a method's timeout before a Live run is reported as timed out (design §10). + public static readonly TimeSpan RunGrace = TimeSpan.FromSeconds(10); + + /// How often the value-change run polls the site for spawned rows. + public static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(1); + + /// Quiet period after the last new row before a value-change run is declared complete. + public static readonly TimeSpan SettleWindow = TimeSpan.FromSeconds(2); + + private readonly ILogger _logger; + private readonly Func _delay; + private readonly Func _now; + + /// Creates the runner. + public FlowDebugRunner(ILogger logger) + : this(logger, Task.Delay, () => DateTimeOffset.UtcNow) + { + } + + /// Creates the runner with an injectable delay and clock (tests). + public FlowDebugRunner( + ILogger logger, + Func delay, + Func now) + { + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _delay = delay ?? throw new ArgumentNullException(nameof(delay)); + _now = now ?? throw new ArgumentNullException(nameof(now)); + } + + /// + /// Live run of an inbound API method (design §7.1): active-node gate, schema validation, + /// synthesized inbound row (actor = operator), traced execution through the real + /// with a trace-sink . + /// + public async Task RunMethodAsync( + IServiceProvider sp, FlowDebugRunMethodCommand cmd, AuthenticatedUser user, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(sp); + ArgumentNullException.ThrowIfNull(cmd); + ArgumentNullException.ThrowIfNull(user); + EnsureActiveNode(sp); + + var repo = sp.GetRequiredService(); + var method = await repo.GetApiMethodByIdAsync(cmd.ApiMethodId, ct).ConfigureAwait(false) + ?? throw new ManagementCommandException($"ApiMethod with ID {cmd.ApiMethodId} not found.", NotFoundCode); + + var resolveRef = await SchemaRefResolver.BuildAsync( + sp.GetService(), [method.ParameterDefinitions], ct).ConfigureAwait(false); + var validation = ParameterValidator.Validate(cmd.Parameters, method.ParameterDefinitions, resolveRef); + if (!validation.IsValid) + { + throw new ManagementCommandException( + validation.ErrorMessage ?? "Parameter validation failed", ValidationFailedCode); + } + + var root = Guid.NewGuid(); + var events = new ConcurrentQueue(); + var startedAt = DateTimeOffset.UtcNow; + var sw = Stopwatch.StartNew(); + var timeout = TimeSpan.FromSeconds(Math.Max(1, method.TimeoutSeconds)); + + var route = sp.GetRequiredService().WithTraceSink(events.Enqueue); + var executor = sp.GetRequiredService(); + + InboundScriptResult result; + using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(ct)) + { + runCts.CancelAfter(timeout + RunGrace); + try + { + result = await executor.ExecuteAsync( + method, validation.Parameters, route, timeout, runCts.Token, + parentExecutionId: root, trace: true).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + result = new InboundScriptResult(false, null, + $"Flow Debugger run timed out after {timeout.TotalSeconds}s + {RunGrace.TotalSeconds}s grace", "TIMEOUT"); + } + } + + var finishedAt = DateTimeOffset.UtcNow; + var requestJson = cmd.Parameters is { } p ? p.GetRawText() : null; + var inboundRow = ScadaBridgeAuditEventFactory.Create( + AuditChannel.ApiInbound, AuditKind.InboundRequest, + result.Success ? AuditStatus.Delivered : AuditStatus.Failed, + occurredAtUtc: startedAt.UtcDateTime, + actor: user.Username, + target: method.Name, + sourceNode: sp.GetService()?.NodeName, + executionId: root, + durationMs: (int)Math.Min(sw.ElapsedMilliseconds, int.MaxValue), + errorMessage: result.ErrorMessage, + requestSummary: requestJson, + responseSummary: result.ResultJson, + extra: new ScriptTraceExtra { Source = SourceTag }.ToJson()); + await WriteCentralAsync(sp, inboundRow, ct).ConfigureAwait(false); + + var ordered = new List(events.Count + 1) { inboundRow }; + ordered.AddRange(events.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId)); + + return new FlowDebugRunResult( + root, result.Success, result.ErrorMessage, result.Success ? null : result.ErrorCode, + ordered, Truncated: false, startedAt, finishedAt); + } + + /// + /// Live run of a value-change trigger (design §5.3/§5.4): traced routed write through the + /// site's normal path, then poll the site's trace rows for the spawned run(s). + /// + public async Task RunValueChangeAsync( + IServiceProvider sp, FlowDebugRunValueChangeCommand cmd, AuthenticatedUser user, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(sp); + ArgumentNullException.ThrowIfNull(cmd); + ArgumentNullException.ThrowIfNull(user); + if (string.IsNullOrWhiteSpace(cmd.AttributeName)) + throw new ManagementCommandException("AttributeName is required.", ValidationFailedCode); + EnsureActiveNode(sp); + + var instance = await sp.GetRequiredService() + .GetInstanceByIdAsync(cmd.InstanceId, ct).ConfigureAwait(false) + ?? throw new ManagementCommandException($"Instance with ID {cmd.InstanceId} not found.", NotFoundCode); + EnforceSiteScope(user, instance.SiteId); + var site = await sp.GetRequiredService() + .GetSiteByIdAsync(instance.SiteId, ct).ConfigureAwait(false) + ?? throw new ManagementCommandException($"Site with ID {instance.SiteId} not found.", NotFoundCode); + + var root = Guid.NewGuid(); + var startedAt = DateTimeOffset.UtcNow; + var sw = Stopwatch.StartNew(); + // IInstanceRouter is the same seam RouteHelper uses (central binds it to the comms layer). + var router = sp.GetRequiredService(); + var write = await router.RouteToSetAttributesAsync( + site.SiteIdentifier, + new RouteToSetAttributesRequest( + Guid.NewGuid().ToString(), instance.UniqueName, + new Dictionary { [cmd.AttributeName] = cmd.Value ?? string.Empty }, + DateTimeOffset.UtcNow, ParentExecutionId: root, Trace: true), + ct).ConfigureAwait(false); + + var rootRow = ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, AuditKind.AttributeWrite, + write.Success ? AuditStatus.Delivered : AuditStatus.Failed, + occurredAtUtc: startedAt.UtcDateTime, + actor: user.Username, + target: cmd.AttributeName, + sourceNode: sp.GetService()?.NodeName, + executionId: root, + sourceSiteId: site.SiteIdentifier, + sourceInstanceId: instance.UniqueName, + durationMs: (int)Math.Min(sw.ElapsedMilliseconds, int.MaxValue), + errorMessage: write.ErrorMessage, + requestSummary: JsonSerializer.Serialize(new { attribute = cmd.AttributeName, value = cmd.Value }, ScriptTraceExtra.JsonOptions), + extra: new ScriptTraceExtra + { + Source = SourceTag, + Writes = [new AttributeWriteDetail(cmd.AttributeName, cmd.Value, write.Success, write.ErrorMessage)], + }.ToJson()); + await WriteCentralAsync(sp, rootRow, ct).ConfigureAwait(false); + + if (!write.Success) + { + return new FlowDebugRunResult(root, false, write.ErrorMessage ?? "Attribute write failed", "WRITE_FAILED", + [rootRow], false, startedAt, DateTimeOffset.UtcNow); + } + + // Poll the site for the traced run(s) the write spawned — matched by ParentExecutionId + // == root, then their children — until a ScriptRun parented to root exists AND the + // set has been quiet for SettleWindow, or WaitSeconds elapses. + var puller = sp.GetRequiredService(); + var deadline = _now() + TimeSpan.FromSeconds(Math.Max(1, cmd.WaitSeconds)); + var known = new Dictionary(); + DateTimeOffset? lastNew = null; + var since = startedAt.UtcDateTime.AddSeconds(-5); + while (!ct.IsCancellationRequested) + { + var ids = new HashSet { root }; + foreach (var e in known.Values) + { + var d = AuditDetailsCodec.Deserialize(e.DetailsJson); + if (d.ExecutionId is { } x) ids.Add(x); + } + var pulled = await puller.PullAsync(site.SiteIdentifier, ids.Take(64).ToList(), since, 500, ct).ConfigureAwait(false); + var added = false; + foreach (var e in pulled) + { + if (known.TryAdd(e.EventId, e)) added = true; + } + if (added) lastNew = _now(); + + var spawned = known.Values.Any(e => + { + var d = AuditDetailsCodec.Deserialize(e.DetailsJson); + return d.Kind == nameof(AuditKind.ScriptRun) && d.ParentExecutionId == root; + }); + var now = _now(); + if (spawned && lastNew is { } ln && now - ln >= SettleWindow) break; + if (now >= deadline) break; + await _delay(PollInterval, ct).ConfigureAwait(false); + } + + var ordered = new List { rootRow }; + ordered.AddRange(known.Values.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId)); + var completed = known.Values.Any(e => + { + var d = AuditDetailsCodec.Deserialize(e.DetailsJson); + return d.Kind == nameof(AuditKind.ScriptRun) && d.ParentExecutionId == root; + }); + return new FlowDebugRunResult( + root, completed, + completed ? null : $"No traced run was spawned within {cmd.WaitSeconds}s (no value-change trigger on '{cmd.AttributeName}', or no device echo).", + completed ? null : NoTriggerCode, + ordered, false, startedAt, DateTimeOffset.UtcNow); + } + + /// Replay (design §7.1): the stored ScriptTrace chain for any execution id. + public async Task ReplayAsync( + IServiceProvider sp, FlowDebugReplayCommand cmd, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(sp); + ArgumentNullException.ThrowIfNull(cmd); + var startedAt = DateTimeOffset.UtcNow; + var events = await sp.GetRequiredService() + .GetTraceEventsAsync(cmd.ExecutionId, ct).ConfigureAwait(false); + if (events.Count == 0) + { + return new FlowDebugRunResult(cmd.ExecutionId, false, + $"No trace rows found for execution {cmd.ExecutionId:D}.", NotFoundCode, + events, false, startedAt, DateTimeOffset.UtcNow); + } + var rootId = AuditDetailsCodec.Deserialize(events[0].DetailsJson).ExecutionId ?? cmd.ExecutionId; + return new FlowDebugRunResult(rootId, true, null, null, events, false, startedAt, DateTimeOffset.UtcNow); + } + + private static void EnsureActiveNode(IServiceProvider sp) + { + var gate = sp.GetService(); + if (gate is not null && !gate.IsActiveNode) + { + throw new ManagementCommandException( + "Live runs execute on the ACTIVE central node only (Component-InboundAPI: method scripts and " + + "Route.To never run on a standby). Open the Central UI or point the CLI at the active node.", + StandbyNodeCode); + } + } + + private static void EnforceSiteScope(AuthenticatedUser user, int siteId) + { + if (user.PermittedSiteIds.Length == 0) return; + if (user.Roles.Contains(Roles.Administrator, StringComparer.OrdinalIgnoreCase)) return; + if (!user.PermittedSiteIds.Contains(siteId.ToString())) + { + throw new SiteScopeViolationException( + $"Access denied: your Deployment role is scoped to sites [{string.Join(", ", user.PermittedSiteIds)}] " + + $"and does not include site {siteId}."); + } + } + + private async Task WriteCentralAsync(IServiceProvider sp, AuditEvent evt, CancellationToken ct) + { + var writer = sp.GetService(); + if (writer is null) return; + try + { + await writer.WriteAsync(evt, ct).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Flow Debugger central audit write failed for {Action}", evt.Action); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs index 3cf0e504..b5b9772e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs @@ -183,7 +183,8 @@ public class ManagementActor : ReceiveActor var clientMessage = cause is ManagementCommandException ? cause.Message : $"An internal error occurred (CorrelationId={correlationId})."; - return new ManagementError(correlationId, clientMessage, "COMMAND_FAILED"); + var errorCode = cause is ManagementCommandException { Code: { Length: > 0 } code } ? code : "COMMAND_FAILED"; + return new ManagementError(correlationId, clientMessage, errorCode); } // Static readonly role sets for the authorization gate. Single-role arms @@ -304,6 +305,10 @@ public class ManagementActor : ReceiveActor or RetryParkedMessageCommand or DiscardParkedMessageCommand or DebugSnapshotCommand => DeployerOnly, + // Flow Debugger (design 2026-09-05 §9): Live runs execute scripts and write attributes + // on operator input, Replay reads the audit chain — Deployer, like the Debug View. + FlowDebugRunMethodCommand or FlowDebugRunValueChangeCommand or FlowDebugReplayCommand => DeployerOnly, + // Two-person secured write. Submit is an Operator action; // approve/reject are Verifier actions. Separation of duties (a write may // not be verified by its submitter) is enforced inside the handler — the @@ -401,6 +406,11 @@ public class ManagementActor : ReceiveActor DeleteDataConnectionCommand cmd => await HandleDeleteDataConnection(sp, cmd, user.Username), MoveDataConnectionCommand cmd => await HandleMoveDataConnection(sp, cmd, user.Username), + // Flow Debugger (design 2026-09-05 §7.1) — one execution path for the UI and the CLI. + FlowDebugRunMethodCommand cmd => await sp.GetRequiredService().RunMethodAsync(sp, cmd, user, CancellationToken.None), + FlowDebugRunValueChangeCommand cmd => await sp.GetRequiredService().RunValueChangeAsync(sp, cmd, user, CancellationToken.None), + FlowDebugReplayCommand cmd => await sp.GetRequiredService().ReplayAsync(sp, cmd, CancellationToken.None), + // External Systems ListExternalSystemsCommand => await HandleListExternalSystems(sp), GetExternalSystemCommand cmd => await HandleGetExternalSystem(sp, cmd), @@ -4207,4 +4217,16 @@ public class ManagementCommandException : Exception /// Initializes a new instance with the given caller-safe error message. /// The error message to surface verbatim to the HTTP/CLI caller. public ManagementCommandException(string message) : base(message) { } + + /// + /// Initializes a new instance with a caller-safe message AND a machine-readable error code + /// that replaces the generic COMMAND_FAILED on the ManagementError reply + /// (Flow Debugger design 2026-09-05 §7.1: STANDBY_NODE, NO_TRIGGER_WITHIN_WINDOW). + /// + /// The error message to surface verbatim to the HTTP/CLI caller. + /// The error code to surface as ManagementError.ErrorCode. + public ManagementCommandException(string message, string code) : base(message) { Code = code; } + + /// Optional machine-readable error code; null = COMMAND_FAILED. + public string? Code { get; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs index 0d70ea5a..3a04aadc 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs @@ -55,6 +55,9 @@ public static class ManagementEndpoints // it is N instance deploys, so it must never inherit the ordinary Ask // timeout or the caller would 504 while the batch is still applying. typeof(MgmtDeploySiteCommand), + // Flow Debugger (design 2026-09-05 §7.1): a Live run lasts up to the method timeout + // (+ grace) or the value-change wait window — never the ordinary Ask timeout. + typeof(FlowDebugRunMethodCommand), typeof(FlowDebugRunValueChangeCommand), ]; /// diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ServiceCollectionExtensions.cs index 1a1221f7..5d727e2a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ServiceCollectionExtensions.cs @@ -14,6 +14,8 @@ public static class ServiceCollectionExtensions public static IServiceCollection AddManagementService(this IServiceCollection services) { services.AddSingleton(); + // Flow Debugger (design 2026-09-05 §7.1): the runner behind the three FlowDebug* commands. + services.AddSingleton(); services.AddOptions() .BindConfiguration("ScadaBridge:ManagementService") .ValidateOnStart(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs new file mode 100644 index 00000000..8ca0642b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs @@ -0,0 +1,242 @@ +using System.Text.Json; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.InboundAPI; +using ZB.MOM.WW.ScadaBridge.ManagementService.FlowDebug; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; + +/// +/// Flow Debugger design 2026-09-05 §7.1: the runner behind the three FlowDebug* commands — +/// standby gate, schema validation before any routing, synthesized inbound row + sink events +/// on a Live run, the value-change poll loop, and Replay from the repository. +/// +public class FlowDebugRunnerTests +{ + private static readonly AuthenticatedUser Deployer = new("ops", "Ops", new[] { "Deployer" }, Array.Empty()); + + private readonly ServiceCollection _services = new(); + private readonly IInboundApiRepository _apiRepo = Substitute.For(); + private readonly IInstanceLocator _locator = Substitute.For(); + private readonly IInstanceRouter _router = Substitute.For(); + private readonly List _central = new(); + private readonly IActiveNodeGate _gate = Substitute.For(); + private readonly InboundScriptExecutor _executor = new(NullLogger.Instance, Substitute.For()); + private DateTimeOffset _clock = new(2026, 9, 5, 12, 0, 0, TimeSpan.Zero); + + public FlowDebugRunnerTests() + { + _gate.IsActiveNode.Returns(true); + var writer = Substitute.For(); + writer.WriteAsync(Arg.Do(_central.Add), Arg.Any()).Returns(Task.CompletedTask); + _services.AddSingleton(_apiRepo); + _services.AddSingleton(_gate); + _services.AddSingleton(writer); + _services.AddSingleton(_locator); + _services.AddSingleton(_router); + _services.AddSingleton(_executor); + _services.AddScoped(sp => new RouteHelper(_locator, _router, sp.GetService(), null)); + } + + private FlowDebugRunner Runner() => new( + NullLogger.Instance, + (d, _) => { _clock += d; return Task.CompletedTask; }, + () => _clock); + + private static JsonElement Json(string s) => JsonDocument.Parse(s).RootElement.Clone(); + + [Fact] + public async Task RunMethod_OnStandby_RefusesWithStandbyCode_BeforeTouchingTheMethod() + { + _gate.IsActiveNode.Returns(false); + var ex = await Assert.ThrowsAsync(() => + Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(1, null), Deployer, CancellationToken.None)); + Assert.Equal(FlowDebugRunner.StandbyNodeCode, ex.Code); + Assert.Contains("ACTIVE central node", ex.Message); + await _apiRepo.DidNotReceiveWithAnyArgs().GetApiMethodByIdAsync(default, default); + } + + [Fact] + public async Task RunMethod_InvalidParameters_FailsBeforeAnyRouteCall() + { + var method = new ApiMethod("M", "return 1;") + { + Id = 3, TimeoutSeconds = 5, + ParameterDefinitions = "{\"type\":\"object\",\"properties\":{\"n\":{\"type\":\"integer\"}},\"required\":[\"n\"]}", + }; + _apiRepo.GetApiMethodByIdAsync(3, Arg.Any()).Returns(method); + + var ex = await Assert.ThrowsAsync(() => + Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(3, Json("{}")), Deployer, CancellationToken.None)); + Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code); + await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default); + Assert.Empty(_central); + } + + [Fact] + public async Task RunMethod_HappyPath_ReturnsInboundRow_ThenSinkEvents_Ordered_AndWritesTheInboundRow() + { + var method = new ApiMethod("M", "x") { Id = 4, TimeoutSeconds = 5 }; + _apiRepo.GetApiMethodByIdAsync(4, Arg.Any()).Returns(method); + _locator.GetSiteIdForInstanceAsync("inst", Arg.Any()).Returns("site-a"); + var child = Guid.NewGuid(); + var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, + occurredAtUtc: DateTime.UtcNow.AddSeconds(1), executionId: child, target: "work"); + RouteToCallRequest? captured = null; + _router.RouteToCallAsync("site-a", Arg.Do(r => captured = r), Arg.Any()) + .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(4, Json("{\"a\":1}")), Deployer, CancellationToken.None); + + Assert.True(result.Completed, result.Error); + Assert.Null(result.ErrorCode); + Assert.True(captured!.Trace); + Assert.Equal(result.RootExecutionId, captured.ParentExecutionId); + + // inbound (synthesized) → RouteCall (sink) → site ScriptRun (sink, from the reply) + Assert.Equal(3, result.Events.Count); + var inbound = result.Events[0].AsRow(); + Assert.Equal(AuditKind.InboundRequest, inbound.Kind); + Assert.Equal("ops", result.Events[0].Actor); + Assert.Equal("M", result.Events[0].Target); + Assert.Equal(result.RootExecutionId, inbound.ExecutionId); + Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(inbound.Extra)!.Source); + Assert.Contains("\"a\":1", inbound.RequestSummary); + Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind); + Assert.Same(siteRun, result.Events[2]); + + // Central writer saw the inbound row AND the always-on RouteCall row. + Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.InboundRequest); + Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.RouteCall); + } + + [Fact] + public async Task RunMethod_UnknownMethod_IsNotFound() + { + var ex = await Assert.ThrowsAsync(() => + Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(99, null), Deployer, CancellationToken.None)); + Assert.Equal(FlowDebugRunner.NotFoundCode, ex.Code); + } + + private (ITemplateEngineRepository Templates, ISiteRepository Sites, IPullTraceEventsClient Puller) WireValueChange() + { + var templates = Substitute.For(); + var sites = Substitute.For(); + var puller = Substitute.For(); + templates.GetInstanceByIdAsync(5, Arg.Any()).Returns(new Instance("Pump1") { Id = 5, SiteId = 2 }); + sites.GetSiteByIdAsync(2, Arg.Any()).Returns(new Site("Site A", "site-a") { Id = 2 }); + _router.RouteToSetAttributesAsync("site-a", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow)); + _services.AddSingleton(templates); + _services.AddSingleton(sites); + _services.AddSingleton(puller); + return (templates, sites, puller); + } + + [Fact] + public async Task RunValueChange_TracedWrite_ThenPollsUntilSpawnedRunSettles() + { + var (_, _, puller) = WireValueChange(); + RouteToSetAttributesRequest? write = null; + _router.RouteToSetAttributesAsync("site-a", Arg.Do(r => write = r), Arg.Any()) + .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow)); + var calls = 0; + puller.PullAsync("site-a", Arg.Any>(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(ci => + { + calls++; + var ids = (IReadOnlyCollection)ci[1]; + var root = ids.First(); + // Second poll onwards: a ScriptRun parented to the root exists. + return calls >= 2 + ? (IReadOnlyList)[ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, + eventId: Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001"), executionId: Guid.NewGuid(), parentExecutionId: root, target: "OnStatus")] + : Array.Empty(); + }); + + var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(), + new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 30), Deployer, CancellationToken.None); + + Assert.True(result.Completed, result.Error); + Assert.True(write!.Trace); + Assert.Equal(result.RootExecutionId, write.ParentExecutionId); + Assert.Equal("Pump1", write.InstanceUniqueName); + Assert.Equal("Running", write.AttributeValues["Status"]); + Assert.Equal(2, result.Events.Count); + Assert.Equal(AuditKind.AttributeWrite, result.Events[0].AsRow().Kind); + Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(result.Events[0].AsRow().Extra)!.Source); + Assert.Equal(AuditKind.ScriptRun, result.Events[1].AsRow().Kind); + // Settled: at least one quiet poll (2 s window) after the last new row, then stopped. + Assert.True(calls >= 3, $"expected settle polls, got {calls}"); + } + + [Fact] + public async Task RunValueChange_NoSpawnWithinWindow_ReportsNoTriggerCode() + { + var (_, _, puller) = WireValueChange(); + puller.PullAsync(default!, default!, default, default, default).ReturnsForAnyArgs(Array.Empty()); + + var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(), + new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 3), Deployer, CancellationToken.None); + + Assert.False(result.Completed); + Assert.Equal(FlowDebugRunner.NoTriggerCode, result.ErrorCode); + Assert.Single(result.Events); // just the synthesized write row + } + + [Fact] + public async Task RunValueChange_OutOfScopeSite_IsRefused() + { + WireValueChange(); + var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "9" }); + await Assert.ThrowsAsync(() => Runner().RunValueChangeAsync( + _services.BuildServiceProvider(), new FlowDebugRunValueChangeCommand(5, "Status", "x"), scoped, CancellationToken.None)); + await _router.DidNotReceiveWithAnyArgs().RouteToSetAttributesAsync(default!, default!, default); + } + + [Fact] + public async Task Replay_ReturnsRepositoryRows_WithRootFromTheFirstRow() + { + var repo = Substitute.For(); + var root = Guid.NewGuid(); + var child = Guid.NewGuid(); + var rows = new List + { + ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered, executionId: root), + ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, executionId: child, parentExecutionId: root), + }; + repo.GetTraceEventsAsync(child, Arg.Any()).Returns(rows); + _services.AddSingleton(repo); + + var result = await Runner().ReplayAsync(_services.BuildServiceProvider(), new FlowDebugReplayCommand(child), CancellationToken.None); + + Assert.True(result.Completed); + Assert.Equal(root, result.RootExecutionId); + Assert.Equal(rows, result.Events); + + var empty = await Runner().ReplayAsync(_services.BuildServiceProvider(), new FlowDebugReplayCommand(Guid.NewGuid()), CancellationToken.None); + Assert.False(empty.Completed); + Assert.Equal(FlowDebugRunner.NotFoundCode, empty.ErrorCode); + } + + [Fact] + public void RoleGate_AllThreeCommands_AreDeployerOnly() + { + foreach (var cmd in new object[] { new FlowDebugRunMethodCommand(1, null), new FlowDebugRunValueChangeCommand(1, "a", "b"), new FlowDebugReplayCommand(Guid.NewGuid()) }) + Assert.Equal(new[] { "Deployer" }, ManagementActor.GetRequiredRoles(cmd)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs index d3b98acf..a289ab04 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/RequiredRoleMatrixTests.cs @@ -138,6 +138,10 @@ public class RequiredRoleMatrixTests ["RetryParkedMessage"] = [Roles.Deployer], ["DiscardParkedMessage"] = [Roles.Deployer], ["DebugSnapshot"] = [Roles.Deployer], + // Flow Debugger (design 2026-09-05 §9): Live runs + Replay are Deployer, like Debug View. + ["FlowDebugRunMethod"] = [Roles.Deployer], + ["FlowDebugRunValueChange"] = [Roles.Deployer], + ["FlowDebugReplay"] = [Roles.Deployer], // ---- Two-person secured write ---------------------------------------------- ["SubmitSecuredWrite"] = [Roles.Operator],