fix: fix expression logic

improvement: flow -> mermaid chart
improvement: flow tenants
improvement: fix nested map statements in flow

.formatter.exs

@@ -137,6 +137,7 @@ locals_without_parens = [
   sum: 3,
   sum: 4,
   table: 1,
+  tenant: 1,
   through: 1,
   timeout: 1,
   timestamps: 0,
@@ -159,6 +160,7 @@ locals_without_parens = [
   validate: 2,
   validate_destination_field?: 1,
   violation_message: 1,
+  wait_for: 1,
   where: 1,
   writable?: 1
 ]

lib/ash/actions/create.ex

@@ -127,6 +127,18 @@ defmodule Ash.Actions.Create do
           Request.resolve(changeset_dependencies, fn %{actor: actor} = context ->
             input = changeset_input.(context) || %{}
 
+            tenant =
+              case tenant do
+                nil ->
+                  nil
+
+                tenant when is_function(tenant) ->
+                  tenant.(context)
+
+                tenant ->
+                  tenant
+              end
+
             changeset =
               case changeset do
                 nil ->
@@ -138,6 +150,18 @@ defmodule Ash.Actions.Create do
                   changeset(changeset, api, action, actor)
               end
 
+            tenant =
+              case tenant do
+                nil ->
+                  nil
+
+                tenant when is_function(tenant) ->
+                  tenant.(context)
+
+                tenant ->
+                  tenant
+              end
+
             changeset =
               if tenant do
                 Ash.Changeset.set_tenant(changeset, tenant)

lib/ash/actions/destroy.ex

@@ -110,6 +110,18 @@ defmodule Ash.Actions.Destroy do
           Request.resolve(changeset_dependencies, fn %{actor: actor} = context ->
             input = changeset_input.(context) || %{}
 
+            tenant =
+              case tenant do
+                nil ->
+                  nil
+
+                tenant when is_function(tenant) ->
+                  tenant.(context)
+
+                tenant ->
+                  tenant
+              end
+
             changeset =
               case changeset do
                 nil ->

lib/ash/actions/read.ex

@@ -153,6 +153,18 @@ defmodule Ash.Actions.Read do
 
               input = query_input.(context) || %{}
 
+              tenant =
+                case tenant do
+                  nil ->
+                    nil
+
+                  tenant when is_function(tenant) ->
+                    tenant.(context)
+
+                  tenant ->
+                    tenant
+                end
+
               query =
                 case query do
                   nil ->
@@ -163,7 +175,7 @@ defmodule Ash.Actions.Read do
                       query,
                       action,
                       actor,
-                      query.tenant,
+                      tenant || query.tenant,
                       authorize?
                     )
                 end

lib/ash/actions/update.ex

@@ -150,6 +150,18 @@ defmodule Ash.Actions.Update do
           Request.resolve(changeset_dependencies, fn %{actor: actor} = context ->
             input = changeset_input.(context) || %{}
 
+            tenant =
+              case tenant do
+                nil ->
+                  nil
+
+                tenant when is_function(tenant) ->
+                  tenant.(context)
+
+                tenant ->
+                  tenant
+              end
+
             changeset =
               case changeset do
                 nil ->

lib/ash/changeset/changeset.ex

@@ -537,7 +537,7 @@ defmodule Ash.Changeset do
           changeset
           |> handle_errors(action.error_handler)
           |> set_actor(opts)
-          |> set_tenant(opts[:tenant] || changeset.tenant)
+          |> set_tenant(opts[:tenant] || changeset.tenant || changeset.data.__metadata__[:tenant])
           |> Map.put(:action, action)
           |> Map.put(:__validated_for_action__, action.name)
           |> cast_params(action, params || %{}, opts)

lib/ash/engine/request.ex

@@ -470,7 +470,13 @@ defmodule Ash.Engine.Request do
     %{request | dependency_data: Map.put(request.dependency_data, dep, value)}
   end
 
-  def store_dependency(request, receiver_path, field, internal? \\ false) do
+  def store_dependency(error, receiver_path, field, internal? \\ false)
+
+  def store_dependency(%{state: :error} = request, _receiver_path, _field, _internal?) do
+    {:ok, request, []}
+  end
+
+  def store_dependency(request, receiver_path, field, internal?) do
     request = do_store_dependency(request, field, receiver_path)
 
     case try_resolve_local(request, field, internal?) do

lib/ash/engine/runner.ex

@@ -277,10 +277,21 @@ defmodule Ash.Engine.Runner do
       else
         GenServer.cast(state.engine_pid, {:spawn_requests, async})
         runner_ref = state.ref
+        engine_pid = state.engine_pid
+        complete? = Enum.all?(state.requests, &(&1.state in [:complete, :error]))
 
         receive do
           {:pid_info, pid_info, ^runner_ref} ->
             %{state | pid_info: pid_info}
+
+          {:DOWN, _, _, ^engine_pid,
+           {:shutdown, %{errored_requests: [], runner_ref: ^runner_ref} = engine_state}} ->
+            log(state, fn -> "Engine complete" end)
+            handle_completion(state, engine_state, complete?, false)
+
+          {:DOWN, _, _, ^engine_pid, {:shutdown, %{runner_ref: ^runner_ref} = engine_state}} ->
+            log(state, fn -> "Engine complete" end)
+            handle_completion(state, engine_state, complete?, true)
         end
       end
 

lib/ash/error/unknown.ex

@@ -13,7 +13,13 @@ defmodule Ash.Error.Unknown do
       message(%{error | errors: [], error: "Something went wrong"})
     end
 
-    def message(%{errors: errors, error: error, path: path, stacktraces?: stacktraces?}) do
+    def message(%{
+          errors: errors,
+          error: error,
+          path: path,
+          stacktraces?: stacktraces?,
+          stacktrace: stacktrace
+        }) do
       errors = List.wrap(errors)
 
       custom_prefix =
@@ -45,7 +51,13 @@ defmodule Ash.Error.Unknown do
           end
         end)
 
+      if stacktrace && stacktrace.stacktrace do
+        Ash.Error.error_messages(errors, custom_message, stacktraces?) <>
+          "\n" <>
+          Exception.format_stacktrace(stacktrace.stacktrace)
+      else
         Ash.Error.error_messages(errors, custom_message, stacktraces?)
       end
     end
   end
+end

lib/ash/flow/chart/mermaid.ex

@@ -0,0 +1,353 @@
+defmodule Ash.Flow.Chart.Mermaid do
+  @moduledoc "Tools to render an Ash.Flow as a mermaid chart."
+  @opts [
+    expand: [
+      type: :boolean,
+      default: true,
+      doc: """
+      If the flow should be fully expanded (all `run_flow` steps will be inlined)
+      """
+    ]
+  ]
+
+  def chart(flow, opts \\ []) do
+    opts = Ash.OptionsHelpers.validate!(opts, @opts)
+
+    # This is a hack that may not work forever
+    # Eventually, we may need a separate mode/option for `build`
+    # that doesn't attempt to substitute arguments into templates
+    args =
+      flow
+      |> Ash.Flow.Info.arguments()
+      |> Map.new(fn argument ->
+        {argument.name, {:_arg, argument.name}}
+      end)
+
+    steps =
+      if opts[:expand] do
+        {:ok, %{steps: steps}} = Ash.Flow.Executor.AshEngine.build(flow, args, [])
+        unwrap(steps)
+      else
+        Ash.Flow.Info.steps(flow)
+      end
+
+    arguments = flow |> Ash.Flow.Info.arguments()
+
+    init = """
+    flowchart TB
+    classDef hidden visibility:hidden
+    """
+
+    init
+    |> add_arguments(arguments)
+    |> add_steps(steps, steps, opts)
+    |> add_links(steps, steps, opts)
+    |> IO.iodata_to_binary()
+  end
+
+  defp add_arguments(message, arguments) do
+    message =
+      message
+      |> add_line("subgraph Arguments")
+      |> add_line("direction TB")
+
+    Enum.reduce(arguments, message, fn argument, message ->
+      question_mark =
+        if argument.allow_nil? do
+          "?"
+        else
+          ""
+        end
+
+      add_line(
+        message,
+        "_arguments.#{argument.name}(\"#{argument.name}#{question_mark}: #{inspect(argument.type)}\")"
+      )
+    end)
+    |> add_line("end")
+  end
+
+  defp unwrap(steps) do
+    Enum.map(steps, fn %{step: step} ->
+      case step do
+        %{steps: steps} ->
+          %{step | steps: unwrap(steps)}
+
+        step ->
+          step
+      end
+    end)
+  end
+
+  defp add_steps(message, steps, all_steps, opts) do
+    Enum.reduce(steps, message, fn step, message ->
+      case step do
+        %Ash.Flow.Step.Map{steps: steps, over: over, output: output} = step ->
+          id = "#{format_name(step)}.element"
+
+          highlight =
+            if output do
+              do_format_name(List.wrap(output))
+            else
+              format_name(List.last(steps))
+            end
+
+          name = format_name(step)
+
+          message =
+            message
+            |> add_line("subgraph #{name} [Map]")
+            |> add_line("direction TB")
+            |> add_line("#{id}(\"Element: #{format_template(over, all_steps)}\")")
+            |> add_steps(steps, all_steps, opts)
+            |> highlight(highlight)
+            |> add_line("end")
+
+          message
+
+        %Ash.Flow.Step.Transaction{steps: steps} = step ->
+          name = format_name(step)
+
+          message
+          |> add_line("subgraph #{name}.subgraph [Transaction]")
+          |> add_line("direction TB")
+          |> add_steps(steps, all_steps, opts)
+          |> add_line("end")
+
+        %Ash.Flow.Step.RunFlow{flow: flow} = step ->
+          returns = Ash.Flow.Info.returns(flow)
+
+          if returns && opts[:expand] do
+            escaped_returns = escape(inspect(Ash.Flow.Info.returns(flow)))
+            name = format_name(step)
+
+            header =
+              if is_atom(returns) do
+                "Gather Value"
+              else
+                "Gather Values"
+              end
+
+            message
+            |> add_line("#{name}(\"#{header}: #{escaped_returns}\")")
+          else
+            message
+          end
+
+        step ->
+          add_line(message, "#{format_name(step)}(\"#{short_name(step)}\")")
+      end
+    end)
+  end
+
+  defp short_name(%Ash.Flow.Step.Custom{custom: {mod, opts}}) do
+    if function_exported?(mod, :short_name, 1) do
+      mod.short_name(opts)
+    else
+      escape(inspect({mod, opts}))
+    end
+  end
+
+  defp short_name(%Ash.Flow.Step.Map{steps: steps, output: output}) do
+    child_step =
+      if output do
+        find_step(steps, output)
+      else
+        List.last(steps)
+      end
+
+    "Element of #{short_name(child_step)}"
+  end
+
+  defp short_name(%Ash.Flow.Step.RunFlow{flow: flow}) do
+    "Run Flow: #{inspect(flow)}"
+  end
+
+  defp short_name(%Ash.Flow.Step.Create{action: action, resource: resource}) do
+    "Create: #{inspect(resource)}.#{action}"
+  end
+
+  defp short_name(%Ash.Flow.Step.Update{action: action, resource: resource}) do
+    "Update: #{inspect(resource)}.#{action}"
+  end
+
+  defp short_name(%Ash.Flow.Step.Destroy{action: action, resource: resource}) do
+    "Destroy: #{inspect(resource)}.#{action}"
+  end
+
+  defp short_name(%Ash.Flow.Step.Read{action: action, resource: resource}) do
+    "Read: #{inspect(resource)}.#{action}"
+  end
+
+  defp highlight(message, id) do
+    add_line(message, "style #{id} fill:#4287f5,stroke:#333,stroke-width:4px")
+  end
+
+  defp add_links(message, steps, all_steps, opts) do
+    Enum.reduce(steps, message, fn step, message ->
+      case step do
+        %Ash.Flow.Step.Map{steps: steps, over: over} = step ->
+          id = "#{format_name(step)}.element"
+
+          message
+          |> add_dependencies(step, all_steps)
+          |> add_deps(over, id, all_steps)
+          |> add_links(steps, all_steps, opts)
+
+        %Ash.Flow.Step.Transaction{steps: steps} = step ->
+          message
+          |> add_dependencies(step, all_steps)
+          |> add_links(steps, all_steps, opts)
+
+        %Ash.Flow.Step.RunFlow{flow: flow} ->
+          returns = Ash.Flow.Info.returns(flow)
+          name = format_name(step)
+
+          message =
+            Enum.reduce(List.wrap(returns), message, fn
+              {key, _}, message ->
+                {source, note} = link_source(all_steps, List.wrap(step.name) ++ List.wrap(key))
+
+                message
+                |> add_link(source, note, name)
+
+              value, message ->
+                {source, note} = link_source(all_steps, List.wrap(step.name) ++ List.wrap(value))
+
+                message
+                |> add_link(source, note, name)
+            end)
+
+          if opts[:expand] do
+            message
+          else
+            message
+            |> add_dependencies(step, all_steps)
+            |> add_links(steps, all_steps, opts)
+          end
+
+        step ->
+          add_dependencies(message, step, all_steps)
+      end
+    end)
+  end
+
+  defp add_link(message, source, nil, name) do
+    add_line(message, "#{source}-->#{name}")
+  end
+
+  defp add_link(message, source, note, name) do
+    add_line(message, "#{source}-->|#{note}|#{name}")
+  end
+
+  defp format_template(template, all_steps) do
+    do_format_template(template, all_steps)
+  end
+
+  defp do_format_template(template, all_steps) when is_map(template) do
+    "%{#{Enum.map_join(template, ", ", fn {key, value} -> "#{do_format_template(key, all_steps)}: #{do_format_template(value, all_steps)}" end)}}"
+  end
+
+  defp do_format_template(template, all_steps) when is_list(template) do
+    "[#{Enum.map_join(template, ", ", &do_format_template(&1, all_steps))}]"
+  end
+
+  defp do_format_template({:_path, value, path}, all_steps) do
+    "get_in(#{do_format_template(value, all_steps)}, #{Enum.map_join(path, ", ", &do_format_template(&1, all_steps))})"
+  end
+
+  defp do_format_template({:_result, step_name}, all_steps) do
+    "result(#{short_name(find_step(all_steps, step_name))})"
+  end
+
+  defp do_format_template({:_element, step_name}, all_steps) do
+    "element(#{short_name(find_step(all_steps, step_name)).name})"
+  end
+
+  defp do_format_template(value, all_steps) when is_tuple(value) do
+    "#{Enum.map_join(value, ", ", &do_format_template(&1, all_steps))}"
+  end
+
+  defp do_format_template(value, _), do: inspect(value)
+
+  def find_step(steps, name) when is_list(steps), do: Enum.find_value(steps, &find_step(&1, name))
+  def find_step(%{name: name} = step, name), do: step
+  def find_step(%{steps: steps}, name), do: find_step(steps, name)
+  def find_step(_, _), do: nil
+
+  defp escape(string) do
+    String.replace(string, "\"", "'")
+  end
+
+  defp add_dependencies(message, step, all_steps) do
+    Enum.reduce(Ash.Flow.Executor.AshEngine.deps_keys(), message, fn key, message ->
+      case Map.fetch(step, key) do
+        {:ok, value} ->
+          add_deps(message, value, format_name(step), all_steps)
+
+        :error ->
+          message
+      end
+    end)
+  end
+
+  defp add_deps(message, template, destination, all_steps) do
+    result_refs = Ash.Flow.result_refs(template)
+    arg_refs = Ash.Flow.arg_refs(template)
+    element_refs = Ash.Flow.element_refs(template)
+
+    message =
+      Enum.reduce(element_refs, message, fn element, message ->
+        add_line(message, "#{do_format_name(element)}.element --> #{destination}")
+      end)
+
+    message =
+      Enum.reduce(arg_refs, message, fn arg, message ->
+        add_line(message, "_arguments.#{arg} -.-> #{destination}")
+      end)
+
+    Enum.reduce(result_refs, message, fn dep, message ->
+      {source, note} = link_source(all_steps, dep)
+
+      add_link(message, source, note, destination)
+    end)
+  end
+
+  defp link_source(all_steps, dep, note \\ nil) do
+    case find_step(all_steps, dep) do
+      %Ash.Flow.Step.Map{steps: steps, output: output} = step ->
+        output_step =
+          if output do
+            find_step(steps, List.wrap(output))
+          else
+            List.last(steps)
+          end
+
+        case output_step do
+          nil ->
+            {format_name(step), note}
+
+          output_step ->
+            link_source(all_steps, output_step.name, "list")
+        end
+
+      step ->
+        {format_name(step), note}
+    end
+  end
+
+  defp add_line(message, line) do
+    [message, "\n", line]
+  end
+
+  defp format_name(step) do
+    do_format_name(step.name)
+  end
+
+  defp do_format_name(name) do
+    name
+    |> List.wrap()
+    |> List.flatten()
+    |> Enum.join(".")
+  end
+end

lib/ash/flow/dsl.ex

@@ -228,8 +228,7 @@ defmodule Ash.Flow.Dsl do
     ],
     examples: [
       """
-      map :create_users do
-        over range(1, arg(:count))
+      map :create_users, range(1, arg(:count)) do
         output :create_user
 
         create :create_user, Org, :create do

lib/ash/flow/flow.ex

@@ -20,6 +20,9 @@ defmodule Ash.Flow do
   @spec run!(any, any, nil | maybe_improper_list | map) :: any
   def run!(flow, input, opts \\ []) do
     case run(flow, input, opts) do
+      {:ok, result, metadata} ->
+        {result, metadata}
+
       {:ok, result} ->
         result
 
@@ -223,6 +226,14 @@ defmodule Ash.Flow do
     {:_result, [prefix | List.wrap(step)]}
   end
 
+  defp do_remap_result_references({:_element, step}, prefix) when is_function(prefix) do
+    {:_element, prefix.(step)}
+  end
+
+  defp do_remap_result_references({:_element, step}, prefix) do
+    {:_element, [prefix | List.wrap(step)]}
+  end
+
   defp do_remap_result_references(action_input, input) when is_tuple(action_input) do
     List.to_tuple(do_remap_result_references(Tuple.to_list(action_input), input))
   end
@@ -268,6 +279,66 @@ defmodule Ash.Flow do
 
   defp do_set_dependent_values(other, _), do: other
 
+  def arg_refs(input) when is_map(input) do
+    Enum.flat_map(input, &arg_refs/1)
+  end
+
+  def arg_refs(input) when is_list(input) do
+    Enum.flat_map(input, &arg_refs/1)
+  end
+
+  def arg_refs({:_arg, name}) do
+    [name]
+  end
+
+  def arg_refs(input) when is_tuple(input) do
+    input
+    |> Tuple.to_list()
+    |> Enum.flat_map(&arg_refs/1)
+  end
+
+  def arg_refs(_), do: []
+
+  def element_refs(input) when is_map(input) do
+    Enum.flat_map(input, &element_refs/1)
+  end
+
+  def element_refs(input) when is_list(input) do
+    Enum.flat_map(input, &element_refs/1)
+  end
+
+  def element_refs({:_element, name}) do
+    [name]
+  end
+
+  def element_refs(input) when is_tuple(input) do
+    input
+    |> Tuple.to_list()
+    |> Enum.flat_map(&element_refs/1)
+  end
+
+  def element_refs(_), do: []
+
+  def result_refs(input) when is_map(input) do
+    Enum.flat_map(input, &result_refs/1)
+  end
+
+  def result_refs(input) when is_list(input) do
+    Enum.flat_map(input, &result_refs/1)
+  end
+
+  def result_refs({:_result, name}) do
+    [name]
+  end
+
+  def result_refs(input) when is_tuple(input) do
+    input
+    |> Tuple.to_list()
+    |> Enum.flat_map(&result_refs/1)
+  end
+
+  def result_refs(_), do: []
+
   def handle_input_template(action_input, input) do
     {val, deps} = do_handle_input_template(action_input, input)
     {val, Enum.uniq(deps)}

lib/ash/flow/step/builtin_step.ex

@@ -2,7 +2,7 @@ defmodule Ash.Flow.Step.BuiltinStep do
   @moduledoc false
   defmacro __using__(fields) do
     quote do
-      defstruct [:name, touches_resources: []] ++ unquote(fields)
+      defstruct [:name, :wait_for, touches_resources: []] ++ unquote(fields)
     end
   end
 end

lib/ash/flow/step/create.ex

@@ -1,6 +1,6 @@
 defmodule Ash.Flow.Step.Create do
   @moduledoc "Runs a create action."
-  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input]
+  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :tenant, :input]
   @shared_opts Ash.Flow.Step.shared_opts()
   @shared_action_opts Ash.Flow.Step.shared_action_opts()
 

lib/ash/flow/step/destroy.ex

@@ -1,6 +1,6 @@
 defmodule Ash.Flow.Step.Destroy do
   @moduledoc "A flow step to run a destroy action."
-  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input, :record]
+  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input, :tenant, :record]
   @shared_opts Ash.Flow.Step.shared_opts()
   @shared_action_opts Ash.Flow.Step.shared_action_opts()
 

lib/ash/flow/step/map.ex

@@ -12,10 +12,10 @@ defmodule Ash.Flow.Step.Map do
             "The value to be iterated over. Will be available inside the `map` step as `element(:map_step_name)`"
         ],
         output: [
-          type: :any,
-          doc:
-            "Which step or steps to use when constructing the output list. Defaults to the last step."
+          type: :atom,
+          doc: "Which step to use when constructing the output list. Defaults to the last step."
         ]
       ]
       |> Ash.OptionsHelpers.merge_schemas(@shared_opts, "Global Options")
+      |> Keyword.delete(:wait_for)
 end

lib/ash/flow/step/read.ex

@@ -1,6 +1,6 @@
 defmodule Ash.Flow.Step.Read do
   @moduledoc "Runs a read action"
-  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input, get?: false]
+  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input, :tenant, get?: false]
   @shared_opts Ash.Flow.Step.shared_opts()
   @shared_action_opts Ash.Flow.Step.shared_action_opts()
 

lib/ash/flow/step/step.ex

@@ -4,7 +4,13 @@ defmodule Ash.Flow.Step do
   """
 
   @callback run(input :: map | nil, opts :: Keyword.t(), context :: map) ::
-              {:ok, term} | {:error, term}
+              {:ok, term}
+              | {:ok, term, %{optional(:notifications) => list(Ash.Notifier.Notification.t())}}
+              | {:error, term}
+
+  @callback describe(opts :: Keyword.t()) :: String.t()
+  @callback short_name(opts :: Keyword.t()) :: String.t()
+  @optional_callbacks [describe: 1, short_name: 1]
 
   defmacro __using__(_) do
     quote do
@@ -20,6 +26,15 @@ defmodule Ash.Flow.Step do
         required: true,
         doc: "The name of the step. Will be used when expressing dependencies, and step inputs."
       ],
+      wait_for: [
+        type: :any,
+        doc: """
+        Ensures that the step happens after the configured step or steps.
+
+        This value is just a template that isn't used, except to determine dependencies, so you can
+        use it like this `wait_for [result(:step_one), result(:step_two)]` or `wait_for result(:step)`.
+        """
+      ],
       touches_resources: [
         type: {:list, :atom},
         doc: """
@@ -47,6 +62,10 @@ defmodule Ash.Flow.Step do
         doc:
           "The api to use when calling the action. Defaults to the api set in the `flow` section."
       ],
+      tenant: [
+        type: :any,
+        doc: "A tenant to use for the operation. May be a template or a literal value."
+      ],
       input: input()
     ]
   end

lib/ash/flow/step/transaction.ex

@@ -13,4 +13,5 @@ defmodule Ash.Flow.Step.Transaction do
         ]
       ]
       |> Ash.OptionsHelpers.merge_schemas(@shared_opts, "Global Options")
+      |> Keyword.delete(:wait_for)
 end

lib/ash/flow/step/update.ex

@@ -1,6 +1,6 @@
 defmodule Ash.Flow.Step.Update do
   @moduledoc "Runs an update action"
-  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input, :record]
+  use Ash.Flow.Step.BuiltinStep, [:resource, :action, :api, :input, :tenant, :record]
   @shared_opts Ash.Flow.Step.shared_opts()
   @shared_action_opts Ash.Flow.Step.shared_action_opts()
 

lib/ash/flow/transformers/validate_unique_names.ex

@@ -7,6 +7,7 @@ defmodule Ash.Flow.Transformers.ValidateUniqueNames do
   def transform(flow, dsl_state) do
     flow
     |> Ash.Flow.Info.steps()
+    |> unnest()
     |> Enum.map(& &1.name)
     |> Enum.group_by(& &1)
     |> Enum.find_value({:ok, dsl_state}, fn
@@ -23,4 +24,14 @@ defmodule Ash.Flow.Transformers.ValidateUniqueNames do
          )}
     end)
   end
+
+  defp unnest(steps) do
+    Enum.flat_map(steps, fn
+      %{steps: steps} = step ->
+        [step | unnest(steps)]
+
+      step ->
+        [step]
+    end)
+  end
 end

lib/ash/query/operator/basic.ex

@@ -46,8 +46,8 @@ defmodule Ash.Query.Operator.Basic do
           end
         end
 
-        defp do_evaluate(_, nil, _), do: :unknown
-        defp do_evaluate(_, _, nil), do: :unknown
+        defp do_evaluate(_, nil, _), do: {:ok, nil}
+        defp do_evaluate(_, _, nil), do: {:ok, nil}
 
         defp do_evaluate(:<>, left, right) do
           {:known, to_string(left) <> to_string(right)}

lib/ash/query/operator/eq.ex

@@ -13,8 +13,8 @@ defmodule Ash.Query.Operator.Eq do
     predicate?: true,
     types: [:same, :any]
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}) do
     {:known, Comp.equal?(left, right)}

lib/ash/query/operator/greater_than.ex

@@ -10,8 +10,8 @@ defmodule Ash.Query.Operator.GreaterThan do
     predicate?: true,
     types: [:same, :any]
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}) do
     {:known, Comp.greater_than?(left, right)}

lib/ash/query/operator/greater_than_or_equal.ex

@@ -10,8 +10,8 @@ defmodule Ash.Query.Operator.GreaterThanOrEqual do
     predicate?: true,
     types: [:same, :any]
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}),
     do: {:known, Comp.greater_or_equal?(left, right)}

lib/ash/query/operator/in.ex

@@ -21,8 +21,8 @@ defmodule Ash.Query.Operator.In do
 
   def new(left, right), do: {:ok, %__MODULE__{left: left, right: right}}
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}) do
     {:known, Enum.any?(right, &Comp.equal?(&1, left))}

lib/ash/query/operator/is_nil.ex

@@ -30,7 +30,7 @@ defmodule Ash.Query.Operator.IsNil do
     end
   end
 
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: is_nil?}) do
     {:known, is_nil(left) == is_nil?}

lib/ash/query/operator/less_than.ex

@@ -21,8 +21,8 @@ defmodule Ash.Query.Operator.LessThan do
 
   alias Ash.Query.Operator.{Eq, IsNil}
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}) do
     {:known, Comp.less_than?(left, right)}

lib/ash/query/operator/less_than_or_equal.ex

@@ -10,8 +10,8 @@ defmodule Ash.Query.Operator.LessThanOrEqual do
     predicate?: true,
     types: [:same, :any]
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}) do
     {:known, Comp.less_or_equal?(left, right)}

lib/ash/query/operator/not_eq.ex

@@ -13,8 +13,8 @@ defmodule Ash.Query.Operator.NotEq do
   alias Ash.Query.Not
   alias Ash.Query.Operator.Eq
 
-  def evaluate(%{left: nil}), do: :unknown
-  def evaluate(%{right: nil}), do: :unknown
+  def evaluate(%{left: nil}), do: {:ok, nil}
+  def evaluate(%{right: nil}), do: {:ok, nil}
 
   def evaluate(%{left: left, right: right}) do
     {:known, Comp.not_equal?(left, right)}

lib/ash/type/type.ex

@@ -526,7 +526,7 @@ defmodule Ash.Type do
         other -> other
       end
     else
-      type.apply_constraints(term, [])
+      {:ok, term}
     end
   end