%% @doc Public, reconnectable invocation API. %% %% Runs are owned by a supervised adk_invocation process rather than by the %% process which starts or subscribes to them. Subscribers receive: %% %% {adk_run_event, RunId, Sequence, Event} %% {adk_run_terminal, RunId, Sequence, Outcome} %% %% Outcome is one of `{completed, Output}', `{paused, Event}', %% `{cancelled, Reason}', or `{failed, Reason}'. Every run commits exactly one %% terminal outcome. A late subscriber receives the bounded event replay and %% the same terminal outcome while the run remains retained. %% %% HTTP and other back-pressure-sensitive consumers should use %% subscribe_credit/2,3 and acknowledge each delivered sequence with ack/2,3. %% A credit subscriber has at most one unacknowledged run message in its %% mailbox. The original subscribe/1,2 push protocol remains unchanged for %% local Erlang processes which already drain their mailboxes promptly. -module(adk_run). -export([start/4, start/5, resume/2, resume/3, subscribe/1, subscribe/2, subscribe_credit/2, subscribe_credit/3, ack/2, ack/3, unsubscribe/1, unsubscribe/2, status/1, await/1, await/2, cancel/1, cancel/2]). -define(DEFAULT_CALL_TIMEOUT, 5000). -type run_id() :: binary(). -type outcome() :: {completed, binary() | adk_content:content()} | {paused, term()} | {cancelled, term()} | {failed, term()}. -export_type([run_id/0, outcome/0]). -spec start(adk_runner:runner(), binary(), binary(), term()) -> {ok, run_id()} | {error, term()}. start(Runner, UserId, SessionId, Message) -> start(Runner, UserId, SessionId, Message, #{}). -spec start(adk_runner:runner(), binary(), binary(), term(), map()) -> {ok, run_id()} | {error, term()}. start(Runner, UserId, SessionId, Message, Opts) when is_binary(UserId), is_binary(SessionId), is_map(Opts) -> RunId = generate_run_id(), Request = #{runner => Runner, user_id => UserId, session_id => SessionId, message => Message}, try adk_invocation_sup:start_invocation(RunId, Request, Opts) of {ok, _Pid} -> {ok, RunId}; {ok, _Pid, _Info} -> {ok, RunId}; {error, Reason} -> {error, Reason} catch exit:{noproc, _} -> {error, invocation_supervisor_not_started}; exit:Reason -> {error, Reason} end; start(_Runner, UserId, SessionId, _Message, Opts) -> {error, {invalid_start_arguments, UserId, SessionId, Opts}}. %% @doc Resume a terminal paused run as a new supervised run. %% %% The paused run remains immutable and retained for inspection. The returned %% run has its own event sequence and terminal outcome, while status/1 links %% the two runs through `resumed_to' and `parent_run_id'. Only one resume may be %% accepted for a paused run. -spec resume(run_id(), term()) -> {ok, run_id()} | {error, term()}. resume(PausedRunId, ToolResponse) -> resume(PausedRunId, ToolResponse, #{}). -spec resume(run_id(), term(), map()) -> {ok, run_id()} | {error, term()}. resume(PausedRunId, ToolResponse, Opts) when is_binary(PausedRunId), is_map(Opts) -> NewRunId = generate_run_id(), call_run(PausedRunId, {resume, NewRunId, ToolResponse, Opts}, ?DEFAULT_CALL_TIMEOUT); resume(PausedRunId, _ToolResponse, Opts) -> {error, {invalid_resume_arguments, PausedRunId, Opts}}. -spec subscribe(run_id()) -> ok | {error, term()}. subscribe(RunId) -> subscribe(RunId, self()). -spec subscribe(run_id(), pid()) -> ok | {error, term()}. subscribe(RunId, Subscriber) when is_pid(Subscriber) -> call_run(RunId, {subscribe, Subscriber}, ?DEFAULT_CALL_TIMEOUT). %% @doc Subscribe with one-message credit starting strictly after Cursor. %% %% The result includes the invocation's latest sequence and terminal state at %% the subscription boundary. If Cursor is older than the bounded replay %% window, no subscription is installed and `{error, {replay_gap, Details}}' %% is returned. Callers must acknowledge each event before another event is %% released. A terminal message does not need acknowledgement. -spec subscribe_credit(run_id(), non_neg_integer()) -> {ok, map()} | {error, term()}. subscribe_credit(RunId, Cursor) -> subscribe_credit(RunId, self(), Cursor). -spec subscribe_credit(run_id(), pid(), non_neg_integer()) -> {ok, map()} | {error, term()}. subscribe_credit(RunId, Subscriber, Cursor) when is_pid(Subscriber), is_integer(Cursor), Cursor >= 0 -> call_run(RunId, {subscribe_credit, Subscriber, Cursor}, ?DEFAULT_CALL_TIMEOUT); subscribe_credit(_RunId, Subscriber, Cursor) -> {error, {invalid_credit_subscription, Subscriber, Cursor}}. %% @doc Return credit for the exact event Sequence just consumed. -spec ack(run_id(), non_neg_integer()) -> ok | {error, term()}. ack(RunId, Sequence) -> ack(RunId, self(), Sequence). -spec ack(run_id(), pid(), non_neg_integer()) -> ok | {error, term()}. ack(RunId, Subscriber, Sequence) when is_pid(Subscriber), is_integer(Sequence), Sequence >= 0 -> call_run(RunId, {ack, Subscriber, Sequence}, ?DEFAULT_CALL_TIMEOUT); ack(_RunId, Subscriber, Sequence) -> {error, {invalid_credit_ack, Subscriber, Sequence}}. -spec unsubscribe(run_id()) -> ok | {error, term()}. unsubscribe(RunId) -> unsubscribe(RunId, self()). -spec unsubscribe(run_id(), pid()) -> ok | {error, term()}. unsubscribe(RunId, Subscriber) when is_pid(Subscriber) -> call_run(RunId, {unsubscribe, Subscriber}, ?DEFAULT_CALL_TIMEOUT). -spec status(run_id()) -> {ok, map()} | {error, term()}. status(RunId) -> case call_run(RunId, status, ?DEFAULT_CALL_TIMEOUT) of Status when is_map(Status) -> {ok, Status}; {error, _} = Error -> Error end. -spec await(run_id()) -> outcome() | {error, term()}. await(RunId) -> await(RunId, infinity). -spec await(run_id(), timeout()) -> outcome() | {error, term()}. await(RunId, Timeout) when Timeout =:= infinity; is_integer(Timeout), Timeout >= 0 -> call_run(RunId, await, Timeout). -spec cancel(run_id()) -> ok | {error, term()}. cancel(RunId) -> cancel(RunId, user_cancelled). -spec cancel(run_id(), term()) -> ok | {error, term()}. cancel(RunId, Reason) -> call_run(RunId, {cancel, Reason}, ?DEFAULT_CALL_TIMEOUT). call_run(RunId, Request, Timeout) when is_binary(RunId) -> try adk_run_registry:lookup(RunId) of {ok, Pid} -> safe_call(Pid, Request, Timeout); {error, not_found} = Error -> Error catch exit:{noproc, _} -> {error, run_registry_not_started}; exit:Reason -> {error, Reason} end; call_run(RunId, _Request, _Timeout) -> {error, {invalid_run_id, RunId}}. safe_call(Pid, Request, Timeout) -> try gen_statem:call(Pid, Request, Timeout) of Reply -> Reply catch exit:{timeout, _} -> {error, timeout}; exit:{noproc, _} -> {error, not_found}; exit:{normal, _} -> {error, not_found}; exit:{shutdown, _} -> {error, not_found}; exit:Reason -> {error, {run_call_failed, Reason}} end. generate_run_id() -> <> = crypto:strong_rand_bytes(16), list_to_binary( io_lib:format( "run-~8.16.0b-~4.16.0b-4~3.16.0b-~4.16.0b-~12.16.0b", [A, B, C band 16#0fff, D band 16#3fff bor 16#8000, E])).