LemonAgent

Core agent runtime for the Lemon umbrella project. LemonAgent provides OTP-native building blocks for AI agents: a supervised GenServer for stateful agent lifecycle management, a stateless agentic loop with streaming events, bounded event streams with backpressure, cooperative abort signaling, context window management, and a subagent supervision/registry infrastructure.

Architecture Overview

                        +--------------------------+
                        |    Your Application      |
                        +--------------------------+
                                   |
                                   v
+-----------------------------------------------------------------+
|  LemonAgent                                                      |
|                                                                 |
|  +-------------+  +-------------+  +--------------------------+ |
|  |    Agent    |  |    Loop     |  |  EventStream / Types     | |
|  |  (GenServer)|  | (stateless) |  |  (events & structures)   | |
|  +------+------+  +------+------+  +--------------------------+ |
|         |                |                                      |
|  +------+------+  +------+------+  +--------------------------+ |
|  | AgentRegistry| | SubagentSup |  |  ToolRegistry            | |
|  | (lookup)    |  | (dynamic)   |  |  (runtime tools)         | |
|  +-------------+  +-------------+  +--------------------------+ |
+-----------------------------------------------------------------+
                                   |
                                   v
+-----------------------------------------------------------------+
|  LemonAi Library (low-level LLM abstractions: streaming, providers)  |
+-----------------------------------------------------------------+

LemonAgent sits between application code and the low-level LemonAi library. Applications interact with agents through LemonAgent's public API or the LemonAgent.Agent GenServer. The LemonAi library handles raw LLM provider communication (streaming, completions, message types).

Supervision Tree

LemonAgent.Supervisor (:one_for_one)
|-- LemonAgent.AbortSignal.TableOwner   (GenServer, owns the abort ETS table)
|-- LemonAgent.AgentRegistry            (Registry, :unique keys)
|-- LemonAgent.SubagentSupervisor       (DynamicSupervisor, :temporary children)
|-- LemonAgent.LoopTaskSupervisor       (Task.Supervisor for loop tasks)
+-- LemonAgent.ToolTaskSupervisor       (Task.Supervisor for tool execution)

The supervisor uses a :one_for_one strategy. Each child is independent:

• AbortSignal.TableOwner -- A long-lived GenServer that owns the :agent_core_abort_signals ETS table and acts as heir so the table survives process restarts.
• AgentRegistry -- An Elixir Registry with :unique keys. Agents register under {session_id, role, index} tuples.
• SubagentSupervisor -- A DynamicSupervisor for spawning subagent LemonAgent.Agent processes as :temporary children.
• LoopTaskSupervisor -- A Task.Supervisor for spawning agent loop tasks via Task.Supervisor.async_nolink/2.
• ToolTaskSupervisor -- A Task.Supervisor for spawning concurrent tool execution tasks.

Module Inventory

Core Agent

Loop and Execution

Events and Context

Types

Key Concepts and Design Patterns

Separation of Concerns: Loop vs. Agent

LemonAgent.Loop is pure, stateless logic. It takes context, config, and callbacks, runs the agentic loop (stream LLM response, execute tools, repeat), and emits events through an EventStream. It has no GenServer state.

LemonAgent.Agent is the stateful GenServer that wraps Loop. It manages conversation history, subscriber lists, steering/follow-up queues, and abort references. It spawns the loop as a supervised task and forwards events to subscribers.

Event-Driven Architecture

All execution emits structured events via LemonAgent.EventStream:

{:agent_start}
{:turn_start}
{:message_start, message}
{:message_update, message, assistant_event}
{:message_end, message}
{:tool_execution_start, id, name, args}
{:tool_execution_update, id, name, args, partial_result}
{:tool_execution_end, id, name, result, is_error}
{:turn_end, message, tool_results}
{:agent_end, new_messages}
{:error, reason, partial_state}
{:canceled, reason}

The {:agent_end, new_messages} event contains only messages created during the current run, not the full conversation history.

Cooperative Abort

LemonAgent.AbortSignal uses an ETS table with read_concurrency: true. Tools check AbortSignal.aborted?(signal) in their execute functions. The loop checks before each LLM call, during retry backoff, while waiting for idle provider streams, and during tool execution. On cancellation, supervised tool workers get up to 250ms to observe the signal and finish resource cleanup before remaining workers are terminated. The abort signal stays set until this drain finishes. Cancellation also stops provider streams, cancels tool timers, and leaves queued steering and follow-up messages available for the next run. Tools owning external processes must additionally clean them up on worker death. Incomplete assistant tool calls are removed from canceled model responses so continuing never sends an unpaired tool transcript.

Every started or queued tool call receives a paired terminal result before the canceled event. Cooperative tools retain their actual final output, trust and details. If a worker must be terminated without returning a final result, its last observed output snapshot is preserved and the result explicitly states that the mutation outcome is unknown (details.cancelled: true, details.outcome: :unknown). Calls that never started use outcome: :not_started. Cancellation is not evidence that a mutation was rolled back.

Canceled event streams drain already accepted events in FIFO order before their single cancellation terminal. Consumers that attach after cancellation still receive the buffered tool results and original cancellation reason; late producer pushes are rejected.

