Erlang ADK v0.7.0

Erlang ADK is an experimental, Erlang-native toolkit for building Gemini-backed agents with OTP processes, supervision, tools, sessions, event streams, and concurrent multi-agent workflows.

Version 0.7.0 is the cumulative Erlang ADK release from the v0.3-v0.7 delivery branches. It adds multimodal and Gemini Live sessions, Runner-global plugins, evaluation, expanded observability, and an authenticated production Phoenix companion on top of the agent, workflow, artifact, memory, context, authentication, MCP, and A2A foundations. The v0.7.0 release contract records both completed behavior and explicit limitations. The final deterministic, packaging, Phoenix, browser, dependency, and provider evidence is recorded in Verification and Testing; billable REST and Live Gemini suites remain explicit opt-in gates in a shell that owns GEMINI_API_KEY.

See the documentation index, changelog, upgrade guide, and security policy before a production rollout.

The detailed ADK behavior-parity matrix maps the official ADK capability families to their Erlang/OTP-native implementation contracts and current branch status.

The project follows the behavior of Google ADK where that behavior maps cleanly to Erlang. It is not a drop-in port and does not claim complete Google ADK feature parity. One agent is one gen_server admission point and one immutable runtime specification. Legacy prompt/delegate turns share a stateful FIFO; Runner and explicit invocation turns use bounded, fair, session-scoped lanes whose provider/tool work runs in supervised lightweight processes. Independent sessions, agents, and workflow branches therefore use BEAM concurrency without making one conversation nondeterministic.

Current scope

Source layout

The Erlang module namespace remains flat, but implementation files are grouped by ownership under one recursive src root. The root contains only the public facade and OTP application shell. Agents, tools, workflows, Live, runtime, context, artifacts, memory, sessions, protocols, integrations, authentication, models, plugins, telemetry, and evaluation each have explicit ownership directories; provider implementations such as Gemini remain below models/. This is a filesystem-only organization: public module atoms and BEAM names do not change. See src/README.md for the exact boundaries and rules for new modules. Test and fixture modules mirror the same ownership hierarchy under a test-profile-only recursive test root; see docs/TEST_LAYOUT.md for placement and discovery rules.

Installation

The supported release toolchain is Erlang/OTP 27; v0.7.0 was verified with OTP 27.3.4.14 and the bundled Rebar3 3.27.0. Older OTP 27 patch levels are not the supported production baseline because 27.3.4.14 contains required TLS security fixes. The core library does not require Elixir or Node.js. The optional Phoenix companion has its own pinned toolchain in examples/phoenix_adk_ui/.tool-versions.

After the release owner creates v0.7.0, use that immutable tag. Before the tag exists, pin the reviewed release-candidate commit SHA instead of a mutable branch. After the package is published to Hex, the equivalent package requirement is {erlang_adk, "0.7.0"}.

{deps, [
    {erlang_adk,
     {git, "https://github.com/hsalap7/erlang_adk.git",
      {tag, "v0.7.0"}}}
]}.

Start the application before spawning agents:

{ok, _StartedApps} = application:ensure_all_started(erlang_adk).

Gemini calls read GEMINI_API_KEY, or you can put api_key in the provider config. The environment variable avoids keeping a secret in source code:

export GEMINI_API_KEY="your_api_key_here"

The default model is the stable Gemini 3.1 Flash-Lite, gemini-3.1-flash-lite.

Quickstart

The repository includes the tool used below. From ./rebar3 shell:

{ok, readme_weather_tool} = c("examples/readme_weather_tool.erl"),

LLMConfig = #{
    provider => adk_llm_gemini,
    instructions => <<"You are a concise weather assistant.">>,
    model => <<"gemini-3.1-flash-lite">>
},

{ok, WeatherPid} = erlang_adk:spawn_agent(
    <<"WeatherAgent">>, LLMConfig, [readme_weather_tool]),
{ok, WeatherResponse} = erlang_adk:prompt(
    WeatherPid, <<"What is the weather in Tokyo?">>),
io:format("~ts~n", [WeatherResponse]),
ok = erlang_adk:stop_agent(WeatherPid).

Provider-reported failures use {error, Reason}. A missing key, for example, returns {error, missing_api_key} rather than a successful error string.

Provider adapters advertise only behavior they actually implement:

{ok, GeminiCapabilities} = adk_llm:capabilities(adk_llm_gemini),
true = maps:get(streaming, GeminiCapabilities),
true = maps:get(function_calling, GeminiCapabilities),
true = maps:get(structured_output, GeminiCapabilities),
true = maps:get(multimodal, GeminiCapabilities),
true = maps:get(thinking, GeminiCapabilities),
true = maps:get(safety_settings, GeminiCapabilities),
true = maps:get(google_search_grounding, GeminiCapabilities),
[google_search] = maps:get(builtin_tools, GeminiCapabilities),
1 = maps:get(content_schema_version, GeminiCapabilities),
false = maps:get(live, GeminiCapabilities),
ok = adk_llm:validate_config(
    #{provider => adk_llm_gemini,
      model => <<"gemini-3.1-flash-lite">>,
      response_mime_type => <<"application/json">>,
      response_schema => #{<<"type">> => <<"object">>}}).

Custom providers need only generate/3 and stream/4; capabilities/0 and validate_config/1 are optional behavior callbacks.

Google Search grounding

Gemini 3.1 Flash-Lite can let the model ground an answer with Google Search by adding the strictly validated built-in tool. This uses the Gemini generateContent/streamGenerateContentgoogleSearch tool, not the Interactions API:

GroundingConfig = #{
    provider => adk_llm_gemini,
    model => <<"gemini-3.1-flash-lite">>,
    builtin_tools => [google_search]
},
GroundingHistory = [
    #{role => user,
      content => <<"What changed in the latest stable Erlang/OTP release?">>}
],
case adk_llm:generate(GroundingConfig, GroundingHistory, []) of
    {provider_result, _} = GroundedResult ->
        {ok, {ok, GroundedAnswer}, ProviderMetadata} =
            adk_provider_result:decode(GroundedResult),
        <<"gemini">> = maps:get(<<"provider">>, ProviderMetadata),
        <<"google_search_grounding">> =
            maps:get(<<"type">>, ProviderMetadata),
        GroundingMetadata = maps:get(<<"metadata">>, ProviderMetadata),
        io:format("~ts~nqueries: ~p~n", [
            GroundedAnswer,
            maps:get(<<"webSearchQueries">>, GroundingMetadata, [])
        ]);
    {ok, UngroundedAnswer} ->
        %% Enabling Search lets Gemini decide whether the prompt needs it.
        io:format("~ts~n", [UngroundedAnswer])
end.

When both Erlang/custom tools and Google Search are configured, the request contains one googleSearch declaration and one functionDeclarations declaration; the two mechanisms are not conflated. A candidate with no groundingMetadata keeps the existing provider return byte-for-byte compatible. A grounded candidate uses the reserved {provider_result, Envelope} result. Agents unwrap its model output before output-schema validation. Runner/event APIs persist this JSON-safe projection:

#{<<"provider_metadata">> =>
      #{<<"schema_version">> => 1,
        <<"provider">> => <<"gemini">>,
        <<"type">> => <<"google_search_grounding">>,
        <<"metadata">> => GroundingMetadata}}

The projection is attached to the event whose Gemini candidate carried it, including tool-call events. SSE list fields are accumulated in arrival order, and the final streamed event receives the combined metadata. Candidate metadata must be a strict JSON map and is capped at 256 KiB; malformed or oversized values fail explicitly instead of being dropped. Direct prompt/2 intentionally returns only the answer, so use Runner/event APIs or direct adk_llm calls when citations or grounding provenance must be retained.

searchEntryPoint.renderedContent can contain provider-supplied HTML. Erlang ADK preserves it as bounded data but never renders it as HTML. The bundled developer console uses textContent/JSON rendering. A custom UI must sanitize untrusted markup and satisfy Google's display terms before showing search suggestions. This release does not add URL Context, Maps grounding, Enterprise Web Search, Gemini Live grounding, automatic citation insertion, or an Interactions-API adapter. See Gemini Google Search grounding, the official Google Search grounding guide, and the GenerateContent API.

Gemini 3.1 Flash-Lite supports model-native thinking. Direct provider calls put the validated option at the provider-config level; an agent puts the same map inside generation_config:

ThinkingConfig = #{
    provider => adk_llm_gemini,
    model => <<"gemini-3.1-flash-lite">>,
    thinking_config => #{thinking_level => high,
                         include_thoughts => true}
},
{ok, ThinkingResult} = adk_llm:generate(
    ThinkingConfig,
    [#{role => user,
       content => <<"Explain why supervisors isolate failures.">>}], []),
