w4c-workflows-api/w4c-workflows-api.Tests/NodeGraphRunnerTests.cs

using System.Text.Json.Nodes;
using w4c_workflows.Models.Nodes;
using w4c_workflows.Services.Nodes;
using w4c_workflows.Services.Nodes.Executors;
using w4c_workflows.Services.Nodes.Interpolation;
using Xunit;

namespace w4c_workflows.Tests;

public class NodeGraphRunnerTests
{
    // ------------------------------------------------------------------ fixtures

    private sealed class FailingNodeExecutor : INodeExecutor
    {
        public string Type => "test.fail";

        public Task<NodeExecutionOutcome> RunAsync(NodeExecutionContext context, CancellationToken ct)
            => Task.FromResult(NodeExecutionOutcome.Failed("boom", "boom_code", "exploded"));
    }

    private static NodeGraphRunner Runner(params INodeExecutor[] extra)
    {
        var executors = new List<INodeExecutor>
        {
            new NoOpNodeExecutor(),
            new SetNodeExecutor(),
            new IfNodeExecutor(),
        };
        executors.AddRange(extra);
        return new NodeGraphRunner(new NodeExecutorRegistry(executors), new NodeParameterInterpolator());
    }

    private static NodeGraph Compile(string yaml, NodeBlueprintCatalog? catalog = null)
    {
        var result = new NodeGraphCompiler(catalog ?? NodeTestData.CoreCatalog()).Compile(NodeTestData.Parse(yaml));
        Assert.True(result.Success, string.Join("\n", result.Errors));
        return result.Graph!;
    }

    private static FlowItem Item(string json) => FlowItem.FromJson(JsonNode.Parse(json)!.AsObject());

    private static NodeBlueprintCatalog FailCatalog(params NodePort[] outputs)
        => NodeTestData.Catalog(NodeTestData.NodeBlueprint("test.fail", outputs: outputs.ToList()));

    // ------------------------------------------------------------------ tests

