CodexEx.AppServer.Client (codex_ex v0.1.0)

Copy Markdown View Source

Public entrypoint for the Codex app-server client.

This module owns the long-lived session process, exposes the thread and turn request helpers, broadcasts parsed server events to subscribers, and can auto-reply to server-initiated requests through register_request_handler/2.

Summary

Types

client_call_error()
config_read_response()
config_write_response()
experimental_feature_enablement_set_response()
experimental_feature_list_response()
hooks_list_response()
initialize_response()
mcp_resource_read_response()
mcp_server_tool_call_response()
model_list_response()
pending_model_list_request()
pending_request_map()
registered_name()
request_handler()
request_handler_reply()
request_result()
request_timeout()
skills_list_response()
state()
subscriber()
subscriber_map()
t()
thread_list_response()
thread_response()
thread_turns_page()
turn_interrupt_response()

Functions

acknowledge_replay_gap_async(client, owner, through_sequence)

Queues replay-gap acknowledgement on the client generation that reported it.

archive_thread(client, thread_id)
broadcast_active_threads(client)

Publishes active threads already loaded by the connected app server.

build_initialize_params(params)

Builds normalized app-server initialization parameters with required client capabilities.

call_mcp_tool(client, thread_id, server, tool, arguments)

Calls one MCP tool in the active thread's server scope.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

clear_thread_goal(client, thread_id)
connect(opts \\ [])
disconnect(client, reason \\ :normal, timeout \\ 15000)
fork_thread(client, thread_id, overrides \\ %{})
get_thread_goal(client, thread_id)
initialize_result(client)
inject_thread_items(client, thread_id, items)

Injects model-visible history items without starting a turn.

interrupt_turn(client, thread_id, turn_id, timeout \\ 15000)
list_experimental_features(client, params \\ %{})
list_hooks(client, cwds \\ [])
list_models(client)
list_skills(client, params \\ %{})
list_thread_turns(client, thread_id, opts \\ [])

Lists one page of a paginated thread's turns in the requested server order.

list_threads(client, params \\ %{})
pending_requests(client)
read_config(client, cwd)

Reads the effective Codex configuration for a working directory.

read_mcp_resource(client, thread_id, server, uri)

Reads one MCP resource in the active thread's server scope.

read_thread(client, thread_id, opts \\ [])
register_request_handler(client, handler)
reply_request(client, request_id, reply, timeout \\ 15000)
resume_thread(client, thread_id, overrides \\ %{}, opts \\ [])
revert_thread(client, thread_id, before_turn_id)
rollback_thread(client, thread_id, num_turns)
run(client, thread_id, input, opts \\ %{}, direct_stream? \\ true)
run_json(client, thread_id, text, output_schema, opts \\ %{})
run_text(client, thread_id, text, opts \\ %{})
set_experimental_feature_enablement(client, enablement)
set_hook_enabled(client, hook_key, enabled?, timeout \\ 15000)
set_thread_goal(client, thread_id, attrs)
set_thread_name(client, thread_id, name)
start_fuzzy_file_search_session(client, session_id, roots)
start_link(opts \\ [])
start_realtime(client, thread_id, sdp)
start_review(client, thread_id, target \\ %{"type" => "uncommittedChanges"}, delivery \\ nil, direct_stream? \\ true)
start_review_request(client, params, timeout \\ 1_800_000)
start_thread(client, params \\ %{})
start_thread_compaction(client, thread_id)
start_turn(client, thread_id, input, opts \\ %{})

Starts a turn and returns its initial protocol state without collecting its event stream.

start_turn_request(client, params, timeout \\ 1_800_000)
steer_turn(client, thread_id, expected_turn_id, input, client_message_id, timeout \\ 15000)
stop_fuzzy_file_search_session(client, session_id)
stop_realtime(client, thread_id)
subscribe(client, subscriber \\ self())
subscribe(client, subscriber, opts)
subscribe_thread_activity()

Subscribes the caller to active-thread signals from shared Codex clients.

trust_hook(client, hook_key, current_hash, timeout \\ 15000)
unarchive_thread(client, thread_id)
unsubscribe(client, subscriber \\ self())
unsubscribe_thread(client, thread_id)

Unsubscribes the app-server connection from a remote thread.

update_fuzzy_file_search_session(client, session_id, query)

Types

client_call_error()

@type client_call_error() :: {:client_call_failed, term()} | {:defer_failed, term()}

config_read_response()

@type config_read_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.ConfigReadResponse{
    config: term(),
    layers: term(),
    origins: term()
  }

config_write_response()