case ThinkingResult of
    Answer when is_binary(Answer) -> io:format("~ts~n", [Answer]);
    ThoughtContent when is_map(ThoughtContent) ->
        {ok, ThoughtContent} = adk_content:validate(ThoughtContent),
        io:format("structured response: ~p~n", [ThoughtContent])
end.

Supported levels are minimal, low, medium, and high; omission uses the model default. minimal is not a guarantee that thinking is disabled. The legacy numeric thinking_budget is accepted for older Gemini models, but the adapter rejects a request containing both a level and a budget. When include_thoughts is true and Gemini returns an optional thought summary, the one-shot result is canonical adk_content so the summary marked <<"thought">> => true cannot be concatenated into the visible answer. Text streaming similarly emits only visible answer deltas; content streaming preserves both kinds of part. Thought summaries are provider output, not an executable adk_plan.

Gemini safety settings are request-scoped. Direct provider calls put safety_settings at the provider-config level; agents put the same list in generation_config:

SafetySettings = [
    #{category => hate_speech, threshold => block_low_and_above},
    #{category => harassment, threshold => block_only_high}
],
ok = adk_llm:validate_config(
    #{provider => adk_llm_gemini,
      model => <<"gemini-3.1-flash-lite">>,
      safety_settings => SafetySettings}).

The adjustable categories are harassment, hate_speech, sexually_explicit, and dangerous_content. Thresholds are off, block_none, block_only_high, block_medium_and_above, block_low_and_above, and unspecified; they encode to Gemini's documented safetySettings REST enums. Categories must be unique. The deprecated civic integrity category, unknown keys, and unknown thresholds fail locally before HTTP. These settings adjust probability filters; they do not disable Gemini's built-in protection for core harms. Applications remain responsible for testing and choosing an appropriate policy. See the official Gemini safety settings and GenerateContent request contracts.

Gemini configuration is strict: a misspelled or unsupported top-level key, or an unknown key inside generation_config, returns an explicit validation error instead of being silently ignored. Documented agent/runtime fields are accepted because the immutable agent config is shared with the provider adapter. Callback-specific application data belongs under callback_config; callbacks receive a least-privilege projection containing only model/generation metadata and that recursively pruned namespace. API keys, credentials, process IDs, references, clients, connections, and store handles are never included. Register observer processes out of band, as in the callback example below.

Gemini provider configs may set request_timeout to a non-negative number of milliseconds (or infinity). It bounds the complete non-streaming HTTP request and each streaming connection/response/data wait. Omitting it keeps the underlying httpc and Gun defaults. When calling Gemini through an agent, keep this below the separate agent_call_timeout and agent_turn_timeout application settings (both 60 seconds by default). request_timeout => infinity removes only the transport limit; direct adk_llm calls do not have the outer agent deadline.

Text returned by prompt/2, correlated delegation, orchestration helpers, Runner, agent-as-tool, and the HTTP client is a UTF-8 binary. Use ~ts to print it as Unicode text; unlike a legacy Erlang charlist, it will not appear as a list of integer code points in the shell or logs.

Agent contracts and scoped instructions

An agent validates immutable input/output contracts before and after the model call. Instructions can read the exact invocation's scoped state, structured output is requested from a capable provider, and output_key is committed in the same final session event as the response:

ContractApp = <<"contract_app">>,
ContractUser = <<"user-1">>,
ContractSession = <<"contract-session">>,
ok = erlang_adk_session:init(),
{ok, _} = erlang_adk_session:create_session(
    ContractApp, ContractUser,
    #{session_id => ContractSession,
      state => #{<<"user:name">> => <<"Ada">>}}),

ContractConfig = #{
    provider => adk_llm_gemini,
    model => <<"gemini-3.1-flash-lite">>,
    instructions =>
        <<"Answer {user:name} concisely using the requested JSON shape.">>,
    input_schema => #{
        type => object,
        properties => #{<<"topic">> => #{type => string, minLength => 1}},
        required => [<<"topic">>],
        additionalProperties => false},
    output_schema => #{
        type => object,
        properties => #{<<"answer">> => #{type => string, minLength => 1}},
        required => [<<"answer">>],
        additionalProperties => false},
    output_key => <<"user:last_answer">>,
    history_policy => exclude,
    generation_config => #{
        max_output_tokens => 128,
        thinking_config => #{thinking_level => low},
        safety_settings =>
            [#{category => harassment,
               threshold => block_medium_and_above}]}
},
{ok, ContractAgent} = erlang_adk:spawn_agent(
    <<"ContractAgent">>, ContractConfig, []),
ContractRunner = adk_runner:new(
    ContractAgent, ContractApp, erlang_adk_session),
ContractInput = jsx:encode(#{<<"topic">> => <<"supervision trees">>}),
{ok, ContractResponse} = adk_runner:run(
    ContractRunner, ContractUser, ContractSession, ContractInput),
#{<<"answer">> := ContractAnswer} =
    jsx:decode(ContractResponse, [return_maps]),
io:format("~ts~n", [ContractAnswer]),
{ok, ContractStored} = erlang_adk_session:get_session(
    ContractApp, ContractUser, ContractSession),
#{<<"user:last_answer">> := #{<<"answer">> := ContractAnswer}} =
    maps:get(state, ContractStored),
ok = erlang_adk:stop_agent(ContractAgent),
ok = erlang_adk_session:delete_session(
    ContractApp, ContractUser, ContractSession).

global_instruction accepts the same static templates and {dynamic, Module, Function} providers as instructions. For a direct or Runner invocation, that agent is the root: its global instruction is prepended to its local instruction, and the same root instruction provider is carried across process boundaries and resolved for each delegated sub-agent against that invocation's scope. A child's own global_instruction is ignored while it runs below that root, but applies if the child is invoked independently as the root of a new tree. Resolution remains bounded and secret-scrubbed; agent configuration is never mutated.

Google's Python ADK currently marks its direct global_instruction field as deprecated in favor of GlobalInstructionPlugin. Erlang ADK retains the explicit root field for tree-scoped compatibility and provides adk_plugin_global_instruction for ordered Runner-wide policy. The plugin amends the checked model request through the normal v0.7 plugin pipeline; it does not mutate agent configuration or replace the root/delegation contract.

history_policy => exclude removes earlier turns but always retains the current input. A schema failure is returned before provider execution or state mutation. Static placeholders support scoped state keys such as {user:name} and exact-scope artifacts such as {artifact.reports/summary.txt}; secret-like keys are rejected. Dynamic instruction providers use {dynamic, Module, Function} and receive only a bounded, read-only scope map.

Tools

A tool module implements adk_tool and returns {ok, Value} or {error, Reason} from execute/2:

-module(readme_weather_tool).
-behaviour(adk_tool).

-export([schema/0, execute/2]).

schema() ->
    #{<<"name">> => <<"get_weather">>,
      <<"description">> => <<"Get the weather for a city">>,
      <<"parameters">> =>
          #{<<"type">> => <<"object">>,
            <<"properties">> =>
                #{<<"city">> => #{<<"type">> => <<"string">>}},
            <<"required">> => [<<"city">>]}}.

execute(#{<<"city">> := City}, Context) ->
    _ = Context,
    {ok, #{<<"city">> => City, <<"forecast">> => <<"sunny">>}}.

For agent-driven calls, Context contains app_name, session_id, user_id, invocation_id, call_id, and state_ref when available. Tool exceptions and {error, Reason} results are returned to the model as failed function responses instead of terminating the agent.

Before the provider sees a catalog, Erlang ADK loads every module/toolset schema, normalizes and compiles its supported JSON Schema, and rejects invalid schemas or duplicate names across Erlang tools, dynamic toolsets, and sub-agents. The whole provider call batch is structurally validated before it is persisted. Each call's arguments are then checked against the compiled schema before runtime policy, confirmation, callbacks, credentials, or side effects. Invalid values are never reflected in the model-visible structural error.

Gemini has two mutually exclusive function-parameter fields. Erlang ADK keeps the normalized provider-neutral contract for local validation, then emits the legacy parameters field only for a deliberately small positive subset that Gemini accepts there. Any schema outside that subset—including oneOf, additionalProperties, a type union, a boolean subschema, a top-level parameters => true | false schema, or an unknown keyword—is sent unchanged as parametersJsonSchema. The provider boundary does not weaken the tool's runtime validation contract.

An adk_toolset descriptor is an immutable catalog snapshot. schemas/1 and resolution reuse its compiled contracts; refresh/1 deliberately returns a replacement descriptor for a mutable MCP/OpenAPI backend. A running agent has no live catalog-swap API, so new remote tools require a refreshed descriptor and replacement agent. A removed or changed operation fails closed as tool_catalog_changed instead of executing against a stale declaration. Trusted local Erlang tool contexts carry private bounded agent ancestry so an AgentTool wrapper cannot reset cycle protection; that metadata is not sent to remote toolsets or model providers.

External-sandbox code execution

adk_code_toolset makes a checked execute_code tool available only through an application-supplied adk_code_executor. Erlang ADK deliberately has no erl_eval, shell, local-interpreter, executable-path, port, or NIF fallback:

#{external_sandbox_required := true,
  in_process_execution := false,
  shell_fallback := false} = adk_code_toolset:capabilities(),
{ok, CodeTools} = adk_code_toolset:new(
    #{executor => {my_sandbox_adapter, SandboxHandle},
      languages => [<<"erlang">>, <<"python">>],
      timeout => 30000,
      max_code_bytes => 65536,
      max_output_bytes => 1048576,
      parallel_safe => false}),
{ok, CodeAgent} = erlang_adk:spawn_agent(
    <<"SandboxedCoder">>,
    #{provider => adk_llm_gemini,
      instructions =>
          <<"Use execute_code only when calculation requires it.">>},
    [CodeTools]).

