codex_ex

Elixir client for the Codex app-server JSON-RPC 2.0 protocol. Owns the client GenServer, session framing, transport abstraction, turn streaming, client pooling, and a fully code-generated protocol binding.

Installation

def deps do
  [
    {:codex_ex, "~> 0.1"}
  ]
end

Requires a codex executable on $PATH for the default stdio transport (or pass executable: / use the websocket transport).

Usage

Get a pooled client and run a turn to completion:

alias CodexEx.AppServer.{Client, ClientManager, Thread}

# Pooled: callers with the same connection options share one client
# (and one underlying `codex app-server` OS process).
{:ok, client} = ClientManager.get_client(transport: :stdio)

# Start a thread and run a prompt, returning the final assistant text.
{:ok, thread} = Client.start_thread(client, %{"cwd" => "/path/to/workspace"})
{:ok, answer} = Thread.run_text(thread, "Summarize the TODOs in this repo.")

Thread.run/3 starts the turn and returns a live TurnStream; wait on it to get the collected result — items, text deltas, token usage, and the final turn:

alias CodexEx.AppServer.TurnStream

{:ok, stream} = Thread.run(thread, [%{"type" => "text", "text" => "Refactor foo/1"}])
{:ok, stream} = TurnStream.wait(stream)
stream.final_text
stream.items
stream.usage

For structured output, pass a JSON schema:

{:ok, %{"languages" => _}} =
  Thread.run_json(thread, "List the languages used in this repo.", %{
    "type" => "object",
    "properties" => %{"languages" => %{"type" => "array", "items" => %{"type" => "string"}}},
    "required" => ["languages"]
  })

To observe streamed events (deltas, item updates, token usage) while a turn runs, subscribe before starting it — every parsed server event arrives as {:codex_app_server_event, message}:

:ok = Client.subscribe(client)

{:ok, _stream} = Thread.run(thread, [%{"type" => "text", "text" => "Go"}])

receive do
  {:codex_app_server_event, message} ->
    CodexEx.AppServer.Message.extract_text_delta(message)
end

Server-initiated requests (tool approvals, user-input elicitation) are answered by a request_handler: function passed to Client.start_link/1 — note that pooled clients from ClientManager don't accept one, so use a dedicated client for approval flows:

{:ok, client} =
  Client.start_link(
    transport: :stdio,
    request_handler: fn _request -> %{"decision" => "approved"} end
  )

Resume an existing thread by id with Client.resume_thread(client, thread_id); fork, archive, goals, model/skill listing, and fuzzy file search are all on Client/Thread — see the module reference below.

Host integration

The package runs its own supervision tree (CodexEx.Application): a Task.Supervisor (CodexEx.TaskSupervisor), a DynamicSupervisor (CodexEx.ClientSupervisor), and the ClientManager singleton.

Optional host configuration:

config :codex_ex,
  # Phoenix.PubSub server for thread-activity broadcasts (nil disables them)
  pubsub: MyApp.PubSub,
  # Starts a shared local observer client at boot
  thread_activity_observer_enabled: true,
  # MFA invoked during local thread-activity reconciliation,
  # must return {:ok, term()} | {:error, term()}
  thread_activity_recovery: {MyApp.SessionRecovery, :recover_pending_sessions, [[runtime_type: "local"]]}

Custom transports implement the CodexEx.AppServer.Transport behaviour and are passed as the :transport option. Remote transports (sessions that outlive the node) additionally implement the optional callbacks remote_transport?/0, reconcile_thread_activity/1, acknowledge/2, acknowledge_replay_gap/2, and session_bootstrap/1.

Process Hierarchy

ClientManager (singleton, named GenServer)
  └─ Client (one per unique connection key, :temporary)
       State: session pid, subscribers, pending requests, request handler
       │
       └─ Session (one per Client, linked)
            State: JSON-RPC framing, pending request map, buffer
            │
            └─ Transport (one per Session)
                 ├─ StdioTransport   — managed proxy when available, standalone fallback
                 ├─ WebSocketTransport — Mint HTTP + Mint.WebSocket (GenServer)
                 └─ any host-provided Transport module (e.g. a remote bridge)

Additionally, each Client.run/4 call spawns a TurnStream GenServer (one per active turn) that collects streamed notifications until the turn completes.

All Clients are started under CodexEx.ClientSupervisor (a DynamicSupervisor) with :temporary restart — they do not restart on crash. ClientManager monitors pooled Clients and reuses only live pids; normal Client shutdown also stops its owned Session and transport.

Module Reference

Core GenServers

Client

