Cairn

Small OTP helpers for message delivery and supervised task callbacks.

Cairn tries not to become a framework for OTP. It is a thin layer for starting lightweight processes, passing messages, running supervised work, and waiting on refs.

Try it

cd cairn
mix deps.get
iex -S mix

Paste this into IEx:

{:ok, pid} = Cairn.Function.start_link(fn value -> value * 2 end)

msg = Cairn.dispatch(pid, 21)

{:ok, %Cairn.Message{payload: {:ok, 42}}} =
  Cairn.Await.message(msg.ref)

Livebook

Run notebooks/cairn_features.livemd for a tour of function workers, fan-out/fan-in, streaming, supervision, server callbacks, task callbacks, and human review.

API

• Cairn.Message.new/3 and Cairn.Message.reply/2
• Cairn.deliver/2 and Cairn.dispatch/2
• Cairn.Server with handle_msg/2 and handle_task/3
• Cairn.Function.start_link/2, Cairn.Function.start_many/1, and supervised child specs
• Cairn.Task.run/2
• Cairn.Await.message/2, any/2, all/2, collect/2, and stream/2

Architecture

Cairn.dispatch/2 --> Cairn.Message --> process mailbox
Cairn.deliver/2  --> Cairn.Message --> process mailbox

Cairn.Function   --> Cairn.Server --> GenServer
Cairn.Task.run/2 --> Task.Supervisor --> Cairn.Server.handle_task/3

caller mailbox --> Cairn.Await --> replies matched by ref

Many Processes

{:ok, pids} =
  1..1_000
  |> Enum.map(fn n -> fn input -> {n, input * n} end end)
  |> Cairn.Function.start_many()

msgs = Cairn.dispatch(pids, 21)
refs = Enum.map(msgs, & &1.ref)

{:ok, replies} = Cairn.Await.all(refs)

Use as many processes as your BEAM node can actually afford. Cairn does not add a pool or scheduler above OTP. If your node can handle one million processes, pids can be one million processes.

Supervision

children = [
  {Cairn.Function, {:classifier, fn text -> MyApp.LLM.classify(text) end}},
  {Cairn.Function, {:retriever, fn query -> MyApp.Search.run(query) end}}
]

Supervisor.start_link(children, strategy: :one_for_one)

AI orchestration

caller --dispatch(prompt)--> search
       --dispatch(prompt)--> classify
       --dispatch(prompt)--> draft

search   --reply(ref)--> caller
classify --reply(ref)--> caller
draft    --reply(ref)--> caller

caller --Await.any/all/collect/stream(refs)--> results

{:ok, classifier} =
  Cairn.Function.start_link(fn text ->
    MyApp.LLM.classify_ticket(text)
  end)

msg = Cairn.dispatch(classifier, "payment failed after upgrade")

{:ok, %Cairn.Message{payload: {:ok, %{team: team, summary: summary}}}} =
  Cairn.Await.message(msg.ref)

{:ok, pids} =
  Cairn.Function.start_many([
    fn topic -> MyApp.Search.notes(topic) end,
    fn topic -> MyApp.LLM.outline(topic) end,
    fn topic -> MyApp.LLM.risks(topic) end
  ])

refs =
  pids
  |> Cairn.dispatch("elixir lightweight processes")
  |> Enum.map(& &1.ref)

{:ok, replies} = Cairn.Await.all(refs)

pids
|> Cairn.dispatch("index my release notes")
|> Enum.map(& &1.ref)
|> Cairn.Await.stream()
|> Enum.each(fn msg -> IO.inspect(msg.payload) end)

case Cairn.Await.collect(refs, 2_000) do
  {:ok, replies} -> replies
  {:partial, replies, missing} -> {replies, missing}
end

Human in the loop

Human decisions are just messages too:

def handle_msg(%Cairn.Message{from: from, ref: ref, payload: {:draft, text}}, state) do
  send(state.ui, {:review, self(), ref, text})
  {:noreply, put_in(state.pending[ref], from)}
end

def handle_msg(%Cairn.Message{payload: {:approved, ref, edits}}, state) do
  case pop_in(state.pending[ref]) do
    {nil, state} ->
      {:noreply, state}

    {from, state} ->
      Cairn.deliver(from, Cairn.Message.new(self(), {:ok, edits}, ref))
      {:noreply, state}
  end
end

Install

def deps do
  [
    {:cairn, "~> 0.1.6"}
  ]
end

Release

Set HEX_API_KEY in GitHub repository secrets. Run the Release workflow with the next unreleased version, without the v prefix.

The workflow updates mix.exs, README, and CHANGELOG.md; runs quality checks; publishes Hex; tags the commit; and creates the GitHub release.

If the tag already exists but Hex does not have that version, the workflow publishes the existing tag instead of rebuilding from main.