The configuration fragment above becomes runnable after the application provides my_sandbox_adapter and SandboxHandle; they are intentionally not invented by the library. Before invoking that adapter, the toolset enforces a language allow-list, UTF-8/input/file/output limits, canonical relative paths, a minimal redacted context, and the normal supervised tool deadline, plugin, policy, and cancellation path. The external service must independently enforce CPU, memory, process, filesystem, network, and cleanup isolation. See External sandbox code execution for the adapter behaviour and deployment requirements.

OpenAPI toolsets

OpenAPI operations use the same tool path as Erlang modules. The compiler accepts an already-decoded OpenAPI 3.0 or 3.1 document, generates deterministic model schemas, and resolves each selected operation into a bounded worker. The repository includes the schema used here:

{ok, PetstoreJson} = file:read_file(
    "examples/readme_petstore_openapi.json"),
PetstoreSpec = jsx:decode(PetstoreJson, [return_maps]),
{ok, Petstore} = adk_openapi_toolset:compile(
    PetstoreSpec,
    #{transport => {adk_openapi_gun_transport, default},
      allowed_hosts => [<<"petstore3.swagger.io">>]}),
{ok, PetstoreTools} = adk_toolset:new(
    adk_openapi_toolset, Petstore),
{ok, PetstoreAgent} = erlang_adk:spawn_agent(
    <<"PetstoreAgent">>,
    #{provider => adk_llm_gemini,
      model => <<"gemini-3.1-flash-lite">>,
      instructions => <<"Use the Petstore operation when needed.">>},
    [PetstoreTools]),
{ok, PetstoreAnswer} = erlang_adk:prompt(
    PetstoreAgent, <<"What is the name and status of pet 1?">>),
io:format("~ts~n", [PetstoreAnswer]),
ok = erlang_adk:stop_agent(PetstoreAgent).

Production transport defaults to HTTPS, re-resolves and validates the target against allowed_hosts, rejects private addresses and redirects, verifies TLS hostnames, and enforces request, response, operation, parameter, schema-depth, and time limits. Set allow_private_hosts => true and explicitly allow <<"http">> only for a trusted local service.

The supported request subset is path/query/header parameters plus JSON request bodies and JSON responses. Local $ref values are resolved. API keys in headers or query parameters, HTTP bearer, and OAuth 2.0 are resolved out of band through adk_openapi_auth_broker; credentials are never model arguments. Cookie parameters or API keys, Basic/Digest auth, mTLS, OpenID Connect security schemes, form/multipart bodies, callbacks, and remote references are rejected explicitly rather than silently weakened.

Asynchronous delegation

Use a reference when several delegated requests may be in flight:

AsyncConfig = #{provider => adk_llm_gemini,
                model => <<"gemini-3.1-flash-lite">>},
{ok, AsyncPid} = erlang_adk:spawn_agent(
    <<"AsyncAgent">>, AsyncConfig, []),
Ref = make_ref(),
ok = erlang_adk:delegate(
    AsyncPid, <<"Summarize OTP in one sentence.">>, self(), Ref),
receive
    {agent_response, Ref, AsyncPid, {ok, AsyncResponse}} ->
        io:format("~ts~n", [AsyncResponse]);
    {agent_response, Ref, AsyncPid, {error, AsyncReason}} ->
        io:format("delegation failed: ~p~n", [AsyncReason])
after 60000 ->
    exit(delegate_timeout)
end,
ok = erlang_adk:stop_agent(AsyncPid).

delegate/2 is fire-and-forget. The older delegate/3 message format remains supported, but delegate/4 is the safe choice for correlation.

Concurrent multi-agent workflows

Sequential execution feeds each response into the next agent. Parallel execution creates monitored Erlang workers, preserves input order, and applies one overall timeout:

{ok, TranslatorPid} = erlang_adk:spawn_agent(
    <<"Translator">>,
    #{provider => adk_llm_gemini,
      instructions => <<"Translate the input to French.">>}, []),
{ok, SummarizerPid} = erlang_adk:spawn_agent(
    <<"Summarizer">>,
    #{provider => adk_llm_gemini,
      instructions => <<"Summarize the input in one sentence.">>}, []),

{ok, PipelineResult} = erlang_adk:sequential(
    [TranslatorPid, SummarizerPid], <<"Erlang processes are lightweight.">>),
ParallelResults = erlang_adk:parallel(
    [TranslatorPid, SummarizerPid], <<"Explain OTP.">>, 60000),
io:format("pipeline=~ts parallel=~p~n",
          [PipelineResult, ParallelResults]),

ok = erlang_adk:stop_agent(TranslatorPid),
ok = erlang_adk:stop_agent(SummarizerPid).

Each parallel result is {AgentPid, Response} or {AgentPid, {error, Reason}}. Requests to the same agent PID retain FIFO turn semantics; their blocking work is mailbox-isolated but not executed concurrently. Concurrency comes from independent agents.

The bounded worker/reviewer loop treats a response as approval only when, after trimming and case normalization, it is exactly APPROVED. It also stops at the iteration bound and then returns the latest draft, even if the reviewer has not approved it:

{ok, WriterPid} = erlang_adk:spawn_agent(
    <<"Writer">>,
    #{provider => adk_llm_gemini,
      instructions => <<"Write and revise the requested draft.">>}, []),
{ok, ReviewerPid} = erlang_adk:spawn_agent(
    <<"Reviewer">>,
    #{provider => adk_llm_gemini,
      instructions => <<"Reply exactly APPROVED when the draft is ready; otherwise critique it.">>}, []),
{ok, FinalDraft} = erlang_adk:loop(
    WriterPid, ReviewerPid, <<"Write a short poem about Erlang.">>, 3),
io:format("~ts~n", [FinalDraft]),
ok = erlang_adk:stop_agent(WriterPid),
ok = erlang_adk:stop_agent(ReviewerPid).

Sub-agents and agent-as-tool

Sub-agents are advertised to the model as function declarations. A description helps the coordinator route well:

{ok, SearchPid} = erlang_adk:spawn_agent(
    <<"SearchSpecialist">>,
    #{provider => adk_llm_gemini,
      instructions => <<"Research the requested Erlang topic.">>}, []),
CoordinatorConfig = #{
    provider => adk_llm_gemini,
    global_instruction =>
        <<"You are part of the Erlang documentation team.">>,
    instructions => <<"Delegate research tasks to SearchSpecialist.">>,
    sub_agents => #{
        <<"SearchSpecialist">> =>
            #{pid => SearchPid,
              description => <<"Researches Erlang and OTP topics">>}
    }
},
{ok, CoordinatorPid} = erlang_adk:spawn_agent(
    <<"Coordinator">>, CoordinatorConfig, []),
{ok, CoordinatedResponse} = erlang_adk:prompt(
    CoordinatorPid, <<"Research supervision trees.">>),
io:format("~ts~n", [CoordinatedResponse]),