Central GenServer wrapping a single app-server connection. Exposes the full domain API (threads, turns, goals, models, skills, fuzzy search) and manages subscriber fan-out for streamed server events.

The client owns its Session, initialize result, thread-scoped subscribers, pending server requests, model-list cache, replay-gap ownership, and optional request handler.

Timeouts:

• Default RPC: 15 seconds
• Turn operations: 30 minutes
• Client call grace: 1 second (added to RPC timeout for GenServer.call)

Deferred reply pattern: All domain calls (thread start/resume/read, turn run, goal ops) use the same flow: the GenServer handle_call captures from, spawns a Task under CodexEx.TaskSupervisor, that task issues the RPC through Session and calls GenServer.reply(from, result). This prevents the Client GenServer from blocking on slow network calls.

Key operations:

• Thread lifecycle: start_thread, resume_thread, read_thread, fork_thread, archive_thread, unarchive_thread, start_thread_compaction, rollback_thread, revert_thread
• Turn lifecycle: run (via TurnStream), start_turn_request, steer_turn, interrupt_turn
• Goals: set_thread_goal, get_thread_goal, clear_thread_goal
• Capabilities: list_models, list_skills, list_threads, list_experimental_features, set_experimental_feature_enablement
• Fuzzy file search: start_fuzzy_file_search_session, update_fuzzy_file_search_session, stop_fuzzy_file_search_session
• Subscription: subscribe/1, unsubscribe/1 — subscribers receive {:codex_app_server_event, Message.t()} for every parsed server event

Server-initiated request handling: When the server sends a request (tool approval, MCP elicitation), the Client parses it via Protocol.Parser, broadcasts to subscribers, and (if a request_handler function is configured) spawns a task to compute the reply. The task calls Client.reply_request/4, which sends the response through Session.

ClientManager

Singleton GenServer that pools Client processes by connection identity.

The effective identity includes transport, launcher, workspace/runner, initialize, and protocol options. Remote daemon transport ids intentionally exclude local-only observer flags so deploys can reattach retained sessions.

On get_client/1, the manager returns a live client for the key or starts one under CodexEx.ClientSupervisor. Shared clients disallow request_handler (it would be ambiguous with multiple consumers).

Session

GenServer implementing JSON-RPC 2.0 framing over a pluggable transport.

The session owns request ids, pending request timers, the partial-line buffer, the transport handle, and ordered transport acknowledgement state.

Protocol details:

• Request IDs are incrementing integers starting at 1
• Outgoing payloads normalized through ClientRequest.encode/1 (generated module)
• Line-based parsing: incoming data split on \n, partial lines buffered
• Per-request timeouts via Process.send_after/3, max 30 minutes
• On transport close: replies all pending requests with {:error, {:transport_closed, reason}}, then stops

Transport selection (via :transport option):

• :stdio → StdioTransport
• :websocket → WebSocketTransport
• :mock → MockTransport
• Any atom → used directly as the transport module

TurnStream

GenServer that collects streamed notifications for a single agent turn. Started by Client.run/4, it correlates items, text deltas, turn events, and token usage into a coherent result.

Public struct:

%TurnStream{
  final_text: binary() | nil,
  final_turn: Turn.t() | nil,
  initial_turn: Turn.t() | nil,
  items: [ThreadItem.t()],
  pid: pid(),
  text_deltas: [binary()],
  thread_id: binary(),
  turn_id: binary() | nil,
  usage: TokenUsage.t() | nil
}

Race handling: Notifications may arrive before the RPC result returns the turn_id. TurnStream buffers these as pending_messages_rev and replays them once the turn ID is known from the RPC result.

Completion semantics: A turn is only complete when BOTH conditions are met:

1. The RPC result has returned (rpc_final_turn)
2. A final turn event has arrived (event_final_turn)

best_final_turn/1 prefers the RPC result over the event turn.

Item merging: AgentMessage items merge by status rank comparison:

• completed / failed / interrupted = rank 3
• inProgress = rank 2
• everything else = rank 1

Higher-ranked status wins; text is replaced only when the new item carries a non-nil text value.

Text delta handling: Deltas are ignored after the RPC final turn arrives with a non-empty final_text, preventing late-arriving deltas from corrupting the result.

Default wait timeout: 30 minutes.

Transport Layer

Transport (behaviour)

Defines the contract for all transport implementations:

