erlang_adk

%% @doc adk_callbacks - Behavior and registry for ADK execution callbacks.
%%
%% Callbacks allow hooking into the execution lifecycle (e.g., on_llm_start,
%% on_tool_call, on_error) for logging, monitoring, or side-effects.
-module(adk_callbacks).

-export([execute/3, run/3]).
-export_type([callback_result/0]).

-define(APP, erlang_adk).
-define(DEFAULT_TIMEOUT_MS, 5000).
-define(DEFAULT_MAX_HEAP_WORDS, 262144).

%% Callback behaviour definition.
-callback on_agent_start(AgentName :: binary(), Input :: term()) -> ok.
-callback on_agent_end(AgentName :: binary(), Output :: term()) -> ok.
-callback on_tool_start(ToolName :: binary(), Args :: map()) -> ok.
-callback on_tool_end(ToolName :: binary(), Result :: term()) -> ok.
-callback on_error(Error :: term()) -> ok.
-callback before_agent(AgentName :: term(), Input :: term()) -> callback_result().
-callback after_agent(AgentName :: term(), Output :: term()) -> callback_result().
-callback before_model(Config :: map(), Memory :: list(), Tools :: list()) -> callback_result().
-callback after_model(Config :: map(), Result :: term()) -> callback_result().
-callback before_tool(ToolName :: binary(), Args :: map(), Context :: map()) -> callback_result().
-callback after_tool(ToolName :: binary(), Args :: map(), Context :: map(), Result :: term()) -> callback_result().

-type callback_result() :: ok | continue | {halt, term()} | {replace, term()}.

-optional_callbacks([
    on_agent_start/2, on_agent_end/2, on_tool_start/2, on_tool_end/2, on_error/1,
    before_agent/2, after_agent/2, before_model/3, after_model/2,
    before_tool/3, after_tool/4
]).

%% @doc Execute a callback hook across all registered handlers.
-spec execute(Handlers :: [module()], Hook :: atom(), Args :: [term()]) -> ok.
execute(Handlers, Hook, Args) ->
    PublicArgs = adk_callback_view:callback_args(Hook, Args),
    lists:foreach(fun(Handler) ->
        _ = invoke(Handler, Hook, PublicArgs)
    end, Handlers),
    ok.

%% @doc Run handlers until one explicitly replaces or halts the operation.
%% Observation-only callbacks return `ok' or `continue'. A callback may return
%% `{replace, Value}' to replace an after-hook result, or `{halt, Value}' to
%% skip the operation wrapped by a before-hook.
-spec run([module()], atom(), [term()]) -> continue | {halt, term()} | {replace, term()}.
run([], _Hook, _Args) ->
    continue;
run(Handlers, Hook, Args) ->
    PublicArgs = adk_callback_view:callback_args(Hook, Args),
    run_public(Handlers, Hook, PublicArgs).

run_public([], _Hook, _Args) ->
    continue;
run_public([Handler | Rest], Hook, Args) ->
    case invoke(Handler, Hook, Args) of
        {halt, _} = Halt -> Halt;
        {replace, _} = Replace -> Replace;
        _ -> run_public(Rest, Hook, Args)
    end.

invoke(Handler, Hook, Args) ->
    case code:ensure_loaded(Handler) of
        {module, Handler} ->
            case erlang:function_exported(Handler, Hook, length(Args)) of
                true ->
                    invoke_isolated(Handler, Hook, Args);
                false -> continue
            end;
        {error, Reason} ->
            Failure = adk_failure:exception(
                        callback, load, error, Reason),
            logger:warning("Unable to load callback ~p: ~p",
                           [Handler, Failure]),
            continue
    end.

invoke_isolated(Handler, Hook, Args) ->
    TimeoutMs = positive_env(callback_timeout_ms, ?DEFAULT_TIMEOUT_MS),
    MaxHeapWords = positive_env(callback_max_heap_words,
                                ?DEFAULT_MAX_HEAP_WORDS),
    Parent = self(),
    ResultRef = make_ref(),
    Deadline = erlang:monotonic_time(millisecond) + TimeoutMs,
    Worker = fun() ->
        Result = safe_apply(Handler, Hook, Args),
        Parent ! {adk_callback_result, ResultRef, self(), Result}
    end,
    Options =
        [monitor,
         {max_heap_size,
          #{size => MaxHeapWords, kill => true, error_logger => false}}],
    {Pid, MonitorRef} = erlang:spawn_opt(Worker, Options),
    await_result(Handler, Hook, Pid, MonitorRef, ResultRef, Deadline).

safe_apply(Handler, Hook, Args) ->
    try erlang:apply(Handler, Hook, Args) of
        Value -> Value
    catch
        Class:Reason:_Stack ->
            Failure = adk_failure:exception(
                        callback, Hook, Class, Reason),
            logger:error("Callback failed in ~p:~p: ~p",
                         [Handler, Hook, Failure]),
            continue
    end.

await_result(Handler, Hook, Pid, MonitorRef, ResultRef, Deadline) ->
    Remaining = remaining_ms(Deadline),
    receive
        {adk_callback_result, ResultRef, Pid, Value} ->
            await_worker_exit(Handler, Hook, Pid, MonitorRef, ResultRef,
                              Deadline, Value);
        {'DOWN', MonitorRef, process, Pid, Reason} ->
            log_worker_failure(Handler, Hook, Reason),
            flush_result(ResultRef, Pid),
            continue
    after Remaining ->
        terminate_worker(Handler, Hook, Pid, MonitorRef, ResultRef, timeout)
    end.

await_worker_exit(Handler, Hook, Pid, MonitorRef, ResultRef, Deadline, Value) ->
    Remaining = remaining_ms(Deadline),
    receive
        {'DOWN', MonitorRef, process, Pid, normal} ->
            Value;
        {'DOWN', MonitorRef, process, Pid, Reason} ->
            log_worker_failure(Handler, Hook, Reason),
            continue
    after Remaining ->
        terminate_worker(Handler, Hook, Pid, MonitorRef, ResultRef, timeout)
    end.

terminate_worker(Handler, Hook, Pid, MonitorRef, ResultRef, timeout) ->
    exit(Pid, kill),
    receive
        {'DOWN', MonitorRef, process, Pid, _Reason} -> ok
    end,
    flush_result(ResultRef, Pid),
    logger:error("Callback timed out in ~p:~p", [Handler, Hook]),
    continue.

flush_result(ResultRef, Pid) ->
    receive
        {adk_callback_result, ResultRef, Pid, _Value} -> ok
    after 0 ->
        ok
    end.

log_worker_failure(_Handler, _Hook, normal) ->
    ok;
log_worker_failure(Handler, Hook, Reason) ->
    Failure = adk_failure:external(callback, worker_down, Reason),
    logger:error("Callback worker failed in ~p:~p: ~p",
                 [Handler, Hook, Failure]).

remaining_ms(Deadline) ->
    max(0, Deadline - erlang:monotonic_time(millisecond)).

positive_env(Key, Default) ->
    case application:get_env(?APP, Key) of
        {ok, Value} when is_integer(Value), Value > 0 -> Value;
        _ -> Default
    end.