AgentToolSchema = adk_agent_tool:schema(
    #{name => <<"SearchSpecialist">>,
      description => <<"Researches Erlang and OTP topics">>}),
{ok, DirectSpecialistResponse} = adk_agent_tool:execute(
    SearchPid, #{<<"prompt">> => <<"Explain rest_for_one.">>}, #{}),
io:format("schema=~p response=~ts~n",
          [AgentToolSchema, DirectSpecialistResponse]),

ok = erlang_adk:stop_agent(CoordinatorPid),
ok = erlang_adk:stop_agent(SearchPid).

The coordinator resolves a restarted sub-agent by its registered binary name, so a stale child PID does not have to crash the parent agent. Agent names use the portable [A-Za-z_][A-Za-z0-9_]* grammar; user is reserved. Spawn-time tree validation rejects alias/name mismatches, duplicate ownership, self-reference, cycles, more than 256 agents, depth above 64, and an unresponsive tree after one two-second deadline. Invocation-time ancestry tracking independently rejects a repeated agent or depth above 64 before its provider runs, including dynamically resolved replacements.

prompt/2 and asynchronous delegate retain the compatibility conversation and execute in one FIFO. erlang_adk:invoke/3, AgentTool, and Runner use fresh invocation history. The same {app_name,user_id,session_id} lane remains FIFO; different lanes may overlap up to max_concurrent_invocations (32 by default) with fair admission. A child owns its provider, model, local instructions, and tool catalog. Only the root global instruction and explicitly scoped safe context cross a delegation boundary—never compatibility memory or provider credentials. The safe scope includes the caller's state snapshot and opaque session-service module (state_ref) so an output_key commits to the exact invocation session rather than the reusable agent's configured default; AgentTool may additionally carry the explicitly configured memory-service reference.

Supervised declarative workflows

adk_workflow executes each accepted workflow under adk_workflow_sup. Potentially blocking actions run in monitored, linked lightweight processes; one absolute deadline covers the entire workflow. State is JSON-safe and each action returns a state delta rather than mutating shared state.

Each coordinator child specification contains only an opaque launch reference. Compiled closures, initial state, options, checkpoints, ledger handles and owner tokens cross a validated one-shot handoff and are omitted from OTP diagnostics. Compound action, cancellation, engine and ledger failures become bounded structural adk_failure values before they reach public status or durable storage; stable atom reasons remain compatible.

Sequential and bounded-parallel workflows use the same compiled public API:

AddOne = fun(State, Context) ->
    null = maps:get(input, Context),
    {output, <<"counted">>,
     #{<<"count">> => maps:get(<<"count">>, State, 0) + 1}}
end,
MarkDone = fun(_State, Context) ->
    <<"counted">> = maps:get(input, Context),
    {output, <<"ready">>, #{<<"done">> => true}}
end,
SequentialSpec = #{
    version => 1,
    id => <<"readme-sequential-v1">>,
    kind => sequential,
    max_steps => 4,
    input_schema =>
        #{<<"type">> => <<"object">>,
          <<"required">> => [<<"count">>]},
    output_schema => #{<<"type">> => <<"string">>},
    steps => [
        #{id => <<"increment">>, run => AddOne},
        #{id => <<"finish">>, run => MarkDone}
    ]
},
{ok, Sequential} = erlang_adk:compile_workflow(SequentialSpec),
{completed, SequentialState, SequentialCheckpoint} =
    erlang_adk:run_workflow(Sequential, #{<<"count">> => 0}),
1 = maps:get(<<"count">>, SequentialState),
true = maps:get(<<"done">>, SequentialState),
true = maps:get(<<"completed">>, SequentialCheckpoint),
<<"ready">> = maps:get(<<"output">>, SequentialCheckpoint),

ParallelSpec = #{
    version => 1,
    id => <<"readme-parallel-v1">>,
    kind => parallel,
    max_concurrency => 2,
    merge => reject_conflicts,
    branches => [
        #{id => <<"left">>,
          run => fun(_State) ->
              {output, <<"left-output">>, #{<<"left">> => 1}}
          end},
        #{id => <<"right">>,
          run => fun(_State) ->
              {output, <<"right-output">>, #{<<"right">> => 2}}
          end}
    ]
},
{ok, Parallel} = erlang_adk:compile_workflow(ParallelSpec),
{completed, ParallelState, ParallelCheckpoint} =
    erlang_adk:run_workflow(Parallel, #{}),
1 = maps:get(<<"left">>, ParallelState),
2 = maps:get(<<"right">>, ParallelState),
#{<<"left">> := <<"left-output">>,
  <<"right">> := <<"right-output">>} =
    maps:get(<<"output">>, ParallelCheckpoint),

RetryTable = ets:new(readme_workflow_retry, [set, public]),
ets:insert(RetryTable, {attempts, 0}),
RetryAction = fun(_State, Context) ->
    Attempt = ets:update_counter(RetryTable, attempts, 1),
    Attempt = maps:get(attempt, Context),
    case Attempt of
        1 -> {error, transient_failure};
        2 -> {output, <<"recovered">>, #{<<"retried">> => true}}
    end
end,
RetrySpec = #{
    version => 1,
    id => <<"readme-workflow-retry-v1">>,
    kind => sequential,
    max_steps => 1,
    steps => [
        #{id => <<"retryable">>, run => RetryAction,
          timeout => 1000,
          retry => #{max_attempts => 2, backoff_ms => 10}}
    ]
},
{ok, Retrying} = erlang_adk:compile_workflow(RetrySpec),
{completed, #{<<"retried">> := true}, RetryCheckpoint} =
    erlang_adk:run_workflow(Retrying, #{}),
<<"recovered">> = maps:get(<<"output">>, RetryCheckpoint),
true = ets:delete(RetryTable).

Parallel branches receive the same immutable input state. The default reject_conflicts merge rejects unequal writes to one key; use ordered_last_wins only when declared branch order should resolve conflicts. Completion order never changes merge order.

Loop, collaborative transfer, and dynamic graph routing are also first-class specifications:

LoopSpec = #{
    version => 1,
    id => <<"readme-loop-v1">>,
    kind => loop,
    max_iterations => 3,
    body => fun(State) ->
        Attempt = maps:get(<<"attempt">>, State, 0) + 1,
        {output, Attempt, #{<<"attempt">> => Attempt}}
    end,
    until => fun(State) -> maps:get(<<"attempt">>, State) >= 2 end
},
{ok, Loop} = erlang_adk:compile_workflow(LoopSpec),
{completed, #{<<"attempt">> := 2}, LoopCheckpoint} =
    erlang_adk:run_workflow(Loop, #{}),
2 = maps:get(<<"output">>, LoopCheckpoint),

TransferSpec = #{
    version => 1,
    id => <<"readme-transfer-v1">>,
    kind => transfer,
    entry => <<"triage">>,
    max_transfers => 1,
    members => #{
        <<"triage">> => #{run => fun(_State, _Context) ->
            {transfer, <<"specialist">>, <<"handoff">>,
             #{<<"triaged">> => true}}
        end},
        <<"specialist">> => #{run => fun(State, Context) ->
            true = maps:get(<<"triaged">>, State),
            <<"handoff">> = maps:get(input, Context),
            {stop, <<"resolved">>,
             #{<<"resolved">> => true}}
        end}
    }
},
{ok, Transfer} = erlang_adk:compile_workflow(TransferSpec),
{completed, #{<<"resolved">> := true}, _} =
    erlang_adk:run_workflow(Transfer, #{}),

GraphSpec = #{
    version => 1,
    id => <<"readme-graph-v1">>,
    kind => graph,
    entry => <<"counter">>,
    max_steps => 5,
    nodes => [#{id => <<"counter">>, run => fun(State) ->
        Count = maps:get(<<"count">>, State, 0) + 1,
        {output, Count, #{<<"count">> => Count}}
    end}],
    edges => #{<<"counter">> => {route, fun(_State, Context) ->
        Count = maps:get(input, Context),
        case Count < 3 of
            true -> <<"counter">>;
            false -> end_node
        end
    end}}
},
{ok, Graph} = erlang_adk:compile_workflow(GraphSpec),
{completed, #{<<"count">> := 3}, GraphCheckpoint} =
    erlang_adk:run_workflow(Graph, #{}),
3 = maps:get(<<"output">>, GraphCheckpoint).

Graph-native fork/join executes declared branch node IDs in bounded lightweight processes and merges their deltas in declared order. A paused graph commits the node delta first; resuming supplies JSON-safe input to the node's edge router, so the paused node is not executed twice:

ForkJoinSpec = #{
    version => 1, id => <<"readme-fork-join-v1">>, kind => graph,
    entry => <<"fork">>, max_steps => 6,
    nodes => [
        #{id => <<"fork">>, type => fork,
          branches => [<<"left">>, <<"right">>], join => <<"join">>,
          merge => reject_conflicts, max_concurrency => 2},
        #{id => <<"left">>, run => fun(_) ->
            {output, <<"left-output">>, #{<<"left">> => 1}}
        end},
        #{id => <<"right">>, run => fun(_) ->
            {output, <<"right-output">>, #{<<"right">> => 2}}
        end},
        #{id => <<"join">>, type => join,
          run => fun(_State, Context) ->
              Outputs = maps:get(input, Context),
              <<"left-output">> = maps:get(<<"left">>, Outputs),
              <<"right-output">> = maps:get(<<"right">>, Outputs),
              {output, Outputs, #{<<"joined">> => true}}
          end}
    ],
    edges => #{<<"left">> => <<"join">>,
               <<"right">> => <<"join">>,
               <<"join">> => end_node}
},
{ok, ForkJoin} = erlang_adk:compile_workflow(ForkJoinSpec),
{completed, #{<<"left">> := 1, <<"right">> := 2,
              <<"joined">> := true}, ForkCheckpoint} =
    erlang_adk:run_workflow(ForkJoin, #{}),
#{<<"left">> := <<"left-output">>,
  <<"right">> := <<"right-output">>} =
    maps:get(<<"output">>, ForkCheckpoint),

