CLI Runners

This module provides infrastructure for wrapping CLI-based AI tools (Codex, Claude, Droid, Kimi, etc.) as subagents. Inspired by the Takopi project's reliable subprocess management patterns.

Overview

CLI Runners enable you to:

• Spawn AI CLI tools as subprocesses with proper lifecycle management
• Stream JSONL events from the CLI's output
• Maintain long-lived sessions with resume capability
• Integrate external agents as collaborators in your main agent loop

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        CodexSubagent                             │
│                   (High-level API)                               │
├─────────────────────────────────────────────────────────────────┤
│                        CodexRunner                               │
│             (Codex-specific event translation)                   │
├─────────────────────────────────────────────────────────────────┤
│                       JsonlRunner                                │
│           (Generic JSONL subprocess GenServer)                   │
├─────────────────────────────────────────────────────────────────┤
│                         Types                                    │
│    (ResumeToken, Action, StartedEvent, ActionEvent, etc.)       │
└─────────────────────────────────────────────────────────────────┘

Quick Start

Basic Usage

alias LemonCliRunners.CodexSubagent

# Start a new Codex session
{:ok, session} = CodexSubagent.start(
  prompt: "Create a GenServer that manages a counter with increment/decrement",
  cwd: "/path/to/project"
)

# Process events as they stream
for event <- CodexSubagent.events(session) do
  case event do
    {:started, token} ->
      IO.puts("Session: #{token.value}")

    {:action, %{kind: :command, title: cmd}, :started, _} ->
      IO.puts("Running: #{cmd}")

    {:action, %{kind: :file_change, title: title}, :completed, ok: true} ->
      IO.puts("Changed: #{title}")

    {:completed, answer, _opts} ->
      IO.puts("Done: #{answer}")

    _ -> :ok
  end
end

One-Shot Execution

# Run synchronously and get the answer
answer = CodexSubagent.run!(
  prompt: "Explain this error: undefined function foo/2",
  cwd: ".",
  on_event: &IO.inspect/1
)

IO.puts(answer)

Session Continuation

# Start initial session
{:ok, session1} = CodexSubagent.start(prompt: "Create a User struct", cwd: ".")
_events = CodexSubagent.events(session1) |> Enum.to_list()

# Continue the session
{:ok, session2} = CodexSubagent.continue(session1, "Add validation for email field")
_events = CodexSubagent.events(session2) |> Enum.to_list()

# Or resume later using the token
token = CodexSubagent.resume_token(session2)
{:ok, session3} = CodexSubagent.resume(token, prompt: "Now add a changeset function")

Event Types

Events are normalized into a simple format:

Action Kinds

Integration as Agent Tool

defmodule MyAgent.Tools do
  alias LemonCliRunners.CodexSubagent
  alias LemonAgent.Types.{AgentTool, AgentToolResult}
  alias LemonAi.Types.TextContent

  def codex_tool(cwd) do
    %AgentTool{
      name: "codex",
      description: "Delegate a complex coding task to a Codex subagent",
      parameters: %{
        "type" => "object",
        "properties" => %{
          "task" => %{
            "type" => "string",
            "description" => "The coding task to perform"
          }
        },
        "required" => ["task"]
      },
      execute: fn _id, %{"task" => task}, _signal, on_update ->
        {:ok, session} = CodexSubagent.start(prompt: task, cwd: cwd)

        # Stream progress updates
        answer = session
        |> CodexSubagent.events()
        |> Enum.reduce("", fn
          {:action, %{title: title}, :completed, ok: true}, acc ->
            if on_update do
              on_update.(%AgentToolResult{
                content: [%TextContent{text: "Completed: #{title}"}]
              })
            end
            acc

          {:completed, answer, _}, _acc ->
            answer

          _, acc ->
            acc
        end)

        %AgentToolResult{
          content: [%TextContent{text: answer}],
          details: %{
            resume_token: CodexSubagent.resume_token(session)
          }
        }
      end
    }
  end
end

Low-Level API

For more control, use the runner directly:

alias LemonCliRunners.CodexRunner
alias LemonCliRunners.Types.ResumeToken

# Start runner
{:ok, pid} = CodexRunner.start_link(
  prompt: "Hello",
  cwd: "/path/to/project",
  timeout: 300_000
)

# Get event stream
stream = CodexRunner.stream(pid)

# Process raw events
for event <- LemonAgent.EventStream.events(stream) do
  case event do
    {:cli_event, event} -> handle_cli_event(event)
    {:agent_end, _} -> :done
    _ -> :ok
  end
end

Implementing New Runners

To add support for a new CLI tool (e.g., Claude):

defmodule LemonCliRunners.ClaudeRunner do
  use LemonCliRunners.JsonlRunner

  alias LemonCliRunners.Types.{EventFactory, ResumeToken}

  @impl true
  def engine, do: "claude"

  @impl true
  def build_command(prompt, resume, _state) do
    args = ["-p", "--output-format", "stream-json"]

    args = case resume do
      %ResumeToken{value: session_id} ->
        args ++ ["--resume", session_id]
      nil ->
        args
    end

    {"claude", args ++ ["--input-format", "text"]}
  end

  @impl true
  def stdin_payload(prompt, _resume, _state), do: String.trim_trailing(prompt) <> "\n"

  @impl true
  def translate_event(data, state) do
    # Convert Claude's JSONL events to CLI runner events
    # ... implementation ...
  end

  @impl true
  def handle_exit_error(code, state) do
    # Handle non-zero exit
  end

  @impl true
  def handle_stream_end(state) do
    # Handle normal exit without completion event
  end
end

Files

Testing

# Run CLI runner tests
mix test apps/lemon_cli_runners/test/lemon_cli_runners/

Design Notes

Session Locking

When resuming a session, the runner acquires a lock (via ETS) to prevent concurrent execution of the same session. This ensures consistency when multiple callers try to resume the same session.

Graceful Shutdown

Subprocess termination follows a graceful pattern:

1. Close stdin to signal end of input
2. Wait for process to exit naturally
3. On timeout: SIGTERM → wait 2s → SIGKILL

Event Translation

Each runner translates tool-specific events to a unified format:

• StartedEvent - Session began with resume token
• ActionEvent - Action lifecycle with phase tracking
• CompletedEvent - Session ended with answer and optional resume

This allows the same UI/progress tracking code to work with any CLI tool.

Adding a New CLI Runner

Implement the LemonCliRunners.JsonlRunner behaviour:

defmodule LemonCliRunners.MyEngineRunner do
  use LemonCliRunners.JsonlRunner

  alias LemonCliRunners.Types.EventFactory
  alias LemonCore.ResumeToken

  @engine "myengine"

  @impl true
  def engine, do: @engine

  @impl true
  def init_state(_prompt, _resume, cwd, _opts) do
    %{factory: EventFactory.new(@engine), last_text: nil}
  end

  @impl true
  def build_command(prompt, resume, _state) do
    args = ["--json", "--output-format", "jsonl"]
    args = if resume, do: args ++ ["--resume", resume.value], else: args
    {"myengine", args ++ ["--", prompt]}
  end

  @impl true
  def translate_event(data, state) do
    case data do
      %{"type" => "init", "session_id" => sid} ->
        token = ResumeToken.new(@engine, sid)
        {started, factory} = EventFactory.started(state.factory, token)
        {[started], %{state | factory: factory}, [found_session: token]}

      %{"type" => "done", "result" => result} ->
        {completed, factory} = EventFactory.completed_ok(state.factory, result || "")
        {[completed], %{state | factory: factory}, [done: true]}

      _ ->
        {[], state, []}
    end
  end

  @impl true
  def handle_exit_error(exit_code, state) do
    {event, factory} = EventFactory.completed_error(state.factory, "failed (rc=#{exit_code})")
    {[event], %{state | factory: factory}}
  end

  @impl true
  def handle_stream_end(state) do
    {event, factory} = EventFactory.completed_error(state.factory, "ended without result")
    {[event], %{state | factory: factory}}
  end
end