    [Fact]
    public async Task Linear_graph_shapes_items_with_per_item_interpolation()
    {
        var yaml = """
            name: set-demo
            tasks:
              - id: pass
                node: { type: core.noop }
                next: shape
              - id: shape
                node: { type: core.set }
                parameters:
                  mode: manual
                  keepOnlySet: true
                  fields:
                    - name: greeting
                      value: "=Hello {{ $json.name }}"
            """;

        var result = await Runner().RunAsync(Compile(yaml), new[] { Item("""{"name":"Ada"}""") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Equal(new[] { "pass", "shape" }, result.ExecutionOrder);

        Assert.Single(result.OutputOf("pass"));
        var shaped = result.OutputOf("shape");
        Assert.Single(shaped);
        Assert.Equal("Hello Ada", shaped[0].Json["greeting"]!.GetValue<string>());
        Assert.Single(shaped[0].Json); // keepOnlySet dropped the input field
    }

    [Fact]
    public async Task If_routes_each_item_to_the_matching_branch()
    {
        var yaml = """
            name: branch
            tasks:
              - id: decide
                node: { type: core.if }
                parameters:
                  left: "={{ $json.status }}"
                  operator: equals
                  right: ok
                edges:
                  - { output: 0, to: yes }
                  - { output: 1, to: no }
              - id: yes
                node: { type: core.noop }
              - id: no
                node: { type: core.noop }
            """;

        var result = await Runner().RunAsync(
            Compile(yaml),
            new[] { Item("""{"status":"ok"}"""), Item("""{"status":"bad"}""") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Single(result.OutputOf("yes"));
        Assert.Single(result.OutputOf("no"));
        Assert.Equal("ok", result.OutputOf("yes")[0].Json["status"]!.GetValue<string>());
        Assert.Equal("bad", result.OutputOf("no")[0].Json["status"]!.GetValue<string>());
    }

    [Fact]
    public async Task Node_reference_reads_an_upstream_nodes_output()
    {
        var yaml = """
            name: ref
            tasks:
              - id: source
                node: { type: core.set }
                parameters:
                  mode: json
                  jsonOutput: { id: 7 }
                next: echo
              - id: echo
                node: { type: core.set }
                parameters:
                  mode: manual
                  keepOnlySet: true
                  fields:
                    - name: copied
                      value: '={{ $("source").id }}'
            """;

        var result = await Runner().RunAsync(Compile(yaml), Array.Empty<FlowItem>());

        Assert.True(result.Succeeded, result.Failure?.Message);
        var echoed = result.OutputOf("echo");
        Assert.Single(echoed);
        Assert.Equal(7, echoed[0].Json["copied"]!.GetValue<long>());
    }

    [Fact]
    public async Task Produced_items_carry_provenance()
    {
        var yaml = """
            name: prov
            tasks:
              - id: pass
                node: { type: core.noop }
                next: shape
              - id: shape
                node: { type: core.set }
                parameters:
                  mode: manual
                  fields:
                    - name: seen
                      value: "true"
            """;

        var result = await Runner().RunAsync(Compile(yaml), new[] { Item("""{"id":1}""") });

        var passOrigin = result.OutputOf("pass")[0].Origin;
        Assert.NotNull(passOrigin);
        Assert.Equal("pass", passOrigin!.NodeId);
        Assert.Equal(0, passOrigin.OutputIndex);

        // The set node passes the input item through, so it keeps the upstream origin.
        Assert.Equal("pass", result.OutputOf("shape")[0].Origin!.NodeId);
    }

    [Fact]
    public async Task A_failure_is_routed_to_the_error_output()
    {
        var yaml = """
            name: error-branch
            tasks:
              - id: work
                node: { type: test.fail }
                onError: recover
              - id: recover
                node: { type: core.noop }
            """;

        var catalog = FailCatalog(NodePort.Main, NodePort.Error);
        var result = await Runner(new FailingNodeExecutor())
            .RunAsync(Compile(yaml, catalog), new[] { Item("""{"x":1}""") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        var recovered = result.OutputOf("recover");
        Assert.Single(recovered);
        Assert.Equal(1, recovered[0].Json["x"]!.GetValue<int>());
        Assert.Equal("boom", recovered[0].Json["error"]!["message"]!.GetValue<string>());
        Assert.Equal("boom_code", recovered[0].Json["error"]!["code"]!.GetValue<string>());
    }

    [Fact]
    public async Task A_failure_without_an_error_output_fails_the_run()
    {
        var yaml = """
            name: hard-fail
            tasks:
              - id: work
                node: { type: test.fail }
            """;

        var result = await Runner(new FailingNodeExecutor())
            .RunAsync(Compile(yaml, FailCatalog(NodePort.Main)), new[] { Item("{}") });

        Assert.False(result.Succeeded);
        Assert.Equal("boom", result.Failure!.Message);
    }

    [Fact]
    public async Task A_failure_with_an_unconnected_error_output_fails_the_run()
    {
        // The blueprint declares an error port, but no branch is wired to it.
        // Merely having the port must not swallow the failure (P1-10 depends on
        // the failure reaching WorkflowRun.Error); only a real error branch does.
        var yaml = """
            name: dangling-error-port
            tasks:
              - id: work
                node: { type: test.fail }
            """;

        var result = await Runner(new FailingNodeExecutor())
            .RunAsync(Compile(yaml, FailCatalog(NodePort.Main, NodePort.Error)), new[] { Item("{}") });

        Assert.False(result.Succeeded);
        Assert.Equal("boom", result.Failure!.Message);
    }

    [Fact]
    public async Task ContinueOnFail_succeeds_without_an_error_output()
    {
        var yaml = """
            name: soft-fail
            tasks:
              - id: work
                node: { type: test.fail }
                continueOnFail: true
            """;

        var result = await Runner(new FailingNodeExecutor())
            .RunAsync(Compile(yaml, FailCatalog(NodePort.Main)), new[] { Item("{}") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Empty(result.OutputOf("work"));
    }

    [Fact]
    public async Task A_missing_executor_fails_the_run()
    {
        var catalog = NodeTestData.Catalog(NodeTestData.NodeBlueprint("test.noexec"));
        var yaml = """
            name: no-executor
            tasks:
              - id: work
                node: { type: test.noexec }
            """;

        var result = await Runner().RunAsync(Compile(yaml, catalog), new[] { Item("{}") });

        Assert.False(result.Succeeded);
        Assert.Equal("missing_executor", result.Failure!.Code);
    }

    [Fact]
    public async Task Fan_out_sends_the_same_items_to_every_target()
    {
        var yaml = """
            name: fan-out
            tasks:
              - id: source
                node: { type: core.noop }
                edges:
                  - { output: 0, to: left }
                  - { output: 0, to: right }
              - id: left
                node: { type: core.noop }
              - id: right
                node: { type: core.noop }
            """;

        var result = await Runner().RunAsync(Compile(yaml), new[] { Item("""{"n":1}""") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Single(result.OutputOf("left"));
        Assert.Single(result.OutputOf("right"));
    }

    [Fact]
    public async Task Filter_routes_matching_and_non_matching_items()
    {
        var yaml = """
            name: filter-demo
            tasks:
              - id: f
                node: { type: core.filter }
                parameters:
                  left: "={{ $json.n }}"
                  operator: greater
                  right: 1
                edges:
                  - { output: 0, to: kept }
                  - { output: 1, to: dropped }
              - id: kept
                node: { type: core.noop }
              - id: dropped
                node: { type: core.noop }
            """;

        var result = await Runner(new FilterNodeExecutor())
            .RunAsync(Compile(yaml), new[] { Item("""{"n":0}"""), Item("""{"n":2}""") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Equal(2, Assert.Single(result.OutputOf("kept")).Json["n"]!.GetValue<int>());
        Assert.Equal(0, Assert.Single(result.OutputOf("dropped")).Json["n"]!.GetValue<int>());
    }

    [Fact]
    public async Task Switch_routes_each_item_to_its_rule_output()
    {
        var yaml = """
            name: switch-demo
            tasks:
              - id: route
                node: { type: core.switch }
                parameters:
                  rules:
                    - { left: "={{ $json.kind }}", operator: equals, right: a }
                    - { left: "={{ $json.kind }}", operator: equals, right: b }
                  fallbackEnabled: true
                edges:
                  - { output: 0, to: a }
                  - { output: 1, to: b }
                  - { output: 4, to: other }
              - id: a
                node: { type: core.noop }
              - id: b
                node: { type: core.noop }
              - id: other
                node: { type: core.noop }
            """;

        var result = await Runner(new SwitchNodeExecutor()).RunAsync(
            Compile(yaml),
            new[] { Item("""{"kind":"a"}"""), Item("""{"kind":"b"}"""), Item("""{"kind":"c"}""") });

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Equal("a", Assert.Single(result.OutputOf("a")).Json["kind"]!.GetValue<string>());
        Assert.Equal("b", Assert.Single(result.OutputOf("b")).Json["kind"]!.GetValue<string>());
        Assert.Equal("c", Assert.Single(result.OutputOf("other")).Json["kind"]!.GetValue<string>());
    }

    [Fact]
    public async Task A_loop_node_iterates_its_body_then_emits_done()
    {
        var yaml = """
            name: loop-demo
            tasks:
              - id: source
                node: { type: core.noop }
                edges:
                  - { output: 0, to: loop }
              - id: loop
                node: { type: core.splitInBatches }
                parameters: { batchSize: 2 }
                edges:
                  - { output: 0, to: body }
                  - { output: 1, to: done }
              - id: body
                node: { type: core.noop }
                edges:
                  - { output: 0, to: loop }
              - id: done
                node: { type: core.noop }
            """;

        var seed = Enumerable.Range(1, 5).Select(i => Item($$"""{"n":{{i}}}""")).ToArray();
        var result = await Runner(new SplitInBatchesNodeExecutor()).RunAsync(Compile(yaml), seed);

        Assert.True(result.Succeeded, result.Failure?.Message);
        // 5 items in batches of 2 → 3 batch iterations + 1 final (done) invocation.
        Assert.Equal(4, result.ExecutionOrder.Count(id => id == "loop"));
        Assert.Equal(3, result.ExecutionOrder.Count(id => id == "body"));
        Assert.Equal(5, result.OutputOf("done").Count); // original items reach the exit
    }

    /// <summary>
    /// P1-3: an entry loop node (zero non-loop-back incoming) must be re-triggered
    /// by its own loop-back edge. Previously readiness compared `arrived ==
    /// incoming` and an entry loop node (incoming 0) silently ran a single batch.
    /// </summary>
    [Fact]
    public async Task An_entry_loop_node_iterates_each_batch()
    {
        var yaml = """
            name: entry-loop
            tasks:
              - id: loop
                node: { type: core.splitInBatches }
                parameters: { batchSize: 2 }
                edges:
                  - { output: 0, to: body }
                  - { output: 1, to: done }
              - id: body
                node: { type: core.noop }
                edges:
                  - { output: 0, to: loop }
              - id: done
                node: { type: core.noop }
            """;

        var seed = Enumerable.Range(1, 5).Select(i => Item($$"""{"n":{{i}}}""")).ToArray();
        var result = await Runner(new SplitInBatchesNodeExecutor()).RunAsync(Compile(yaml), seed);

        Assert.True(result.Succeeded, result.Failure?.Message);
        Assert.Equal(4, result.ExecutionOrder.Count(id => id == "loop"));
        Assert.Equal(3, result.ExecutionOrder.Count(id => id == "body"));
        Assert.Equal(5, result.OutputOf("done").Count);
    }

    /// <summary>Records the run iteration each invocation observed.</summary>
    private sealed class RecordingRunIndexExecutor : INodeExecutor
    {
        public List<int> RunIndices { get; } = new();
        public List<int> InterpolatedRunIndices { get; } = new();

        public string Type => "test.record";

        public Task<NodeExecutionOutcome> RunAsync(NodeExecutionContext context, CancellationToken ct)
        {
            RunIndices.Add(context.RunIndex);
            InterpolatedRunIndices.Add(
                context.Parameters.TryGetPropertyValue("seen", out var value) && value != null
                    ? value.GetValue<int>()
                    : -1);
            return Task.FromResult(NodeExecutionOutcome.Single(context.Input(0).ToList()));
        }
    }

    /// <summary>
    /// The run iteration index is real, not the hard-coded 0 it used to be: a node
    /// re-triggered by its loop-back edge sees an incrementing
    /// <see cref="NodeExecutionContext.RunIndex"/> and <c>$runIndex</c> resolves to it.
    /// </summary>
    [Fact]
    public async Task Loop_body_sees_an_incrementing_run_index()
    {
        var recording = new RecordingRunIndexExecutor();
        var loop = new NodeGraphNode
        {
            Id = "loop",
            Blueprint = NodeTestData.CoreCatalog().Get("core.splitInBatches")!,
            Parameters = new JsonObject { ["batchSize"] = 2 },
        };
        var body = new NodeGraphNode
        {
            Id = "body",
            Blueprint = NodeTestData.NodeBlueprint("test.record"),
            Parameters = new JsonObject { ["seen"] = "={{ $runIndex }}" },
        };
        var graph = new NodeGraph
        {
            Nodes = new[] { loop, body },
            Edges = new[]
            {
                new NodeGraphEdge { FromNodeId = "loop", FromOutput = 0, ToNodeId = "body", ToInput = 0 },
                new NodeGraphEdge { FromNodeId = "body", FromOutput = 0, ToNodeId = "loop", ToInput = 0, IsLoopBack = true },
            },
            EntryNodeId = "loop",
        };

        var seed = Enumerable.Range(1, 5).Select(i => Item($$"""{"n":{{i}}}""")).ToArray();
        var result = await Runner(new SplitInBatchesNodeExecutor(), recording).RunAsync(graph, seed);

        Assert.True(result.Succeeded, result.Failure?.Message);
        // 5 items in batches of 2 → the body runs on iterations 0, 1 and 2 (the
        // fourth loop invocation is the empty "done" pass that emits no batch).
        Assert.Equal(new[] { 0, 1, 2 }, recording.RunIndices);
        Assert.Equal(new[] { 0, 1, 2 }, recording.InterpolatedRunIndices);
    }
}