ApprovalSpec = #{
    version => 1, id => <<"readme-graph-approval-v1">>, kind => graph,
    entry => <<"approval">>, max_steps => 3,
    nodes => [
        #{id => <<"approval">>, run => fun(_) ->
            {pause, human_approval, <<"Approve this action">>,
             #{<<"approval_requested">> => true}}
        end},
        #{id => <<"accepted">>, run => fun(_) ->
            {ok, #{<<"accepted">> => true}}
        end}
    ],
    edges => #{
        <<"approval">> => {route, fun(_State, Context) ->
            true = maps:get(<<"approved">>, maps:get(input, Context)),
            <<"accepted">>
        end},
        <<"accepted">> => end_node}
},
{ok, Approval} = erlang_adk:compile_workflow(ApprovalSpec),
{paused, _PauseDetails, ApprovalCheckpoint} =
    erlang_adk:run_workflow(Approval, #{}),
{ok, ApprovalRef} = erlang_adk:resume_workflow(
    Approval, ApprovalCheckpoint,
    #{resume_input => #{<<"approved">> => true}}),
{completed, #{<<"approval_requested">> := true,
              <<"accepted">> := true}, _} =
    erlang_adk:await_workflow(ApprovalRef).

For recovery after the workflow coordinator, application, or BEAM restarts, use a stable invocation ID backed by the Mnesia ledger. The compiled workflow is application code and must be reconstructed with the same id, version, and kind before resuming:

{ok, LedgerHandle} = adk_invocation_ledger_mnesia:init(#{}),
DurableOptions = #{
    ledger => {adk_invocation_ledger_mnesia, LedgerHandle},
    lease_ms => 30000,
    timeout => 120000
},
{ok, DurableInvocationId, DurableApprovalRef} =
    erlang_adk:start_workflow_invocation(Approval, #{}, DurableOptions),
{paused, _DurablePause, _DurableCheckpoint} =
    erlang_adk:await_workflow(DurableApprovalRef),
{ok, #{phase := paused, owned := false}} =
    erlang_adk:workflow_invocation_status(
      DurableInvocationId, DurableOptions),

%% After reconstructing Approval in a restarted application:
{ok, DurableResumedRef} = erlang_adk:resume_workflow_invocation(
    DurableInvocationId, Approval,
    DurableOptions#{resume_input => #{<<"approved">> => true}}),
{completed, #{<<"approval_requested">> := true,
              <<"accepted">> := true}, _} =
    erlang_adk:await_workflow(DurableResumedRef),
ok = erlang_adk:delete_workflow_invocation(
       DurableInvocationId, DurableOptions).

The ledger checkpoints before the engine receives permission to start the next action and renews a fenced ownership lease while work is active. Renew, checkpoint and finish require the matching token and Now < lease_until in one Mnesia transaction; equality is expired. Expiry permits one atomic takeover even if an old local PID is still alive, after which its token is stale. Recovery is at-least-once only for an action whose external side effect happened before its result was durably committed; use the stable invocation/step context as an idempotency key. See durable workflow invocations for restart, replication, retention, and encryption-at-rest guidance.

branch and dynamic nodes can only select IDs listed in their targets; loop nodes require body, done, and max_iterations. Typed agent, tool, and nested workflow nodes are compiled from application-owned descriptors. No route may introduce an MFA, source string, or undeclared node at runtime. Completed fork branches and completed graph nodes are checkpointed individually; only an in-flight action that has not reached a commit boundary has at-least-once replay semantics. In particular, if one graph-fork branch pauses, an in-flight sibling is cancelled and, because it has no committed result, runs again after resume. Concurrent side-effecting branches must use the invocation/step context as an external idempotency key; a fork checkpoint does not make arbitrary external effects transactional.

A transfer is an ownership handoff, unlike a sub-agent tool call which returns to its caller. Every accepted handoff emits an adk_event action named <<"transfer_to_agent">> and consumes the transfer budget. Agent-backed actions can use {agent, RegisteredName, Prompt} or {agent, RegisteredName, Prompt, DecisionFun}; the binary name is resolved at dispatch time so a supervised replacement is used after an agent restart. Before invocation, the resolved process must report the same canonical runtime name through adk_agent:get_runtime/1; a registry alias mismatch fails closed as agent_identity_mismatch.

Checkpoints are committed only at deterministic boundaries and retain the remaining budgets. Cancellation kills active workflow workers; resuming does not replay already committed steps or replenish consumed attempts:

Parent = self(),
WaitForRelease = fun(_State) ->
    Parent ! {workflow_waiting, self()},
    receive
        continue -> {ok, #{<<"released">> => true}}
    end
end,
ResumeSpec = #{
    version => 1,
    id => <<"readme-resume-v1">>,
    kind => sequential,
    max_steps => 2,
    steps => [#{id => <<"wait">>, run => WaitForRelease}]
},
{ok, ResumeCompiled} = erlang_adk:compile_workflow(ResumeSpec),
{ok, WorkflowRef} = erlang_adk:start_workflow(ResumeCompiled, #{}),
FirstWorker = receive {workflow_waiting, Pid1} -> Pid1 end,
FirstWorkerMonitor = erlang:monitor(process, FirstWorker),
{ok, #{state := running}} = erlang_adk:workflow_status(WorkflowRef),
{ok, Checkpoint} = erlang_adk:workflow_checkpoint(WorkflowRef),
ok = erlang_adk:cancel_workflow(WorkflowRef, revise_later),
{cancelled, revise_later, Checkpoint} =
    erlang_adk:await_workflow(WorkflowRef),
receive
    {'DOWN', FirstWorkerMonitor, process, FirstWorker, _} -> ok
end,

{ok, ResumedRef} =
    erlang_adk:resume_workflow(ResumeCompiled, Checkpoint),
SecondWorker = receive {workflow_waiting, Pid2} -> Pid2 end,
SecondWorker ! continue,
{completed, #{<<"released">> := true}, _} =
    erlang_adk:await_workflow(ResumedRef).

The action-result contract is {ok, Delta}, normal continuation {output, Output, Delta}, explicit terminal {stop, Output, Delta}, {route, TargetNode, Delta}, {transfer, TargetMember, NextInput, Delta}, or {error, Reason}. Legacy {complete, Output, Delta} remains terminal for compatibility. Graph actions additionally accept {pause, Reason, Summary, Delta}. A normal output is checkpointed and becomes the next sequential step, graph successor, or join node's Context.input; parallel/fork outputs are maps keyed by declared branch ID. Workflow-level input_schema and output_schema are compiled once. Individual action maps accept timeout and retry => #{max_attempts, backoff_ms}; each attempt runs in a monitored lightweight process under the workflow's absolute deadline. Retry counts cover one live execution of an uncommitted node and reset if that node is later reconstructed from a checkpoint, so external effects still need the documented idempotency key. Runtime options include timeout or an absolute monotonic deadline, max_steps, max_transfers, max_concurrency, and terminal retention_ms.

Explicit planning and replanning

Model-native thinking and executable plans are deliberately separate. Gemini thinking configuration controls how that provider produces an answer; it does not expose private reasoning as an executable plan. For an application-owned plan, use a trusted adk_planner and adk_plan_executor. The executor maps a small declared JSON action vocabulary to application operations, while the runtime supplies monitored lightweight processes, one absolute deadline, step/replan/heap/byte budgets, cancellation, and secret-pruned results.

The repository examples are deterministic and executable from ./rebar3 shell:

{ok, readme_planner} = c("examples/readme_planner.erl"),
{ok, readme_plan_executor} = c("examples/readme_plan_executor.erl"),

Planner = #{module => readme_planner,
            target => undefined,
            config => #{}},
Executor = #{module => readme_plan_executor,
             target => undefined,
             config => #{}},
Goal = #{<<"task">> => <<"prepare release">>},
Context = #{<<"invocation_id">> => <<"planning-readme">>},

{ok, PlanningResult} = erlang_adk:run_planning(
    Planner, Executor, Goal, Context,
    #{max_steps => 4,
      max_replans => 1,
      timeout_ms => 5000}),
<<"completed">> = maps:get(<<"status">>, PlanningResult),
1 = maps:get(<<"steps_executed">>, PlanningResult),
#{<<"goal">> := Goal,
  <<"invocation_id">> := <<"planning-readme">>} =
    maps:get(<<"result">>, PlanningResult).

For caller-controlled cancellation, use start_planning/5, cancel_planning/2, and await_planning/2. The returned opaque reference is owned by the starting process: another process cannot await or cancel it, and owner death stops the runtime and its active callback. Planner and executor module atoms always come from trusted application configuration; values in a plan can never select a module or evaluate Erlang/source/shell code. See the planning runtime contract for the canonical plan schema, adapter behaviours, failure values, and all limits.

Legacy graph helpers

Graphs are deterministic Erlang workflows. Compilation validates the entry point and deterministic edges; execution has a cycle bound:

Graph0 = adk_graph:new(),
Increment = fun(State) ->
    #{<<"count">> => maps:get(<<"count">>, State, 0) + 1}
end,
Graph1 = adk_graph:add_node(Graph0, counter, Increment),
Next = fun(State) ->
    case maps:get(<<"count">>, State) < 3 of
        true -> counter;
        false -> end_node
    end
end,
Graph2 = adk_graph:add_conditional_edge(Graph1, counter, Next),
Graph3 = adk_graph:set_entry_point(Graph2, counter),
{ok, CompiledGraph} = adk_graph:compile(Graph3),
{ok, FinalState} = adk_graph:run(
    CompiledGraph, #{<<"count">> => 0}, #{max_steps => 10}),
3 = maps:get(<<"count">>, FinalState).

adk_graph_node:agent_node/3, function_node/1, and tool_node/1 are helpers for event-based graphs. Agent nodes include configured instructions and model callbacks.

Event Runner

The Runner stores user, model, and tool events in one session and clears temp: state only at a final or error boundary:

RunnerConfig = #{provider => adk_llm_gemini,
                 instructions => <<"Be concise.">>},
{ok, RunnerAgentPid} = erlang_adk:spawn_agent(
    <<"RunnerAgent">>, RunnerConfig, []),
Runner = adk_runner:new(
    RunnerAgentPid, <<"readme_app">>, erlang_adk_session,
    #{run_timeout => 120000,
      max_llm_calls => 32,
      max_tool_rounds => 16,
      streaming_mode => text,
      max_stream_output_bytes => 16777216,
      admission_control => #{overflow => queue,
                             queue_timeout => 5000},
      runtime_policy =>
          #{id => <<"readme-runner-policy">>,
            agents => #{allow => [<<"RunnerAgent">>]},
            tools => #{allow => []},
            max_argument_bytes => 32768,
            max_content_bytes => 262144},
      context_policy => #{max_bytes => 32768,
                          max_tokens => 8192,
                          max_request_bytes => 65536,
                          max_request_tokens => 16384,
                          overflow => truncate}}),
{ok, RunnerResponse} = adk_runner:run(
    Runner, <<"user-1">>, <<"session-1">>, <<"Hello">>),
io:format("~ts~n", [RunnerResponse]).

streaming_mode => text makes the Runner use the provider's UTF-8 delta stream. content selects canonical adk_content deltas, and the default none uses one-shot generation. Provider I/O executes in the independently supervised invocation worker rather than the agent gen_server mailbox, so the agent remains responsive to runtime inspection and cancellation routing. The short preparation/finalization calls are still serialized by that agent.

max_stream_output_bytes bounds one model round's accumulated callback payload (16 MiB by default, with a 64 MiB hard ceiling). Canonical content is also subject to its part, text, inline-data, URI, and function-payload limits. An unsupported provider content stream, invalid UTF-8/content delta, empty stream, or exceeded bound fails the invocation explicitly.

run_timeout is the overall synchronous Runner deadline. A non-streaming model round invoked through adk_agent:run_with_events/3 has a separate 60-second gen_server call limit by default, so a larger Runner deadline can cover several model/tool rounds but does not itself extend one non-streaming model round. Applications that need a different agent call limit can set the erlang_adk application environment key agent_call_timeout to a non-negative number of milliseconds (or infinity); adk_agent:prompt/2 and run_with_events/3,4 use it. Blocking direct-turn workers have the separate agent_turn_timeout setting, also 60 seconds by default, so work is cancelled and cleaned up even when a caller process survives its gen_server call timeout. Set both values when intentionally allowing longer turns; explicit infinity is an opt-in for either boundary.

Use one active Runner invocation per session. Independent agents and sessions can run concurrently, but sharing one session between simultaneous invocations can interleave event history and invocation-temp state.

max_llm_calls and max_tool_rounds are independent invocation budgets. They default to 32 and 16 respectively and prevent an otherwise valid model/tool conversation from looping forever.

Runner always canonicalizes and secret-prunes model history. context_policy adds filtering and selected-event budgets before each model call. overflow => truncate keeps the newest complete context exchanges; it never separates a tool call from its responses. error fails explicitly, while compress requires an owner-bound monitored compressor with timeout, heap, event-count, and output-size bounds. Author, invocation, content type, time-range, partial, and final-event filters are supported. The current invocation's user event may not be filtered out.

max_request_bytes and max_request_tokens are a second fail-closed boundary over the complete sanitized provider envelope: effective instructions and generation settings, selected history, retrieved memory, current input, multimodal parts, tool declarations, and framing. A violation returns {request_context_budget_exceeded, Details} before provider I/O. Context-build telemetry is emitted as [erlang_adk, context, build]; complete-envelope telemetry uses [erlang_adk, context, envelope]. Both expose bounded counts, estimates, and deterministic context fingerprints without prompt content or credentials. Exact provider tokenization remains a provider adapter concern. See the context guide for compaction, provider-prefix cache, and least-authority tool context contracts.

Admission control and runtime policy

The application always supervises one adk_admission_control process. A Runner opts into it with admission_control => #{...}. reject returns a busy error immediately; queue uses a bounded queue and an absolute monotonic deadline derived from queue_timeout. Limits are global and per agent. The invocation worker owns its permit, so normal completion releases it explicitly and an untrappable cancellation, timeout, or crash releases it through the controller's process monitor.

Configure node-wide limits before application startup (for example in sys.config):

[{erlang_adk,
  [{admission_control,
    #{global_limit => 256,
      default_agent_limit => 16,
      agent_limits => #{<<"BatchAgent">> => 4},
      overflow => reject,
      max_queue => 1024,
      default_queue_timeout => 30000}}]}].

