From a5979b5cfd0f9077faaa11fbeff1b25c2a8d67bc Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 8 Sep 2026 01:15:57 -0400 Subject: [PATCH] =?UTF-8?q?feat(management):=20FlowDebugRunner.RunScriptAs?= =?UTF-8?q?ync=20=E2=80=94=20write-kind=20vs=20direct=20traced=20call=20by?= =?UTF-8?q?=20trigger=20kind=20(flow=20debugger=20script=20entry)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29 --- .../FlowDebug/FlowDebugRunner.cs | 137 ++++++++++++++- .../FlowDebugRunnerTests.cs | 163 ++++++++++++++++++ 2 files changed, 299 insertions(+), 1 deletion(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs index b5bcf529..5473b842 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/FlowDebug/FlowDebugRunner.cs @@ -12,6 +12,8 @@ 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.Commons.Types.Scripts; +using ZB.MOM.WW.ScadaBridge.DeploymentManager; using ZB.MOM.WW.ScadaBridge.InboundAPI; using ZB.MOM.WW.ScadaBridge.Security; @@ -39,6 +41,18 @@ public sealed class FlowDebugRunner /// Error code when the method / instance / site does not exist. public const string NotFoundCode = "NOT_FOUND"; + /// Error code when a direct script call (design 2026-09-08 §4.4) fails at the site. + public const string CallFailedCode = "CALL_FAILED"; + + /// Error code when a Live run exceeds the method's / script's timeout plus . + public const string TimeoutCode = "TIMEOUT"; + + /// + /// Timeout for a direct script call whose script declares none — mirrors + /// SiteRuntimeOptions.ScriptExecutionTimeoutSeconds's default (the runner has no site options). + /// + public const int DefaultScriptTimeoutSeconds = 30; + /// /// Arch-review F2: 's code for a routed call refused /// because the target site is outside the operator's site scope. InboundAPI cannot @@ -145,7 +159,7 @@ public sealed class FlowDebugRunner catch (OperationCanceledException) when (!ct.IsCancellationRequested) { result = new InboundScriptResult(false, null, - $"Flow Debugger run timed out after {timeout.TotalSeconds}s + {RunGrace.TotalSeconds}s grace", "TIMEOUT"); + $"Flow Debugger run timed out after {timeout.TotalSeconds}s + {RunGrace.TotalSeconds}s grace", TimeoutCode); } } @@ -218,6 +232,127 @@ public sealed class FlowDebugRunner .ConfigureAwait(false); } + /// + /// Live run of a site script by name (design 2026-09-08 §4): resolves the script's effective + /// definition from the flattened instance configuration and branches on its trigger kind — + /// write kinds go through , every other kind is a traced + /// with one operator-attributed + /// root row. Calling directly + /// rather than through is deliberate: the helper emits its own + /// always-on RouteCall row with no operator actor, which would give the graph two roots. + /// + public async Task RunScriptAsync( + IServiceProvider sp, FlowDebugRunScriptCommand cmd, AuthenticatedUser user, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(sp); + ArgumentNullException.ThrowIfNull(cmd); + ArgumentNullException.ThrowIfNull(user); + if (string.IsNullOrWhiteSpace(cmd.ScriptName)) + throw new ManagementCommandException("ScriptName 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); + + // Design §2: the flattened configuration is the one place canonical names, effective triggers and + // effective parameter definitions all live (composition + inheritance + instance overrides applied). + var flattened = await sp.GetRequiredService() + .FlattenAndValidateAsync(instance.Id, ct, validateScripts: false).ConfigureAwait(false); + if (flattened.IsFailure) + throw new ManagementCommandException($"Could not resolve instance '{instance.UniqueName}': {flattened.Error}", NotFoundCode); + var script = flattened.Value.Configuration.Scripts + .FirstOrDefault(s => string.Equals(s.CanonicalName, cmd.ScriptName, StringComparison.OrdinalIgnoreCase)) + ?? throw new ManagementCommandException($"Script '{cmd.ScriptName}' is not defined on instance '{instance.UniqueName}'.", NotFoundCode); + + if (ScriptTriggerAttribute.IsWriteKind(script.TriggerType)) + { + if (!ScriptTriggerAttribute.TryRead(script.TriggerType, script.TriggerConfiguration, out var attribute)) + { + throw new ManagementCommandException( + $"Script '{script.CanonicalName}' has a {script.TriggerType} trigger with no readable attribute; fix its trigger configuration.", + ValidationFailedCode); + } + + if (cmd.Value is null) + throw new ManagementCommandException($"Value is required to fire '{script.CanonicalName}' (writes '{attribute}').", ValidationFailedCode); + return await WriteAndPollAsync( + sp, user, instance, site, attribute!, cmd.Value, cmd.WaitSeconds, script.CanonicalName, ct).ConfigureAwait(false); + } + + var resolveRef = await SchemaRefResolver.BuildAsync( + sp.GetService(), [script.ParameterDefinitions], ct).ConfigureAwait(false); + var validation = ParameterValidator.Validate(cmd.Parameters, script.ParameterDefinitions, resolveRef); + if (!validation.IsValid) + throw new ManagementCommandException(validation.ErrorMessage ?? "Parameter validation failed", ValidationFailedCode); + + var root = Guid.NewGuid(); + var startedAt = DateTimeOffset.UtcNow; + var sw = Stopwatch.StartNew(); + // Review follow-up: the root row is synthesized here, so the copy handed back must be + // redacted by this method — WriteCentralAsync's writer only redacts its own private copy. + var redactor = sp.GetService(); + var collector = new FlowTraceCollector(redactor: redactor); + var timeout = TimeSpan.FromSeconds(Math.Max(1, script.ExecutionTimeoutSeconds ?? DefaultScriptTimeoutSeconds)); + var router = sp.GetRequiredService(); + var request = new RouteToCallRequest( + Guid.NewGuid().ToString(), instance.UniqueName, script.CanonicalName, + validation.Parameters.Count == 0 ? null : validation.Parameters, + DateTimeOffset.UtcNow, ParentExecutionId: root, Trace: true); + + RouteToCallResponse response; + var timedOut = false; + using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(ct)) + { + runCts.CancelAfter(timeout + RunGrace); + try + { + response = await router.RouteToCallAsync(site.SiteIdentifier, request, runCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + timedOut = true; + response = new RouteToCallResponse(request.CorrelationId, false, null, + $"Flow Debugger run timed out after {timeout.TotalSeconds}s + {RunGrace.TotalSeconds}s grace", DateTimeOffset.UtcNow); + } + } + + foreach (var e in response.TraceEvents ?? []) + collector.Add(e); + var events = collector.Complete(); + + var rootRow = ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, AuditKind.RouteCall, + response.Success ? AuditStatus.Delivered : AuditStatus.Failed, + occurredAtUtc: startedAt.UtcDateTime, + actor: user.Username, + target: $"{instance.UniqueName}.{script.CanonicalName}", + sourceNode: sp.GetService()?.NodeName, + executionId: root, + sourceSiteId: site.SiteIdentifier, + sourceInstanceId: instance.UniqueName, + durationMs: (int)Math.Min(sw.ElapsedMilliseconds, int.MaxValue), + errorMessage: response.ErrorMessage, + requestSummary: cmd.Parameters is { } p ? p.GetRawText() : null, + responseSummary: response.ReturnValue is null ? null : JsonSerializer.Serialize(response.ReturnValue, ScriptTraceExtra.JsonOptions), + extra: new ScriptTraceExtra { Source = SourceTag, ChildExecutionId = response.ChildExecutionId }.ToJson()); + await WriteCentralAsync(sp, rootRow, ct).ConfigureAwait(false); + + // Arch-review F5: emission order is authoritative — the root row carries central's clock and + // the 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. + var ordered = new List(events.Count + 1) { redactor?.Apply(rootRow) ?? rootRow }; + ordered.AddRange(events); + return new FlowDebugRunResult( + root, response.Success, response.ErrorMessage, + response.Success ? null : timedOut ? TimeoutCode : CallFailedCode, + ordered, Truncated: collector.Truncated, startedAt, DateTimeOffset.UtcNow); + } + /// /// The traced attribute write + spawned-run poll shared by and /// the write-kind branch of the script entry's RunScriptAsync (design 2026-09-08 §4.3). The diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs index c7ce6c99..586c0f18 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/FlowDebugRunnerTests.cs @@ -10,9 +10,12 @@ 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; 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.Commons.Types.Flattening; +using ZB.MOM.WW.ScadaBridge.DeploymentManager; using ZB.MOM.WW.ScadaBridge.InboundAPI; using ZB.MOM.WW.ScadaBridge.ManagementService.FlowDebug; @@ -448,6 +451,166 @@ public class FlowDebugRunnerTests Assert.Equal(501, result.Events.Count); // inbound row + the 500 kept } + // --- Flow Debugger script entry (design 2026-09-08) --- + // + // The runner resolves the script from the FLATTENED configuration and branches on its + // server-side trigger kind: ValueChange/Conditional fire by WRITING the monitored attribute + // (the shared WriteAndPollAsync helper), every other kind is a direct traced routed call + // with ONE operator-attributed RouteCall root row. + + /// Wires instance 5 / site-a like plus a flattened configuration holding the given scripts. + private (ITemplateEngineRepository Templates, ISiteRepository Sites, IPullTraceEventsClient Puller, IFlatteningPipeline Pipeline) + WireScript(params ResolvedScript[] scripts) + { + var (templates, sites, puller) = WireValueChange(); + var pipeline = Substitute.For(); + var config = new FlattenedConfiguration { InstanceUniqueName = "Pump1", TemplateId = 10, SiteId = 2, Scripts = scripts }; + pipeline.FlattenAndValidateAsync(5, Arg.Any(), false, null) + .Returns(Result.Success(new FlatteningPipelineResult(config, "hash", new ValidationResult()))); + _services.AddSingleton(pipeline); + return (templates, sites, puller, pipeline); + } + + private static ResolvedScript CallScript(string name, string? parameterDefinitions = null, int? timeout = null) => + new() { CanonicalName = name, Code = "return 1;", TriggerType = "Call", ParameterDefinitions = parameterDefinitions, ExecutionTimeoutSeconds = timeout }; + + private static ResolvedScript WriteScript(string name, string kind, string attribute) => + new() { CanonicalName = name, Code = "return 1;", TriggerType = kind, TriggerConfiguration = $$"""{"attributeName":"{{attribute}}"}""" }; + + [Fact] + public async Task RunScript_WriteKind_WritesTheTriggerAttribute_ThenPolls() + { + var (_, _, puller, _) = WireScript(WriteScript("OnLevel", "Conditional", "Level")); + 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)); + puller.PullAsync("site-a", Arg.Any>(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(ci => (IReadOnlyList)[ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, + eventId: Guid.Parse("bbbbbbbb-0000-0000-0000-000000000001"), + executionId: Guid.NewGuid(), parentExecutionId: ((IReadOnlyCollection)ci[1]).First(), target: "OnLevel")]); + + var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "onlevel", null, "7", WaitSeconds: 10), Deployer, CancellationToken.None); + + Assert.True(result.Completed, result.Error); + Assert.Equal("7", write!.AttributeValues["Level"]); + Assert.True(write.Trace); + Assert.Equal(AuditKind.AttributeWrite, result.Events[0].AsRow().Kind); + await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default); + } + + [Fact] + public async Task RunScript_WriteKind_WithoutValue_FailsValidation_BeforeAnyWrite() + { + WireScript(WriteScript("OnLevel", "ValueChange", "Level")); + var ex = await Assert.ThrowsAsync(() => Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "OnLevel", null, null), Deployer, CancellationToken.None)); + Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code); + await _router.DidNotReceiveWithAnyArgs().RouteToSetAttributesAsync(default!, default!, default); + } + + [Fact] + public async Task RunScript_CallKind_RoutesATracedCall_AndReturnsRootRouteCallRow_ThenSiteRows() + { + WireScript(CallScript("Manual", """{"type":"object","properties":{"n":{"type":"integer"}},"required":["n"]}""", timeout: 5)); + var child = Guid.NewGuid(); + var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, + executionId: child, target: "Manual"); + RouteToCallRequest? captured = null; + _router.RouteToCallAsync("site-a", Arg.Do(r => captured = r), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 42, null, DateTimeOffset.UtcNow, child, [siteRun])); + + var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "Manual", Json("{\"n\":3}"), null), Deployer, CancellationToken.None); + + Assert.True(result.Completed, result.Error); + Assert.Null(result.ErrorCode); + Assert.True(captured!.Trace); + Assert.Equal("Pump1", captured.InstanceUniqueName); + Assert.Equal("Manual", captured.ScriptName); + Assert.Equal(result.RootExecutionId, captured.ParentExecutionId); + Assert.Equal(3L, Convert.ToInt64(captured.Parameters!["n"])); + + Assert.Equal(2, result.Events.Count); + var root = result.Events[0].AsRow(); + Assert.Equal(AuditKind.RouteCall, root.Kind); + Assert.Equal(result.RootExecutionId, root.ExecutionId); + Assert.Equal("ops", result.Events[0].Actor); + Assert.Equal("Pump1.Manual", result.Events[0].Target); + Assert.Equal("site-a", root.SourceSiteId); + Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(root.Extra)!.Source); + Assert.Equal(child, ScriptTraceExtra.TryParse(root.Extra)!.ChildExecutionId); + Assert.Contains("\"n\":3", root.RequestSummary); + Assert.Contains("42", root.ResponseSummary); + Assert.Same(siteRun, result.Events[1]); + Assert.Single(_central); // the root row was written centrally + } + + [Fact] + public async Task RunScript_CallKind_InvalidParameters_FailsBeforeRouting() + { + WireScript(CallScript("Manual", """{"type":"object","properties":{"n":{"type":"integer"}},"required":["n"]}""")); + var ex = await Assert.ThrowsAsync(() => Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "Manual", Json("{}"), null), Deployer, CancellationToken.None)); + Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code); + await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default); + Assert.Empty(_central); + } + + [Fact] + public async Task RunScript_CallKind_SiteFailure_ReportsCallFailed_WithTheRootRow() + { + WireScript(CallScript("Manual")); + _router.RouteToCallAsync("site-a", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, false, null, "Script 'Manual' not found on instance 'Pump1'.", DateTimeOffset.UtcNow)); + + var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "Manual", null, null), Deployer, CancellationToken.None); + + Assert.False(result.Completed); + Assert.Equal(FlowDebugRunner.CallFailedCode, result.ErrorCode); + Assert.Contains("not found", result.Error); + Assert.Single(result.Events); + Assert.Equal(AuditStatus.Failed, result.Events[0].AsRow().Status); + } + + [Fact] + public async Task RunScript_UnknownScript_IsNotFound() + { + WireScript(CallScript("Manual")); + var ex = await Assert.ThrowsAsync(() => Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "Nope", null, null), Deployer, CancellationToken.None)); + Assert.Equal(FlowDebugRunner.NotFoundCode, ex.Code); + Assert.Contains("Nope", ex.Message); + } + + [Fact] + public async Task RunScript_OutOfScopeSite_IsRefused_BeforeFlattening() + { + var (_, _, _, pipeline) = WireScript(CallScript("Manual")); + var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "9" }); + await Assert.ThrowsAsync(() => Runner().RunScriptAsync( + _services.BuildServiceProvider(), new FlowDebugRunScriptCommand(5, "Manual", null, null), scoped, CancellationToken.None)); + await pipeline.DidNotReceiveWithAnyArgs().FlattenAndValidateAsync(default, default, default, default); + } + + [Fact] + public async Task RunScript_CallKind_RootRow_IsRedacted_WhenARedactorIsRegistered() + { + WireScript(CallScript("Manual")); + _router.RouteToCallAsync("site-a", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 1, null, DateTimeOffset.UtcNow)); + _services.AddSingleton(new UpperRedactor()); + + var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(), + new FlowDebugRunScriptCommand(5, "Manual", null, null), Deployer, CancellationToken.None); + + var root = Assert.Single(result.Events); + var written = Assert.Single(_central); + Assert.Equal(written.DetailsJson!.ToUpperInvariant(), root.DetailsJson); + Assert.NotEqual(written.DetailsJson, root.DetailsJson); + } + /// Upper-cases DetailsJson so a test can see the filtered copy land. private sealed class UpperRedactor : IAuditRedactor {