aion_flow

-module(aion@activity).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/aion/activity.gleam").
-export([new/5, retry/2, timeout/2, heartbeat/2, label/3, task_queue/2, node/2, execution_tier/2, name/1, labels/1, input/1, input_codec/1, output_codec/1, runner/1, retry_policy/1, timeout_duration/1, heartbeat_interval/1, selected_task_queue/1, selected_node/1, selected_tier/1, type_ref/2, type_ref_codec/1, declare/4, tier_to_string/1, declaration_name/1, declaration_tier/1, declaration_input_type/1, declaration_output_type/1]).
-export_type([backoff/0, retry_policy/0, tier/0, activity/2, type_ref/1, declaration/0]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

?MODULEDOC(" Typed activity values and policy configuration.\n").

-type backoff() :: {exponential,
        aion@duration:duration(),
        float(),
        aion@duration:duration()} |
    {linear,
        aion@duration:duration(),
        aion@duration:duration(),
        aion@duration:duration()} |
    {fixed, aion@duration:duration()}.

-type retry_policy() :: {retry_policy, integer(), backoff()}.

-type tier() :: in_vm | remote_python | remote_rust.

-opaque activity(DPB, DPC) :: {activity,
        binary(),
        DPB,
        aion@codec:codec(DPB),
        aion@codec:codec(DPC),
        fun((DPB) -> {ok, DPC} | {error, aion@error:activity_error()}),
        gleam@option:option(retry_policy()),
        gleam@option:option(aion@duration:duration()),
        gleam@option:option(aion@duration:duration()),
        list({binary(), binary()}),
        gleam@option:option(binary()),
        gleam@option:option(binary()),
        gleam@option:option(tier())}.

-opaque type_ref(DPD) :: {type_ref, binary(), aion@codec:codec(DPD)}.

-opaque declaration() :: {declaration, binary(), tier(), binary(), binary()}.

-file("src/aion/activity.gleam", 91).
?DOC(
    " Build a typed activity value with no retry, timeout, or heartbeat config.\n"
    "\n"
    " Absence of config is intentional data: there are no hidden defaults. In\n"
    " particular, an activity with no `retry` decorator carries no policy and an\n"
    " activity with no `execution_tier` decorator dispatches on the remote wire.\n"
).
-spec new(
    binary(),
    DPE,
    aion@codec:codec(DPE),
    aion@codec:codec(DPG),
    fun((DPE) -> {ok, DPG} | {error, aion@error:activity_error()})
) -> activity(DPE, DPG).
new(Name, Input, Input_codec, Output_codec, Run) ->
    {activity,
        Name,
        Input,
        Input_codec,
        Output_codec,
        Run,
        none,
        none,
        none,
        [],
        none,
        none,
        none}.

-file("src/aion/activity.gleam", 118).
?DOC(
    " Attach an explicit retry policy to an activity.\n"
    "\n"
    " Later calls replace earlier retry policy values; the SDK does not merge or\n"
    " synthesize policy fields.\n"
).
-spec retry(activity(DPM, DPN), retry_policy()) -> activity(DPM, DPN).
retry(Activity, Policy) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        {some, Policy},
        erlang:element(8, Activity),
        erlang:element(9, Activity),
        erlang:element(10, Activity),
        erlang:element(11, Activity),
        erlang:element(12, Activity),
        erlang:element(13, Activity)}.

-file("src/aion/activity.gleam", 136).
?DOC(" Attach an explicit timeout duration to an activity.\n").
-spec timeout(activity(DPS, DPT), aion@duration:duration()) -> activity(DPS, DPT).
timeout(Activity, Timeout_duration) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        erlang:element(7, Activity),
        {some, Timeout_duration},
        erlang:element(9, Activity),
        erlang:element(10, Activity),
        erlang:element(11, Activity),
        erlang:element(12, Activity),
        erlang:element(13, Activity)}.