@callback open(opts :: keyword()) :: {:ok, handle :: term()} | {:error, term()}
@callback send(handle :: term(), data :: binary()) :: :ok | {:error, term()}
@callback close(handle :: term()) :: :ok
@callback normalize_message(raw :: term(), handle :: term()) ::
            {:data, binary()}
            | {:data, binary(), non_neg_integer()}
            | {:closed, term()}
            | {:closed, term(), non_neg_integer()}
            | {:replay_gap, map()}
            | :ignore

The Session calls open/1 at init, send/2 for outgoing requests, and normalize_message/2 for every message the transport process delivers.

StdioTransport

Local transport for the codex executable. It prefers codex app-server proxy against the managed control socket and starts a standalone codex app-server when the proxy is unavailable, unless proxy_only? was requested.

• Default executable: System.find_executable("codex")
• Standalone default args: ["app-server", "--enable", "realtime_conversation"]
• Port options: :binary, :exit_status, :use_stdio, :hide
• normalize_message/2: {port, {:data, data}} → {:data, data}, {port, {:exit_status, _}} → {:closed, ...}

WebSocketTransport

GenServer wrapping Mint HTTP + Mint.WebSocket for persistent connections.

Retry logic:

• HTTP 429 and 503 trigger automatic retry with exponential backoff
• Respects Retry-After header when present
• Default: 2 retry attempts, 250ms base delay, 5s max handshake timeout
• WSS connections auto-configure CA certs via :public_key.cacerts_get()

Connection lifecycle:

• Handles WebSocket ping/pong automatically
• Acknowledges remote close frames before shutting down
• Monitors owner pid; cleans up on owner death

Remote transports (host-provided)

A host application can bridge to remote runners by implementing the Transport behaviour plus its optional remote callbacks. The reference implementation lives in the host app as App.Runtime.RemoteClient.CodexTransport.

Handle:

%{
  runner_id: binary(),
  owner: pid(),
  transport_id: binary(),
  workspace_id: binary(),
  workspace_root: binary(),
  initialize_bootstrap: :fresh | {:reattached, map(), boolean()},
  next_request_id: pos_integer(),
  acknowledged_through: non_neg_integer(),
  replay: list(),
  replay_gap: non_neg_integer() | nil,
  pending_requests: list()
}

Event subscriptions (on open):

1. codex.session.message — incoming data from remote runner
2. codex.session.closed — remote session terminated
3. codex.session.replay_gap — bounded replay overflow boundary
4. remote_client.connection — remote client disconnected

open/1 subscribes to all four event patterns, then sends codex.session.reattach with the stable client-manager transport id. The daemon either opens a fresh app-server or returns cached initialization state plus retained events, unresolved server requests, the acknowledgement watermark, the next JSON-RPC request id, and an opaque generation for its surviving process. Send and ACK requests echo that generation so delayed work cannot affect a replacement with the same stable transport id. send/2 forwards the decoded JSON payload through codex.session.send; an ambiguous Router timeout or post-push disconnect closes only the BEAM-side owner with a tagged delivery_uncertain reason so the durable client-message correlation is not retried. An ambiguous server-response send remains unresolved for reattach instead of acknowledging its retained request. Sequenced terminal closes pass through the retained event acknowledgement barrier before the BEAM Session stops, allowing the daemon to retain child exits across WebSocket outages without losing their replay tail. close/1 only unsubscribes; the daemon owns app-server lifetime across BEAM deploys.

Protocol Layer

The protocol layer has two parts: a code generator that produces typed Elixir modules from JSON Schema, and hand-written infrastructure that parses, routes, and bridges between generated types and the domain model.

Generated Code

Located in lib/codex_ex/app_server/protocol/generated/ under three namespaces:

Each generated module provides:

• A defstruct with fields matching the JSON Schema properties
• decode/1 — JSON map → struct (via Codec.decode_object/3)
• encode/1 — struct → JSON map (via Codec.encode_object/2)

Envelope modules (ServerNotification, ServerRequest, etc.) are special: they contain a @method_specs map keyed by method name, with each entry pointing to the params module for that method. They provide known_method?/1 for dispatch and decode/1 which resolves the method and decodes params through the appropriate module.

Protocol.Generator

Mix-time code generator that reads JSON Schema files from priv/schema/ and writes Elixir source to lib/codex_ex/app_server/protocol/generated/.

Pipeline:

1. Load and parse all .json schema files (excluding bundle files)
2. Build a module index mapping schema titles to module names
3. For each schema:
   • Object schemas → struct module with @field_specs, nested defmodule for definitions, decode/1 and encode/1
   • Envelope schemas (titles in ClientNotification, ClientRequest, ServerNotification, ServerRequest) → dispatch module with @method_specs, known_method?/1, method-aware decode/1 / encode/1
   • Other schemas → passthrough module (decode returns value, encode calls Codec.encode_value(:plain, ...))
