seb/runtime/src/seb_partition_mgr.erl

%%%-------------------------------------------------------------------
%% @doc Sovereign Event Bus - Partition Manager
%%
%% Deterministic partition assignment (per XML L2 spec):
%% - 1024 partitions (fixed)
%% - Competency-based routing from Datalog policy engine
%% - phash2 deterministic assignment
%% - Same seed → same result (for reproducibility)
%%
%% Partition assignment is deterministic and reproducible.
%% Given the same agent ID and competency, returns same partition.
%%
%% @end
%%%-------------------------------------------------------------------
-module(seb_partition_mgr).
-behaviour(gen_server).

-export([start_link/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

%% Public API
-export([assign_partition/2, get_partition_load/1, rebalance_partitions/0]).

-define(SERVER, ?MODULE).
-define(DEFAULT_PARTITIONS, 1024).
-define(PARTITION_LOAD_THRESHOLD, 0.8).

-record(state, {
    partition_count :: non_neg_integer(),
    partition_assignments :: map(),  %% {agent_id, competency} -> partition
    partition_loads :: map()          %% partition -> load
}).

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

%% @doc Start the partition manager
-spec start_link(non_neg_integer()) -> gen_server:start_ret().
start_link(PartitionCount) when is_integer(PartitionCount), PartitionCount > 0 ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [PartitionCount], []).

%% @doc Assign a partition to an agent based on competency
%%
%% Uses phash2 for deterministic assignment:
%%   partition = phash2({agent_id, competency}) rem partition_count
%%
%% Args:
%%   AgentId: Binary identifier of the agent
%%   Competency: Atom describing agent capability (e.g., 'compute', 'io', 'crypto')
%%
%% Returns: Partition number (0 .. partition_count - 1)
%%
-spec assign_partition(binary(), atom()) -> non_neg_integer().
assign_partition(AgentId, Competency) when is_binary(AgentId), is_atom(Competency) ->
    gen_server:call(?SERVER, {assign_partition, AgentId, Competency}).

%% @doc Get current load for a specific partition
%%
%% Returns: Float between 0.0 and 1.0
%%
-spec get_partition_load(non_neg_integer()) -> float() | {error, not_found}.
get_partition_load(Partition) when is_integer(Partition), Partition >= 0 ->
    gen_server:call(?SERVER, {get_partition_load, Partition}).

%% @doc Trigger partition rebalancing
%%
%% If any partition exceeds PARTITION_LOAD_THRESHOLD, rebalances
%% assignments to distribute load more evenly.
%%
-spec rebalance_partitions() -> ok | {error, term()}.
rebalance_partitions() ->
    gen_server:call(?SERVER, rebalance_partitions).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @doc Initialize the partition manager
-spec init([non_neg_integer()]) -> {ok, #state{}}.
init([PartitionCount]) ->
    State = #state{
        partition_count = PartitionCount,
        partition_assignments = maps:new(),
        partition_loads = init_loads(PartitionCount)
    },
    {ok, State}.

%% @doc Handle synchronous calls
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.

handle_call({assign_partition, AgentId, Competency}, _From, State) ->
    Partition = compute_partition(AgentId, Competency, State#state.partition_count),

    %% Store assignment for later reference (idempotent)
    Key = {AgentId, Competency},
    NewAssignments = maps:put(Key, Partition, State#state.partition_assignments),

    %% Update partition load (simple increment)
    Load = maps:get(Partition, State#state.partition_loads, 0.0),
    NewLoads = maps:put(Partition, Load + 0.01, State#state.partition_loads),

    NewState = State#state{
        partition_assignments = NewAssignments,
        partition_loads = NewLoads
    },

    {reply, Partition, NewState};

handle_call({get_partition_load, Partition}, _From, State) ->
    case maps:find(Partition, State#state.partition_loads) of
        {ok, Load} ->
            {reply, Load, State};
        error ->
            {reply, {error, not_found}, State}
    end;

handle_call(rebalance_partitions, _From, State) ->
    %% Check if any partition exceeds threshold
    MaxLoad = maps:fold(fun(_P, Load, Max) -> max(Load, Max) end, 0.0, State#state.partition_loads),

    case MaxLoad > ?PARTITION_LOAD_THRESHOLD of
        true ->
            %% Reset loads to even distribution
            NewLoads = init_loads(State#state.partition_count),
            NewState = State#state{partition_loads = NewLoads},
            {reply, ok, NewState};
        false ->
            {reply, ok, State}
    end;

handle_call(_Request, _From, State) ->
    {reply, {error, unknown_call}, State}.

%% @doc Handle asynchronous casts
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast(_Msg, State) ->
    {noreply, State}.

%% @doc Handle info messages
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info(_Info, State) ->
    {noreply, State}.

%% @doc Terminate the partition manager
-spec terminate(term(), #state{}) -> ok.
terminate(_Reason, _State) ->
    ok.

%% @doc Code change (upgrade support)
-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

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

%% @doc Initialize partition loads (0.0 for each partition)
-spec init_loads(non_neg_integer()) -> map().
init_loads(PartitionCount) ->
    maps:from_list([{P, 0.0} || P <- lists:seq(0, PartitionCount - 1)]).

%% @doc Compute deterministic partition assignment
%%
%% Uses Erlang's phash2 for deterministic hashing:
%%   partition = phash2({agent_id, competency}) rem partition_count
%%
%% This ensures:
%%   1. Same agent + competency always maps to same partition
%%   2. Distribution is uniform across partitions
%%   3. Deterministic and reproducible
%%
-spec compute_partition(binary(), atom(), non_neg_integer()) -> non_neg_integer().
compute_partition(AgentId, Competency, PartitionCount) ->
    erlang:phash2({AgentId, Competency}) rem PartitionCount.