File size: 10,869 Bytes

ebed3db

%%%-------------------------------------------------------------------
%% @doc Sovereign Event Bus - Agent Lifecycle FSM
%%
%% 4-State Corrected FSM (per XML L2 spec):
%%   1. active - Processing events normally
%%   2. draining - Rejecting new events, processing queue
%%   3. checkpointed - Committed offset to L0 kernel, ready to stop
%%   4. stopped - Shutdown complete
%%
%% State Transitions:
%%   active -> draining:    shutdown/0 called or supervisor timeout
%%   draining -> checkpointed: queue empty AND offset committed to L0
%%   checkpointed -> stopped: final cleanup
%%   (all states can jump to stopped on fatal error)
%%
%% Drain Timeout: 30 seconds (per XML)
%% Offset Commit: Via seb_kernel_nif NIF
%%
%% @end
%%%-------------------------------------------------------------------
-module(seb_agent_fsm).
-behaviour(gen_statem).

-export([start_link/2]).
-export([init/1, callback_mode/0, terminate/3]).
-export([active/3, draining/3, checkpointed/3, stopped/3]).

%% Public API
-export([shutdown/1, get_state/1, queue_event/2, commit_offset/2]).

-record(data, {
    agent_id :: binary(),
    config :: map(),
    queue :: queue:queue(),
    current_offset :: non_neg_integer(),
    prior_offset :: non_neg_integer(),
    drain_timer :: reference() | undefined,
    drain_start :: integer() | undefined
}).

-define(DRAIN_TIMEOUT_MS, 30000).
-define(MAX_QUEUE_SIZE, 10000).

%%%===================================================================
%%% API
%%%===================================================================

%% @doc Start an agent FSM
%%
%% AgentId: Binary identifier
%% Config: Map with runtime configuration
%%
-spec start_link(binary(), map()) -> gen_statem:start_ret().
start_link(AgentId, Config) when is_binary(AgentId), is_map(Config) ->
    gen_statem:start_link(?MODULE, {AgentId, Config}, []).

%% @doc Initiate shutdown sequence (active -> draining)
-spec shutdown(pid()) -> ok | {error, term()}.
shutdown(Pid) when is_pid(Pid) ->
    gen_statem:call(Pid, shutdown).

%% @doc Get current FSM state
-spec get_state(pid()) -> atom().
get_state(Pid) when is_pid(Pid) ->
    gen_statem:call(Pid, get_state).

%% @doc Queue an event for processing
%%
%% Returns: ok | {error, queue_full}
-spec queue_event(pid(), term()) -> ok | {error, queue_full}.
queue_event(Pid, Event) when is_pid(Pid) ->
    gen_statem:call(Pid, {queue_event, Event}).

%% @doc Commit current offset to L0 kernel
%%
%% Offset: Event offset to commit
%% AgentPid: Self pid (for error reporting)
%%
-spec commit_offset(pid(), non_neg_integer()) -> ok | {error, term()}.
commit_offset(Pid, Offset) when is_pid(Pid), is_integer(Offset), Offset >= 0 ->
    gen_statem:call(Pid, {commit_offset, Offset}).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @doc Initialize the FSM
-spec init({binary(), map()}) -> gen_statem:init_ret().
init({AgentId, Config}) ->
    Data = #data{
        agent_id = AgentId,
        config = Config,
        queue = queue:new(),
        current_offset = 0,
        prior_offset = 0,
        drain_timer = undefined,
        drain_start = undefined
    },
    {ok, active, Data}.

%% @doc Use state_functions callback mode
-spec callback_mode() -> gen_statem:callback_mode().
callback_mode() ->
    state_functions.

%%%===================================================================
%%% State: active - Processing events normally
%%%===================================================================

%% @doc active/3 - Entry point and event handler
-spec active(gen_statem:event_type(), term(), #data{}) -> gen_statem:event_handler_ret().

%% Shutdown request - transition to draining
active(call, shutdown, Data) ->
    {next_state, draining, Data#data{
        drain_start = erlang:monotonic_time(millisecond)
    }, [{reply, ok}]};