runtime_policy is optional, but once configured it is fail-closed. An omitted agent/tool allow selector means no names are allowed; use allow => all explicitly when intended, and deny always wins. Binary content is measured as UTF-8 payload bytes, while structured content and final resolved tool arguments are measured as canonical JSON. Finite defaults are 64 KiB for arguments and 1 MiB for content.

Tool policy runs after Erlang module/OpenAPI/MCP/dynamic-toolset resolution but before global/local tool callbacks and human approval. A denied tool therefore cannot execute, invoke policy-bypass callbacks, or become an approval request. It becomes a bounded structural tool error so the model can continue. Input, tool-result, and final-output budget failures never persist the rejected value.

Every denial is persisted and streamed as a canonical system event whose actions contains <<"runtime_policy_decision">>. The immutable decision has IDs, policy fingerprint, operation, subject, outcome/reason, and byte counts; it never contains arguments, content, credentials, exception terms, pids, or references. Audit events deliberately bypass mutable on_event plugins. See Runtime safety and admission control for the direct API, telemetry fields, release semantics, and Phoenix ownership guidance.

For application-facing asynchronous work, prefer the supervised run API. The run is not owned by the process that starts it or by a browser subscriber, so a caller disconnect does not terminate useful work:

Each dynamic invocation child specification contains only a fresh opaque reference. The run ID, Runner handle, request/prompt, and options cross a validated one-shot handoff after the empty child starts; a rejected handoff or duplicate-registration loser is cleaned up synchronously. OTP diagnostic status exposes only bounded lifecycle counters, while the explicit adk_run:status/1 API continues to return the caller's run ID and outcome.