@type config_write_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.ConfigWriteResponse{
    file_path: term(),
    overridden_metadata: term(),
    status: term(),
    version: term()
  }

experimental_feature_enablement_set_response()

@type experimental_feature_enablement_set_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.ExperimentalFeatureEnablementSetResponse{
    enablement: term()
  }

experimental_feature_list_response()

@type experimental_feature_list_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.ExperimentalFeatureListResponse{
    data: term(),
    next_cursor: term()
  }

hooks_list_response()

@type hooks_list_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.HooksListResponse{data: term()}

initialize_response()

@type initialize_response() ::
  %CodexEx.AppServer.Protocol.Generated.V1.InitializeResponse{
    codex_home: term(),
    platform_family: term(),
    platform_os: term(),
    user_agent: term()
  }

mcp_resource_read_response()

@type mcp_resource_read_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.McpResourceReadResponse{
    contents: term()
  }

mcp_server_tool_call_response()

@type mcp_server_tool_call_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.McpServerToolCallResponse{
    _meta: term(),
    content: term(),
    is_error: term(),
    structured_content: term()
  }

model_list_response()

@type model_list_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.ModelListResponse{
    data: term(),
    next_cursor: term()
  }

pending_model_list_request()

@type pending_model_list_request() :: %{ref: reference(), callers: [GenServer.from()]}

pending_request_map()

@type pending_request_map() :: %{optional(term()) => CodexEx.AppServer.Message.t()}

registered_name()

@type registered_name() ::
  atom() | {atom(), node()} | {:global, term()} | {:via, atom(), term()}

request_handler()

@type request_handler() :: (term() -> request_handler_reply())

request_handler_reply()

@type request_handler_reply() ::
  CodexEx.AppServer.Message.supported_reply_payload()
  | {:ok, CodexEx.AppServer.Message.supported_reply_payload()}
  | {:error, term()}

request_result()

@type request_result() ::
  {:ok, CodexEx.AppServer.Message.supported_reply_payload()} | {:error, term()}

request_timeout()

@type request_timeout() :: timeout()

skills_list_response()

@type skills_list_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.SkillsListResponse{
    data: term()
  }

state()

@type state() :: %{
  initialize_result: initialize_response(),
  model_list_cache: model_list_response() | nil,
  model_list_pending: pending_model_list_request() | nil,
  pending_requests: pending_request_map(),
  replay_gap: map() | nil,
  replay_gap_owners: MapSet.t(pid()),
  broadcasts_thread_activity?: boolean(),
  thread_activity_runner_id: binary() | nil,
  thread_activity_workspace_id: binary() | nil,
  request_handler: request_handler() | nil,
  session: CodexEx.AppServer.Session.t(),
  strict_protocol: boolean(),
  subscribers: subscriber_map()
}

subscriber()

@type subscriber() :: %{
  monitor_ref: reference(),
  thread_id: binary() | nil | :all,
  reconciles_replay_gap?: boolean()
}

subscriber_map()

@type subscriber_map() :: %{optional(pid()) => subscriber()}

t()

@type t() :: pid() | registered_name()

thread_list_response()

@type thread_list_response() :: %{
  data: [CodexEx.AppServer.ThreadSnapshot.t()],
  next_cursor: term()
}

thread_response()

@type thread_response() :: %CodexEx.AppServer.Thread{
  client: term(),
  id: term(),
  settings: term(),
  snapshot: term()
}

thread_turns_page()

@type thread_turns_page() :: %{
  turns: [CodexEx.AppServer.Turn.t()],
  next_cursor: binary() | nil
}

turn_interrupt_response()

@type turn_interrupt_response() ::
  %CodexEx.AppServer.Protocol.Generated.V2.TurnInterruptResponse{}

Functions

acknowledge_replay_gap_async(client, owner, through_sequence)

@spec acknowledge_replay_gap_async(pid(), pid(), non_neg_integer()) :: :ok

Queues replay-gap acknowledgement on the client generation that reported it.

archive_thread(client, thread_id)

@spec archive_thread(t(), binary()) :: :ok | {:error, term()}

broadcast_active_threads(client)

@spec broadcast_active_threads(t()) :: :ok | {:error, term()}

Publishes active threads already loaded by the connected app server.

build_initialize_params(params)

@spec build_initialize_params(map()) :: map()

Builds normalized app-server initialization parameters with required client capabilities.

call_mcp_tool(client, thread_id, server, tool, arguments)

@spec call_mcp_tool(GenServer.server(), binary(), binary(), binary(), map()) ::
  {:ok, mcp_server_tool_call_response()} | {:error, term()}

Calls one MCP tool in the active thread's server scope.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

