Merge branch 'worktree-agent-a33b0136b18715bde' into feat/flow-debugger-script-entry

This commit is contained in:
Joseph Doherty
2026-09-08 01:16:34 -04:00
2 changed files with 299 additions and 1 deletions
@@ -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
/// <summary>Error code when the method / instance / site does not exist.</summary>
public const string NotFoundCode = "NOT_FOUND";
/// <summary>Error code when a direct script call (design 2026-09-08 §4.4) fails at the site.</summary>
public const string CallFailedCode = "CALL_FAILED";
/// <summary>Error code when a Live run exceeds the method's / script's timeout plus <see cref="RunGrace"/>.</summary>
public const string TimeoutCode = "TIMEOUT";
/// <summary>
/// Timeout for a direct script call whose script declares none — mirrors
/// <c>SiteRuntimeOptions.ScriptExecutionTimeoutSeconds</c>'s default (the runner has no site options).
/// </summary>
public const int DefaultScriptTimeoutSeconds = 30;
/// <summary>
/// Arch-review F2: <see cref="InboundScriptExecutor"/>'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);
}
/// <summary>
/// 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 <see cref="WriteAndPollAsync"/>, every other kind is a traced
/// <see cref="IInstanceRouter.RouteToCallAsync"/> with one operator-attributed
/// <see cref="AuditKind.RouteCall"/> root row. Calling <see cref="IInstanceRouter"/> directly
/// rather than through <see cref="RouteHelper"/> is deliberate: the helper emits its own
/// always-on RouteCall row with no operator actor, which would give the graph two roots.
/// </summary>
public async Task<FlowDebugRunResult> 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<ITemplateEngineRepository>()
.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<ISiteRepository>()
.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<IFlatteningPipeline>()
.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<ISharedSchemaRepository>(), [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<IAuditRedactor>();
var collector = new FlowTraceCollector(redactor: redactor);
var timeout = TimeSpan.FromSeconds(Math.Max(1, script.ExecutionTimeoutSeconds ?? DefaultScriptTimeoutSeconds));
var router = sp.GetRequiredService<IInstanceRouter>();
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<INodeIdentityProvider>()?.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<AuditEvent>(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);
}
/// <summary>
/// The traced attribute write + spawned-run poll shared by <see cref="RunValueChangeAsync"/> and
/// the write-kind branch of the script entry's <c>RunScriptAsync</c> (design 2026-09-08 §4.3). The
@@ -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.
/// <summary>Wires instance 5 / site-a like <see cref="WireValueChange"/> plus a flattened configuration holding the given scripts.</summary>
private (ITemplateEngineRepository Templates, ISiteRepository Sites, IPullTraceEventsClient Puller, IFlatteningPipeline Pipeline)
WireScript(params ResolvedScript[] scripts)
{
var (templates, sites, puller) = WireValueChange();
var pipeline = Substitute.For<IFlatteningPipeline>();
var config = new FlattenedConfiguration { InstanceUniqueName = "Pump1", TemplateId = 10, SiteId = 2, Scripts = scripts };
pipeline.FlattenAndValidateAsync(5, Arg.Any<CancellationToken>(), false, null)
.Returns(Result<FlatteningPipelineResult>.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<RouteToSetAttributesRequest>(r => write = r), Arg.Any<CancellationToken>())
.Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
puller.PullAsync("site-a", Arg.Any<IReadOnlyCollection<Guid>>(), Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(ci => (IReadOnlyList<AuditEvent>)[ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
eventId: Guid.Parse("bbbbbbbb-0000-0000-0000-000000000001"),
executionId: Guid.NewGuid(), parentExecutionId: ((IReadOnlyCollection<Guid>)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<ManagementCommandException>(() => 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<RouteToCallRequest>(r => captured = r), Arg.Any<CancellationToken>())
.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<ManagementCommandException>(() => 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<RouteToCallRequest>(), Arg.Any<CancellationToken>())
.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<ManagementCommandException>(() => 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<SiteScopeViolationException>(() => 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<RouteToCallRequest>(), Arg.Any<CancellationToken>())
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 1, null, DateTimeOffset.UtcNow));
_services.AddSingleton<IAuditRedactor>(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);
}
/// <summary>Upper-cases <c>DetailsJson</c> so a test can see the filtered copy land.</summary>
private sealed class UpperRedactor : IAuditRedactor
{