LemonCliRunners.JsonlRunner behaviour (lemon_cli_runners v0.1.0)

Base GenServer for running CLI tools that emit JSONL events.

This module provides the infrastructure for spawning CLI subprocesses, reading their JSONL output, and emitting events through an EventStream.

Architecture

The JsonlRunner is a behaviour module that concrete runners implement. It handles:

Subprocess spawning with proper signal handling
JSONL line parsing from stdout
Concurrent stderr draining
Graceful shutdown (SIGTERM → SIGKILL)
Session locking to prevent concurrent runs of the same session
Event translation to unified CLI runner events

Usage

Implement the callbacks:

defmodule MyRunner do
  use LemonCliRunners.JsonlRunner

  @impl true
  def engine, do: "my_engine"

  @impl true
  def build_command(prompt, resume, state) do
    {"my-cli", ["exec", "--json"]}
  end

  @impl true
  def translate_event(data, state) do
    # Convert parsed JSON to CLI runner events
  end
end

Then start it:

{:ok, pid} = MyRunner.start_link(prompt: "Hello", cwd: "/path")

# Subscribe to events
for event <- LemonAgent.EventStream.events(stream) do
  handle_event(event)
end

Session Locking

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

Summary

Types

resume_option()

runner_state()

start_option()

Callbacks

build_command(prompt, resume, state)

Build the command and arguments to execute

decode_line(line)

Decode a JSON line into a data structure

engine()

Return the engine identifier (e.g., 'codex', 'claude')

env(state)

Optional environment variables

handle_exit_error(exit_code, state)

Handle non-zero exit code

handle_stream_end(state)

Handle stream end without completion event

init_state(prompt, resume)

Create initial runner state

init_state(prompt, resume, cwd)

Create initial runner state with cwd context

init_state(prompt, resume, cwd, opts)

Create initial runner state with cwd + raw options context

stdin_payload(prompt, resume, state)

Return bytes to send to stdin (or nil for no input)

translate_event(data, state)

Translate decoded data into CLI runner events.

Functions

cancel(pid, reason \\ :user_requested)

Cancel a running runner

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

run(module, opts)

Run synchronously and collect all events.

start_link(module, opts)

Start a JSONL runner.

stream(pid)

Get the event stream from a running runner

Types

resume_option()

@type resume_option() :: LemonCore.ResumeToken.t() | nil

runner_state()

@type runner_state() :: term()

start_option()

@type start_option() ::
  {:prompt, String.t()}
  | {:resume, resume_option()}
  | {:cwd, String.t()}
  | {:env, [{String.t(), String.t()}]}
  | {:timeout, timeout()}
  | {:model, String.t()}
  | {:system_prompt, String.t()}
  | {:extra_tools, [term()]}
  | {:owner, pid()}

Callbacks

build_command(prompt, resume, state)

@callback build_command(
  prompt :: String.t(),
  resume :: resume_option(),
  state :: runner_state()
) :: {command :: String.t(), args :: [String.t()]}

Build the command and arguments to execute

decode_line(line)

@callback decode_line(line :: binary()) :: {:ok, term()} | {:error, term()}

Decode a JSON line into a data structure

engine()

@callback engine() :: String.t()

Return the engine identifier (e.g., 'codex', 'claude')

env(state)

(optional)

@callback env(state :: runner_state()) :: [{String.t(), String.t()}] | nil

Optional environment variables

handle_exit_error(exit_code, state)

@callback handle_exit_error(exit_code :: integer(), state :: runner_state()) ::
  {events :: [LemonCliRunners.Types.cli_event()], state :: runner_state()}

Handle non-zero exit code

handle_stream_end(state)

@callback handle_stream_end(state :: runner_state()) ::
  {events :: [LemonCliRunners.Types.cli_event()], state :: runner_state()}

Handle stream end without completion event

init_state(prompt, resume)

@callback init_state(prompt :: String.t(), resume :: resume_option()) :: runner_state()

Create initial runner state

init_state(prompt, resume, cwd)

(optional)

@callback init_state(prompt :: String.t(), resume :: resume_option(), cwd :: String.t()) ::
  runner_state()

Create initial runner state with cwd context

init_state(prompt, resume, cwd, opts)

(optional)

@callback init_state(
  prompt :: String.t(),
  resume :: resume_option(),
  cwd :: String.t(),
  opts :: keyword()
) :: runner_state()

Create initial runner state with cwd + raw options context

stdin_payload(prompt, resume, state)

@callback stdin_payload(
  prompt :: String.t(),
  resume :: resume_option(),
  state :: runner_state()
) :: binary() | nil

Return bytes to send to stdin (or nil for no input)

translate_event(data, state)

@callback translate_event(data :: term(), state :: runner_state()) ::
  {events :: [LemonCliRunners.Types.cli_event()], state :: runner_state(),
   opts :: keyword()}

Translate decoded data into CLI runner events.

Returns a tuple of {events, updated_state, options} where options can include:

:found_session - ResumeToken extracted from this event
:done - true if this is the final event

Functions

cancel(pid, reason \\ :user_requested)

@spec cancel(pid(), term()) :: :ok

Cancel a running runner

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

run(module, opts)

@spec run(module(), [start_option()]) :: {:ok, [term()]} | {:error, term()}

Run synchronously and collect all events.

Returns {:ok, events} or {:error, reason}.

start_link(module, opts)

@spec start_link(module(), [start_option()]) :: GenServer.on_start()

Start a JSONL runner.

Options

:prompt - The prompt to send (required)
:resume - ResumeToken for session continuation (optional)
:cwd - Working directory (default: current directory)
:env - Additional environment variables
:timeout - Subprocess timeout in ms (default from :lemon_cli_runners, :cli_timeout_ms)
:owner - Owner process to monitor (default: caller)

Returns {:ok, pid} where pid is the runner GenServer. Use stream/1 to get the event stream.

stream(pid)

@spec stream(pid()) :: LemonAgent.EventStream.t()

Get the event stream from a running runner