clear_thread_goal(client, thread_id)

@spec clear_thread_goal(t(), binary()) :: :ok | {:error, term()}

connect(opts \\ [])

@spec connect(keyword()) :: GenServer.on_start()

disconnect(client, reason \\ :normal, timeout \\ 15000)

@spec disconnect(t(), term(), timeout()) :: :ok

fork_thread(client, thread_id, overrides \\ %{})

@spec fork_thread(t(), binary(), map()) ::
  {:ok, CodexEx.AppServer.Thread.t()} | {:error, term()}

get_thread_goal(client, thread_id)

@spec get_thread_goal(t(), binary()) ::
  {:ok, CodexEx.AppServer.ThreadGoal.t() | nil} | {:error, term()}

initialize_result(client)

@spec initialize_result(t()) ::
  {:ok, initialize_response()} | {:error, {:client_call_failed, term()}}

inject_thread_items(client, thread_id, items)

@spec inject_thread_items(t(), binary(), [map()]) :: :ok | {:error, term()}

Injects model-visible history items without starting a turn.

interrupt_turn(client, thread_id, turn_id, timeout \\ 15000)

@spec interrupt_turn(t(), binary(), binary(), timeout()) ::
  {:ok, turn_interrupt_response()} | {:error, term()}

list_experimental_features(client, params \\ %{})

@spec list_experimental_features(t(), map()) ::
  {:ok, experimental_feature_list_response()}
  | {:error, {:client_call_failed, term()} | term()}

list_hooks(client, cwds \\ [])

@spec list_hooks(t(), [binary()]) ::
  {:ok, hooks_list_response()}
  | {:error, {:client_call_failed, term()} | term()}

list_models(client)

@spec list_models(t()) ::
  {:ok, model_list_response()}
  | {:error, {:client_call_failed, term()} | term()}

list_skills(client, params \\ %{})

@spec list_skills(t(), map()) ::
  {:ok, skills_list_response()}
  | {:error, {:client_call_failed, term()} | term()}

list_thread_turns(client, thread_id, opts \\ [])

@spec list_thread_turns(t(), binary(), keyword()) ::
  {:ok, thread_turns_page()} | {:error, term()}

Lists one page of a paginated thread's turns in the requested server order.

list_threads(client, params \\ %{})

@spec list_threads(t(), map()) ::
  {:ok, thread_list_response()}
  | {:error, {:client_call_failed, term()} | term()}

pending_requests(client)

@spec pending_requests(t()) :: {:ok, [term()]} | {:error, term()}

read_config(client, cwd)

@spec read_config(t(), binary() | nil) ::
  {:ok, config_read_response()} | {:error, term()}

Reads the effective Codex configuration for a working directory.

read_mcp_resource(client, thread_id, server, uri)

@spec read_mcp_resource(GenServer.server(), binary(), binary(), binary()) ::
  {:ok, mcp_resource_read_response()} | {:error, term()}

Reads one MCP resource in the active thread's server scope.

read_thread(client, thread_id, opts \\ [])

@spec read_thread(t(), binary(), keyword()) ::
  {:ok, CodexEx.AppServer.ThreadSnapshot.t()} | {:error, term()}

register_request_handler(client, handler)

@spec register_request_handler(t(), (term() -> term()) | nil) ::
  :ok | {:error, {:client_call_failed, term()}}

reply_request(client, request_id, reply, timeout \\ 15000)

@spec reply_request(t(), term(), request_result(), timeout()) ::
  :ok | {:error, term()}

resume_thread(client, thread_id, overrides \\ %{}, opts \\ [])

@spec resume_thread(t(), binary(), map(), keyword()) ::
  {:ok, thread_response()} | {:error, term()}

revert_thread(client, thread_id, before_turn_id)

@spec revert_thread(t(), binary(), binary()) ::
  {:ok, CodexEx.AppServer.Thread.t()} | {:error, term()}

rollback_thread(client, thread_id, num_turns)

@spec rollback_thread(t(), binary(), non_neg_integer()) ::
  {:ok, CodexEx.AppServer.Thread.t()} | {:error, term()}

run(client, thread_id, input, opts \\ %{}, direct_stream? \\ true)

@spec run(t(), binary(), [map()], map(), boolean()) ::
  {:ok, CodexEx.AppServer.TurnStream.t()} | {:error, term()}

run_json(client, thread_id, text, output_schema, opts \\ %{})

@spec run_json(t(), binary(), binary(), map(), map()) ::
  {:ok, term()} | {:error, term()}

run_text(client, thread_id, text, opts \\ %{})

@spec run_text(t(), binary(), binary(), map()) :: {:ok, binary()} | {:error, term()}