Steering and Follow-up Queues

The Agent GenServer provides two message queues:

• Steering (steer/2): Messages injected mid-run. Already-started tools finish, but queued calls are skipped with error results before another model turn. All calls receive paired results before the steering message enters context.
• Follow-up (follow_up/2): Messages queued for after the agent would naturally stop (no more tool calls). A 50ms long-poll closes the race where a follow-up is enqueued just as the run ends.

Both queues support two consumption modes: :one_at_a_time (default) or :all.

Provider Recovery

The loop retries transient provider failures with capped exponential backoff, including failures after partial text or tool-call output. A retry reuses the pre-attempt context, re-resolves credentials, and never executes tools from a failed response. Failed attempts are not committed as assistant messages. The loop emits {:provider_retry, details} before each wait and {:provider_retry_end, details} on success, exhaustion, or cancellation; clients can show the attempt count and delay without treating a retry as a terminal failure.

Retry decisions and retry-after delays use the provider's structured AssistantMessage.error_details when available, not only the user-facing error string. Opaque transport messages can therefore recover without making authentication or quota failures retryable.

Direct agents default to three retries, starting at 2 seconds and capped at 30 seconds (max_retries, retry_base_delay_ms, retry_max_delay_ms). max_retries: 0 disables retries. Explicit provider retry-after hints are honored up to 120 seconds rather than shortened to the backoff cap. Coding sessions use their existing [runtime.retry] settings, with one session-level retry budget rather than stacked retry wrappers.

Authentication, invalid-request and context-overflow errors are not transient retries. Coding sessions handle overflow through their compaction-and-continue recovery boundary; the core loop reports the original failure to that boundary.

Tool Ordering

Tools default to sequential execution. Set parallel_safe: true only when overlapping invocations cannot conflict with mutations or depend on earlier calls. Contiguous safe calls run concurrently up to max_tool_concurrency; a sequential call waits for the preceding group and finishes before later calls start. Results enter model context in the assistant's original call order, regardless of completion order. Unknown tools and invalid arguments are returned as model-readable error results, so the model can correct them.

Registry Pattern

Agents register in LemonAgent.AgentRegistry under {session_id, role, index} tuples. This enables structured lookup across sessions and roles:

LemonAgent.AgentRegistry.lookup({session_id, :research, 0})
LemonAgent.AgentRegistry.list_by_session(session_id)
LemonAgent.AgentRegistry.list_by_role(:research)

Backpressure

EventStream.push/2 is synchronous and returns :ok | {:error, :overflow | :canceled}. Producers can use this for flow control. push_async/2 is fire-and-forget. Drop strategies (:error, :drop_oldest, :drop_newest) control overflow behavior.

Introspection

LemonAgent emits introspection events via LemonCore.Introspection.record/3 for observability. Events include :agent_loop_started, :agent_turn_observed, :agent_loop_ended, :tool_use_observed, and :assistant_turn_observed. Payloads never include prompt or response content.

Telemetry

The library emits telemetry events under the [:lemon_agent, ...] prefix:

• [:lemon_agent, :loop, :start] / [:lemon_agent, :loop, :end] -- Agent loop lifecycle.
• [:lemon_agent, :tool_task, :start] / [:lemon_agent, :tool_task, :end] / [:lemon_agent, :tool_task, :error] -- Individual tool execution.
• [:lemon_agent, :tool_result, :emit] -- Tool result emission.
• [:lemon_agent, :context, :size] / [:lemon_agent, :context, :warning] / [:lemon_agent, :context, :truncated] -- Context management.
• [:lemon_agent, :subagent, :spawn] / [:lemon_agent, :subagent, :end] -- Subagent lifecycle.

Configuration

Application environment keys under :lemon_agent:

The :cli_timeout_ms, :cli_session_lock_max_age_ms, and :cli_cancel_grace_ms keys were removed with the vendor CLI runners; subagent tasks now run natively in-process.

Usage Examples

Creating and Using an Agent

# Define a tool
read_tool = LemonAgent.new_tool(
  name: "read_file",
  description: "Read the contents of a file",
  parameters: %{
    "type" => "object",
    "properties" => %{"path" => %{"type" => "string"}},
    "required" => ["path"]
  },
  execute: fn _id, %{"path" => path}, _signal, _on_update ->
    case File.read(path) do
      {:ok, content} ->
        LemonAgent.new_tool_result(content: [LemonAgent.text_content(content)])
      {:error, reason} ->
        {:error, reason}
    end
  end
)

# Start an agent
{:ok, agent} = LemonAgent.new_agent(
  model: %{provider: :anthropic, id: "claude-sonnet-4-20250514"},
  system_prompt: "You are a helpful assistant.",
  tools: [read_tool],
  convert_to_llm: fn msgs ->
    Enum.filter(msgs, &match?(%{role: role} when role in [:user, :assistant, :tool_result], &1))
  end
)

# Subscribe to events
LemonAgent.subscribe(agent, self())

# Send a prompt
:ok = LemonAgent.prompt(agent, "Read the README.md file")