%% Queue an event (reject if queue full)
active(call, {queue_event, Event}, Data) ->
    case queue:len(Data#data.queue) >= ?MAX_QUEUE_SIZE of
        true ->
            {keep_state_and_data, [{reply, {error, queue_full}}]};
        false ->
            NewQueue = queue:in(Event, Data#data.queue),
            NewData = Data#data{queue = NewQueue},
            Actions = [
                {reply, ok},
                {next_event, internal, process_queue}
            ],
            {keep_state, NewData, Actions}
    end;

%% Commit offset (accepts in active state)
active(call, {commit_offset, Offset}, Data) ->
    case commit_to_kernel(Offset) of
        ok ->
            {keep_state, Data#data{current_offset = Offset}, [{reply, ok}]};
        {error, Reason} ->
            {keep_state_and_data, [{reply, {error, Reason}}]}
    end;

%% Get state
active(call, get_state, _Data) ->
    {keep_state_and_data, [{reply, active}]};

%% Internal: process queue
active(internal, process_queue, Data) ->
    case queue:out(Data#data.queue) of
        {{value, Event}, NewQueue} ->
            case process_event(Event) of
                ok ->
                    {keep_state, Data#data{queue = NewQueue}, [{next_event, internal, process_queue}]};
                {error, _Reason} ->
                    {keep_state, Data#data{queue = NewQueue}, [{next_event, internal, process_queue}]}
            end;
        {empty, _Queue} ->
            keep_state_and_data
    end.

%%%===================================================================
%%% State: draining - Rejecting new events, processing queue
%%%===================================================================

%% @doc draining/3 - Entry point and event handler
-spec draining(gen_statem:event_type(), term(), #data{}) -> gen_statem:event_handler_ret().

%% Drain timeout - force transition to checkpointed
draining(info, {drain_timeout, _Ref}, Data) ->
    {next_state, checkpointed, Data#data{drain_timer = undefined}};

%% Reject new events while draining
draining(call, {queue_event, _Event}, _Data) ->
    {keep_state_and_data, [{reply, {error, agent_draining}}]};

%% Shutdown request - already draining
draining(call, shutdown, _Data) ->
    {keep_state_and_data, [{reply, ok}]};

%% Commit offset while draining
draining(call, {commit_offset, Offset}, Data) ->
    case commit_to_kernel(Offset) of
        ok ->
            NewData = Data#data{current_offset = Offset},
            case queue:is_empty(NewData#data.queue) of
                true ->
                    {next_state, checkpointed, NewData, [{reply, ok}]};
                false ->
                    {keep_state, NewData, [{reply, ok}]}
            end;
        {error, Reason} ->
            {keep_state_and_data, [{reply, {error, Reason}}]}
    end;

%% Get state
draining(call, get_state, _Data) ->
    {keep_state_and_data, [{reply, draining}]};

%% Process remaining queue items
draining(internal, process_queue, Data) ->
    case queue:out(Data#data.queue) of
        {{value, Event}, NewQueue} ->
            case process_event(Event) of
                ok ->
                    {keep_state, Data#data{queue = NewQueue}, [{next_event, internal, process_queue}]};
                {error, _Reason} ->
                    {keep_state, Data#data{queue = NewQueue}, [{next_event, internal, process_queue}]}
            end;
        {empty, Queue} ->
            case Data#data.drain_timer of
                undefined ->
                    Timer = erlang:send_after(?DRAIN_TIMEOUT_MS, self(), {drain_timeout, self()}),
                    {keep_state, Data#data{queue = Queue, drain_timer = Timer}};
                _AlreadySet ->
                    {keep_state, Data#data{queue = Queue}}
            end
    end;

%% On state entry, start draining
draining(enter, _PrevState, Data) ->
    {keep_state, Data, [{next_event, internal, process_queue}]}.

%%%===================================================================
%%% State: checkpointed - Offset committed, ready to stop
%%%===================================================================

%% @doc checkpointed/3 - Entry point and event handler
-spec checkpointed(gen_statem:event_type(), term(), #data{}) -> gen_statem:event_handler_ret().

%% Reject all operations in checkpointed state
checkpointed(call, {queue_event, _Event}, _Data) ->
    {keep_state_and_data, [{reply, {error, agent_checkpointed}}]};

checkpointed(call, shutdown, _Data) ->
    {keep_state_and_data, [{reply, ok}]};

checkpointed(call, {commit_offset, _Offset}, _Data) ->
    {keep_state_and_data, [{reply, {error, already_checkpointed}}]};

%% Get state
checkpointed(call, get_state, _Data) ->
    {keep_state_and_data, [{reply, checkpointed}]};

%% On state entry, transition to stopped
checkpointed(enter, _PrevState, Data) ->
    {next_state, stopped, Data}.

%%%===================================================================
%%% State: stopped - Shutdown complete
%%%===================================================================

%% @doc stopped/3 - Final state, no more transitions
-spec stopped(gen_statem:event_type(), term(), #data{}) -> gen_statem:event_handler_ret().

stopped(call, get_state, _Data) ->
    {keep_state_and_data, [{reply, stopped}]};

stopped(call, _Request, _Data) ->
    {keep_state_and_data, [{reply, {error, agent_stopped}}]}.

%%%===================================================================
%%% Cleanup
%%%===================================================================

%% @doc Terminate callback
-spec terminate(term(), gen_statem:state(), #data{}) -> ok.
terminate(_Reason, _State, #data{drain_timer = Timer}) ->
    case Timer of
        undefined -> ok;
        Ref -> erlang:cancel_timer(Ref)
    end.

%%%===================================================================
%%% Internal Functions
%%%===================================================================

%% @doc Process a single event
%%
%% In a real implementation, this would:
%%   1. Extract intent/context/authority from event
%%   2. Call policy gate (seb_datalog_bridge)
%%   3. Route to execution adapter
%%   4. Seal with WORM
%%
%% For now, we just succeed.
-spec process_event(term()) -> ok | {error, term()}.
process_event(_Event) ->
    ok.

%% @doc Commit offset to L0 kernel via NIF
%%
%% This bridges to the Ada kernel interface (seb_kernel_nif).
%% Verifies offset monotonicity before committing.
%%
-spec commit_to_kernel(non_neg_integer()) -> ok | {error, term()}.
commit_to_kernel(Offset) when is_integer(Offset), Offset >= 0 ->
    try
        %% Call NIF function: seb_kernel_nif:commit_offset/1
        %% Per XML spec, this commits the offset to the L0 kernel
        try seb_kernel_nif:commit_offset(Offset) of
            ok -> ok;
            Error -> {error, Error}
        catch
            _:Reason -> {error, {nif_error, Reason}}
        end
    catch
        _Type:_Reason ->
            {error, nif_unavailable}
    end.