lemon_agent

# 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

| Module | File | Purpose |
|--------|------|---------|
| `LemonAgent` | `lib/agent_core.ex` | Top-level API facade. Delegates to `Agent` for lifecycle operations (`new_agent/1`, `prompt/2`, `abort/1`, `wait_for_idle/2`, `get_state/1`). Provides convenience constructors (`new_tool/1`, `new_tool_result/1`, `text_content/1`, `image_content/2`, `get_text/1`). Wraps `Loop.stream/4` and `Loop.stream_continue/3` as `agent_loop/4` and `agent_loop_continue/3`. |
| `LemonAgent.Agent` | `lib/agent_core/agent.ex` | GenServer for stateful agent management. Handles prompts, streaming, subscriber broadcasting, steering/follow-up queues, abort signals, and waiter notification. |
| `LemonAgent.AgentRegistry` | `lib/agent_core/agent_registry.ex` | Registry wrapper for agent lookup by `{session_id, role, index}` keys. Supports `via/1` tuples, `lookup/1`, `list_by_session/1`, `list_by_role/1`. |
| `LemonAgent.SubagentSupervisor` | `lib/agent_core/subagent_supervisor.ex` | DynamicSupervisor for spawning subagent processes. `start_subagent/1`, `stop_subagent/1`, `stop_subagent_by_key/1`, `list_subagents/0`, `stop_all/0`. |
| `LemonAgent.Application` | `lib/agent_core/application.ex` | OTP application with the supervision tree above. |

### Loop and Execution

| Module | File | Purpose |
|--------|------|---------|
| `LemonAgent.Loop` | `lib/agent_core/loop.ex` | Stateless agent loop: `agent_loop/5`, `agent_loop_continue/4`, `stream/4`, `stream_continue/3`. Orchestrates prompt injection, LLM streaming, tool call execution, steering, and follow-up in a recursive inner/outer loop. |
| LemonAgent.Loop.Streaming (internal) | `lib/agent_core/loop/streaming.ex` | LLM response streaming. Calls `LemonAi.stream/3` (or a custom `stream_fn`), processes SSE events, builds partial `AssistantMessage`, emits `message_start`/`message_update`/`message_end` events. |
| LemonAgent.Loop.ToolCalls (internal) | `lib/agent_core/loop/tool_calls.ex` | Supervised tool execution. Only tools opting into `parallel_safe: true` overlap; sequential tools are ordering barriers. Supports `max_tool_concurrency`, steering, abort, crash and timeout recovery. |

### Events and Context

| Module | File | Purpose |
|--------|------|---------|
| `LemonAgent.EventStream` | `lib/agent_core/event_stream.ex` | GenServer-based async event producer/consumer. Bounded queue with backpressure (`push/2` returns `:ok` or `{:error, :overflow}`). Owner monitoring, task linking, configurable timeout. Drop strategies: `:error`, `:drop_oldest`, `:drop_newest`. |
| `LemonAgent.Context` | `lib/agent_core/context.ex` | Context window management. `truncate/2` keeps retained messages chronological, enforces both bookends limits, and treats assistant tool-call/result transcripts atomically; also provides `estimate_size/2`, `estimate_tokens/1`, `make_transform/1`, `stats/2`, and `check_size/3`. |
| `LemonAgent.AbortSignal` | `lib/agent_core/abort_signal.ex` | ETS-based cooperative abort signaling. `new/0`, `abort/1`, `aborted?/1`, `clear/1`. Used by the loop and tool execution to check for cancellation. |
| LemonAgent.AbortSignal.TableOwner (internal) | `lib/agent_core/abort_signal/table_owner.ex` | GenServer that owns the abort signal ETS table and acts as heir for table survival. |
| `LemonAgent.Proxy` | `lib/agent_core/proxy.ex` | SSE proxy stream function for routing LLM calls through a server. Reconstructs partial `AssistantMessage` from bandwidth-optimized SSE events. Includes `ProxyStreamOptions` struct. |
| `LemonAgent.ModelRuntime.ProviderConfiguration` | `lib/lemon_agent/model_runtime/provider_configuration.ex` | Shared comment-preserving provider fallback/pool mutation boundary. Preview results carry an opaque config revision; applies can reject concurrent edits, and destructive operations require exact confirmation without returning credential references. |

### Types

| Module | File | Purpose |
|--------|------|---------|
| `LemonAgent.Types` | `lib/agent_core/types.ex` | Core type definitions: `thinking_level`, `agent_message`, `agent_event`. |
| `LemonAgent.Types.AgentTool` | (nested in types.ex) | Tool definition: `name`, `description`, `parameters` (JSON Schema), `label`, `execute` (4-arity function). |
| `LemonAgent.Types.AgentToolResult` | (nested in types.ex) | Tool result: `content` (list of text/image blocks), `details`, `trust` (`:trusted` or `:untrusted`). |
| `LemonAgent.Types.AgentContext` | (nested in types.ex) | Conversation context: `system_prompt`, `messages`, `tools`. |
| `LemonAgent.Types.AgentState` | (nested in types.ex) | Runtime state: `system_prompt`, `model`, `thinking_level`, `tools`, `messages`, `is_streaming`, `stream_message`, `pending_tool_calls`, `error`. |
| `LemonAgent.Types.AgentLoopConfig` | (nested in types.ex) | Loop config: `model`, `convert_to_llm`, `transform_context`, `get_api_key`, `get_steering_messages`, `get_follow_up_messages`, `max_tool_concurrency`, `stream_options`, `stream_fn`. |

## 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:

```elixir
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`:

| Key | Type | Default | Description |
|-----|------|---------|-------------|
| `:queue_call_timeout_ms` | `pos_integer() \| :infinity` | `1_800_000` (30 min) | GenServer call timeout for loop queue polling |
| `:event_stream_cancel_grace_ms` | `pos_integer()` | `100` | Grace period before force-killing an EventStream's attached task |

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

```elixir
# 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

```elixir
# 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

```elixir
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

```elixir
{: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

```elixir
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

```elixir
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

```elixir
{: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

| Dependency | Type | Purpose |
|-----------|------|---------|
| `ai` | umbrella | Low-level LLM API abstractions (streaming, providers, message types) |
| `lemon_core` | umbrella | Shared primitives (telemetry, introspection, ResumeToken) |
| `req` | hex (~> 0.5) | HTTP client used by `LemonAgent.Proxy` |
| `jason` | hex (~> 1.4) | JSON encoding/decoding |
| `stream_data` | hex (~> 1.1, test only) | Property-based testing |

## Testing

### Running Tests

```bash
# 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.