-file("src/aion/activity.gleam", 157).
?DOC(" Attach an explicit heartbeat interval to an activity.\n").
-spec heartbeat(activity(DPY, DPZ), aion@duration:duration()) -> activity(DPY, DPZ).
heartbeat(Activity, Heartbeat_interval) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        erlang:element(7, Activity),
        erlang:element(8, Activity),
        {some, Heartbeat_interval},
        erlang:element(10, Activity),
        erlang:element(11, Activity),
        erlang:element(12, Activity),
        erlang:element(13, Activity)}.

-file("src/aion/activity.gleam", 185).
?DOC(
    " Attach a display label to an activity.\n"
    "\n"
    " Labels are human-meaningful key/value hints (for example `#(\"brief\",\n"
    " \"IP-001\")` or `#(\"repo\", \"ablative-io/yggdrasil\")`) that ride with the\n"
    " dispatch to the worker and surface in its logs and the dashboard. They are\n"
    " display metadata only: the engine never interprets them and they never\n"
    " affect routing, replay, or the recorded history. Repeated calls accumulate\n"
    " in call order; nothing is deduplicated or overwritten.\n"
).
-spec label(activity(DQE, DQF), binary(), binary()) -> activity(DQE, DQF).
label(Activity, Key, Value) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        erlang:element(7, Activity),
        erlang:element(8, Activity),
        erlang:element(9, Activity),
        lists:append(erlang:element(10, Activity), [{Key, Value}]),
        erlang:element(11, Activity),
        erlang:element(12, Activity),
        erlang:element(13, Activity)}.

-file("src/aion/activity.gleam", 218).
?DOC(
    " Select the task queue this activity is dispatched on (per-activity override).\n"
    "\n"
    " The task queue is the routing pool inside the workflow's namespace that a\n"
    " worker subscribes to; selecting it lets one workflow mix activities across\n"
    " pools (for example a `\"norn\"` step and a `\"gpu\"` step). This is the\n"
    " highest-precedence selection: it overrides any workflow-level default.\n"
    "\n"
    " Absence is intentional data, exactly like retry/timeout/heartbeat: an\n"
    " activity built with `new` and no `task_queue` decorator carries no\n"
    " selection, so the engine resolves it to the workflow-level default when one\n"
    " is set, else the named `\"default\"` task queue. Later calls replace earlier\n"
    " values; the SDK does not merge.\n"
).
-spec task_queue(activity(DQK, DQL), binary()) -> activity(DQK, DQL).
task_queue(Activity, Name) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        erlang:element(7, Activity),
        erlang:element(8, Activity),
        erlang:element(9, Activity),
        erlang:element(10, Activity),
        {some, Name},
        erlang:element(12, Activity),
        erlang:element(13, Activity)}.

-file("src/aion/activity.gleam", 247).
?DOC(
    " Pin this activity's dispatch to a specific node (per-activity affinity).\n"
    "\n"
    " A node is a concrete worker host inside the activity's `(namespace,\n"
    " task_queue)` pool. Pinning narrows the dispatch from \"any worker in the\n"
    " pool\" to that one host — for an activity that must run where its state or\n"
    " hardware lives, for example.\n"
    "\n"
    " Affinity is OPTIONAL and has NO workflow-level default: an activity built\n"
    " with `new` and no `node` decorator carries no pin (`None`), so the engine\n"
    " dispatches it to any worker in the pool. There is no precedence to resolve —\n"
    " the activity either pins a node or it does not. Later calls replace earlier\n"
    " values; the SDK does not merge.\n"
).
-spec node(activity(DQQ, DQR), binary()) -> activity(DQQ, DQR).
node(Activity, Name) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        erlang:element(7, Activity),
        erlang:element(8, Activity),
        erlang:element(9, Activity),
        erlang:element(10, Activity),
        erlang:element(11, Activity),
        {some, Name},
        erlang:element(13, Activity)}.