{ok, RunId} = adk_run:start(
    Runner, <<"user-1">>, <<"session-supervised">>,
    <<"Explain supervisors.">>,
    #{retention_ms => 60000, max_buffered_events => 256}),
ok = adk_run:subscribe(RunId),
ReceiveRun = fun Loop(Id) ->
    receive
        {adk_run_event, Id, Sequence, Event} ->
            io:format("~p ~p~n", [Sequence, adk_event:to_map(Event)]),
            Loop(Id);
        {adk_run_terminal, Id, _Sequence, Outcome} ->
            Outcome
    after 120000 ->
        {error, subscriber_timeout}
    end
end,
{completed, SupervisedText} = ReceiveRun(RunId),
io:format("~ts~n", [SupervisedText]),
{ok, #{state := completed}} = adk_run:status(RunId).

Late subscribers receive the bounded replay followed by the same terminal outcome while the run is retained. adk_run:await/1,2 waits without subscribing; adk_run:cancel/1,2 propagates cancellation to the Runner worker. For a slow network or broker boundary, use adk_run:subscribe_credit/2,3 with the last consumed sequence and return credit through adk_run:ack/2,3; this permits at most one unacknowledged run message per subscriber and reports a replay_gap instead of silently skipping events. The original subscribe/1,2 push API is preserved for local Erlang processes that continuously drain their mailbox.

With provider streaming, each delta is persisted as #adk_event{partial = true}. One is_final = true event contains the complete response, and the completed run outcome contains that same response exactly once. A UI should append/render partials provisionally, then replace that provisional view with the final snapshot; it must not append the final snapshot to the partial text. Partial events remain replayable but are excluded from the next model turn's history. Exactly one terminal outcome is committed: {completed, Output} (a UTF-8 binary or canonical adk_content), {paused, Event}, {cancelled, Reason}, or {failed, Reason}.

Resume a supervised paused run through adk_run, not by turning the approval into a new user prompt:

{paused, _PauseEvent} = adk_run:await(PausedRunId, 120000),
{ok, ResumedRunId} = adk_run:resume(
    PausedRunId,
    #{<<"approved">> => true, <<"approver">> => <<"operator@example.com">>},
    #{retention_ms => 60000, max_buffered_events => 256}),
{completed, ResumedText} = adk_run:await(ResumedRunId, 120000).

The original paused run stays immutable. Its status exposes resumed_to; the new run exposes parent_run_id. A second resume returns {error, {already_resumed, ResumedRunId}}.

Ambient/background invocation

Register an application-owned Runner once, then synchronously submit each delivery into a bounded queue. Submission returns immediately with a stable event reference; the invocation and its retry lifecycle are owned by supervised lightweight processes rather than by the HTTP/queue consumer:

ok = adk_ambient:register_trigger(
    <<"inbox-events">>, Runner,
    #{max_concurrency => 8,
      max_queue => 256,
      event_timeout => 120000,
      retention_ms => 300000,
      max_retained => 1000,
      max_event_bytes => 1048576,
      session_policy => #{mode => per_event,
                          user_id => <<"background-worker">>,
                          prefix => <<"inbox-">>},
      retry => #{max_attempts => 3,
                 initial_delay => 250,
                 max_delay => 5000,
                 backoff_factor => 2.0,
                 attempt_timeout => 30000,
                 max_heap_words => 1000000,
                 jitter => full}}),

Delivery = #{payload => <<"Summarize message 42">>,
             idempotency_key => <<"mailbox:42:v1">>,
             metadata => #{source => <<"local-queue">>}},
{ok, AmbientRef} = adk_ambient:submit(<<"inbox-events">>, Delivery),
{completed, #{run_id := _AmbientRunId,
              output := AmbientOutput}} =
    adk_ambient:await(AmbientRef, 120000),
io:format("~ts~n", [AmbientOutput]),
{ok, #{state := terminal, attempts := _AttemptCount}} =
    adk_ambient:status(AmbientRef).

Every event must carry a non-empty idempotency_key. A duplicate returns {ok, ExistingRef, duplicate} and cannot start a second invocation while the original result is retained. Dedupe and trigger registration are node-local and bounded by retention_ms, max_retained, and the application-wide ambient_runtime.max_events; a durable broker should redeliver until its adapter has successfully called submit/2. Retried invocations reuse the same session, so tools called by a retry must also be idempotent.

Session ownership cannot be implicit. per_event derives a separate stable session from the idempotency key; explicit requires session => #{user_id => ..., session_id => ...} on every event; shared uses one configured session and should only be selected when interleaved history is intended. Per-event timeout_ms may shorten, but never extend, the registered event deadline. Queue wait, retry backoff, and run execution consume that one absolute deadline. cancel/2 covers both queued and active events and publishes a terminal result only after the run and admission permit are cleaned up.

The bundled schedule source owns one fixed-delay timer and also submits through the same bounded path:

{ok, SchedulePid} = adk_trigger_schedule:start(
    <<"inbox-events">>, <<"hourly-inbox-summary">>, 3600000,
    #{payload => <<"Summarize the last hour">>},
    #{initial_delay_ms => 3600000}),
{ok, #{type := schedule, interval_ms := 3600000}} =
    adk_trigger_schedule:status(SchedulePid),
ok = adk_trigger_schedule:stop(SchedulePid),
ok = adk_ambient:unregister_trigger(<<"inbox-events">>).

adk_trigger_source is the provider-neutral OTP behaviour for additional sources. Pub/Sub, Eventarc, Kafka, RabbitMQ, cron services, and cloud push verification remain deployment adapters; the core deliberately has no cloud SDK dependency. Adapters apply backpressure by calling submit/2, acknowledge only after acceptance, and use the provider delivery ID as the idempotency key. The bundled scheduler is fixed-delay, not a cron parser. See the ambient runtime contract for every option, outcome, cleanup guarantee, and adapter responsibility.

The lower-level Runner stream remains available when a caller deliberately wants to own and drain the mailbox protocol:

Drain = fun Loop(StreamPid) ->
    receive
        {adk_event, StreamPid, Event} ->
            io:format("event: ~p~n", [adk_event:to_map(Event)]),
            Loop(StreamPid);
        {adk_done, StreamPid} -> ok;
        {adk_paused, StreamPid, PauseEvent} -> {paused, PauseEvent};
        {adk_error, StreamPid, Reason} -> {error, Reason}
    after 120000 ->
        {error, timeout}
    end
end,
{ok, StreamPid} = adk_runner:run_async(
    Runner, <<"user-1">>, <<"session-2">>, <<"Explain supervisors.">>),
ok = Drain(StreamPid),
ok = erlang_adk:stop_agent(RunnerAgentPid).

Human approval pause/resume

A side-effecting adk_tool can export require_confirmation/0 for a static gate or require_confirmation/2 for an argument-dependent gate. The latter takes precedence and returns false, true, or a bounded map such as #{required => true, hint => <<"Publish production">>}. Runner evaluates it after schema validation and runtime policy, but before tool/plugin lifecycle callbacks or execution. A dynamic toolset is materialized after policy so its validated resolved call can declare confirmation; that resolver may obtain a scoped credential but must not perform the business side effect. A confirmation is a serial barrier even when neighboring calls are parallel-safe.

The repository's release example requires confirmation for a real publication but lets dry_run => true proceed. Use the supervised run API so the approval survives the original caller:

{ok, readme_release_tool} = c("examples/readme_release_tool.erl"),
ReleaseConfig = #{
    provider => adk_llm_gemini,
    model => <<"gemini-3.1-flash-lite">>,
    instructions =>
        <<"Call publish_release once for production with dry_run=false. "
          "After its result, answer without calling it again.">>
},
{ok, ReleasePid} = erlang_adk:spawn_agent(
    <<"ReleaseAgent">>, ReleaseConfig, [readme_release_tool]),
ReleaseRunner = adk_runner:new(
    ReleasePid, <<"readme_app">>, erlang_adk_session),
{ok, ReleaseRun} = adk_run:start(
    ReleaseRunner, <<"user-1">>, <<"release-session">>,
    <<"Publish the prepared production release.">>),
case adk_run:await(ReleaseRun, 120000) of
    {paused, ConfirmationPause} ->
        ConfirmationMap = adk_event:to_map(ConfirmationPause),
        #{<<"pause">> :=
              #{<<"details">> :=
                    #{<<"type">> := <<"tool_confirmation">>,
                      <<"action_id">> := _OpaqueActionId}}} =
            maps:get(<<"actions">>, ConfirmationMap),
        {ok, ApprovedReleaseRun} = adk_run:resume(
            ReleaseRun, #{<<"confirmed">> => true}),
        case adk_run:await(ApprovedReleaseRun, 120000) of
            {completed, ReleaseAnswer} ->
                io:format("~ts~n", [ReleaseAnswer]);
            {paused, NextPause} ->
                io:format("tool requested another pause: ~p~n",
                          [adk_event:to_map(NextPause)])
        end;
    {completed, ReleaseAnswer} ->
        io:format("model completed without using the tool: ~ts~n",
                  [ReleaseAnswer])
end,
ok = erlang_adk:stop_agent(ReleasePid),

%% Long-running work is a separate pause type. This compatibility example
%% lets the model request an explicit human response itself.
ApprovalConfig = #{
    provider => adk_llm_gemini,
    instructions =>
        <<"Before any destructive action, call request_human_approval.">>
},
{ok, ApprovalPid} = erlang_adk:spawn_agent(
    <<"ApprovalAgent">>, ApprovalConfig, [adk_long_running_tool]),
ApprovalRunner = adk_runner:new(
    ApprovalPid, <<"readme_app">>, erlang_adk_session),
ApprovalDrain = fun Loop(ApprovalStreamPid) ->
    receive
        {adk_event, ApprovalStreamPid, ApprovalEvent} ->
            io:format("event: ~p~n", [adk_event:to_map(ApprovalEvent)]),
            Loop(ApprovalStreamPid);
        {adk_done, ApprovalStreamPid} -> ok;
        {adk_paused, ApprovalStreamPid, NextPause} -> {paused, NextPause};
        {adk_error, ApprovalStreamPid, ResumeReason} -> {error, ResumeReason}
    after 120000 ->
        {error, timeout}
    end