set_experimental_feature_enablement(client, enablement)

@spec set_experimental_feature_enablement(t(), map()) ::
  {:ok, experimental_feature_enablement_set_response()}
  | {:error, {:client_call_failed, term()} | term()}

set_hook_enabled(client, hook_key, enabled?, timeout \\ 15000)

@spec set_hook_enabled(t(), binary(), boolean(), timeout()) ::
  {:ok, config_write_response()}
  | {:error, {:client_call_failed, term()} | term()}

set_thread_goal(client, thread_id, attrs)

@spec set_thread_goal(t(), binary(), map()) ::
  {:ok, CodexEx.AppServer.ThreadGoal.t()} | {:error, term()}

set_thread_name(client, thread_id, name)

@spec set_thread_name(t(), binary(), binary()) :: :ok | {:error, term()}

start_fuzzy_file_search_session(client, session_id, roots)

@spec start_fuzzy_file_search_session(t(), binary(), [binary()]) ::
  {:ok, :started} | {:error, {:client_call_failed, term()} | term()}

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

start_realtime(client, thread_id, sdp)

@spec start_realtime(t(), binary(), binary()) :: :ok | {:error, term()}

start_review(client, thread_id, target \\ %{"type" => "uncommittedChanges"}, delivery \\ nil, direct_stream? \\ true)

@spec start_review(t(), binary(), map(), binary() | nil, boolean()) ::
  {:ok, CodexEx.AppServer.TurnStream.t()} | {:error, term()}

start_review_request(client, params, timeout \\ 1_800_000)

@spec start_review_request(t(), map(), timeout()) ::
  {:ok, CodexEx.AppServer.Turn.t()} | {:error, term()}

start_thread(client, params \\ %{})

@spec start_thread(t(), map()) ::
  {:ok, CodexEx.AppServer.Thread.t()} | {:error, term()}

start_thread_compaction(client, thread_id)

@spec start_thread_compaction(t(), binary()) :: :ok | {:error, term()}

start_turn(client, thread_id, input, opts \\ %{})

@spec start_turn(t(), binary(), [map()], map()) ::
  {:ok, CodexEx.AppServer.Turn.t()} | {:error, term()}

Starts a turn and returns its initial protocol state without collecting its event stream.

start_turn_request(client, params, timeout \\ 1_800_000)

@spec start_turn_request(t(), map(), timeout()) ::
  {:ok, CodexEx.AppServer.Turn.t()} | {:error, term()}

steer_turn(client, thread_id, expected_turn_id, input, client_message_id, timeout \\ 15000)

@spec steer_turn(t(), binary(), binary(), [map()], binary(), timeout()) ::
  {:ok, binary()} | {:error, term()}

stop_fuzzy_file_search_session(client, session_id)

@spec stop_fuzzy_file_search_session(t(), binary()) ::
  :ok | {:error, {:client_call_failed, term()} | term()}

stop_realtime(client, thread_id)

@spec stop_realtime(t(), binary()) :: :ok | {:error, term()}

subscribe(client, subscriber \\ self())

@spec subscribe(t(), pid()) :: :ok | {:error, {:client_call_failed, term()}}

subscribe(client, subscriber, opts)

@spec subscribe(t(), pid(), keyword()) ::
  :ok | {:error, {:client_call_failed, term()}}

subscribe_thread_activity()

@spec subscribe_thread_activity() ::
  :ok | {:error, :pubsub_not_configured | {:already_registered, pid()}}

Subscribes the caller to active-thread signals from shared Codex clients.

Requires a Phoenix.PubSub server configured as config :codex_ex, :pubsub.

trust_hook(client, hook_key, current_hash, timeout \\ 15000)

@spec trust_hook(t(), binary(), binary(), timeout()) ::
  {:ok, config_write_response()}
  | {:error, {:client_call_failed, term()} | term()}

unarchive_thread(client, thread_id)

@spec unarchive_thread(t(), binary()) ::
  {:ok, CodexEx.AppServer.Thread.t()} | {:error, term()}

unsubscribe(client, subscriber \\ self())

@spec unsubscribe(t(), pid()) :: :ok | {:error, {:client_call_failed, term()}}

unsubscribe_thread(client, thread_id)

@spec unsubscribe_thread(t(), binary()) :: {:ok, struct()} | {:error, term()}

Unsubscribes the app-server connection from a remote thread.

update_fuzzy_file_search_session(client, session_id, query)

@spec update_fuzzy_file_search_session(t(), binary(), binary()) ::
  {:ok, :updated} | {:error, {:client_call_failed, term()} | term()}