-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).