Current section
Files
Jump to
Current section
Files
lib/jido_ai/reasoning/react/strategy.ex
defmodule Jido.AI.Reasoning.ReAct.Strategy do
@moduledoc """
ReAct strategy delegated to an internal per-parent worker agent.
The parent strategy remains the public orchestration boundary (`ask/await/ask_sync`),
while runtime execution is delegated to a lazily spawned child worker tagged
`:react_worker`.
## Delegation Model
1. Parent receives `"ai.react.query"` and prepares runtime config/context.
2. Parent lazily spawns internal worker on first request (if needed).
3. Parent emits `"ai.react.worker.start"` to worker.
4. Worker streams `Jido.AI.Reasoning.ReAct` events and emits `"ai.react.worker.event"` to parent.
5. Parent applies runtime events to parent state and emits external lifecycle/LLM/tool signals.
## Worker Lifecycle
- Child tag is fixed as `:react_worker`.
- Single active run is enforced (`:reject` busy policy).
- Worker crash during active request marks the request failed.
- No machine-driven fallback path is retained.
## Trace Retention
Parent stores per-request runtime event history in `request_traces` with a hard cap:
%{request_id => %{events: [event, ...], truncated?: boolean()}}
Once 2000 events are stored for a request, `truncated?` is set to `true`
and new events are not appended.
"""
use Jido.Agent.Strategy
alias Jido.Agent
alias Jido.Agent.Directive, as: AgentDirective
alias Jido.Agent.Strategy.State, as: StratState
alias Jido.AI.Observe
alias Jido.AI.Output
alias Jido.AI.Query
alias Jido.AI.Directive
alias Jido.AI.Effects
alias Jido.AI.Request.Stream, as: RequestStream
alias Jido.AI.Reasoning.ReAct.PendingInput
alias Jido.AI.Reasoning.ReAct.RequestTransformer
alias Jido.AI.Reasoning.ReAct.State, as: ReActState
alias Jido.AI.Reasoning.ReAct.Config, as: ReActRuntimeConfig
alias Jido.AI.Reasoning.ReAct.ToolSelection
alias Jido.AI.Signal
alias Jido.AI.Error
alias Jido.AI.Reasoning.Helpers
alias Jido.AI.Context, as: AIContext
alias Jido.AI.ToolAdapter
alias Jido.AI.Turn
alias Jido.Thread
alias Jido.Thread.Agent, as: ThreadAgent
@type config :: %{
tools: [module()],
reqllm_tools: [ReqLLM.Tool.t()],
actions_by_name: %{String.t() => module()},
request_transformer: module() | nil,
system_prompt: String.t(),
model: String.t(),
max_iterations: pos_integer(),
max_tokens: pos_integer(),
streaming: boolean(),
base_tool_context: map(),
base_req_http_options: list(),
base_llm_opts: keyword(),
provider_opt_keys_by_string: %{optional(String.t()) => atom()},
request_policy: :reject,
stream_timeout_ms: non_neg_integer(),
stream_receive_timeout_ms: pos_integer(),
stream_timeout_ms: non_neg_integer(),
stream_receive_timeout_ms: pos_integer(),
tool_timeout_ms: pos_integer(),
tool_max_retries: non_neg_integer(),
tool_retry_backoff_ms: non_neg_integer(),
effect_policy: map(),
output: Output.t() | nil,
observability: map(),
runtime_adapter: true,
runtime_task_supervisor: pid() | atom() | nil,
agent_id: String.t() | nil
}
@default_model :fast
@default_max_iterations 10
@default_max_tokens 4_096
@request_trace_cap 2000
@applied_context_ops_cap 128
@default_context_ref "default"
@worker_tag :react_worker
@source "/ai/react/strategy"
@reqllm_generation_opt_keys_by_string ReqLLM.Provider.Options.all_generation_keys()
|> Enum.map(&{Atom.to_string(&1), &1})
|> Map.new()
@default_system_prompt """
You are a helpful AI assistant using the ReAct (Reason-Act) pattern.
When you need to perform an action, use the available tools.
When you have enough information to answer, provide your final answer directly.
Think step by step and explain your reasoning.
"""
@start :ai_react_start
@llm_result :ai_react_llm_result
@tool_result :ai_react_tool_result
@llm_partial :ai_react_llm_partial
@cancel :ai_react_cancel
@steer :ai_react_steer
@inject :ai_react_inject
@request_error :ai_react_request_error
@register_tool :ai_react_register_tool
@unregister_tool :ai_react_unregister_tool
@set_tool_context :ai_react_set_tool_context
@set_system_prompt :ai_react_set_system_prompt
@context_modify :ai_react_context_modify
@runtime_event :ai_react_runtime_event
@worker_event :ai_react_worker_event
@worker_child_started :ai_react_worker_child_started
@worker_child_exit :ai_react_worker_child_exit
@doc "Returns the action atom for starting a ReAct conversation."
@spec start_action() :: :ai_react_start
def start_action, do: @start
@doc "Returns the legacy action atom for handling LLM results (no-op in delegated mode)."
@spec llm_result_action() :: :ai_react_llm_result
def llm_result_action, do: @llm_result
@doc "Returns the action atom for registering a tool dynamically."
@spec register_tool_action() :: :ai_react_register_tool
def register_tool_action, do: @register_tool
@doc "Returns the action atom for unregistering a tool."
@spec unregister_tool_action() :: :ai_react_unregister_tool
def unregister_tool_action, do: @unregister_tool
@doc "Returns the legacy action atom for handling tool results (no-op in delegated mode)."
@spec tool_result_action() :: :ai_react_tool_result
def tool_result_action, do: @tool_result
@doc "Returns the legacy action atom for handling streaming deltas (no-op in delegated mode)."
@spec llm_partial_action() :: :ai_react_llm_partial
def llm_partial_action, do: @llm_partial
@doc "Returns the action atom for request cancellation."
@spec cancel_action() :: :ai_react_cancel
def cancel_action, do: @cancel
@doc "Returns the action atom for steering an active ReAct request."
@spec steer_action() :: :ai_react_steer
def steer_action, do: @steer
@doc "Returns the action atom for injecting user-style input into an active ReAct request."
@spec inject_action() :: :ai_react_inject
def inject_action, do: @inject
@doc "Returns the action atom for handling request rejections."
@spec request_error_action() :: :ai_react_request_error
def request_error_action, do: @request_error
@doc "Returns the action atom for updating tool context."
@spec set_tool_context_action() :: :ai_react_set_tool_context
def set_tool_context_action, do: @set_tool_context
@doc "Returns the action atom for updating the base system prompt."
@spec set_system_prompt_action() :: :ai_react_set_system_prompt
def set_system_prompt_action, do: @set_system_prompt
@doc "Returns the canonical action atom for context lifecycle operations."
@spec context_modify_action() :: :ai_react_context_modify
def context_modify_action, do: @context_modify
@doc "Returns the legacy action atom for direct runtime stream events (no-op in delegated mode)."
@spec runtime_event_action() :: :ai_react_runtime_event
def runtime_event_action, do: @runtime_event
@action_specs %{
@start => %{
schema:
Zoi.object(%{
query: Query.schema(),
request_id: Zoi.string() |> Zoi.optional(),
tool_context: Zoi.map() |> Zoi.optional(),
tools: Zoi.any() |> Zoi.optional(),
allowed_tools: Zoi.list(Zoi.string()) |> Zoi.optional(),
request_transformer: Zoi.atom() |> Zoi.optional(),
stream_to: Zoi.any() |> Zoi.optional(),
stream_receive_timeout_ms: Zoi.integer() |> Zoi.optional(),
stream_timeout_ms: Zoi.integer() |> Zoi.optional(),
req_http_options: Zoi.list(Zoi.any()) |> Zoi.optional(),
llm_opts: Zoi.any() |> Zoi.optional(),
max_iterations: Zoi.integer() |> Zoi.optional(),
output: Zoi.any() |> Zoi.optional(),
extra_refs: Zoi.map() |> Zoi.optional()
}),
doc: "Start a delegated ReAct conversation with a user query",
name: "ai.react.start"
},
@cancel => %{
schema:
Zoi.object(%{
request_id: Zoi.string() |> Zoi.optional(),
reason: Zoi.atom() |> Zoi.default(:user_cancelled)
}),
doc: "Cancel an in-flight ReAct request",
name: "ai.react.cancel"
},
@steer => %{
schema:
Zoi.object(%{
content: Zoi.string(),
expected_request_id: Zoi.string() |> Zoi.optional(),
source: Zoi.any() |> Zoi.optional(),
extra_refs: Zoi.map() |> Zoi.optional()
}),
doc: "Steer an active delegated ReAct request with additional user input",
name: "ai.react.steer"
},
@inject => %{
schema:
Zoi.object(%{
content: Zoi.string(),
expected_request_id: Zoi.string() |> Zoi.optional(),
source: Zoi.any() |> Zoi.optional(),
extra_refs: Zoi.map() |> Zoi.optional()
}),
doc: "Inject user-style input into an active delegated ReAct request",
name: "ai.react.inject"
},
@request_error => %{
schema:
Zoi.object(%{
request_id: Zoi.string(),
reason: Zoi.atom(),
message: Zoi.string()
}),
doc: "Handle request rejection event",
name: "ai.react.request_error"
},
@register_tool => %{
schema: Zoi.object(%{tool_module: Zoi.atom()}),
doc: "Register a new tool dynamically at runtime",
name: "ai.react.register_tool"
},
@unregister_tool => %{
schema: Zoi.object(%{tool_name: Zoi.string()}),
doc: "Unregister a tool by name",
name: "ai.react.unregister_tool"
},
@set_tool_context => %{
schema: Zoi.object(%{tool_context: Zoi.map()}),
doc: "Update the persistent base tool context",
name: "ai.react.set_tool_context"
},
@set_system_prompt => %{
schema: Zoi.object(%{system_prompt: Zoi.string()}),
doc: "Update the persistent base system prompt",
name: "ai.react.set_system_prompt"
},
@context_modify => %{
schema:
Zoi.object(%{
op_id: Zoi.string() |> Zoi.optional(),
context_ref: Zoi.string() |> Zoi.optional(),
operation: Zoi.map()
}),
doc: "Modify context lifecycle using canonical operation envelopes",
name: "ai.react.context.modify"
},
@worker_event => %{
schema: Zoi.object(%{request_id: Zoi.string(), event: Zoi.map()}),
doc: "Handle delegated ReAct runtime event envelopes",
name: "ai.react.worker.event"
},
@worker_child_started => %{
schema:
Zoi.object(%{
parent_id: Zoi.string() |> Zoi.optional(),
child_id: Zoi.string() |> Zoi.optional(),
child_module: Zoi.any() |> Zoi.optional(),
tag: Zoi.any(),
pid: Zoi.any(),
meta: Zoi.map() |> Zoi.default(%{})
}),
doc: "Handle worker child started lifecycle signal",
name: "jido.agent.child.started"
},
@worker_child_exit => %{
schema:
Zoi.object(%{
tag: Zoi.any(),
pid: Zoi.any(),
reason: Zoi.any()
}),
doc: "Handle worker child exit lifecycle signal",
name: "jido.agent.child.exit"
},
# Legacy compatibility actions kept as no-op adapters.
@llm_result => %{
schema: Zoi.object(%{call_id: Zoi.string(), result: Zoi.any()}),
doc: "Legacy no-op in delegated ReAct mode",
name: "ai.react.llm_result"
},
@tool_result => %{
schema: Zoi.object(%{call_id: Zoi.string(), tool_name: Zoi.string(), result: Zoi.any()}),
doc: "Legacy no-op in delegated ReAct mode",
name: "ai.react.tool_result"
},
@llm_partial => %{
schema:
Zoi.object(%{
call_id: Zoi.string(),
delta: Zoi.string(),
chunk_type: Zoi.atom() |> Zoi.default(:content)
}),
doc: "Legacy no-op in delegated ReAct mode",
name: "ai.react.llm_partial"
},
@runtime_event => %{
schema: Zoi.object(%{request_id: Zoi.string(), event: Zoi.map()}),
doc: "Legacy no-op in delegated ReAct mode",
name: "ai.react.runtime_event"
}
}
@impl true
def action_spec(action), do: Map.get(@action_specs, action)
@impl true
def signal_routes(_ctx) do
[
{"ai.react.query", {:strategy_cmd, @start}},
{"ai.react.cancel", {:strategy_cmd, @cancel}},
{"ai.react.steer", {:strategy_cmd, @steer}},
{"ai.react.inject", {:strategy_cmd, @inject}},
{"ai.request.error", {:strategy_cmd, @request_error}},
{"ai.react.register_tool", {:strategy_cmd, @register_tool}},
{"ai.react.unregister_tool", {:strategy_cmd, @unregister_tool}},
{"ai.react.set_tool_context", {:strategy_cmd, @set_tool_context}},
{"ai.react.set_system_prompt", {:strategy_cmd, @set_system_prompt}},
{"ai.react.context.modify", {:strategy_cmd, @context_modify}},
{"ai.react.worker.event", {:strategy_cmd, @worker_event}},
{"jido.agent.child.started", {:strategy_cmd, @worker_child_started}},
{"jido.agent.child.exit", {:strategy_cmd, @worker_child_exit}},
{"ai.llm.delta", Jido.Actions.Control.Noop},
{"ai.llm.response", Jido.Actions.Control.Noop},
{"ai.tool.result", Jido.Actions.Control.Noop},
{"ai.request.started", Jido.Actions.Control.Noop},
{"ai.request.completed", Jido.Actions.Control.Noop},
{"ai.request.failed", Jido.Actions.Control.Noop},
{"ai.usage", Jido.Actions.Control.Noop}
]
end
@impl true
def snapshot(%Agent{} = agent, _ctx) do
state = StratState.get(agent, %{})
status = snapshot_status(state[:status])
config = state[:config] || %{}
%Jido.Agent.Strategy.Snapshot{
status: status,
done?: status in [:success, :failure],
result: state[:result],
details: build_snapshot_details(state, config)
}
end
defp snapshot_status(:completed), do: :success
defp snapshot_status(:error), do: :failure
defp snapshot_status(:idle), do: :idle
defp snapshot_status(_), do: :running
defp build_snapshot_details(state, config) do
conversation = state |> snapshot_context(config) |> AIContext.to_messages()
trace_summary =
state
|> Map.get(:request_traces, %{})
|> Enum.map(fn {request_id, trace} ->
{request_id, %{events: length(trace.events), truncated?: trace.truncated?}}
end)
|> Map.new()
%{
phase: state[:status],
iteration: state[:iteration],
termination_reason: state[:termination_reason],
streaming_text: state[:streaming_text],
streaming_thinking: state[:streaming_thinking],
thinking_trace: state[:thinking_trace],
usage: state[:usage],
output: state[:output],
duration_ms: calculate_duration(state[:started_at]),
tool_calls: format_tool_calls(state[:pending_tool_calls] || []),
tool_results: state[:tool_results] || [],
current_llm_call_id: state[:current_llm_call_id],
active_request_id: state[:active_request_id],
checkpoint_token: state[:checkpoint_token],
cancel_reason: state[:cancel_reason],
worker_pid: state[:react_worker_pid],
worker_status: state[:react_worker_status],
trace_summary: trace_summary,
model: config[:model],
request_transformer:
case config[:request_transformer] do
module when is_atom(module) -> Atom.to_string(module)
_ -> nil
end,
max_iterations: config[:max_iterations],
max_tokens: config[:max_tokens],
streaming: config[:streaming],
stream_receive_timeout_ms: config[:stream_receive_timeout_ms],
request_policy: config[:request_policy],
runtime_adapter: true,
tool_timeout_ms: config[:tool_timeout_ms],
tool_max_retries: config[:tool_max_retries],
tool_retry_backoff_ms: config[:tool_retry_backoff_ms],
available_tools: Enum.map(Map.get(config, :tools, []), & &1.name()),
conversation: conversation
}
|> Enum.reject(fn {_k, v} -> is_nil(v) or v == "" or v == %{} or v == [] end)
|> Map.new()
end
defp calculate_duration(nil), do: nil
defp calculate_duration(started_at), do: System.monotonic_time(:millisecond) - started_at
defp format_tool_calls([]), do: []
defp format_tool_calls(pending_tool_calls) do
Enum.map(pending_tool_calls, fn tc ->
result = fetch_map_value(tc, :result)
%{
id: fetch_map_value(tc, :id),
name: fetch_map_value(tc, :name),
arguments: fetch_map_value(tc, :arguments) || %{},
status: if(result == nil, do: :running, else: :completed),
result: result
}
end)
end
defp completed_tool_result_entry(state, tool_call_id, tool_name, tool_result) do
%{
id: tool_call_id,
name: tool_name,
arguments: pending_tool_arguments(state, tool_call_id),
result: tool_result
}
end
defp put_completed_tool_result(state, tool_result_entry) do
existing =
case Map.get(state, :tool_results, []) do
results when is_list(results) -> results
_ -> []
end
updated =
case Enum.split_while(existing, &(fetch_map_value(&1, :id) != tool_result_entry[:id])) do
{_before, []} ->
existing ++ [tool_result_entry]
{before, [_old | after_results]} ->
before ++ [tool_result_entry | after_results]
end
Map.put(state, :tool_results, updated)
end
defp pending_tool_arguments(state, tool_call_id) do
state
|> Map.get(:pending_tool_calls, [])
|> Enum.find_value(%{}, fn call ->
if fetch_map_value(call, :id) == tool_call_id do
fetch_map_value(call, :arguments) || %{}
else
false
end
end)
end
@impl true
def init(%Agent{} = agent, ctx) do
config = build_config(agent, ctx)
active_context_ref = initial_active_context_ref(agent)
base_context = initial_context(agent, config)
projected_context = project_context_from_core_thread(agent, active_context_ref, base_context)
projection_cursor_seq = core_thread_last_seq(agent)
state =
%{
status: :idle,
iteration: 0,
context: projected_context,
run_context: nil,
active_context_ref: active_context_ref,
pending_tool_calls: [],
tool_results: [],
final_answer: nil,
result: nil,
current_llm_call_id: nil,
termination_reason: nil,
run_tool_context: %{},
run_req_http_options: [],
run_llm_opts: [],
active_request_id: nil,
pending_input_server: nil,
last_pending_input_control: nil,
cancel_reason: nil,
usage: %{},
started_at: nil,
streaming_text: "",
streaming_thinking: "",
thinking_trace: [],
checkpoint_token: nil,
pending_context_op: nil,
applied_context_ops: [],
projection_cursor_seq: projection_cursor_seq,
request_traces: %{},
react_worker_pid: nil,
react_worker_status: :missing,
pending_worker_start: nil,
agent_id: Map.get(agent, :id)
}
|> Helpers.apply_to_state([Helpers.update_config(config)])
agent = put_strategy_state(agent, state)
{agent, []}
end
@impl true
def cmd(%Agent{} = agent, instructions, ctx) do
{agent, directives_rev} =
Enum.reduce(instructions, {agent, []}, fn instruction, {acc_agent, acc_directives} ->
case process_instruction(acc_agent, instruction, ctx) do
{new_agent, new_directives} ->
{new_agent, Enum.reverse(new_directives, acc_directives)}
:noop ->
{acc_agent, acc_directives}
end
end)
{agent, Enum.reverse(directives_rev)}
end
defp process_instruction(
agent,
%Jido.Instruction{action: action, params: params} = instruction,
ctx
) do
case normalize_action(action) do
@start ->
state = StratState.get(agent, %{})
config = state[:config] || %{}
provider_opt_keys_by_string = config[:provider_opt_keys_by_string] || %{}
run_tool_context = Map.get(params, :tool_context) || %{}
run_req_http_options =
params |> Map.get(:req_http_options, []) |> normalize_req_http_options()
run_llm_opts =
params |> Map.get(:llm_opts, []) |> normalize_llm_opts(provider_opt_keys_by_string)
agent
|> set_run_tool_context(run_tool_context)
|> set_run_req_http_options(run_req_http_options)
|> set_run_llm_opts(run_llm_opts)
|> process_start(params)
@cancel ->
process_cancel(agent, params)
@steer ->
process_pending_input_control(agent, params, :steer)
@inject ->
process_pending_input_control(agent, params, :inject)
@request_error ->
process_request_error(agent, params)
@register_tool ->
process_register_tool(agent, params)
@unregister_tool ->
process_unregister_tool(agent, params)
@set_tool_context ->
process_set_tool_context(agent, params)
@set_system_prompt ->
process_set_system_prompt(agent, params)
@context_modify ->
process_context_modify(agent, params)
@worker_event ->
process_worker_event(agent, params)
@worker_child_started ->
process_worker_child_started(agent, params)
@worker_child_exit ->
process_worker_child_exit(agent, params)
# Legacy compatibility no-ops in delegated mode.
legacy when legacy in [@llm_result, @tool_result, @llm_partial, @runtime_event] ->
{agent, []}
_ ->
Helpers.maybe_execute_action_instruction(agent, instruction, ctx)
end
end
defp process_start(agent, %{query: query} = params) when is_binary(query) or is_list(query) do
state = StratState.get(agent, %{})
config = state[:config] || %{}
request_id = Map.get(params, :request_id, generate_call_id())
run_id = request_id
if busy?(state, config) do
directive =
Directive.EmitRequestError.new!(%{
request_id: request_id,
reason: :busy,
message: "Agent is busy (status: #{state[:status]})"
})
{agent, [directive]}
else
run_tool_context = Map.get(state, :run_tool_context, %{})
effective_tool_context = Map.merge(config[:base_tool_context] || %{}, run_tool_context)
run_req_http_options = Map.get(state, :run_req_http_options, [])
base_req_http_options = normalize_req_http_options(config[:base_req_http_options])
effective_req_http_options = base_req_http_options ++ run_req_http_options
run_llm_opts = Map.get(state, :run_llm_opts, [])
provider_opt_keys_by_string = config[:provider_opt_keys_by_string] || %{}
base_llm_opts = normalize_llm_opts(config[:base_llm_opts], provider_opt_keys_by_string)
effective_llm_opts = Keyword.merge(base_llm_opts, run_llm_opts)
state_snapshot = normalize_map_opt(Map.get(agent, :state, %{}))
state = PendingInput.stop(state)
with {:ok, effective_tools} <- resolve_request_tools(config, params),
{:ok, request_transformer} <- resolve_request_transformer(config, params),
{:ok, output} <- resolve_request_output(config, params),
{:ok, pending_input_server} <- PendingInput.start(request_id) do
runtime_config =
runtime_config_from_strategy(config,
req_http_options: effective_req_http_options,
llm_opts: effective_llm_opts,
tools: effective_tools,
stream_receive_timeout_ms: Map.get(params, :stream_receive_timeout_ms),
stream_timeout_ms: Map.get(params, :stream_timeout_ms),
max_iterations: Map.get(params, :max_iterations),
request_transformer: request_transformer,
output: output,
pending_input_server: pending_input_server
)
base_context = strategy_context(state, config)
extra_refs = Map.get(params, :extra_refs, %{})
run_context = AIContext.append_user(base_context, query, refs: normalize_refs(extra_refs))
runtime_state = runtime_state_from_context(run_context, query, request_id, run_id)
context_ref = Map.get(state, :active_context_ref, @default_context_ref)
{agent, state} =
append_ai_message_event(
agent,
state,
context_ref,
%{role: :user, content: query},
request_id,
run_id,
nil,
extra_refs
)
worker_start_payload = %{
request_id: request_id,
run_id: run_id,
query: query,
config: runtime_config,
state: runtime_state,
task_supervisor: config[:runtime_task_supervisor],
context:
Map.merge(effective_tool_context, %{
state: state_snapshot,
request_id: request_id,
run_id: run_id,
agent_id: state[:agent_id] || Map.get(agent, :id),
observability: config[:observability] || %{},
effect_policy: config[:effect_policy] || Effects.default_policy()
})
}
{new_state, directives} = ensure_worker_start(state, worker_start_payload)
new_state =
new_state
|> Map.put(:status, :awaiting_llm)
|> Map.put(:active_request_id, request_id)
|> Map.put(:pending_input_server, pending_input_server)
|> Map.put(:last_pending_input_control, nil)
|> Map.put(:current_llm_call_id, nil)
|> Map.put(:iteration, 1)
|> Map.put(:result, nil)
|> Map.put(:output, %{})
|> Map.put(:termination_reason, nil)
|> Map.put(:started_at, System.monotonic_time(:millisecond))
|> Map.put(:streaming_text, "")
|> Map.put(:streaming_thinking, "")
|> Map.put(:pending_tool_calls, [])
|> Map.put(:tool_results, [])
|> Map.put(:cancel_reason, nil)
|> Map.put(:checkpoint_token, nil)
|> Map.put(:run_context, run_context)
|> ensure_request_trace(request_id)
{put_strategy_state(agent, new_state), directives}
else
{:error, :pending_input_unavailable} ->
directive =
Directive.EmitRequestError.new!(%{
request_id: request_id,
reason: :runtime,
message: "Failed to start pending input queue"
})
{agent, [directive]}
{:error, reason, message} ->
directive =
Directive.EmitRequestError.new!(%{
request_id: request_id,
reason: reason,
message: message
})
{agent, [directive]}
end
end
end
defp process_start(agent, _params), do: {agent, []}
defp process_pending_input_control(agent, %{content: content} = params, kind)
when is_binary(content) and kind in [:steer, :inject] do
state = StratState.get(agent, %{})
case PendingInput.accept_control(state, params, kind) do
{:ok, new_state} ->
{put_strategy_state(agent, new_state), []}
{:error, new_state} ->
{put_strategy_state(agent, new_state), []}
end
end
defp process_pending_input_control(agent, _params, _kind), do: {agent, []}
defp process_cancel(agent, params) do
state = StratState.get(agent, %{})
request_id = Map.get(params, :request_id, state[:active_request_id])
reason = Map.get(params, :reason, :user_cancelled)
should_cancel? =
is_binary(request_id) and request_id == state[:active_request_id] and
is_pid(state[:react_worker_pid]) and
Process.alive?(state[:react_worker_pid])
directives =
if should_cancel? do
[
AgentDirective.emit_to_pid(
worker_cancel_signal(request_id, reason),
state[:react_worker_pid]
)
]
else
[]
end
new_state =
if should_cancel? do
Map.put(state, :cancel_reason, reason)
else
state
end
{put_strategy_state(agent, new_state), directives}
end
defp process_request_error(agent, %{request_id: request_id, reason: reason, message: message}) do
state = StratState.get(agent, %{})
new_state =
Map.put(state, :last_request_error, %{
request_id: request_id,
reason: reason,
message: message
})
{put_strategy_state(agent, new_state), []}
end
defp process_request_error(agent, _params), do: {agent, []}
defp process_register_tool(agent, %{tool_module: module}) when is_atom(module) do
state = StratState.get(agent, %{})
config = state[:config]
new_tools = [module | config[:tools]] |> Enum.uniq()
new_actions_by_name = Map.put(config[:actions_by_name], module.name(), module)
new_reqllm_tools = ToolAdapter.from_actions(new_tools)
new_state =
Helpers.apply_to_state(
state,
Helpers.update_tools_config(new_tools, new_actions_by_name, new_reqllm_tools)
)
{put_strategy_state(agent, new_state), []}
end
defp process_register_tool(agent, _params), do: {agent, []}
defp process_unregister_tool(agent, %{tool_name: tool_name}) when is_binary(tool_name) do
state = StratState.get(agent, %{})
config = state[:config]
new_tools = Enum.reject(config[:tools], fn m -> m.name() == tool_name end)
new_actions_by_name = Map.delete(config[:actions_by_name], tool_name)
new_reqllm_tools = ToolAdapter.from_actions(new_tools)
new_state =
Helpers.apply_to_state(
state,
Helpers.update_tools_config(new_tools, new_actions_by_name, new_reqllm_tools)
)
{put_strategy_state(agent, new_state), []}
end
defp process_unregister_tool(agent, _params), do: {agent, []}
defp process_set_tool_context(agent, %{tool_context: new_context}) when is_map(new_context) do
state = StratState.get(agent, %{})
new_state =
Helpers.apply_to_state(state, [
Helpers.set_config_field(:base_tool_context, new_context)
])
{put_strategy_state(agent, new_state), []}
end
defp process_set_tool_context(agent, _params), do: {agent, []}
defp process_set_system_prompt(agent, %{system_prompt: prompt}) when is_binary(prompt) do
state = StratState.get(agent, %{})
run_context = Map.get(state, :run_context)
base_context = strategy_context(state, state[:config] || %{})
new_state =
Helpers.apply_to_state(state, [
Helpers.set_config_field(:system_prompt, prompt)
])
|> Map.put(:context, %{base_context | system_prompt: prompt})
|> then(fn updated ->
if match?(%AIContext{}, run_context) do
Map.put(updated, :run_context, %{run_context | system_prompt: prompt})
else
updated
end
end)
{put_strategy_state(agent, new_state), []}
end
defp process_set_system_prompt(agent, _params), do: {agent, []}
defp process_context_modify(agent, params) when is_map(params) do
state = StratState.get(agent, %{})
case normalize_context_operation(params, state) do
{:ok, context_op} ->
if active_run?(state) do
{put_strategy_state(agent, Map.put(state, :pending_context_op, context_op)), []}
else
{agent, new_state} = apply_context_op(agent, state, context_op)
{put_strategy_state(agent, new_state), []}
end
:error ->
{agent, []}
end
end
defp process_context_modify(agent, _params), do: {agent, []}
defp initial_context(%Agent{} = agent, config) do
state = agent.state || %{}
case Map.fetch(state, :context) do
{:ok, nil} ->
AIContext.new(system_prompt: config[:system_prompt])
{:ok, source_context} ->
case AIContext.coerce(source_context) do
{:ok, %AIContext{system_prompt: nil} = context} ->
%{context | system_prompt: config[:system_prompt]}
{:ok, %AIContext{} = context} ->
context
:error ->
raise ArgumentError,
"invalid initial_state[:context]; expected Jido.AI.Context"
end
:error ->
if legacy_thread_context?(Map.get(state, :thread)) do
raise ArgumentError,
"initial_state[:thread] is no longer supported for AI context; use initial_state[:context] with Jido.AI.Context"
else
AIContext.new(system_prompt: config[:system_prompt])
end
end
end
defp legacy_thread_context?(%{} = value) do
has_entries_key? = Map.has_key?(value, :entries) or Map.has_key?(value, "entries")
has_system_prompt_key? =
Map.has_key?(value, :system_prompt) or Map.has_key?(value, "system_prompt")
has_entries_key? and has_system_prompt_key?
end
defp legacy_thread_context?(_), do: false
defp active_run?(state) do
is_binary(state[:active_request_id]) and state[:status] in [:awaiting_llm, :awaiting_tool]
end
defp apply_context_op(agent, state, %{op_id: op_id} = context_op) do
if context_op_applied?(state, op_id) do
{agent, Map.put(state, :pending_context_op, nil)}
else
do_apply_context_op(agent, state, context_op)
end
end
defp do_apply_context_op(agent, state, %{
op_id: op_id,
context_ref: context_ref,
operation: %{type: :replace} = operation
}) do
context = operation.result_context
state =
state
|> maybe_sync_config_prompt(context)
|> Map.put(:context, context)
|> Map.put(:active_context_ref, context_ref)
|> Map.put(:pending_context_op, nil)
|> record_applied_context_op(op_id)
{agent, state} =
append_ai_context_operation_event(agent, state, %{
op_id: op_id,
context_ref: context_ref,
operation: operation
})
{agent, state}
end
defp do_apply_context_op(agent, state, %{
op_id: op_id,
context_ref: context_ref,
operation: %{type: :switch} = operation
}) do
projected_context =
project_context_from_core_thread(
agent,
context_ref,
fresh_projection_context(state[:config] || %{})
)
state =
state
|> Map.put(:active_context_ref, context_ref)
|> Map.put(:context, projected_context)
|> Map.put(:pending_context_op, nil)
|> record_applied_context_op(op_id)
{agent, state} =
append_ai_context_operation_event(agent, state, %{
op_id: op_id,
context_ref: context_ref,
operation: operation
})
{agent, state}
end
defp maybe_sync_config_prompt(state, %AIContext{system_prompt: prompt})
when is_binary(prompt) do
Helpers.apply_to_state(state, [
Helpers.set_config_field(:system_prompt, prompt)
])
end
defp maybe_sync_config_prompt(state, _context), do: state
defp maybe_apply_pending_context_op(agent, state) do
case Map.get(state, :pending_context_op) do
%{operation: %{}} = context_op -> apply_context_op(agent, state, context_op)
_ -> {agent, state}
end
end
defp maybe_apply_pending_context_op_after_terminal(agent, state, kind)
when kind in [:request_completed, :request_failed, :request_cancelled] do
maybe_apply_pending_context_op(agent, state)
end
defp maybe_apply_pending_context_op_after_terminal(agent, state, _kind), do: {agent, state}
defp normalize_context_operation(params, state) do
operation = fetch_map_value(params, :operation)
with true <- is_map(operation),
{:ok, type} <- normalize_context_operation_type(fetch_map_value(operation, :type)),
{:ok, reason} <- normalize_context_operation_reason(fetch_map_value(operation, :reason)),
{:ok, context_ref} <- normalize_context_ref(params, operation, state),
{:ok, op_id} <- normalize_context_op_id(params, operation) do
base =
%{
op_id: op_id,
context_ref: context_ref,
operation: %{
type: type,
reason: reason,
base_seq: normalize_optional_integer(fetch_map_value(operation, :base_seq)),
meta: normalize_optional_map(fetch_map_value(operation, :meta))
}
}
normalize_context_operation_payload(base, operation)
else
_ -> :error
end
end
defp normalize_context_operation_payload(base, operation) do
case base.operation.type do
:replace ->
result_payload =
fetch_map_value(operation, :result_context) ||
fetch_map_value(operation, :context)
case AIContext.coerce(result_payload) do
{:ok, context} ->
{:ok, put_in(base, [:operation, :result_context], context)}
:error ->
:error
end
:switch ->
{:ok, put_in(base, [:operation, :result_context], nil)}
end
end
defp normalize_context_operation_type(type) when type in [:replace, :switch], do: {:ok, type}
defp normalize_context_operation_type("replace"), do: {:ok, :replace}
defp normalize_context_operation_type("switch"), do: {:ok, :switch}
defp normalize_context_operation_type(_), do: :error
defp normalize_context_operation_reason(reason)
when reason in [:manual, :restore, :compaction, :system],
do: {:ok, reason}
defp normalize_context_operation_reason("manual"), do: {:ok, :manual}
defp normalize_context_operation_reason("restore"), do: {:ok, :restore}
defp normalize_context_operation_reason("compaction"), do: {:ok, :compaction}
defp normalize_context_operation_reason("system"), do: {:ok, :system}
defp normalize_context_operation_reason(nil), do: {:ok, :manual}
defp normalize_context_operation_reason(_), do: :error
defp normalize_context_ref(params, operation, state) do
ref =
fetch_map_value(params, :context_ref) ||
fetch_map_value(operation, :context_ref) ||
Map.get(state, :active_context_ref) ||
@default_context_ref
if is_binary(ref) and ref != "", do: {:ok, ref}, else: :error
end
defp normalize_context_op_id(params, operation) do
op_id =
fetch_map_value(params, :op_id) ||
fetch_map_value(operation, :op_id) ||
fetch_map_value(params, :signal_id) ||
"op_#{Jido.Util.generate_id()}"
if is_binary(op_id) and op_id != "", do: {:ok, op_id}, else: :error
end
defp context_op_applied?(state, op_id) when is_binary(op_id) do
op_id in Map.get(state, :applied_context_ops, [])
end
defp context_op_applied?(_state, _op_id), do: false
defp record_applied_context_op(state, op_id) when is_binary(op_id) do
existing = Map.get(state, :applied_context_ops, [])
updated =
[op_id | Enum.reject(existing, &(&1 == op_id))] |> Enum.take(@applied_context_ops_cap)
Map.put(state, :applied_context_ops, updated)
end
defp record_applied_context_op(state, _op_id), do: state
defp maybe_append_ai_message_event_from_runtime(agent, state, event) do
context_ref = Map.get(state, :active_context_ref, @default_context_ref)
request_id = event_field(event, :request_id, state[:active_request_id])
run_id = event_field(event, :run_id, request_id)
signal_id = event_field(event, :id)
data = event_field(event, :data, %{})
case event_kind(event) do
:input_injected ->
refs =
data
|> event_field(:refs, %{})
|> normalize_event_message_refs(runtime_event_refs(event, request_id))
|> maybe_put_ref(:source, event_field(data, :source))
append_ai_message_event(
agent,
state,
context_ref,
%{role: :user, content: event_field(data, :content, "")},
request_id,
run_id,
signal_id,
refs || %{}
)
:llm_completed ->
turn_type = event_field(data, :turn_type, :final_answer)
text = event_field(data, :text, "")
tool_calls = event_field(data, :tool_calls, [])
thinking = event_field(data, :thinking_content)
assistant_tool_calls = if turn_type == :tool_calls, do: tool_calls, else: nil
append_ai_message_event(
agent,
state,
context_ref,
%{
role: :assistant,
content: text,
tool_calls: assistant_tool_calls,
thinking: thinking
},
request_id,
run_id,
signal_id
)
:tool_completed ->
tool_call_id = event_field(data, :tool_call_id, event_field(event, :tool_call_id, ""))
tool_name = event_field(data, :tool_name, event_field(event, :tool_name, ""))
tool_result = normalize_tool_result(event_field(data, :result, {:error, :unknown, []}))
content = Turn.format_tool_result_content(tool_result)
append_ai_message_event(
agent,
state,
context_ref,
%{role: :tool, content: content, tool_call_id: tool_call_id, name: tool_name},
request_id,
run_id,
signal_id
)
_ ->
{agent, state}
end
end
defp append_ai_message_event(
agent,
state,
context_ref,
%{} = message,
request_id,
run_id,
signal_id,
extra_refs \\ %{}
) do
payload =
message
|> Map.put(:context_ref, context_ref)
|> Map.put(:request_id, request_id)
|> Map.put(:run_id, run_id)
|> Enum.reject(fn {_k, v} -> is_nil(v) end)
|> Map.new()
refs =
extra_refs
|> sanitize_extra_refs()
|> Map.merge(%{
request_id: request_id,
run_id: run_id
})
|> maybe_put_ref(:signal_id, signal_id)
append_core_thread_entry(agent, state, :ai_message, payload, refs)
end
defp append_ai_context_operation_event(agent, state, %{
op_id: op_id,
context_ref: context_ref,
operation: operation
}) do
serialized_operation =
operation
|> Map.update(:result_context, nil, fn
%AIContext{} = context -> serialize_context(context)
other -> other
end)
|> Enum.reject(fn {_k, v} -> is_nil(v) end)
|> Map.new()
payload = %{
op_id: op_id,
context_ref: context_ref,
operation: serialized_operation
}
refs = %{op_id: op_id, context_ref: context_ref}
append_core_thread_entry(agent, state, :ai_context_operation, payload, refs)
end
defp append_core_thread_entry(agent, state, kind, payload, refs)
when is_map(payload) and is_map(refs) do
agent =
ThreadAgent.append(agent, %{
kind: kind,
payload: payload,
refs: refs
})
{agent, Map.put(state, :projection_cursor_seq, core_thread_last_seq(agent))}
end
defp project_context_from_core_thread(agent, context_ref, %AIContext{} = fallback_context) do
case ThreadAgent.get(agent) do
%Thread{} = thread ->
project_context_from_entries(Thread.to_list(thread), context_ref, fallback_context)
_ ->
fallback_context
end
end
defp project_context_from_entries(entries, context_ref, %AIContext{} = fallback_context)
when is_list(entries) do
{anchor_context, anchor_seq} =
Enum.reduce(entries, {fallback_context, -1}, fn entry, {current_context, current_anchor_seq} ->
with :ai_context_operation <- fetch_map_value(entry, :kind),
payload when is_map(payload) <- fetch_map_value(entry, :payload),
^context_ref <- fetch_map_value(payload, :context_ref),
operation when is_map(operation) <- fetch_map_value(payload, :operation),
:replace <- normalize_operation_type(fetch_map_value(operation, :type)),
{:ok, context} <- AIContext.coerce(fetch_map_value(operation, :result_context)) do
{context, fetch_map_value(entry, :seq) || current_anchor_seq}
else
_ -> {current_context, current_anchor_seq}
end
end)
Enum.reduce(entries, anchor_context, fn entry, acc ->
seq = fetch_map_value(entry, :seq) || -1
with true <- seq > anchor_seq,
:ai_message <- fetch_map_value(entry, :kind),
payload when is_map(payload) <- fetch_map_value(entry, :payload),
^context_ref <- fetch_map_value(payload, :context_ref) do
refs = fetch_map_value(entry, :refs)
apply_projected_ai_message(acc, payload, refs)
else
_ -> acc
end
end)
end
defp apply_projected_ai_message(%AIContext{} = context, payload, refs) do
case normalize_message_role(fetch_map_value(payload, :role)) do
:user ->
case fetch_map_value(payload, :content) do
content when is_binary(content) ->
AIContext.append_user(context, content, refs: normalize_refs(refs))
_ ->
context
end
:assistant ->
content = fetch_map_value(payload, :content)
tool_calls = fetch_map_value(payload, :tool_calls)
thinking = fetch_map_value(payload, :thinking)
opts = if is_binary(thinking) and thinking != "", do: [thinking: thinking], else: []
opts = Keyword.put(opts, :refs, normalize_refs(refs))
AIContext.append_assistant(
context,
normalize_content(content),
normalize_optional_list(tool_calls),
opts
)
:tool ->
case {fetch_map_value(payload, :tool_call_id), fetch_map_value(payload, :name)} do
{tool_call_id, name} when is_binary(tool_call_id) and is_binary(name) ->
AIContext.append_tool_result(
context,
tool_call_id,
name,
normalize_content(fetch_map_value(payload, :content)),
refs: normalize_refs(refs)
)
_ ->
context
end
_ ->
context
end
end
defp normalize_message_role(role) when role in [:user, :assistant, :tool], do: role
defp normalize_message_role("user"), do: :user
defp normalize_message_role("assistant"), do: :assistant
defp normalize_message_role("tool"), do: :tool
defp normalize_message_role(_), do: :unknown
defp normalize_operation_type(type) when type in [:replace, :switch], do: type
defp normalize_operation_type("replace"), do: :replace
defp normalize_operation_type("switch"), do: :switch
defp normalize_operation_type(_), do: :unknown
defp serialize_context(%AIContext{} = context) do
%{
id: context.id,
system_prompt: context.system_prompt,
entries: Enum.map(context.entries, &Map.from_struct/1)
}
end
defp initial_active_context_ref(%Agent{state: state}) when is_map(state) do
case Map.get(state, :active_context_ref) do
ref when is_binary(ref) and ref != "" -> ref
_ -> @default_context_ref
end
end
defp initial_active_context_ref(_), do: @default_context_ref
defp core_thread_last_seq(%Agent{} = agent) do
agent
|> ThreadAgent.get()
|> core_thread_last_seq()
end
defp core_thread_last_seq(%Thread{} = thread) do
case Thread.last(thread) do
%{seq: seq} when is_integer(seq) -> seq
_ -> -1
end
end
defp core_thread_last_seq(_), do: -1
defp maybe_put_ref(refs, _key, nil), do: refs
defp maybe_put_ref(refs, key, value), do: Map.put(refs, key, value)
defp sanitize_extra_refs(extra_refs) when is_map(extra_refs) do
Map.drop(extra_refs, [:request_id, :run_id, :signal_id])
end
defp sanitize_extra_refs(_extra_refs), do: %{}
defp fetch_map_value(%{} = map, key) when is_atom(key) do
Map.get(map, key, Map.get(map, Atom.to_string(key)))
end
defp fetch_map_value(_map, _key), do: nil
defp normalize_optional_integer(value) when is_integer(value), do: value
defp normalize_optional_integer(_), do: nil
defp normalize_optional_map(value) when is_map(value), do: value
defp normalize_optional_map(_), do: %{}
defp normalize_optional_list(value) when is_list(value), do: value
defp normalize_optional_list(_), do: nil
defp normalize_content(value) when is_binary(value), do: value
defp normalize_content(nil), do: ""
defp normalize_content(value), do: inspect(value)
defp normalize_refs(refs) when is_map(refs) and map_size(refs) > 0, do: refs
defp normalize_refs(_), do: nil
defp process_worker_child_started(agent, %{tag: tag, pid: pid}) when is_pid(pid) do
state = StratState.get(agent, %{})
if react_worker_tag?(tag) do
pending = state[:pending_worker_start]
base_state =
state
|> Map.put(:react_worker_pid, pid)
|> Map.put(:react_worker_status, :ready)
if is_map(pending) do
directive = AgentDirective.emit_to_pid(worker_start_signal(pending), pid)
new_state =
base_state
|> Map.put(:pending_worker_start, nil)
|> Map.put(:react_worker_status, :running)
{put_strategy_state(agent, new_state), [directive]}
else
{put_strategy_state(agent, base_state), []}
end
else
{agent, []}
end
end
defp process_worker_child_started(agent, _params), do: {agent, []}
defp process_worker_child_exit(agent, %{tag: tag, pid: pid, reason: reason}) do
state = StratState.get(agent, %{})
if react_worker_tag?(tag) do
tracked? = worker_pid_matches?(state[:react_worker_pid], pid)
if tracked? do
request_id = state[:active_request_id]
base_state =
state
|> Map.put(:react_worker_pid, nil)
|> Map.put(:react_worker_status, :missing)
|> Map.put(:pending_worker_start, nil)
if is_binary(request_id) and state[:status] in [:awaiting_llm, :awaiting_tool] do
error = {:react_worker_exit, reason}
stream_to = request_stream_to(agent, request_id)
failure_signal =
Signal.RequestFailed.new!(%{
request_id: request_id,
error: error,
run_id: request_id
})
RequestStream.send_event(
stream_to,
RequestStream.failed_event(request_id, error, reason: :react_worker_exit)
)
Jido.AgentServer.cast(self(), failure_signal)
failed_state =
base_state
|> Map.put(:status, :error)
|> Map.put(:termination_reason, :error)
|> Map.put(:result, error)
|> Map.put(:active_request_id, nil)
|> PendingInput.stop()
|> Map.put(:run_context, nil)
|> Map.delete(:run_tool_context)
|> Map.delete(:run_req_http_options)
|> Map.delete(:run_llm_opts)
{agent, failed_state} = maybe_apply_pending_context_op(agent, failed_state)
{put_strategy_state(agent, failed_state), []}
else
{put_strategy_state(agent, base_state), []}
end
else
{agent, []}
end
else
{agent, []}
end
end
defp process_worker_event(agent, %{event: event} = params) when is_map(event) do
state = StratState.get(agent, %{})
event = normalize_event_map(event)
request_id = event_field(event, :request_id, params[:request_id] || state[:active_request_id])
state = append_trace_event(state, request_id, event)
{new_state, signals} = apply_runtime_event(state, event)
Enum.each(signals, &Jido.AgentServer.cast(self(), &1))
RequestStream.send_event(request_stream_to(agent, request_id), event)
kind = event_kind(event)
{agent, new_state} = maybe_append_ai_message_event_from_runtime(agent, new_state, event)
{agent, new_state} = maybe_apply_pending_context_op_after_terminal(agent, new_state, kind)
new_state = maybe_mark_worker_ready(new_state, kind)
agent = put_strategy_state(agent, new_state)
{agent, directives} = maybe_apply_runtime_effects(agent, event, new_state)
{agent, directives}
end
defp process_worker_event(agent, _params), do: {agent, []}
defp apply_runtime_event(state, event) do
kind = event_kind(event)
iteration = event_field(event, :iteration, state[:iteration] || 0)
request_id = event_field(event, :request_id, state[:active_request_id])
run_id = event_field(event, :run_id, request_id)
llm_call_id = event_field(event, :llm_call_id, state[:current_llm_call_id])
data = event_field(event, :data, %{})
base_state =
state
|> Map.put(:active_request_id, request_id)
|> Map.put(:iteration, iteration)
|> Map.put(:current_llm_call_id, llm_call_id)
case kind do
:request_started ->
query = event_field(data, :query, "")
started_state =
base_state
|> Map.put(:status, :awaiting_llm)
|> Map.put(:result, nil)
|> Map.put(:termination_reason, nil)
|> Map.put(:started_at, event_field(event, :at_ms, System.monotonic_time(:millisecond)))
|> Map.put(:streaming_text, "")
|> Map.put(:streaming_thinking, "")
|> Map.put(:tool_results, [])
|> ensure_request_trace(request_id)
signal =
Signal.RequestStarted.new!(%{request_id: request_id, query: query, run_id: request_id})
emit_runtime_telemetry(state, :request_started, request_id, run_id, iteration, llm_call_id, event, data)
{started_state, [signal]}
:llm_started ->
emit_runtime_telemetry(state, :llm_started, request_id, run_id, iteration, llm_call_id, event, data)
{Map.put(base_state, :status, :awaiting_llm), []}
:llm_delta ->
chunk_type = event_field(data, :chunk_type, :content)
delta = event_field(data, :delta, "")
updated =
case chunk_type do
:thinking ->
Map.update(base_state, :streaming_thinking, delta, &(&1 <> delta))
_ ->
Map.update(base_state, :streaming_text, delta, &(&1 <> delta))
end
signal =
Signal.LLMDelta.new!(
llm_delta_signal_data(event, request_id, run_id, iteration, llm_call_id, delta, chunk_type)
)
emit_runtime_telemetry(state, :llm_delta, request_id, run_id, iteration, llm_call_id, event, data)
{updated, [signal]}
:llm_completed ->
turn_type = event_field(data, :turn_type, :final_answer)
text = event_field(data, :text, "")
thinking_content = event_field(data, :thinking_content)
reasoning_details = event_field(data, :reasoning_details)
tool_calls = event_field(data, :tool_calls, [])
usage = event_field(data, :usage, %{})
call_id = llm_call_id || event_field(data, :call_id, "")
model = event_field(data, :model, config_model(state))
pending_tool_calls =
Enum.map(tool_calls, fn tc ->
%{
id: event_field(tc, :id, ""),
name: event_field(tc, :name, ""),
arguments: event_field(tc, :arguments, %{}),
result: nil
}
end)
refs = runtime_event_refs(event, request_id)
updated =
base_state
|> Map.put(:status, if(turn_type == :tool_calls, do: :awaiting_tool, else: :awaiting_llm))
|> Map.put(:pending_tool_calls, pending_tool_calls)
|> append_assistant_to_run_context(
turn_type,
text,
tool_calls,
thinking_content,
reasoning_details,
refs
)
|> Map.update(:usage, usage || %{}, fn existing ->
merge_usage(existing, usage || %{})
end)
|> maybe_append_thinking_trace(thinking_content)
|> maybe_put_result(turn_type, text)
llm_signal =
Signal.LLMResponse.new!(%{
call_id: call_id,
result:
{:ok,
%{
type: turn_type,
text: text,
thinking_content: thinking_content,
reasoning_details: reasoning_details,
tool_calls: tool_calls,
model: model,
usage: usage
}, []},
metadata: runtime_signal_metadata(request_id, run_id, iteration, :generate_text)
})
usage_signal = maybe_usage_signal(call_id, model, usage, request_id, run_id, iteration)
emit_runtime_telemetry(state, :llm_completed, request_id, run_id, iteration, call_id, event, data)
{updated, Enum.reject([llm_signal, usage_signal], &is_nil/1)}
:input_injected ->
refs =
data
|> event_field(:refs, %{})
|> normalize_event_message_refs(runtime_event_refs(event, request_id))
updated =
base_state
|> Map.put(:status, :awaiting_llm)
|> Map.put(:result, nil)
|> append_user_to_run_context(event_field(data, :content, ""), refs)
{updated, []}
kind when kind in [:output_started, :output_validated, :output_repair, :output_failed] ->
updated = Map.put(base_state, :output, data)
emit_runtime_telemetry(state, kind, request_id, run_id, iteration, llm_call_id, event, data)
{updated, []}
:tool_started ->
tool_call_id = event_field(data, :tool_call_id, event_field(event, :tool_call_id, ""))
tool_name = event_field(data, :tool_name, event_field(event, :tool_name, ""))
arguments = event_field(data, :arguments, [])
updated = Map.put(base_state, :status, :awaiting_tool)
signal =
Signal.ToolStarted.new!(%{
call_id: tool_call_id,
tool_name: tool_name,
arguments: arguments,
metadata: runtime_signal_metadata(request_id, run_id, iteration, :tool_execute)
})
emit_runtime_telemetry(state, :tool_started, request_id, run_id, iteration, llm_call_id, event, data)
{updated, [signal]}
:tool_completed ->
tool_call_id = event_field(data, :tool_call_id, event_field(event, :tool_call_id, ""))
tool_name = event_field(data, :tool_name, event_field(event, :tool_name, ""))
tool_result = normalize_tool_result(event_field(data, :result, {:error, :unknown, []}))
tool_result_entry = completed_tool_result_entry(base_state, tool_call_id, tool_name, tool_result)
refs = runtime_event_refs(event, request_id)
updated =
base_state
|> Map.update(:pending_tool_calls, [], fn pending ->
Enum.map(pending, fn tc ->
if tc.id == tool_call_id, do: %{tc | result: tool_result}, else: tc
end)
end)
|> put_completed_tool_result(tool_result_entry)
|> append_tool_result_to_run_context(tool_call_id, tool_name, tool_result, refs)
signal =
Signal.ToolResult.new!(%{
call_id: tool_call_id,
tool_name: tool_name,
result: tool_result,
metadata: runtime_signal_metadata(request_id, run_id, iteration, :tool_execute)
})
emit_runtime_telemetry(state, :tool_completed, request_id, run_id, iteration, llm_call_id, event, data)
{updated, [signal]}
:request_completed ->
result = event_field(data, :result)
termination_reason = event_field(data, :termination_reason, :final_answer)
usage = event_field(data, :usage, %{})
updated =
base_state
|> PendingInput.stop()
|> Map.put(:status, :completed)
|> Map.put(:result, result)
|> Map.put(:termination_reason, termination_reason)
|> Map.put(:usage, usage || %{})
|> commit_run_context()
|> Map.put(:active_request_id, nil)
|> Map.delete(:run_tool_context)
|> Map.delete(:run_req_http_options)
|> Map.delete(:run_llm_opts)
signal =
Signal.RequestCompleted.new!(%{
request_id: request_id,
result: result,
run_id: request_id
})
emit_runtime_telemetry(state, :request_completed, request_id, run_id, iteration, llm_call_id, event, data)
{updated, [signal]}
:request_failed ->
error = event_field(data, :error, :unknown_error)
updated =
base_state
|> PendingInput.stop()
|> Map.put(:status, :error)
|> Map.put(:result, error)
|> Map.put(:termination_reason, :error)
|> Map.put(:run_context, nil)
|> Map.put(:active_request_id, nil)
|> Map.delete(:run_tool_context)
|> Map.delete(:run_req_http_options)
|> Map.delete(:run_llm_opts)
signal =
Signal.RequestFailed.new!(%{request_id: request_id, error: error, run_id: request_id})
emit_runtime_telemetry(state, :request_failed, request_id, run_id, iteration, llm_call_id, event, data)
{updated, [signal]}
:request_cancelled ->
reason = event_field(data, :reason, :cancelled)
error = {:cancelled, reason}
updated =
base_state
|> PendingInput.stop()
|> Map.put(:status, :error)
|> Map.put(:result, error)
|> Map.put(:termination_reason, :cancelled)
|> Map.put(:cancel_reason, reason)
|> Map.put(:run_context, nil)
|> Map.put(:active_request_id, nil)
|> Map.delete(:run_tool_context)
|> Map.delete(:run_req_http_options)
|> Map.delete(:run_llm_opts)
signal =
Signal.RequestFailed.new!(%{request_id: request_id, error: error, run_id: request_id})
emit_runtime_telemetry(state, :request_cancelled, request_id, run_id, iteration, llm_call_id, event, data)
{updated, [signal]}
:checkpoint ->
token = event_field(data, :token)
updated =
base_state
|> Map.put(:checkpoint_token, token)
|> then(fn state_after_checkpoint ->
if state[:status] in [:completed, :error] and is_nil(state[:active_request_id]) do
Map.put(state_after_checkpoint, :active_request_id, nil)
else
state_after_checkpoint
end
end)
{updated, []}
_ ->
{base_state, []}
end
end
defp maybe_append_thinking_trace(state, nil), do: state
defp maybe_append_thinking_trace(state, ""), do: state
defp maybe_append_thinking_trace(state, thinking_content) do
trace_entry = %{
call_id: state[:current_llm_call_id],
iteration: state[:iteration],
thinking: thinking_content
}
Map.update(state, :thinking_trace, [trace_entry], fn trace -> trace ++ [trace_entry] end)
end
defp maybe_put_result(state, :final_answer, result), do: Map.put(state, :result, result)
defp maybe_put_result(state, _turn_type, _result), do: state
defp maybe_usage_signal(_call_id, _model, usage, _request_id, _run_id, _iteration) when usage in [%{}, nil], do: nil
defp maybe_usage_signal(call_id, model, usage, request_id, run_id, iteration) do
input_tokens = Map.get(usage, :input_tokens, 0)
output_tokens = Map.get(usage, :output_tokens, 0)
Signal.Usage.new!(%{
call_id: call_id,
model: Jido.AI.model_label(model),
input_tokens: input_tokens,
output_tokens: output_tokens,
total_tokens: input_tokens + output_tokens,
metadata: runtime_signal_metadata(request_id, run_id, iteration, :generate_text)
})
end
defp merge_usage(existing, incoming) do
Jido.AI.Usage.merge(existing, incoming)
end
defp event_kind(event) do
case event_field(event, :kind) do
kind when is_atom(kind) -> kind
kind when is_binary(kind) -> runtime_kind_from_string(kind)
_ -> :unknown
end
end
defp runtime_kind_from_string("request_started"), do: :request_started
defp runtime_kind_from_string("llm_started"), do: :llm_started
defp runtime_kind_from_string("llm_delta"), do: :llm_delta
defp runtime_kind_from_string("llm_completed"), do: :llm_completed
defp runtime_kind_from_string("tool_started"), do: :tool_started
defp runtime_kind_from_string("tool_completed"), do: :tool_completed
defp runtime_kind_from_string("input_injected"), do: :input_injected
defp runtime_kind_from_string("checkpoint"), do: :checkpoint
defp runtime_kind_from_string("request_completed"), do: :request_completed
defp runtime_kind_from_string("request_failed"), do: :request_failed
defp runtime_kind_from_string("request_cancelled"), do: :request_cancelled
defp runtime_kind_from_string(_), do: :unknown
defp event_field(map, key, default \\ nil) when is_map(map) do
Map.get(map, key, Map.get(map, Atom.to_string(key), default))
end
defp llm_delta_signal_data(event, request_id, run_id, iteration, llm_call_id, delta, chunk_type) do
%{
call_id: llm_call_id || "",
delta: delta,
chunk_type: chunk_type,
metadata: runtime_signal_metadata(request_id, run_id, iteration, :generate_text)
}
|> maybe_put(:seq, event_field(event, :seq))
|> maybe_put(:run_id, run_id)
|> maybe_put(:request_id, request_id)
|> maybe_put(:iteration, iteration)
end
defp maybe_put(map, _key, nil), do: map
defp maybe_put(map, key, value), do: Map.put(map, key, value)
defp normalize_event_map(event) when is_map(event), do: event
defp config_model(state) do
state
|> Map.get(:config, %{})
|> Map.get(:model)
end
defp runtime_config_from_strategy(config, opts) do
provider_opt_keys_by_string = config[:provider_opt_keys_by_string] || %{}
req_http_options =
opts
|> Keyword.get(:req_http_options, config[:base_req_http_options] || [])
|> normalize_req_http_options()
llm_opts =
opts
|> Keyword.get(:llm_opts, config[:base_llm_opts] || [])
|> normalize_llm_opts(provider_opt_keys_by_string)
tools = Keyword.get(opts, :tools, config[:actions_by_name] || %{})
request_transformer = Keyword.get(opts, :request_transformer, config[:request_transformer])
output = Keyword.get(opts, :output, config[:output])
max_iterations = resolve_max_iterations_opt(opts, config[:max_iterations])
stream_timeout_ms =
resolve_stream_timeout_ms_opt(
opts,
Map.get(config, :stream_timeout_ms, Map.get(config, :stream_receive_timeout_ms, 0))
)
runtime_opts = %{
model: config[:model],
system_prompt: config[:system_prompt],
tools: tools,
request_transformer: request_transformer,
max_iterations: max_iterations,
max_tokens: config[:max_tokens],
streaming: config[:streaming],
stream_timeout_ms: stream_timeout_ms,
req_http_options: req_http_options,
llm_opts: llm_opts,
tool_timeout_ms: config[:tool_timeout_ms],
tool_max_retries: config[:tool_max_retries],
tool_retry_backoff_ms: config[:tool_retry_backoff_ms],
emit_telemetry?: get_in(config, [:observability, :emit_telemetry?]),
redact_tool_args?: get_in(config, [:observability, :redact_tool_args?]),
capture_deltas?: get_in(config, [:observability, :emit_llm_deltas?]),
pending_input_server: Keyword.get(opts, :pending_input_server),
runtime_task_supervisor: config[:runtime_task_supervisor],
effect_policy: config[:effect_policy],
output: output
}
ReActRuntimeConfig.new(runtime_opts)
end
defp set_run_tool_context(agent, context) when is_map(context) do
state = StratState.get(agent, %{})
put_strategy_state(agent, Map.put(state, :run_tool_context, context))
end
defp set_run_req_http_options(agent, req_http_options) when is_list(req_http_options) do
state = StratState.get(agent, %{})
put_strategy_state(agent, Map.put(state, :run_req_http_options, req_http_options))
end
defp set_run_llm_opts(agent, llm_opts) when is_list(llm_opts) do
state = StratState.get(agent, %{})
put_strategy_state(agent, Map.put(state, :run_llm_opts, llm_opts))
end
defp normalize_action({inner, _meta}), do: normalize_action(inner)
defp normalize_action(action), do: action
defp ensure_worker_start(state, worker_start_payload) do
if is_pid(state[:react_worker_pid]) and Process.alive?(state[:react_worker_pid]) do
directive =
AgentDirective.emit_to_pid(
worker_start_signal(worker_start_payload),
state[:react_worker_pid]
)
new_state =
state
|> Map.put(:pending_worker_start, nil)
|> Map.put(:react_worker_status, :running)
{new_state, [directive]}
else
spawn_directive =
AgentDirective.spawn_agent(Jido.AI.Reasoning.ReAct.Worker.Agent, @worker_tag)
new_state =
state
|> Map.put(:react_worker_pid, nil)
|> Map.put(:react_worker_status, :starting)
|> Map.put(:pending_worker_start, worker_start_payload)
{new_state, [spawn_directive]}
end
end
defp worker_start_signal(payload) do
Jido.Signal.new!("ai.react.worker.start", payload, source: @source)
end
defp worker_cancel_signal(request_id, reason) do
Jido.Signal.new!("ai.react.worker.cancel", %{request_id: request_id, reason: reason}, source: @source)
end
defp react_worker_tag?(tag), do: tag == @worker_tag or tag == Atom.to_string(@worker_tag)
defp worker_pid_matches?(expected, actual) when is_pid(expected) and is_pid(actual),
do: expected == actual
defp worker_pid_matches?(_expected, _actual), do: true
defp busy?(state, config) do
config[:request_policy] == :reject and state[:status] in [:awaiting_llm, :awaiting_tool] and
is_binary(state[:active_request_id])
end
defp maybe_mark_worker_ready(state, kind) when kind in [:request_completed, :request_failed, :request_cancelled] do
Map.put(state, :react_worker_status, :ready)
end
defp maybe_mark_worker_ready(state, _kind), do: state
defp maybe_apply_runtime_effects(agent, event, state) do
case event_kind(event) do
:tool_completed ->
data = event_field(event, :data, %{})
result = normalize_tool_result(event_field(data, :result, {:error, :unknown, []}))
policy = effect_policy_from_state(state)
{agent, directives, _stats, _filtered_result} =
Effects.apply_result(agent, result, policy)
{agent, directives}
_ ->
{agent, []}
end
end
defp runtime_state_from_context(%AIContext{} = context, query, request_id, run_id)
when (is_binary(query) or is_list(query)) and is_binary(request_id) and is_binary(run_id) do
ReActState.new(query, context.system_prompt, request_id: request_id, run_id: run_id)
|> Map.put(:context, context)
end
defp strategy_context(state, config) do
case Map.get(state, :context) do
%AIContext{} = context ->
context
_ ->
AIContext.new(system_prompt: config[:system_prompt])
end
end
defp fresh_projection_context(config) when is_map(config) do
AIContext.new(system_prompt: config[:system_prompt])
end
defp fresh_projection_context(_), do: AIContext.new()
defp snapshot_context(state, config) do
case Map.get(state, :run_context) do
%AIContext{} = context -> context
_ -> strategy_context(state, config)
end
end
defp append_user_to_run_context(state, content, refs) when is_binary(content) do
context = Map.get(state, :run_context) || Map.get(state, :context)
case context do
%AIContext{} = context ->
Map.put(state, :run_context, AIContext.append_user(context, content, refs: normalize_refs(refs)))
_ ->
state
end
end
defp append_assistant_to_run_context(
state,
turn_type,
text,
tool_calls,
thinking_content,
reasoning_details,
refs
) do
context = Map.get(state, :run_context) || Map.get(state, :context)
case context do
%AIContext{} = context ->
assistant_tool_calls = if turn_type == :tool_calls, do: tool_calls, else: nil
assistant_opts =
[]
|> maybe_put_assistant_context_opt(:thinking, thinking_content)
|> maybe_put_assistant_context_opt(:reasoning_details, reasoning_details)
|> maybe_put_assistant_context_opt(:refs, normalize_refs(refs))
Map.put(
state,
:run_context,
AIContext.append_assistant(context, text, assistant_tool_calls, assistant_opts)
)
_ ->
state
end
end
defp append_tool_result_to_run_context(state, tool_call_id, tool_name, tool_result, refs) do
context = Map.get(state, :run_context) || Map.get(state, :context)
case context do
%AIContext{} = context when is_binary(tool_call_id) and is_binary(tool_name) ->
content = Turn.format_tool_result_content(tool_result)
Map.put(
state,
:run_context,
AIContext.append_tool_result(context, tool_call_id, tool_name, content, refs: normalize_refs(refs))
)
_ ->
state
end
end
defp maybe_put_assistant_context_opt(opts, _key, nil), do: opts
defp maybe_put_assistant_context_opt(opts, _key, ""), do: opts
defp maybe_put_assistant_context_opt(opts, key, value), do: Keyword.put(opts, key, value)
defp runtime_event_refs(event, fallback_request_id) do
%{}
|> maybe_put_ref(:request_id, event_field(event, :request_id, fallback_request_id))
|> maybe_put_ref(:run_id, event_field(event, :run_id, fallback_request_id))
|> maybe_put_ref(:signal_id, event_field(event, :id))
|> normalize_refs()
end
defp commit_run_context(state) do
case Map.get(state, :run_context) do
%AIContext{} = context -> state |> Map.put(:context, context) |> Map.put(:run_context, nil)
_ -> state
end
end
defp ensure_request_trace(state, request_id) when is_binary(request_id) do
traces = Map.get(state, :request_traces, %{})
trace = Map.get(traces, request_id, %{events: [], truncated?: false})
Map.put(state, :request_traces, Map.put(traces, request_id, trace))
end
defp ensure_request_trace(state, _request_id), do: state
defp append_trace_event(state, request_id, event) when is_binary(request_id) do
traces = Map.get(state, :request_traces, %{})
trace = Map.get(traces, request_id, %{events: [], truncated?: false})
updated_trace =
cond do
trace.truncated? ->
trace
length(trace.events) < @request_trace_cap ->
%{trace | events: trace.events ++ [event]}
true ->
%{trace | truncated?: true}
end
Map.put(state, :request_traces, Map.put(traces, request_id, updated_trace))
end
defp append_trace_event(state, _request_id, _event), do: state
defp normalize_event_message_refs(%{} = refs, fallback_refs) when is_map(fallback_refs) do
Map.merge(fallback_refs, refs)
end
defp normalize_event_message_refs(%{} = refs, _fallback_refs), do: refs
defp normalize_event_message_refs(_refs, fallback_refs), do: fallback_refs
defp build_config(agent, ctx) do
opts = ctx[:strategy_opts] || []
observability_overrides = opts |> Keyword.get(:observability, %{}) |> normalize_map_opt()
tool_context_opt = opts |> Keyword.get(:tool_context, %{}) |> normalize_map_opt()
agent_effect_policy =
Keyword.get(opts, :agent_effect_policy, Keyword.get(opts, :effect_policy, %{}))
strategy_effect_policy = Keyword.get(opts, :strategy_effect_policy, %{})
tools_modules =
case Keyword.fetch(opts, :tools) do
{:ok, mods} when is_list(mods) ->
mods
:error ->
raise ArgumentError,
"Jido.AI.Reasoning.ReAct.Strategy requires :tools option (list of Jido.Action modules)"
end
actions_by_name = Map.new(tools_modules, &{&1.name(), &1})
reqllm_tools = ToolAdapter.from_actions(tools_modules)
raw_model = Keyword.get(opts, :model, Map.get(agent.state, :model, @default_model))
resolved_model = resolve_model_spec(raw_model)
provider_opt_keys_by_string = provider_opt_keys_by_string(resolved_model)
request_policy = validate_request_policy!(Keyword.get(opts, :request_policy, :reject))
effect_policy = Effects.intersect_policies(agent_effect_policy, strategy_effect_policy)
%{
tools: tools_modules,
reqllm_tools: reqllm_tools,
actions_by_name: actions_by_name,
request_transformer: validate_request_transformer_opt!(Keyword.get(opts, :request_transformer)),
system_prompt: normalize_system_prompt_opt(opts),
model: resolved_model,
max_iterations: Keyword.get(opts, :max_iterations, @default_max_iterations),
max_tokens: Keyword.get(opts, :max_tokens, @default_max_tokens),
streaming: Keyword.get(opts, :streaming, true),
stream_receive_timeout_ms:
opts
|> Keyword.get(:stream_receive_timeout_ms, Keyword.get(opts, :stream_timeout_ms, 30_000))
|> normalize_stream_receive_timeout_ms(30_000),
request_policy: request_policy,
stream_timeout_ms: Keyword.get(opts, :stream_timeout_ms, 0),
tool_timeout_ms: Keyword.get(opts, :tool_timeout_ms, 15_000),
tool_max_retries: Keyword.get(opts, :tool_max_retries, 1),
tool_retry_backoff_ms: Keyword.get(opts, :tool_retry_backoff_ms, 200),
effect_policy: effect_policy,
runtime_adapter: true,
runtime_task_supervisor: Keyword.get(opts, :runtime_task_supervisor),
observability:
Map.merge(
%{
emit_telemetry?: true,
emit_lifecycle_signals?: true,
redact_tool_args?: true,
emit_llm_deltas?: true
},
observability_overrides
),
agent_id: agent.id,
base_tool_context: Map.get(agent.state, :tool_context) || tool_context_opt,
base_req_http_options: opts |> Keyword.get(:req_http_options, []) |> normalize_req_http_options(),
base_llm_opts: opts |> Keyword.get(:llm_opts, []) |> normalize_llm_opts(provider_opt_keys_by_string),
output: opts |> Keyword.get(:output) |> Output.new!(),
provider_opt_keys_by_string: provider_opt_keys_by_string
}
end
defp normalize_system_prompt_opt(opts) do
case Keyword.fetch(opts, :system_prompt) do
:error ->
@default_system_prompt
{:ok, prompt} when is_binary(prompt) and prompt != "" ->
prompt
{:ok, prompt} when prompt in [nil, false, ""] ->
nil
{:ok, other} ->
raise ArgumentError,
"invalid system_prompt: expected binary, nil, or false, got #{inspect(other)}"
end
end
defp resolve_model_spec(model), do: Jido.AI.resolve_model(model)
defp validate_request_policy!(:reject), do: :reject
defp validate_request_policy!(other) do
raise ArgumentError,
"unsupported request_policy #{inspect(other)} for ReAct; supported values: [:reject]"
end
defp normalize_map_opt(%{} = value), do: value
defp normalize_map_opt({:%{}, _meta, pairs}) when is_list(pairs), do: Map.new(pairs)
defp normalize_map_opt(_), do: %{}
defp resolve_request_tools(config, params) do
base_tools = config[:actions_by_name] || %{}
override_tools = Map.get(params, :tools)
allowed_tools = Map.get(params, :allowed_tools)
case ToolSelection.resolve(base_tools, override_tools, allowed_tools) do
{:ok, tools} ->
{:ok, tools}
{:error, :invalid_tools} ->
{:error, :invalid_tools, "Invalid tools override for this request"}
{:error, :invalid_allowed_tools} ->
{:error, :invalid_allowed_tools, "allowed_tools must be a list of tool names"}
{:error, {:unknown_allowed_tools, unknown}} ->
{:error, :unknown_allowed_tools, "Unknown allowed_tools: #{Enum.join(unknown, ", ")}"}
{:error, {:invalid_action, module, reason}} ->
{:error, :invalid_tools, "Invalid tool #{inspect(module)}: #{inspect(reason)}"}
end
end
defp resolve_request_transformer(config, params) do
case validate_request_transformer(Map.get(params, :request_transformer, config[:request_transformer])) do
{:ok, module} ->
{:ok, module}
{:error, message} ->
{:error, :invalid_request_transformer, message}
end
end
defp resolve_request_output(config, params) do
case Map.fetch(params, :output) do
:error ->
{:ok, config[:output]}
{:ok, raw} when raw in [:raw, "raw"] ->
{:ok, nil}
{:ok, raw} ->
case Output.new(raw) do
{:ok, output} -> {:ok, output}
{:error, reason} -> {:error, :invalid_output, "Invalid output config: #{inspect(reason)}"}
end
end
end
defp validate_request_transformer_opt!(request_transformer) do
case validate_request_transformer(request_transformer) do
{:ok, module} ->
module
{:error, message} ->
raise ArgumentError, message
end
end
defp validate_request_transformer(nil), do: {:ok, nil}
defp validate_request_transformer(request_transformer) do
case RequestTransformer.validate(request_transformer) do
{:ok, module} ->
{:ok, module}
{:error, {:request_transformer_not_loaded, module}} ->
{:error, "Request transformer #{inspect(module)} is not loaded"}
{:error, {:request_transformer_missing_callback, module}} ->
{:error, "Request transformer #{inspect(module)} must implement transform_request/4"}
{:error, :invalid_request_transformer} ->
{:error, "request_transformer must be a module implementing transform_request/4"}
end
end
defp resolve_stream_timeout_ms_opt(opts, default) when is_list(opts) do
case Keyword.fetch(opts, :stream_timeout_ms) do
{:ok, value} when is_integer(value) and value >= 0 ->
value
_ ->
case Keyword.fetch(opts, :stream_receive_timeout_ms) do
{:ok, value} when is_integer(value) and value > 0 ->
value
_ ->
default
end
end
end
defp resolve_max_iterations_opt(opts, default) when is_list(opts) do
case Keyword.fetch(opts, :max_iterations) do
{:ok, value} when is_integer(value) and value > 0 -> value
_ -> default
end
end
defp normalize_stream_receive_timeout_ms(value, _default)
when is_integer(value) and value > 0 do
value
end
defp normalize_stream_receive_timeout_ms(_value, default), do: default
defp effect_policy_from_state(state) do
state
|> normalize_map_opt()
|> Map.get(:config, %{})
|> normalize_map_opt()
|> Map.get(:effect_policy, Effects.default_policy())
end
defp normalize_tool_result(result), do: Error.normalize_result(result, :tool_error, "Tool execution failed")
defp request_stream_to(agent, request_id) when is_binary(request_id) do
get_in(agent.state, [:requests, request_id, :stream_to])
end
defp request_stream_to(_agent, _request_id), do: nil
defp runtime_signal_metadata(request_id, run_id, iteration, operation) do
%{
request_id: request_id,
run_id: run_id,
iteration: iteration,
origin: :worker_runtime,
operation: operation,
strategy: :react
}
|> Enum.reject(fn {_key, value} -> is_nil(value) end)
|> Map.new()
end
defp emit_runtime_telemetry(state, kind, request_id, run_id, iteration, llm_call_id, event, data) do
obs_cfg = get_in(state, [:config, :observability]) || %{}
usage = event_field(data, :usage, %{}) || %{}
metadata =
%{
agent_id: get_in(state, [:config, :agent_id]),
request_id: request_id,
run_id: run_id,
iteration: iteration,
llm_call_id: llm_call_id,
tool_call_id: event_field(event, :tool_call_id),
tool_name: event_field(event, :tool_name),
model: telemetry_model_label(state, data),
origin: :worker_runtime,
operation: telemetry_operation(kind),
strategy: :react,
termination_reason: telemetry_termination_reason(kind, data),
error_type: telemetry_error_type(kind, data)
}
measurements = %{
duration_ms: event_field(data, :duration_ms, 0),
input_tokens: Map.get(usage, :input_tokens, 0),
output_tokens: Map.get(usage, :output_tokens, 0),
total_tokens: Map.get(usage, :total_tokens, Map.get(usage, :input_tokens, 0) + Map.get(usage, :output_tokens, 0)),
retry_count: max(event_field(data, :attempts, 1) - 1, 0),
queue_ms: 0
}
case kind do
:request_started ->
Observe.emit(obs_cfg, Observe.request(:start), measurements, metadata)
:request_completed ->
Observe.emit(obs_cfg, Observe.request(:complete), measurements, metadata)
:request_failed ->
Observe.emit(obs_cfg, Observe.request(:failed), measurements, metadata)
:request_cancelled ->
Observe.emit(obs_cfg, Observe.request(:cancelled), measurements, metadata)
:llm_started ->
Observe.emit(obs_cfg, Observe.llm(:start), measurements, metadata)
:llm_delta ->
Observe.emit(obs_cfg, Observe.llm(:delta), measurements, metadata, feature_gate: :llm_deltas)
:llm_completed ->
Observe.emit(obs_cfg, Observe.llm(:complete), measurements, metadata)
:output_started ->
Observe.emit(obs_cfg, Observe.output(:start), measurements, metadata)
:output_validated ->
Observe.emit(obs_cfg, Observe.output(:validated), measurements, metadata)
:output_repair ->
Observe.emit(obs_cfg, Observe.output(:repair), measurements, metadata)
:output_failed ->
Observe.emit(obs_cfg, Observe.output(:error), measurements, metadata)
:tool_started ->
Observe.emit(obs_cfg, Observe.tool(:start), measurements, metadata)
:tool_completed ->
emit_tool_completed_telemetry(
obs_cfg,
metadata,
measurements,
normalize_tool_result(event_field(data, :result))
)
end
end
defp telemetry_model_label(state, data) do
case event_field(data, :model) do
nil -> Jido.AI.model_label(config_model(state))
model -> Jido.AI.model_label(model)
end
end
defp emit_tool_completed_telemetry(obs_cfg, metadata, measurements, {:ok, _result, _effects}) do
Observe.emit(obs_cfg, Observe.tool(:complete), measurements, metadata)
end
defp emit_tool_completed_telemetry(obs_cfg, metadata, measurements, {:error, %{type: :timeout}, _effects}) do
Observe.emit(obs_cfg, Observe.tool(:timeout), measurements, metadata)
Observe.emit(obs_cfg, Observe.tool(:error), measurements, Map.put(metadata, :termination_reason, :error))
end
defp emit_tool_completed_telemetry(obs_cfg, metadata, measurements, {:error, %{type: type}, _effects}) do
Observe.emit(obs_cfg, Observe.tool(:error), measurements, %{metadata | error_type: type, termination_reason: :error})
end
defp telemetry_operation(:tool_started), do: :tool_execute
defp telemetry_operation(:tool_completed), do: :tool_execute
defp telemetry_operation(:output_started), do: :structured_output
defp telemetry_operation(:output_validated), do: :structured_output
defp telemetry_operation(:output_repair), do: :structured_output
defp telemetry_operation(:output_failed), do: :structured_output
defp telemetry_operation(_kind), do: :generate_text
defp telemetry_termination_reason(kind, _data) when kind in [:output_validated], do: :complete
defp telemetry_termination_reason(kind, _data) when kind in [:output_failed], do: :error
defp telemetry_termination_reason(:request_completed, data), do: event_field(data, :termination_reason, :complete)
defp telemetry_termination_reason(:request_failed, _data), do: :error
defp telemetry_termination_reason(:request_cancelled, _data), do: :cancelled
defp telemetry_termination_reason(:tool_completed, data),
do: if(match?({:ok, _, _}, normalize_tool_result(event_field(data, :result))), do: :complete, else: :error)
defp telemetry_termination_reason(_kind, _data), do: nil
defp telemetry_error_type(:request_failed, data), do: infer_error_type(event_field(data, :error))
defp telemetry_error_type(:output_failed, _data), do: :output_validation
defp telemetry_error_type(:tool_completed, data), do: infer_error_type(event_field(data, :result))
defp telemetry_error_type(_kind, _data), do: nil
defp infer_error_type({:error, %{type: type}, _effects}) when is_atom(type), do: type
defp infer_error_type({:error, %{code: type}, _effects}) when is_atom(type), do: type
defp infer_error_type(%{type: type}) when is_atom(type), do: type
defp infer_error_type(%{code: type}) when is_atom(type), do: type
defp infer_error_type({:cancelled, _}), do: :cancelled
defp infer_error_type(_), do: nil
defp normalize_req_http_options(req_http_options) when is_list(req_http_options),
do: req_http_options
defp normalize_req_http_options(_), do: []
defp normalize_llm_opts(llm_opts, provider_opt_keys_by_string) when is_list(llm_opts) do
normalize_llm_opt_pairs(llm_opts, provider_opt_keys_by_string)
end
defp normalize_llm_opts(llm_opts, provider_opt_keys_by_string) when is_map(llm_opts) do
llm_opts
|> Enum.map(fn {key, value} ->
normalized_key = normalize_llm_opt_key(key)
normalized_value =
normalize_llm_opt_value(normalized_key, value, provider_opt_keys_by_string)
{normalized_key, normalized_value}
end)
|> normalize_llm_opt_pairs(provider_opt_keys_by_string)
end
defp normalize_llm_opts(_llm_opts, _provider_opt_keys_by_string), do: []
defp normalize_llm_opt_pairs(pairs, provider_opt_keys_by_string) when is_list(pairs) do
pairs
|> Enum.reduce([], fn
{key, value}, acc when is_atom(key) and not is_nil(key) ->
normalized_value = normalize_llm_opt_value(key, value, provider_opt_keys_by_string)
[{key, normalized_value} | acc]
_other, acc ->
acc
end)
|> Enum.reverse()
end
defp normalize_llm_opt_key(key) when is_atom(key), do: key
defp normalize_llm_opt_key(key) when is_binary(key) do
Map.get(@reqllm_generation_opt_keys_by_string, key) || maybe_to_existing_atom(key)
end
defp normalize_llm_opt_key(_), do: nil
defp normalize_llm_opt_value(:provider_options, value, provider_opt_keys_by_string) do
normalize_provider_options(value, provider_opt_keys_by_string)
end
defp normalize_llm_opt_value(_key, value, _provider_opt_keys_by_string), do: value
defp normalize_provider_options(value, provider_opt_keys_by_string) when is_list(value) do
normalize_provider_option_pairs(value, provider_opt_keys_by_string)
end
defp normalize_provider_options(value, provider_opt_keys_by_string) when is_map(value) do
value
|> Enum.map(fn {key, entry_value} ->
{normalize_provider_opt_key(key, provider_opt_keys_by_string), entry_value}
end)
|> normalize_provider_option_pairs(provider_opt_keys_by_string)
end
defp normalize_provider_options(value, _provider_opt_keys_by_string), do: value
defp normalize_provider_option_pairs(pairs, _provider_opt_keys_by_string) do
pairs
|> Enum.reduce([], fn
{key, value}, acc when is_atom(key) and not is_nil(key) ->
[{key, value} | acc]
_other, acc ->
acc
end)
|> Enum.reverse()
end
defp normalize_provider_opt_key(key, _provider_opt_keys_by_string) when is_atom(key), do: key
defp normalize_provider_opt_key(key, provider_opt_keys_by_string) when is_binary(key) do
Map.get(provider_opt_keys_by_string, key) || maybe_to_existing_atom(key)
end
defp normalize_provider_opt_key(_key, _provider_opt_keys_by_string), do: nil
defp maybe_to_existing_atom(key) when is_binary(key) do
try do
String.to_existing_atom(key)
rescue
ArgumentError -> nil
end
end
defp provider_opt_keys_by_string(model_spec),
do: Jido.AI.provider_opt_keys(model_spec)
defp generate_call_id, do: "req_#{Jido.Util.generate_id()}"
defp put_strategy_state(%Agent{} = agent, state) when is_map(state) do
%{agent | state: Map.put(agent.state, StratState.key(), state)}
end
@doc """
Returns the list of currently registered tools for the given agent.
"""
@spec list_tools(Agent.t()) :: [module()]
def list_tools(%Agent{} = agent) do
state = StratState.get(agent, %{})
config = state[:config] || %{}
config[:tools] || []
end
end