Add src/hk_browser_session.erl

src/hk_browser_session.erl

@@ -0,0 +1,415 @@
+%%%-------------------------------------------------------------------
+%%% @doc HK-BROWSER :: Chromium Controller (one process per session).
+%%%
+%%% One `hk_browser_session' gen_server owns exactly one headless
+%%% Chromium OS process (its own `--user-data-dir', its own
+%%% `--remote-debugging-port') and every tab opened within it. This
+%%% is the process boundary the architecture calls for: browser
+%%% state (which tabs exist, what URL each is on) lives *only* here,
+%%% reached exclusively through the typed operations below -- never
+%%% mixed into agent state or read directly from another process.
+%%%
+%%% Every operation:
+%%%   1. is validated (tab exists, session is `ready') before any
+%%%      Chromium interaction is attempted;
+%%%   2. drives Chromium through `hk_cdp_client' (never through ad
+%%%      hoc process state);
+%%%   3. publishes a structured `hk_event' recording what happened,
+%%%      win or lose -- per the agent_browser_interface requirement
+%%%      that "every operation must produce a structured event."
+%%%
+%%% Click/type are implemented via `Runtime.evaluate' (DOM method
+%%% calls and property assignment) rather than CDP's `Input' domain
+%%% (synthetic OS-level mouse/keyboard events). This is a documented
+%%% simplification, not a hidden one: it's adequate for
+%%% programmatic interaction with most pages, but it will not
+%%% exercise a page's native pointer/keyboard event handlers the way
+%%% `Input.dispatchMouseEvent'/`Input.dispatchKeyEvent' would. A
+%%% production build would add an `Input'-domain code path behind
+%%% the same typed operation names.
+%%% @end
+%%%-------------------------------------------------------------------
+-module(hk_browser_session).
+-behaviour(gen_server).
+
+-include("hyperkitty.hrl").
+
+-export([start_link/1]).
+-export([open_tab/2, close_tab/2, navigate/3, back/2, forward/2, reload/2,
+         read_page/2, query_element/3, click/3, type/4, scroll/3,
+         screenshot/2, extract_links/2, close_session/1, describe/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
+
+-define(CHROME_READY_TIMEOUT_MS, 15000).
+-define(NAV_TIMEOUT_MS, 30000).
+
+-record(tab, {
+    tab_id     :: binary(),
+    target_id  :: binary(),
+    cdp_pid    :: pid(),
+    url        :: binary() | undefined,
+    title      :: binary() | undefined,
+    status     :: opening | loading | idle | closed,
+    created_at_ms :: integer(),
+    updated_at_ms :: integer()
+}).
+
+-record(state, {
+    session_id      :: binary(),
+    owner_agent_id  :: binary() | undefined,
+    profile         :: binary(),
+    debug_port      :: inet:port_number(),
+    user_data_dir   :: string(),
+    os_port         :: port(),
+    status = starting :: starting | ready | busy | closing | closed | crashed,
+    tabs = #{}      :: #{binary() => #tab{}},
+    cdp_pid_to_tab  = #{} :: #{pid() => binary()},
+    waiters = #{}   :: #{{binary(), atom()} => gen_server:from()},
+    created_at_ms   :: integer()
+}).
+
+%% Start args: #{session_id, owner_agent_id, profile}
+start_link(Args) ->
+    gen_server:start_link(?MODULE, Args, []).
+
+open_tab(Pid, Url) -> gen_server:call(Pid, {open_tab, Url}, ?CHROME_READY_TIMEOUT_MS).
+close_tab(Pid, TabId) -> gen_server:call(Pid, {close_tab, TabId}).
+navigate(Pid, TabId, Url) -> gen_server:call(Pid, {navigate, TabId, Url}, ?NAV_TIMEOUT_MS).
+back(Pid, TabId) -> gen_server:call(Pid, {history, TabId, back}, ?NAV_TIMEOUT_MS).
+forward(Pid, TabId) -> gen_server:call(Pid, {history, TabId, forward}, ?NAV_TIMEOUT_MS).
+reload(Pid, TabId) -> gen_server:call(Pid, {reload, TabId}, ?NAV_TIMEOUT_MS).
+read_page(Pid, TabId) -> gen_server:call(Pid, {read_page, TabId}).
+query_element(Pid, TabId, Selector) -> gen_server:call(Pid, {query_element, TabId, Selector}).
+click(Pid, TabId, Selector) -> gen_server:call(Pid, {click, TabId, Selector}).
+type(Pid, TabId, Selector, Text) -> gen_server:call(Pid, {type, TabId, Selector, Text}).
+scroll(Pid, TabId, DeltaY) -> gen_server:call(Pid, {scroll, TabId, DeltaY}).
+screenshot(Pid, TabId) -> gen_server:call(Pid, {screenshot, TabId}).
+extract_links(Pid, TabId) -> gen_server:call(Pid, {extract_links, TabId}).
+close_session(Pid) -> gen_server:call(Pid, close_session).
+describe(Pid) -> gen_server:call(Pid, describe).
+
+%%% gen_server callbacks
+
+init(#{session_id := SessionId} = Args) ->
+    process_flag(trap_exit, true),
+    Profile = maps:get(profile, Args, <<"default">>),
+    OwnerAgentId = maps:get(owner_agent_id, Args, undefined),
+    Port = find_free_port(),
+    UserDataDir = filename:join(["/tmp/hyperkitty_profiles", binary_to_list(SessionId)]),
+    ok = filelib:ensure_dir(UserDataDir ++ "/."),
+    ChromeExe = application:get_env(hyperkitty, chromium_executable,
+                                     "/usr/bin/chromium"),
+    Args0 = [
+        "--headless=new",
+        "--remote-debugging-port=" ++ integer_to_list(Port),
+        "--remote-debugging-address=127.0.0.1",
+        "--user-data-dir=" ++ UserDataDir,
+        "--no-sandbox",
+        "--disable-gpu",
+        "--disable-dev-shm-usage",
+        "about:blank"
+    ],
+    OsPort = erlang:open_port({spawn_executable, ChromeExe},
+                               [{args, Args0}, exit_status, binary, stderr_to_stdout,
+                                {line, 4096}]),
+    State = #state{
+        session_id = SessionId,
+        owner_agent_id = OwnerAgentId,
+        profile = Profile,
+        debug_port = Port,
+        user_data_dir = UserDataDir,
+        os_port = OsPort,
+        created_at_ms = hk_id:timestamp_ms()
+    },
+    case hk_chrome_http:wait_for_ready(Port, ?CHROME_READY_TIMEOUT_MS) of
+        ok ->
+            ok = hk_browser_registry:register(SessionId, self()),
+            emit(State, <<"browser.session.created">>, #{profile => Profile, debug_port => Port}),
+            {ok, State#state{status = ready}};
+        {error, timeout} ->
+            catch erlang:port_close(OsPort),
+            {stop, chrome_launch_timeout}
+    end.
+
+handle_call(describe, _From, State) ->
+    {reply, {ok, session_map(State)}, State};
+
+handle_call({open_tab, Url}, _From, State = #state{status = ready}) ->
+    case hk_chrome_http:new_target(State#state.debug_port, Url) of
+        {ok, #{target_id := TargetId, ws_url := WsUrl}} ->
+            {ok, CdpPid} = hk_cdp_client:start_link(WsUrl, self()),
+            _ = hk_cdp_client:command(CdpPid, <<"Page.enable">>, #{}),
+            _ = hk_cdp_client:command(CdpPid, <<"Runtime.enable">>, #{}),
+            TabId = hk_id:new(<<"tab">>),
+            Now = hk_id:timestamp_ms(),
+            Tab = #tab{tab_id = TabId, target_id = TargetId, cdp_pid = CdpPid,
+                       url = Url, status = loading, created_at_ms = Now, updated_at_ms = Now},
+            State2 = State#state{
+                tabs = maps:put(TabId, Tab, State#state.tabs),
+                cdp_pid_to_tab = maps:put(CdpPid, TabId, State#state.cdp_pid_to_tab)
+            },
+            emit(State2, <<"browser.tab.opened">>, #{tab_id => TabId, url => Url}),
+            {reply, {ok, TabId}, State2};
+        {error, Reason} ->
+            emit(State, <<"browser.tab.open_failed">>, #{url => Url, reason => term_bin(Reason)}),
+            {reply, {error, Reason}, State}
+    end;
+
+handle_call({close_tab, TabId}, _From, State) ->
+    case maps:find(TabId, State#state.tabs) of
+        {ok, Tab} ->
+            hk_cdp_client:close(Tab#tab.cdp_pid),
+            _ = hk_chrome_http:close_target(State#state.debug_port, Tab#tab.target_id),
+            State2 = drop_tab(TabId, Tab, State),
+            emit(State2, <<"browser.tab.closed">>, #{tab_id => TabId}),
+            {reply, ok, State2};
+        error ->
+            {reply, {error, no_such_tab}, State}
+    end;
+
+handle_call({navigate, TabId, Url}, From, State) ->
+    with_tab(TabId, State, fun(Tab) ->
+        emit(State, <<"browser.navigation.started">>, #{tab_id => TabId, url => Url}),
+        _ = hk_cdp_client:command(Tab#tab.cdp_pid, <<"Page.navigate">>, #{<<"url">> => Url}),
+        Tab2 = Tab#tab{url = Url, status = loading, updated_at_ms = hk_id:timestamp_ms()},
+        State2 = put_tab(Tab2, State),
+        Key = {TabId, load_event},
+        {noreply, State2#state{waiters = maps:put(Key, From, State2#state.waiters)}}
+    end);
+
+handle_call({history, TabId, Direction}, From, State) ->
+    with_tab(TabId, State, fun(Tab) ->
+        {ok, HistResult} = hk_cdp_client:command(Tab#tab.cdp_pid, <<"Page.getNavigationHistory">>, #{}),
+        #{<<"currentIndex">> := Idx, <<"entries">> := Entries} = HistResult,
+        TargetIdx = case Direction of back -> Idx - 1; forward -> Idx + 1 end,
+        case TargetIdx >= 0 andalso TargetIdx < length(Entries) of
+            true ->
+                Entry = lists:nth(TargetIdx + 1, Entries),
+                EntryId = maps:get(<<"id">>, Entry),
+                _ = hk_cdp_client:command(Tab#tab.cdp_pid, <<"Page.navigateToHistoryEntry">>,
+                                           #{<<"entryId">> => EntryId}),
+                emit(State, <<"browser.navigation.started">>,
+                     #{tab_id => TabId, direction => Direction}),
+                Key = {TabId, load_event},
+                {noreply, State#state{waiters = maps:put(Key, From, State#state.waiters)}};
+            false ->
+                {reply, {error, no_history_entry}, State}
+        end
+    end);
+
+handle_call({reload, TabId}, From, State) ->
+    with_tab(TabId, State, fun(Tab) ->
+        _ = hk_cdp_client:command(Tab#tab.cdp_pid, <<"Page.reload">>, #{}),
+        emit(State, <<"browser.navigation.started">>, #{tab_id => TabId, reload => true}),
+        Key = {TabId, load_event},
+        {noreply, State#state{waiters = maps:put(Key, From, State#state.waiters)}}
+    end);
+
+handle_call({read_page, TabId}, _From, State) ->
+    with_tab_sync(TabId, State, fun(Tab) ->
+        eval_result(hk_cdp_client:command(Tab#tab.cdp_pid, <<"Runtime.evaluate">>,
+            #{<<"expression">> => <<"document.body ? document.body.innerText : ''">>,
+              <<"returnByValue">> => true}))
+    end, fun(Result) ->
+        emit(State, <<"browser.page.read">>, #{tab_id => TabId, chars => text_len(Result)})
+    end);
+
+handle_call({query_element, TabId, Selector}, _From, State) ->
+    Expr = query_element_js(Selector),
+    with_tab_sync(TabId, State, fun(Tab) ->
+        eval_result(hk_cdp_client:command(Tab#tab.cdp_pid, <<"Runtime.evaluate">>,
+            #{<<"expression">> => Expr, <<"returnByValue">> => true}))
+    end, fun(_Result) ->
+        emit(State, <<"browser.element.queried">>, #{tab_id => TabId, selector => Selector})
+    end);
+
+handle_call({click, TabId, Selector}, _From, State) ->
+    Expr = click_js(Selector),
+    with_tab_sync(TabId, State, fun(Tab) ->
+        eval_result(hk_cdp_client:command(Tab#tab.cdp_pid, <<"Runtime.evaluate">>,
+            #{<<"expression">> => Expr, <<"returnByValue">> => true}))
+    end, fun(_Result) ->
+        emit(State, <<"browser.action.click">>, #{tab_id => TabId, selector => Selector})
+    end);
+
+handle_call({type, TabId, Selector, Text}, _From, State) ->
+    Expr = type_js(Selector, Text),
+    with_tab_sync(TabId, State, fun(Tab) ->
+        eval_result(hk_cdp_client:command(Tab#tab.cdp_pid, <<"Runtime.evaluate">>,
+            #{<<"expression">> => Expr, <<"returnByValue">> => true}))
+    end, fun(_Result) ->
+        emit(State, <<"browser.action.type">>, #{tab_id => TabId, selector => Selector,
+                                                  chars => byte_size(Text)})
+    end);
+
+handle_call({scroll, TabId, DeltaY}, _From, State) ->
+    Expr = list_to_binary(io_lib:format("window.scrollBy(0, ~w); 'ok'", [DeltaY])),
+    with_tab_sync(TabId, State, fun(Tab) ->
+        eval_result(hk_cdp_client:command(Tab#tab.cdp_pid, <<"Runtime.evaluate">>,
+            #{<<"expression">> => Expr, <<"returnByValue">> => true}))
+    end, fun(_Result) ->
+        emit(State, <<"browser.action.scroll">>, #{tab_id => TabId, delta_y => DeltaY})
+    end);
+
+handle_call({screenshot, TabId}, _From, State) ->
+    with_tab_sync(TabId, State, fun(Tab) ->
+        case hk_cdp_client:command(Tab#tab.cdp_pid, <<"Page.captureScreenshot">>,
+                                     #{<<"format">> => <<"png">>}) of
+            {ok, #{<<"data">> := B64}} -> {ok, B64};
+            {error, Reason} -> {error, Reason}
+        end
+    end, fun(_Result) ->
+        emit(State, <<"browser.screenshot.captured">>, #{tab_id => TabId})
+    end);
+
+handle_call({extract_links, TabId}, _From, State) ->
+    Expr = <<"JSON.stringify(Array.from(document.querySelectorAll('a[href]'))"
+             ".map(a => ({href: a.href, text: a.textContent.trim()})))">>,
+    with_tab_sync(TabId, State, fun(Tab) ->
+        case eval_result(hk_cdp_client:command(Tab#tab.cdp_pid, <<"Runtime.evaluate">>,
+                #{<<"expression">> => Expr, <<"returnByValue">> => true})) of
+            {ok, Json} when is_binary(Json) ->
+                try {ok, jsx:decode(Json, [return_maps])}
+                catch _:_ -> {ok, []} end;
+            Other -> Other
+        end
+    end, fun(Result) ->
+        Count = case Result of L when is_list(L) -> length(L); _ -> 0 end,
+        emit(State, <<"browser.links.extracted">>, #{tab_id => TabId, count => Count})
+    end);
+
+handle_call(close_session, _From, State) ->
+    State2 = State#state{status = closing},
+    maps:foreach(fun(_TabId, Tab) -> hk_cdp_client:close(Tab#tab.cdp_pid) end, State2#state.tabs),
+    catch erlang:port_close(State2#state.os_port),
+    emit(State2, <<"browser.session.closed">>, #{}),
+    {stop, normal, ok, State2#state{status = closed}}.
+
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+%% CDP event fan-in: correlate `Page.loadEventFired' back to whichever
+%% navigate/back/forward/reload call is waiting on this tab.
+handle_info({cdp_event, CdpPid, <<"Page.loadEventFired">>, _Params}, State) ->
+    case maps:find(CdpPid, State#state.cdp_pid_to_tab) of
+        {ok, TabId} ->
+            Key = {TabId, load_event},
+            case maps:take(Key, State#state.waiters) of
+                {From, Waiters2} ->
+                    emit(State, <<"browser.navigation.completed">>, #{tab_id => TabId}),
+                    gen_server:reply(From, ok),
+                    {noreply, State#state{waiters = Waiters2}};
+                error ->
+                    {noreply, State}
+            end;
+        error ->
+            {noreply, State}
+    end;
+handle_info({cdp_event, _CdpPid, _Method, _Params}, State) ->
+    {noreply, State}; %% other CDP events are available for future views (console, network, ...)
+handle_info({Port, {exit_status, Status}}, State = #state{os_port = Port, status = St})
+        when St =/= closing ->
+    emit(State, <<"browser.session.crashed">>, #{exit_status => Status}),
+    {stop, {error, {chrome_process_exited, Status}}, State#state{status = crashed}};
+handle_info({Port, {exit_status, _Status}}, State = #state{os_port = Port}) ->
+    {noreply, State};
+handle_info({Port, {data, _}}, State = #state{os_port = Port}) ->
+    {noreply, State};
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+%%% internal
+
+find_free_port() ->
+    {ok, Socket} = gen_tcp:listen(0, [{reuseaddr, true}]),
+    {ok, Port} = inet:port(Socket),
+    ok = gen_tcp:close(Socket),
+    Port.
+
+with_tab(TabId, State, Fun) ->
+    case maps:find(TabId, State#state.tabs) of
+        {ok, Tab} -> Fun(Tab);
+        error -> {reply, {error, no_such_tab}, State}
+    end.
+
+%% For operations that complete synchronously against the CDP client
+%% (i.e. not waiting on a later async event like page load): run
+%% `EvalFun(Tab) -> {ok, Result} | {error, Reason}', emit `OnOk(Result)'
+%% only on success, and always reply from this call.
+with_tab_sync(TabId, State, EvalFun, OnOk) ->
+    case maps:find(TabId, State#state.tabs) of
+        {ok, Tab} ->
+            case EvalFun(Tab) of
+                {ok, Result} ->
+                    OnOk(Result),
+                    {reply, {ok, Result}, State};
+                {error, Reason} ->
+                    emit(State, <<"browser.action.failed">>,
+                         #{tab_id => TabId, reason => term_bin(Reason)}),
+                    {reply, {error, Reason}, State}
+            end;
+        error ->
+            {reply, {error, no_such_tab}, State}
+    end.
+
+eval_result({ok, #{<<"exceptionDetails">> := Details}}) ->
+    {error, {js_exception, Details}};
+eval_result({ok, #{<<"result">> := #{<<"value">> := Value}}}) ->
+    {ok, Value};
+eval_result({ok, #{<<"result">> := _}}) ->
+    {ok, undefined};
+eval_result({error, Reason}) ->
+    {error, Reason}.
+
+query_element_js(Selector) ->
+    SelJson = jsx:encode(Selector),
+    <<"(() => { const el = document.querySelector(", SelJson/binary, "); "
+      "return el ? JSON.stringify({exists:true, tag: el.tagName, "
+      "text:(el.textContent||'').trim().slice(0,500)}) : "
+      "JSON.stringify({exists:false}); })()">>.
+
+click_js(Selector) ->
+    SelJson = jsx:encode(Selector),
+    <<"(() => { const el = document.querySelector(", SelJson/binary, "); "
+      "if (!el) return 'not_found'; el.click(); return 'ok'; })()">>.
+
+type_js(Selector, Text) ->
+    SelJson = jsx:encode(Selector),
+    TextJson = jsx:encode(Text),
+    <<"(() => { const el = document.querySelector(", SelJson/binary, "); "
+      "if (!el) return 'not_found'; el.focus(); el.value = ", TextJson/binary, "; "
+      "el.dispatchEvent(new Event('input', {bubbles:true})); "
+      "el.dispatchEvent(new Event('change', {bubbles:true})); return 'ok'; })()">>.
+
+put_tab(Tab, State) ->
+    State#state{tabs = maps:put(Tab#tab.tab_id, Tab, State#state.tabs)}.
+
+drop_tab(TabId, Tab, State) ->
+    State#state{
+        tabs = maps:remove(TabId, State#state.tabs),
+        cdp_pid_to_tab = maps:remove(Tab#tab.cdp_pid, State#state.cdp_pid_to_tab)
+    }.
+
+session_map(State) ->
+    #browser_session{
+        session_id = State#state.session_id,
+        owner_agent_id = State#state.owner_agent_id,
+        profile = State#state.profile,
+        status = State#state.status,
+        tabs = maps:keys(State#state.tabs),
+        created_at_ms = State#state.created_at_ms,
+        updated_at_ms = hk_id:timestamp_ms()
+    }.
+
+emit(State, Category, Data) ->
+    Event = hk_event:new(Category, {browser_session, State#state.session_id}, Data),
+    catch hk_event_bus:publish(Event).
+
+term_bin(Term) ->
+    iolist_to_binary(io_lib:format("~p", [Term])).
+
+text_len(B) when is_binary(B) -> byte_size(B);
+text_len(_) -> 0.