# Wait for completion
:ok = LemonAgent.wait_for_idle(agent)

# Get final state
state = LemonAgent.get_state(agent)

Steering and Follow-up

# Inject a message mid-run (after current tool batch)
:ok = LemonAgent.Agent.steer(agent, %LemonAi.Types.UserMessage{
  role: :user,
  content: "Actually, use a different approach",
  timestamp: System.system_time(:millisecond)
})

# Queue a message for after the run completes
:ok = LemonAgent.Agent.follow_up(agent, %LemonAi.Types.UserMessage{
  role: :user,
  content: "Now summarize the results",
  timestamp: System.system_time(:millisecond)
})

Using the Loop Directly

alias LemonAgent.{Loop, Types}

context = Types.AgentContext.new(
  system_prompt: "You are helpful",
  tools: tools
)

config = %Types.AgentLoopConfig{
  model: model,
  convert_to_llm: &my_convert/1,
  stream_options: %LemonAi.Types.StreamOptions{max_tokens: 4000}
}

user_msg = %LemonAi.Types.UserMessage{
  role: :user,
  content: "Hello!",
  timestamp: System.system_time(:millisecond)
}

Loop.stream([user_msg], context, config)
|> Enum.each(&IO.inspect/1)

Spawning Subagents

{:ok, pid} = LemonAgent.SubagentSupervisor.start_subagent(
  registry_key: {session_id, :research, 0},
  model: model,
  system_prompt: "Research assistant",
  convert_to_llm: &my_convert/1
)

LemonAgent.Agent.prompt(pid, "Research this topic")
:ok = LemonAgent.Agent.wait_for_idle(pid)
state = LemonAgent.Agent.get_state(pid)
LemonAgent.SubagentSupervisor.stop_subagent(pid)

Context Management

transform = LemonAgent.Context.make_transform(
  max_messages: 50,
  max_chars: 200_000
)

config = %LemonAgent.Types.AgentLoopConfig{
  transform_context: transform,
  model: model,
  convert_to_llm: &my_convert/1
}

# Standalone usage
size = LemonAgent.Context.estimate_size(messages, system_prompt)
{truncated, dropped} = LemonAgent.Context.truncate(messages, max_messages: 50)
stats = LemonAgent.Context.stats(messages, system_prompt)

Proxy Streaming

config = %LemonAgent.Types.AgentLoopConfig{
  model: model,
  convert_to_llm: &my_convert/1,
  stream_fn: fn model, context, opts ->
    LemonAgent.Proxy.stream_proxy(model, context, %LemonAgent.Proxy.ProxyStreamOptions{
      auth_token: get_auth_token(),
      proxy_url: "https://genai.example.com",
      reasoning: opts.reasoning
    })
  end
}

Using EventStream Directly

{:ok, stream} = LemonAgent.EventStream.start_link(
  owner: self(),
  max_queue: 1000,
  timeout: 60_000
)

LemonAgent.EventStream.push(stream, {:custom_event, data})
LemonAgent.EventStream.complete(stream, final_messages)

{:ok, messages} = LemonAgent.EventStream.result(stream, 30_000)
%{queue_size: n, max_queue: m, dropped: d} = LemonAgent.EventStream.stats(stream)

Dependencies

Testing

Running Tests

# All agent_core tests
mix test apps/lemon_agent

# Specific test file
mix test apps/lemon_agent/test/lemon_agent/agent_test.exs

# Run with integration tests
mix test apps/lemon_agent --include integration

Test Organization

Tests are organized to mirror the source structure:

apps/lemon_agent/test/
|-- agent_core_test.exs               Top-level module tests
|-- agent_core_module_test.exs         Module-level tests
|-- agent_registry_test.exs            Registry tests
|-- subagent_supervisor_test.exs       Subagent supervisor tests
|-- agent_core/
|   |-- agent_test.exs                Agent GenServer tests
|   |-- agent_queue_test.exs          Steering/follow-up queue tests
|   |-- abort_signal_test.exs         Abort signal unit tests
|   |-- abort_signal_concurrency_test.exs
|   |-- application_test.exs          Application startup tests
|   |-- application_supervision_test.exs
|   |-- context_test.exs              Context management tests
|   |-- context_property_test.exs     Property-based context tests
|   |-- event_stream_test.exs         EventStream unit tests
|   |-- event_stream_concurrency_test.exs
|   |-- event_stream_edge_cases_test.exs
|   |-- event_stream_runner_test.exs
|   |-- proxy_test.exs                Proxy stream tests
|   |-- proxy_error_test.exs
|   |-- proxy_stream_integration_test.exs
|   |-- types_test.exs
|   |-- property_test.exs
|   |-- supervision_test.exs
|   |-- tool_supervision_test.exs
|   |-- telemetry_test.exs
|   |-- loop/
|   |   |-- tool_calls_test.exs
|   |   |-- streaming_test.exs
|   |-- loop_test.exs
|   |-- loop_abort_test.exs
|   |-- loop_edge_cases_test.exs
|   +-- loop_additional_edge_cases_test.exs

Integration tests that require external services are tagged with @tag :integration and excluded from the default test run.