defmodule Oi do @doc_header """ Oi means Orchid integration. """ @external_resource readme = Path.join([__DIR__, "../README.md"]) @doc_main readme |> File.read!() |> String.split("") |> Enum.fetch!(1) @moduledoc @doc_header <> @doc_main @type name :: String.t() alias Oi.{Compile, Compiled, Result} alias Oi.Dispatch.{Config, Drafting, Orchestrator} alias Oi.Topology.{Graph, Cluster} alias Oi.Topology.Graph.{Edge, PortRef, Node} @doc """ Compile graph into static bundles + plan. Pure topology — no interventions. Reusable across different intervention sets and inputs. """ @spec compile(Graph.t(), Cluster.t()) :: {:ok, Compiled.t()} | {:error, :cycle_detected} def compile(graph, cluster \\ %Oi.Topology.Cluster{}) do with {:ok, bundles} <- Compile.Bundle.compile_graph(graph, cluster), {:ok, plan} <- Compile.Planning.build(bundles) do {:ok, %Compiled{bundles: bundles, plan: plan, edges: graph.edges}} else {:error, _} = err -> err end end @doc """ Execute compiled plan with inputs and interventions. ## Unified `:data` format (recommended) Oi.execute(compiled, data: %{ step1: %{in: "foo"}, step2: %{result: {:override, "bar"}} }) Supports tuple-key format as well: `%{{:step, :port} => value}`. Ports with incoming edges → intervention; ports without → memory. Values pass through as-is. Wrapped values ({:override, v}, {:offset, v}, {:custom, v}) are preserved for downstream intervention handling. ## Legacy `:inputs` / `:interventions` (still supported) Oi.execute(compiled, inputs: %{"step1|in" => "foo"}, interventions: %{{:port, :step1, :in} => {:override, "Bar"}} ) ## Other options * `:executor` — `Oi.Executor.Sync` (default), `Oi.Executor.TaskSup`, or `Oi.Executor.Pool` * `:executor_opts` — passed to the executor (e.g. `[sup: MyTaskSup]`) * `:orchid_adapters` — OrchidPlugin pipeline * `:orchid_baggage` — merged into Orchid run baggage """ @spec execute(Compiled.t(), keyword()) :: {:ok, Result.t()} | {:error, term()} def execute(%Compiled{} = compiled, opts \\ []) do {initial_memory_io, interventions_io} = build_drafting_inputs(compiled, opts) drafting = Drafting.new(initial_memory_io, interventions_io) conf = prepare_config(opts) case Orchestrator.dispatch(compiled.plan, drafting, conf) do {:ok, final_drafting} -> {:ok, Result.new(final_drafting.memory)} {:error, _} = err -> err end end def run(struct, opts \\ []) def run(%Compiled{} = compiled, opts) do execute(compiled, opts) end def run(%Graph{} = graph, opts) do cluster = Keyword.get(opts, :cluster, %Cluster{}) execute_opts = Keyword.delete(opts, :cluster) with {:ok, compiled} <- compile(graph, cluster) do execute(compiled, execute_opts) end end @typedoc """ Unified user-facing data for Oi.execute/2. Replaces the separate `:inputs` / `:interventions` opts. ## Format A — tuple keys %{{:step1, :in} => "foo", {:step2, :result} => {:override, "bar"}} ## Format B — nested %{step1: %{in: "foo"}, step2: %{result: {:override, "bar"}}} """ @type data :: %{ optional({Node.id(), Node.node_port()}) => term() } | %{ optional(Node.id()) => %{ optional(Node.node_port()) => term() } } @doc """ Splits unified `data` into `{memory, interventions}` by topology. For each `{node, port}`: * has incoming edge → intervention (data originates inside the graph) * no incoming edge → memory (external input, no upstream producer) Values pass through as-is — no wrapping, no io_key conversion. """ @spec resolve_data(data(), MapSet.t(Edge.t())) :: {%{{Node.id(), Node.node_port()} => term()}, %{{Node.id(), Node.node_port()} => term()}} def resolve_data(data, edges) when is_map(data) do flat = flatten_data(data) Enum.reduce(flat, {%{}, %{}}, fn {{node_id, port} = key, value}, {mem, intv} -> if has_upstream?(edges, node_id, port) do {mem, Map.put(intv, key, value)} else {Map.put(mem, key, value), intv} end end) end # -- helpers defp flatten_data(data) do if Enum.any?(data, fn {_k, v} -> is_map(v) end) do for {node, ports} <- data, {port, val} <- ports, into: %{}, do: {{node, port}, val} else data end end defp has_upstream?(edges, node_id, port) do Enum.any?(edges, fn e -> e.to_node == node_id and e.to_port == port end) end defp build_drafting_inputs(%Compiled{edges: edges}, opts) do data = Keyword.get(opts, :data, %{}) {memory_raw, interventions_raw} = resolve_data(data, edges) memory_io = Map.new(memory_raw, fn {{n, p}, v} -> {PortRef.to_orchid_key({:port, n, p}), wrap_orchid_param(PortRef.to_orchid_key({:port, n, p}), v)} end) interventions_io = Map.new(interventions_raw, fn {{n, p}, v} -> {PortRef.to_orchid_key({:port, n, p}), wrap_orchid_param(PortRef.to_orchid_key({:port, n, p}), v)} end) {memory_io, interventions_io} end # TODO # Add automatic oi_name => Orchid.Symbiont scope_id inject. defp prepare_config(opts) do baggage = opts |> Keyword.get(:orchid_baggage, []) |> Enum.into(%{}) Config.new( opts |> Keyword.drop([:interventions, :orchid_baggage, :data]) |> Keyword.put(:orchid_baggage, baggage) ) end defp wrap_orchid_param(name, %Orchid.Param{} = p), do: %{p | name: name} defp wrap_orchid_param(name, val), do: Orchid.Param.new(name, :any, val) end