-file("src/aion/activity.gleam", 278).
?DOC(
    " Select the execution tier this activity dispatches on.\n"
    "\n"
    " `InVm` routes the dispatch through the engine's in-VM path: the activity's\n"
    " runner executes ONCE, live, inside a linked child process of the workflow\n"
    " process — no remote worker, no task-queue routing — while its recorded\n"
    " history (`ActivityScheduled`/`ActivityStarted`/terminal) and replay\n"
    " semantics stay byte-identical to a remote dispatch. Remote tiers\n"
    " (`RemotePython`/`RemoteRust`) keep the remote worker wire; selecting one is\n"
    " equivalent to selecting nothing today.\n"
    "\n"
    " Absence is intentional data, exactly like retry/timeout/heartbeat: an\n"
    " activity built with `new` and no `execution_tier` decorator carries no\n"
    " selection (`None`) and dispatches on the remote wire — today's exact\n"
    " behavior. Later calls replace earlier values; the SDK does not merge.\n"
).
-spec execution_tier(activity(DQW, DQX), tier()) -> activity(DQW, DQX).
execution_tier(Activity, Tier) ->
    {activity,
        erlang:element(2, Activity),
        erlang:element(3, Activity),
        erlang:element(4, Activity),
        erlang:element(5, Activity),
        erlang:element(6, Activity),
        erlang:element(7, Activity),
        erlang:element(8, Activity),
        erlang:element(9, Activity),
        erlang:element(10, Activity),
        erlang:element(11, Activity),
        erlang:element(12, Activity),
        {some, Tier}}.

-file("src/aion/activity.gleam", 296).
?DOC(" Return the activity name used by the engine dispatch boundary.\n").
-spec name(activity(any(), any())) -> binary().
name(Activity) ->
    erlang:element(2, Activity).

-file("src/aion/activity.gleam", 301).
?DOC(" Return the display labels attached to the activity, in call order.\n").
-spec labels(activity(any(), any())) -> list({binary(), binary()}).
labels(Activity) ->
    erlang:element(10, Activity).

-file("src/aion/activity.gleam", 306).
?DOC(" Return the typed input captured by the activity value.\n").
-spec input(activity(DRL, any())) -> DRL.
input(Activity) ->
    erlang:element(3, Activity).

-file("src/aion/activity.gleam", 311).
?DOC(" Return the typed input codec captured by the activity value.\n").
-spec input_codec(activity(DRP, any())) -> aion@codec:codec(DRP).
input_codec(Activity) ->
    erlang:element(4, Activity).

-file("src/aion/activity.gleam", 316).
?DOC(" Return the typed output codec captured by the activity value.\n").
-spec output_codec(activity(any(), DRV)) -> aion@codec:codec(DRV).
output_codec(Activity) ->
    erlang:element(5, Activity).

-file("src/aion/activity.gleam", 321).
?DOC(" Return the typed runner captured by the activity value.\n").
-spec runner(activity(DRZ, DSA)) -> fun((DRZ) -> {ok, DSA} |
    {error, aion@error:activity_error()}).
runner(Activity) ->
    erlang:element(6, Activity).

-file("src/aion/activity.gleam", 328).
?DOC(" Return the explicitly attached retry policy, if one exists.\n").
-spec retry_policy(activity(any(), any())) -> gleam@option:option(retry_policy()).
retry_policy(Activity) ->
    erlang:element(7, Activity).

-file("src/aion/activity.gleam", 333).
?DOC(" Return the explicitly attached timeout duration, if one exists.\n").
-spec timeout_duration(activity(any(), any())) -> gleam@option:option(aion@duration:duration()).
timeout_duration(Activity) ->
    erlang:element(8, Activity).

-file("src/aion/activity.gleam", 338).
?DOC(" Return the explicitly attached heartbeat interval, if one exists.\n").
-spec heartbeat_interval(activity(any(), any())) -> gleam@option:option(aion@duration:duration()).
heartbeat_interval(Activity) ->
    erlang:element(9, Activity).

