executor.Plugin: task execution interface reference

The executor.Plugin interface is the extension point for all task execution logic in Aether. Every distinct type of work — HTTP calls, shell commands, data transforms, or any custom logic — is implemented as a Plugin. The engine dispatches tasks to plugins via the broker; plugins return structured outputs and an exit code that the engine maps to a Phase.

Plugin interface

type Plugin interface {
    Type() string
    Schema() model.ExecutorSchema
    Execute(ctx context.Context, req *ExecuteRequest) (*model.ExecOutputs, error)
}

func() string

Returns the unique executor type identifier string (e.g. "http", "shell", "echo"). This string must match the executor.type field in workflow task definitions.

func() model.ExecutorSchema

Declares the executor’s input/output contract. Use SchemaOf[Config, Output]() to derive this from struct types rather than writing field names by hand.

func(ctx, *ExecuteRequest) (*model.ExecOutputs, error)

Runs the task. Use OutputFrom for type-safe output construction. Return ExecCodeSuspended to pause execution until Engine.Resume() is called.

ExecuteRequest

type ExecuteRequest struct {
    // Identifiers
    TaskRunID     string
    WorkflowRunID string
    TaskName      string
    TemplateName  string

    // Runtime inputs and constraints
    Inputs     *model.Inputs
    Resources  *model.Resources
    Timeout    string  // e.g. "30m"; empty = no deadline beyond ctx
    RetryCount int     // retries consumed; 0 = first attempt
}

*model.Inputs

Fully resolved task inputs. Use BindInputs(req.Inputs, &myConfig) to deserialize into a typed config struct.

string

Duration string forwarded from TaskAssignment. The plugin should respect this independently of the context — for example, pass it to a subprocess or remote API call.

int

Number of retries already consumed. 0 = first attempt. Useful for logging or backoff logic inside the plugin.

ExecOutputs and ExecCode

type ExecOutputs struct {
    Code       ExecCode
    Message    string
    Parameters []Parameter
}

The Code field drives the engine’s phase assignment:

ExecCode	Value	Resulting Phase
`ExecCodeSucceeded`	`0`	`Succeeded`
`ExecCodeSuspended`	`1`	`Suspended`
`ExecCodeFailed`	`2`	`Failed`
`ExecCodeError`	`3`	`Error`
`ExecCodeTimeout`	`4`	`Timeout`

The engine is the sole Phase writer. Plugins return an ExecCode integer; the engine maps it to a Phase. Users can override this mapping with phaseConditions expressions on the task, except for Skipped and Cancelled which are exclusively engine semantics.

OutputFrom helper

OutputFrom converts a flat output struct to *model.ExecOutputs by reflecting on its fields. The json tag on each field becomes the parameter name.

// OutputFrom converts an output struct to *model.ExecOutputs.
// The struct must be flat — no embedded fields, no pointer fields, no nested structs.
func OutputFrom(output any) (*model.ExecOutputs, error)

Example usage:

type HttpOutput struct {
    StatusCode int    `json:"status_code"`
    Body       string `json:"body"`
}

func (p *HttpPlugin) Execute(ctx context.Context, req *executor.ExecuteRequest) (*model.ExecOutputs, error) {
    // ... do work ...
    out, err := executor.OutputFrom(HttpOutput{StatusCode: 200, Body: responseBody})
    if err != nil {
        return nil, err
    }
    out.Code = model.ExecCodeSucceeded
    return out, nil
}

BindInputs helper

BindInputs maps model.Inputs.Parameters into a typed config struct by matching parameter names to json struct tags.

// BindInputs maps inputs.Parameters into dst by matching parameter Name against json tags.
func BindInputs(inputs *model.Inputs, dst any) error

Example:

type HttpConfig struct {
    URL    string `json:"url"`
    Method string `json:"method"`
}

func (p *HttpPlugin) Execute(ctx context.Context, req *executor.ExecuteRequest) (*model.ExecOutputs, error) {
    var cfg HttpConfig
    if err := executor.BindInputs(req.Inputs, &cfg); err != nil {
        return nil, err
    }
    // use cfg.URL, cfg.Method
}

SchemaOf

SchemaOf derives an ExecutorSchema by reflecting on Config and Output struct types. Use executor.DynamicOutputs as the output type for executors with runtime-determined outputs.

func SchemaOf[C, O any](execType, version, description string) model.ExecutorSchema

Example:

type EchoConfig struct{}
type EchoOutput struct {
    Message string `json:"message" desc:"The echoed message"`
}

func (p *EchoPlugin) Schema() model.ExecutorSchema {
    return executor.SchemaOf[EchoConfig, EchoOutput]("echo", "v1", "Echoes inputs back as outputs")
}

executor.Registry

The Registry manages registered plugins, routing task dispatch by Type(). It is created automatically when you call WithExecutor(), or you can create one manually and pass it with WithExecutorRegistry().

func NewRegistry() *Registry
func (r *Registry) Register(plugin Plugin) error
func (r *Registry) RegisterSchema(schema model.ExecutorSchema)
func (r *Registry) Get(executorType string) (Plugin, bool)
func (r *Registry) GetSchema(executorType string) (model.ExecutorSchema, bool)
func (r *Registry) Types() []string
func (r *Registry) Schemas() []model.ExecutorSchema

func(Plugin) error

Adds a plugin. Returns an error if the type identifier is already registered. Also caches the plugin’s Schema() result.

func(model.ExecutorSchema)

Registers a schema without a local plugin instance. Used by distributed brokers to propagate remote worker schemas to the master for validation.

Minimal executor example

type EchoPlugin struct{}

func (p *EchoPlugin) Type() string { return "echo" }

func (p *EchoPlugin) Schema() model.ExecutorSchema {
    return executor.SchemaOf[struct{}, executor.DynamicOutputs]("echo", "v1", "Echoes inputs as outputs")
}

func (p *EchoPlugin) Execute(ctx context.Context, req *executor.ExecuteRequest) (*model.ExecOutputs, error) {
    // Read the special "outputs" parameter declared in the workflow
    var outputs []model.Parameter
    if req.Inputs != nil {
        for _, p := range req.Inputs.Parameters {
            if p.Name == "outputs" {
                // parse and return declared outputs
            }
        }
    }
    return &model.ExecOutputs{
        Code:       model.ExecCodeSucceeded,
        Parameters: outputs,
    }, nil
}

Inject with WithExecutor

engine, err := aether.New(
    aether.WithExecutor(&EchoPlugin{}), // required; call multiple times for multiple types
    // ...
)

Protocol

Go API

executor.Plugin: task execution interface reference

Plugin interface

ExecuteRequest

ExecOutputs and ExecCode

OutputFrom helper

BindInputs helper

SchemaOf

executor.Registry

Minimal executor example

Inject with WithExecutor

​Plugin interface

​ExecuteRequest

​ExecOutputs and ExecCode

​OutputFrom helper

​BindInputs helper

​SchemaOf

​executor.Registry

​Minimal executor example

​Inject with WithExecutor

Plugin interface

ExecuteRequest

ExecOutputs and ExecCode

OutputFrom helper

BindInputs helper

SchemaOf

executor.Registry

Minimal executor example

Inject with WithExecutor