end,
ApprovalResult = adk_runner:run(
    ApprovalRunner, <<"user-1">>, <<"approval-session">>,
    <<"Delete the obsolete deployment.">>),
case ApprovalResult of
    {paused, PauseEvent} ->
        io:format("approval requested: ~p~n", [adk_event:to_map(PauseEvent)]),
        PauseMap = adk_event:to_map(PauseEvent),
        #{<<"pause">> :=
              #{<<"continuation_id">> := ContinuationId}} =
            maps:get(<<"actions">>, PauseMap),
        {ok, ResumePid} = adk_runner:resume(
            ApprovalRunner, <<"user-1">>, <<"approval-session">>,
            ContinuationId,
            #{approved => true, approver => <<"operator@example.com">>}),
        ok = ApprovalDrain(ResumePid);
    {ok, UnexpectedFinal} ->
        io:format("model completed without pausing: ~ts~n", [UnexpectedFinal]);
    {error, ApprovalReason} ->
        io:format("approval workflow failed: ~p~n", [ApprovalReason])
end,
ok = erlang_adk:stop_agent(ApprovalPid).

Resume a generic confirmation with exactly #{<<"confirmed">> => true | false}. Approval re-resolves the tool and rechecks current policy before executing exactly once. Rejection executes no tool or tool callback; it records a correlated model-visible rejection and continues the invocation. An invalid response does not consume the continuation, while a valid continuation is single-use. The opaque action ID is a digest and contains no raw arguments. Non-Runner agent execution (prompt/2, fresh invoke/3, delegation, and AgentTool-backed child calls) and typed workflow tool actions have no durable approval channel, so a required confirmation fails closed as tool_confirmation_requires_runner; they never silently execute. Structured “modify arguments” confirmation responses are not supported.

adk_long_running_tool uses the distinct long-running suspension contract. A pause is not an error; resume records a matching function response with the original invocation ID, thought signature, and Gemini function-call ID.

Use resume/5 with the public continuation/invocation ID whenever concurrent invocations can share a session. resume/4 is a compatibility convenience and returns {error, {ambiguous_paused_invocation, InvocationIds}} if more than one pause exists. A continuation is consumed atomically; replay returns {error, no_paused_invocation}. Temp state is retained while paused and removed after the resumed invocation completes or fails. Erlang maps returned by tools are normalized to JSON-safe binary keys, so the atom-keyed approval map above round-trips through event JSON safely.

Generic long-running operations

A pause-capable tool can suspend without keeping a worker process blocked:

adk_suspension:long_running(
    <<"export-42">>, <<"The external export is running.">>).

The application records correlated non-terminal updates while the continuation remains resumable, then supplies one terminal response:

PauseMap = adk_event:to_map(PauseEvent),
InvocationId = maps:get(<<"invocation_id">>, PauseMap),
{ok, _ProgressEvent} = adk_runner:update_long_running(
    Runner, <<"user-1">>, <<"export-session">>, InvocationId,
    <<"export-42">>,
    #{<<"operation_id">> => <<"export-42">>,
      <<"status">> => <<"running">>,
      <<"progress">> => 50}),
{ok, _ResumeStream} = adk_runner:resume(
    Runner, <<"user-1">>, <<"export-session">>, InvocationId,
    #{<<"operation_id">> => <<"export-42">>,
      <<"status">> => <<"completed">>,
      <<"result">> => #{<<"artifact_id">> => <<"export.zip">>}}).

Operation ID and terminal status are checked before the continuation is claimed. ETS and Mnesia sessions atomically order progress against terminal resume, so a racing update cannot attach to a consumed or different operation. Custom session backends must implement add_event_if_state/6 to support progress updates.

Session state

State prefixes have these lifetimes:

• no prefix: one session;
• user:: future and existing sessions for the same app/user;
• app:: future and existing sessions for the same app;
• temp:: one Runner invocation, retained across a pause but cleared at final/error.

Create a session before updating it:

ok = erlang_adk_session:init(),
AppName = <<"state_demo">>,
UserId = <<"user-1">>,
SessionId = <<"state-session-1">>,
{ok, _CreatedSession} = erlang_adk_session:create_session(
    AppName, UserId, #{session_id => SessionId}),
ok = erlang_adk_session:update_state(
    AppName, UserId, SessionId,
    #{<<"theme">> => <<"local">>,
      <<"user:preferences">> => <<"dark">>,
      <<"app:release">> => <<"0.3.0">>,
      <<"temp:lookup">> => <<"in-flight">>}),
{ok, StoredSession} = erlang_adk_session:get_session(
    AppName, UserId, SessionId),
StoredState = maps:get(state, StoredSession),
<<"dark">> = maps:get(<<"user:preferences">>, StoredState),

{ok, FutureSession} = erlang_adk_session:create_session(
    AppName, UserId, #{session_id => <<"state-session-2">>}),
FutureState = maps:get(state, FutureSession),
<<"dark">> = maps:get(<<"user:preferences">>, FutureState),
<<"0.3.0">> = maps:get(<<"app:release">>, FutureState),
error = maps:find(<<"theme">>, FutureState),

ok = erlang_adk_session:clear_temp_state(AppName, UserId, SessionId),
ok = erlang_adk_session:delete_session(AppName, UserId, SessionId),
ok = erlang_adk_session:delete_session(
    AppName, UserId, <<"state-session-2">>).

update_state/4 and add_event/4 return {error, not_found} for a missing session. Creation is idempotent and does not overwrite an existing session during concurrent first use.

For a local disk-backed Mnesia session backend, initialize Mnesia and pass it to the Runner:

ok = erlang_adk_session_mnesia:init(),
{ok, PersistentAgentPid} = erlang_adk:spawn_agent(
    <<"PersistentAgent">>, #{provider => adk_llm_gemini}, []),
MnesiaRunner = adk_runner:new(
    PersistentAgentPid, <<"persistent_app">>, erlang_adk_session_mnesia),
{ok, PersistentResponse} = adk_runner:run(
    MnesiaRunner, <<"user-1">>, <<"persistent-session">>, <<"Hello">>),
io:format("~ts~n", [PersistentResponse]),
ok = erlang_adk:stop_agent(PersistentAgentPid).

init/0 creates disc_copies on the current node only. A distributed deployment must separately configure the Mnesia cluster and add table replicas on the other nodes.

Mnesia is no longer started by the default application path. To make it the configured backend and fail application startup if it cannot initialize, set the backend before starting erlang_adk:

_ = application:stop(erlang_adk),
ok = application:set_env(
    erlang_adk, session_backend, erlang_adk_session_mnesia),
{ok, _} = application:ensure_all_started(erlang_adk).

The older agent session_id option persists that agent's conversation history through the backend's legacy save/2 and load/1 API. Runner sessions are the recommended event/state API.

For authenticated, snapshot-bound pagination and non-destructive rewind, use adk_session_query. Cursor secrets belong in application secret storage and must be stable across the nodes that serve a cursor:

QuerySecret = adk_session_query:new_cursor_secret(),
QueryApp = <<"query_app">>,
QueryUser = <<"user-1">>,
SourceSession = <<"query-a">>,
BranchSession = <<"query-a-before-answer">>,
{ok, _} = erlang_adk_session:create_session(
    QueryApp, QueryUser, #{session_id => SourceSession}),
ok = erlang_adk_session:add_event(
    QueryApp, QueryUser, SourceSession,
    adk_event:with_state_delta(
        adk_event:new(<<"user">>, <<"question">>),
        #{<<"step">> => 1})),
ok = erlang_adk_session:add_event(
    QueryApp, QueryUser, SourceSession,
    adk_event:with_state_delta(
        adk_event:new(<<"agent">>, <<"answer">>, #{is_final => true}),
        #{<<"step">> => 2})),

{ok, SessionPage} = adk_session_query:list(
    erlang_adk_session, QueryApp, QueryUser,
    #{limit => 10, cursor_secret => QuerySecret}),
[_ | _] = maps:get(sessions, SessionPage),
{ok, EventPage} = adk_session_query:get(
    erlang_adk_session, QueryApp, QueryUser, SourceSession,
    #{event_limit => 1, cursor_secret => QuerySecret}),
#{next_cursor := EventCursor} = maps:get(event_page, EventPage),
true = is_binary(EventCursor),

{ok, #{session_id := BranchSession, events_copied := 1}} =
    adk_session_query:rewind(
        erlang_adk_session, QueryApp, QueryUser, SourceSession,
        {index, 1}, #{target_session_id => BranchSession}),
{ok, SourceAfterRewind} = erlang_adk_session:get_session(
    QueryApp, QueryUser, SourceSession),
2 = length(maps:get(events, SourceAfterRewind)),
ok = erlang_adk_session:delete_session(
    QueryApp, QueryUser, SourceSession),
ok = erlang_adk_session:delete_session(
    QueryApp, QueryUser, BranchSession).