4. Format all output files with the project's .formatter.exs

Value spec type algebra:

:plain                      # passthrough (scalars, untyped maps)
{:array, value_spec()}      # list of typed elements
{:module, module()}         # nested object with its own decode/encode
{:nullable, value_spec()}   # nil | inner_spec

Protocol.Codec

Runtime encode/decode engine used by all generated modules.

• decode_object/3 — takes module, field specs, and a JSON map; stringifies keys, walks each field spec calling decode_value/2 recursively, returns struct(module, attrs)
• encode_object/2 — takes a struct and field specs; omits nil optional fields, encodes values recursively, returns a string-keyed map
• Recursive type resolution: :plain passes through, {:module, mod} delegates to mod.decode/1 or mod.encode/1, {:array, spec} maps over lists, {:nullable, spec} handles nil

Protocol.Parser

Routes raw JSON-RPC payloads to the appropriate typed struct.

Dispatch logic:

• Notifications: if ServerNotification.known_method?(method) → %ServerNotification{}; else (and not strict) → %GenericNotification{}
• Requests: if ServerRequest.known_method?(method) → full ServerRequest.decode(payload); else → %GenericServerRequest{}
• Strict mode: unknown methods return {:error, {:unknown_method, kind, method}} instead of falling back to generic types

Protocol.Verifier

Drift detection: regenerates protocol modules to a temp directory and diffs against the committed generated/ tree. Reports missing, unexpected, or changed files. Used in CI to ensure generated code stays in sync with the schema snapshot.

Fallback Types

• GenericNotification — %{method, params} for unrecognized notification methods
• GenericServerRequest — %{id, method, params} for unrecognized request methods
• UnmatchedResponse — %{id, payload} for RPC responses that don't match a pending request

Domain Model

Hand-written structs that normalize generated protocol types into stable, typed values used throughout the agents subsystem. All follow the same pattern: from_protocol/1 accepts either a generated struct or a plain map, normalizes via ProtocolValue, and returns {:ok, struct} or {:error, {tag, reason}}.

Thread

Service object pairing a Client pid with a ThreadSnapshot.

%Thread{client: Client.t(), id: binary(), snapshot: ThreadSnapshot.t()}

Provides convenience methods that delegate to Client: refresh, fork, archive, unarchive, set_goal, get_goal, clear_goal, list_turns_page, run, run_text, run_json. Full history reads use paginated turns; legacy full-history reads fail explicitly.

ThreadSnapshot

Full typed snapshot of thread state returned by the server.

Fields: id, name, status, cwd, preview, source, thread_source, cli_version, model_provider, ephemeral, created_at, updated_at, turns (list of Turn.t()), git_info (GitInfo.t() | nil), history_mode, agent_nickname, agent_role, path.

Nested: ThreadSnapshot.GitInfo — %{branch, origin_url, sha}.

Turn

Snapshot of a single conversation turn.

Fields: id, thread_id, status, items (list of ThreadItem.t()), error (Turn.Error.t() | nil).

Status helpers: completed?/1, failed?/1, interrupted?/1.

Legacy input synthesis: if a legacy server turn payload includes an input field but no userMessage item exists in items, the module synthesizes one and prepends it.

ThreadItem

Two variants:

• ThreadItem.AgentMessage — %{id, type, text, phase, status} for "agentMessage" items
• ThreadItem.Generic — %{id, type, attrs} fallback for all other item types

ThreadGoal

Thread goal with budget tracking.

Fields: thread_id, objective, status (:active | :paused | :budget_limited | :complete), token_budget, tokens_used, time_used_seconds, created_at, updated_at.

TokenUsage

Token usage reported via thread/tokenUsage/updated notifications.

Fields: last (Breakdown.t()), total (Breakdown.t()), model_context_window.

TokenUsage.Breakdown: input_tokens, output_tokens, cached_input_tokens, reasoning_output_tokens, total_tokens.

Types

Shared type definitions:

@type json_scalar :: nil | boolean() | integer() | float() | binary()
@type json_value  :: json_scalar() | [json_value()] | %{optional(binary()) => json_value()}
@type json_object :: %{optional(binary()) => json_value()}
@type timestamp   :: integer() | float()

Infrastructure Modules

Message

Extraction helpers for protocol-native messages. Works uniformly across ServerNotification, ServerRequest, GenericNotification, GenericServerRequest, and UnmatchedResponse.

Extractors:

• method_name/1, request_id/1, resolved_request_id/1
• thread_id/1, turn_id/1, item_id/1 — deep extraction with fallback to nested thread.id, turn.id, item.id
• extract_item/1 → ThreadItem.from_protocol/1
• extract_turn/1 → Turn.from_protocol/2 (requires thread_id)
• extract_text_delta/1 — only for item/agentMessage/delta method
• extract_token_usage/1 — only for thread/tokenUsage/updated method

All field access goes through ProtocolValue for struct/map duality.

ProtocolValue

Permissive key-lookup bridge between generated structs and plain maps.

Problem solved: generated protocol structs use atom keys (often camelCase from the JSON Schema), while the domain model expects snake_case atoms. ProtocolValue.fetch/3 and get/4 try multiple key forms:

1. Direct lookup: atom key, string key, camelCase string
2. Fallback scan: all map keys checked against underscore / camelCase / extra_keys variants

Also provides to_json_value/1 for recursive struct → JSON-safe map conversion and normalize_map/1 to strip struct metadata.

SchemaSnapshot

Exports the JSON Schema from the codex CLI to priv/schema/.

Pipeline:

1. Find codex executable
2. Run codex app-server generate-json-schema --experimental --out <tmp>
3. Replace existing snapshot directory atomically
4. Report file count

Used as a development-time step before running Protocol.Generator.

Data Flow

Outgoing RPC (e.g. start a thread)

Caller
  → Client.start_thread/2  (GenServer.call)
  → handle_call captures `from`, spawns Task
  → Task calls Session.request(session, "thread/start", params, timeout)
  → Session builds JSON-RPC envelope {id: N, method, params}
  → Session encodes via ClientRequest.encode/1
  → Session calls transport_module.send(transport, json_line)
  → Transport delivers bytes (Port / WebSocket / RemoteClient RPC)
  ...
  → Transport receives response bytes
  → Session.normalize_message → {:data, bytes}
  → Session parses newline-delimited JSON
  → Session matches response id to pending request, replies caller
  → Task receives {:ok, result}, calls GenServer.reply(from, {:ok, thread})

Incoming Server Events (streamed turn)

Transport receives notification bytes
  → Session parses, sends {:notification, payload} to Client
  → Client parses via Protocol.Parser.parse(:notification, payload)
  → Client broadcasts {:codex_app_server_event, message} to all subscribers
  → If active TurnStream exists for this thread:
      TurnStream receives the event, correlates by turn_id
      → item/started, item/completed → merge into items map
      → item/agentMessage/delta → append to text_deltas
      → thread/tokenUsage/updated → update usage
      → terminal turn event → set event_final_turn from its status
      → When both rpc_final_turn AND event_final_turn set → mark done
      → Waiters receive {:ok, %TurnStream{}} with final state

Server-Initiated Requests (tool approval)

Transport receives request bytes
  → Session parses, sends {:request, id, payload} to Client
  → Client parses via Protocol.Parser.parse(:request, payload)
  → Client broadcasts to subscribers
  → If request_handler configured:
      Spawns Task → request_handler.(message) → reply map
      → Client.reply_request(client, request_id, reply)
      → Session sends JSON-RPC response with matching id

Cross-Cutting Patterns

Client Pooling

ClientManager reuses clients by their effective connection identity. Two callers with the same transport, launcher, workspace/runner, initialize, and protocol options get the same Client pid. Dead pooled clients are removed by monitor and replaced on the next request.

Stale Client Recovery

Host applications typically wrap thread operations with a stale-client retry (see App.Runtime.Agents.SessionRemoteThreadOps.with_stale_client_retry/5 in the reference host): if the initial attempt returns a stale-session error, reconnect with a fresh client and retry once.

Strict vs. Permissive Protocol

The strict_protocol flag (passed through to Protocol.Parser) controls whether unknown methods are errors or silently wrapped in generic types. Production runs permissive; tests can enable strict mode to catch schema drift.

Schema → Code Pipeline

codex CLI
  → SchemaSnapshot.snapshot/1 → priv/schema/*.json
  → Generator.generate/1 → protocol/generated/{shared,v1,v2}/*.ex
  → Verifier.verify/1 (CI) — regenerates to temp dir, diffs against committed code

Timeout Hierarchy

File Inventory

Hand-Written

Generated

Located in protocol/generated/{shared,v1,v2}/. Each file is a single-module Elixir source with defstruct, decode/1, encode/1. Regenerated by Protocol.Generator from JSON Schema snapshots in priv/schema/. Do not edit manually — changes will be overwritten and flagged by Protocol.Verifier in CI.