Oi (oi v0.2.0)

Copy Markdown

Oi means Orchid integration.

Core concepts

  • Workspace — pure data struct holding graph, cluster, interventions. No process, no GenServer. Compiled and dispatched via pure functions.

  • Compile + Dispatch — two-phase pipeline:

    1. Oi.compile/1 — topology → static bundles (reusable)
    2. Oi.execute/2 — bind interventions → plan → execute via pluggable executor

  • Executor — pluggable task execution strategy. Built-in: Sync (serial), TaskSup (Task.Supervisor), Pool (NimblePool)(in future).

  • Session — optional process tree per tenant, wrapping OrchidSymbiont.Runtime for multi-tenant isolation.

Quick start

Define a step

defmodule MyApp.Steps.Upcase do
  use Oi.Step, name: :upcase

  manifest(
    inputs: [:text],
    outputs: [result: :string]
  )

  routine text, _opts do
    text |> String.upcase() |> ok()
  end
end

Build and run

graph =
  Graph.new()
  |> Graph.add_node(%Node{
    id: :up,
    container: MyApp.Steps.Upcase,
    inputs: [:text],
    outputs: [:result]
  })
 
{:ok, compiled} = Oi.compile(graph)
 
{:ok, result} =
  Oi.execute(compiled,
    inputs: %{"up|text" => "hello"}
  )
 
{:ok, res} = Oi.Result.reify(result, "up|result")
# => "HELLO"

Multi-tenant with Session

Oi.Runtime.Session.start("tenant-1")
Oi.Runtime.Session.start("tenant-2")

ws = Oi.Workspace.new("tenant-1", graph)
{:ok, compiled} = Oi.compile(ws)
Oi.execute(compiled,
  executor: Oi.Executor.TaskSup,
  executor_opts: [sup: Oi.Runtime.Session.tasks_tuple("tenant-1")]
)

Symbiont step

defmodule MyApp.Steps.Predict do
  use Oi.Step, name: :predict, symbiont?: true

  manifest(
    inputs: [:features],
    outputs: [prediction: :string],
    models: [:model]
  )

  routine features, models, _opts do
    {:ok, result} = OrchidSymbiont.call(models.model, {:predict, features})
    ok(result)
  end
end

Summary

Types

name()

Functions

compile(graph, cluster \\ %Oi.Topology.Cluster{})

Compile graph into static bundles + plan.

execute(compiled, opts \\ [])

Execute compiled plan with inputs and interventions.

run(struct, opts \\ [])

Types

name()

@type name() :: String.t()

Functions

compile(graph, cluster \\ %Oi.Topology.Cluster{})

@spec compile(Oi.Topology.Graph.t(), Oi.Topology.Cluster.t()) ::
  {:ok, Oi.Compiled.t()} | {:error, :cycle_detected}

Compile graph into static bundles + plan.

Pure topology — no interventions. Reusable across different intervention sets and inputs.

execute(compiled, opts \\ [])

@spec execute(
  Oi.Compiled.t(),
  keyword()
) :: {:ok, Oi.Result.t()} | {:error, term()}

Execute compiled plan with inputs and interventions.

Options

  • :inputs — map of external io_key => payload, seeded into drafting memory
  • :interventions — map of {:port, node, port} => {type, payload}
  • :executor — Oi.Executor.Sync (default), Oi.Executor.TaskSup, or Oi.Executor.Pool
  • :executor_opts — passed to the executor (e.g. [sup: MyTaskSup])
  • :plugins — OrchidPlugin pipeline
  • :orchid_baggage — merged into Orchid run baggage

Examples

# Compile once
{:ok, compiled} = Oi.compile(graph, cluster)

# Execute with inputs + interventions
{:ok, result} = Oi.execute(compiled,
  inputs: %{"step1|in" => "foo"},
  interventions: %{{:port, :step1, :in} => {:override, "Bar"}}
)

# Same compiled plan, different inputs
{:ok, result_b} = Oi.execute(compiled, inputs: %{"step1|in" => "baz"})

# With Task.Supervisor
{:ok, result} = Oi.execute(compiled,
  executor: Oi.Executor.TaskSup,
  executor_opts: [sup: Oi.Runtime.Session.tasks_tuple("svs-1")]
)

run(struct, opts \\ [])