-file("src/aion/activity.gleam", 348).
?DOC(
    " Return the explicitly selected per-activity task queue, if one exists.\n"
    "\n"
    " Absence (`None`) means no override was set, so the engine resolves the\n"
    " dispatch to the workflow-level default, else the named `\"default\"` queue.\n"
).
-spec selected_task_queue(activity(any(), any())) -> gleam@option:option(binary()).
selected_task_queue(Activity) ->
    erlang:element(11, Activity).

-file("src/aion/activity.gleam", 357).
?DOC(
    " Return the explicitly pinned per-activity node affinity, if one exists.\n"
    "\n"
    " Absence (`None`) means no pin was set, so the engine dispatches the activity\n"
    " to any worker in its `(namespace, task_queue)` pool. Unlike the task queue,\n"
    " node affinity has no workflow-level default: `None` is final.\n"
).
-spec selected_node(activity(any(), any())) -> gleam@option:option(binary()).
selected_node(Activity) ->
    erlang:element(12, Activity).

-file("src/aion/activity.gleam", 365).
?DOC(
    " Return the explicitly selected execution tier, if one exists.\n"
    "\n"
    " Absence (`None`) means no `execution_tier` decorator was applied, so the\n"
    " dispatch takes the remote worker wire — today's exact behavior.\n"
).
-spec selected_tier(activity(any(), any())) -> gleam@option:option(tier()).
selected_tier(Activity) ->
    erlang:element(13, Activity).

-file("src/aion/activity.gleam", 383).
?DOC(" Bind a value type's name to its codec for use in an activity declaration.\n").
-spec type_ref(binary(), aion@codec:codec(DTJ)) -> type_ref(DTJ).
type_ref(Type_name, Value_codec) ->
    {type_ref, Type_name, Value_codec}.

-file("src/aion/activity.gleam", 388).
?DOC(" Return the codec captured by a type reference.\n").
-spec type_ref_codec(type_ref(DTM)) -> aion@codec:codec(DTM).
type_ref_codec(Reference) ->
    erlang:element(3, Reference).

-file("src/aion/activity.gleam", 416).
?DOC(
    " Declare an activity from its name, tier, and typed input/output references.\n"
    "\n"
    " This is the only per-activity artifact an author hand-writes; the plumbing\n"
    " is generated from it. The side-effecting body (the runner) is written\n"
    " separately and referenced by the generated wrapper — codegen never\n"
    " synthesizes a body (the determinism boundary is unchanged).\n"
).
-spec declare(binary(), tier(), type_ref(any()), type_ref(any())) -> declaration().
declare(Name, Tier, Input, Output) ->
    {declaration,
        Name,
        Tier,
        erlang:element(2, Input),
        erlang:element(2, Output)}.

-file("src/aion/activity.gleam", 431).
?DOC(" Render a tier as the canonical string the generator's wire form uses.\n").
-spec tier_to_string(tier()) -> binary().
tier_to_string(Tier) ->
    case Tier of
        in_vm ->
            <<"in_vm"/utf8>>;

        remote_python ->
            <<"remote_python"/utf8>>;

        remote_rust ->
            <<"remote_rust"/utf8>>
    end.

-file("src/aion/activity.gleam", 440).
?DOC(" Return the engine-facing name of a declared activity.\n").
-spec declaration_name(declaration()) -> binary().
declaration_name(Declaration) ->
    erlang:element(2, Declaration).

-file("src/aion/activity.gleam", 445).
?DOC(" Return the tier a declared activity runs on.\n").
-spec declaration_tier(declaration()) -> tier().
declaration_tier(Declaration) ->
    erlang:element(3, Declaration).

-file("src/aion/activity.gleam", 450).
?DOC(" Return the input value type name of a declared activity.\n").
-spec declaration_input_type(declaration()) -> binary().
declaration_input_type(Declaration) ->
    erlang:element(4, Declaration).

-file("src/aion/activity.gleam", 455).
?DOC(" Return the output value type name of a declared activity.\n").
-spec declaration_output_type(declaration()) -> binary().
declaration_output_type(Declaration) ->
    erlang:element(5, Declaration).