executor

package
v2.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 6, 2026 License: GPL-3.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func CloseExecutor 

func CloseExecutor(exec Executor) error

CloseExecutor safely closes an executor if it implements io.Closer. Returns nil if executor doesn't implement io.Closer or is nil. This should be called after executor.Run() completes to release resources.

func CreateTask 

func CreateTask(
	dagName string,
	yamlDefinition string,
	op exec.DispatchOperation,
	runID string,
	opts ...TaskOption,
) *exec.DispatchTask

CreateTask creates a dispatch task from this DAG for distributed execution. It constructs a task with the given operation and run ID, setting the DAG's name as both the root DAG and target, and includes the DAG's YAML definition.

func RegisterExecutor 

func RegisterExecutor(executorType string, factory ExecutorFactory, validator core.StepValidator, caps core.ExecutorCapabilities)

RegisterExecutor registers a new executor type with its factory, validator, and capabilities.

func ResolveBaseConfig 

func ResolveBaseConfig(baseConfigData []byte, fallbackPath string) string

ResolveBaseConfig returns the base config content for a DAG task. It prefers embedded BaseConfigData from the DAG, falling back to reading the file at fallbackPath.

func UnregisterExecutor 

func UnregisterExecutor(executorType string)

UnregisterExecutor removes a registered executor type.

func WithSubWorkflowRunner 

func WithSubWorkflowRunner(ctx context.Context, runner SubWorkflowRunner) context.Context

WithSubWorkflowRunner injects a child workflow runner into ctx.

Types

type ChatMessageHandler 

type ChatMessageHandler interface {
	SetContext([]exec.LLMMessage)
	GetMessages() []exec.LLMMessage
}

ChatMessageHandler is an interface for executors that handle chat session messages.

type DAGExecutor 

type DAGExecutor interface {
	Executor

	// SetParams sets the parameters for running a sub DAG.
	SetParams(RunParams)
}

DAGExecutor is an interface for sub DAG executors.

type DeclaredOutputsProvider 

type DeclaredOutputsProvider interface {
	OutputsProvider
	PublishesDeclaredOutputs() bool
}

DeclaredOutputsProvider marks executor outputs as available to strict step output references.

type Executor 

type Executor interface {
	SetStdout(out io.Writer)
	SetStderr(out io.Writer)
	Kill(sig os.Signal) error
	Run(ctx context.Context) error
}

Executor is an interface for executing steps in a DAG.

func NewExecutor 

func NewExecutor(ctx context.Context, step core.Step) (Executor, error)

NewExecutor creates a new Executor based on the step's executor type.

type ExecutorFactory 

type ExecutorFactory func(ctx context.Context, step core.Step) (Executor, error)

ExecutorFactory is a function type that creates an Executor based on the step configuration.

type ExitCoder 

type ExitCoder interface {
	ExitCode() int
}

ExitCoder is an interface for executors that can return an exit code.

type NodeStatusDeterminer 

type NodeStatusDeterminer interface {
	DetermineNodeStatus() (core.NodeStatus, error)
}

NodeStatusDeterminer is an interface for reporting the status of a node execution.

type OutputsProvider 

type OutputsProvider interface {
	GetOutputs() map[string]any
}

OutputsProvider is implemented by executors that publish DAG/action outputs.

type ParallelExecutor 

type ParallelExecutor interface {
	Executor

	// SetParamsList sets the parameters for running multiple sub DAGs in parallel.
	SetParamsList([]RunParams)
}

ParallelExecutor is an interface for parallel step executors.

type PushBackAware 

type PushBackAware interface {
	SetPushBackContext(inputs map[string]string, iteration int)
}

PushBackAware is implemented by executors that can incorporate push-back feedback into their conversation flow.

type PushBackPreviousStdoutAware 

type PushBackPreviousStdoutAware interface {
	SetPushBackPreviousStdout(path string)
}

PushBackPreviousStdoutAware is implemented by executors that can consume the previous stdout log path for push-back re-execution.

type RunParams 

type RunParams struct {
	RunID          string
	Params         string
	DAGName        string
	WorkerSelector map[string]string
}

RunParams holds the parameters for running a sub DAG.

type StatusDetailsProvider added in v2.12.0 

type StatusDetailsProvider interface {
	GetStatusDetails() []exec.NodeStatusDetail
}

StatusDetailsProvider reports independently tracked executions within a node.

type Stopper 

type Stopper interface {
	Stop(cmdutil.TerminationIntent) error
}

Stopper is implemented by executors that can handle lifecycle stop intent directly instead of receiving only a legacy OS signal.

type SubDAGExecutor 

type SubDAGExecutor struct {
	// DAG is the sub DAG to execute.
	// For local DAGs, this DAG's Location will be set to a temporary file.
	DAG *core.DAG
	// contains filtered or unexported fields
}

SubDAGExecutor is a helper for executing sub DAGs. It handles both regular DAGs and local DAGs (defined in the same file).

func NewSubDAGExecutor 

func NewSubDAGExecutor(ctx context.Context, childName string) (*SubDAGExecutor, error)

NewSubDAGExecutor creates a new SubDAGExecutor. It handles the logic for finding the DAG - either from the database or from local DAGs defined in the parent.

func NewSubDAGExecutorForDAG 

func NewSubDAGExecutorForDAG(ctx context.Context, dag *core.DAG) (*SubDAGExecutor, error)

NewSubDAGExecutorForDAG creates a SubDAGExecutor for an already-loaded DAG.

func (*SubDAGExecutor) Cleanup 

func (e *SubDAGExecutor) Cleanup(ctx context.Context) error

Cleanup removes any temporary files created for local DAGs. This should be called after the sub DAG execution is complete.

func (*SubDAGExecutor) Execute 

func (e *SubDAGExecutor) Execute(ctx context.Context, runParams RunParams, workDir string) (*exec.RunStatus, error)

Execute executes the sub DAG and returns the result. This is useful for parallel execution where results need to be collected.

func (*SubDAGExecutor) Kill 

func (e *SubDAGExecutor) Kill(sig os.Signal) error

Kill cancels all running sub DAG executions.

func (*SubDAGExecutor) Retry 

func (e *SubDAGExecutor) Retry(ctx context.Context, runParams RunParams, stepName, workDir string, path exec.RetryPath) (*exec.RunStatus, error)

Retry executes a parent-managed step retry for a previously started sub DAG.

func (*SubDAGExecutor) Reuse 

func (e *SubDAGExecutor) Reuse(ctx context.Context, runParams RunParams, workDir string) (*exec.RunStatus, error)

Reuse returns the persisted result of a child run without executing it.

func (*SubDAGExecutor) SetExternalStepRetry 

func (e *SubDAGExecutor) SetExternalStepRetry(enabled bool)

func (*SubDAGExecutor) SetWorkerSelector 

func (e *SubDAGExecutor) SetWorkerSelector(selector map[string]string)

SetWorkerSelector sets a per-invocation worker selector for the sub DAG.

func (*SubDAGExecutor) SetWorkspaceSeed 

func (e *SubDAGExecutor) SetWorkspaceSeed(seed WorkspaceSeed)

func (*SubDAGExecutor) Stop 

func (e *SubDAGExecutor) Stop(intent cmdutil.TerminationIntent) error

Stop cancels all running sub DAG executions according to the requested lifecycle intent.

type SubRunProvider 

type SubRunProvider interface {
	GetSubRuns() []exec.SubDAGRun
}

SubRunProvider is an interface for executors that spawn sub-DAG runs. This is used by executors like chat (with tools) to report sub-runs for UI drill-down functionality.

type SubWorkflowCancelIntent 

type SubWorkflowCancelIntent struct {
	Mode   SubWorkflowCancelMode
	Signal os.Signal
}

SubWorkflowCancelIntent carries runtime-owned cancellation intent.

type SubWorkflowCancelMode 

type SubWorkflowCancelMode string

SubWorkflowCancelMode describes how a child workflow should be stopped. 

const (
	// SubWorkflowCancelModeGraceful requests a graceful stop of the child workflow.
	SubWorkflowCancelModeGraceful SubWorkflowCancelMode = "graceful"
	// SubWorkflowCancelModeForce requests a forced stop of the child workflow.
	SubWorkflowCancelModeForce SubWorkflowCancelMode = "force"
)

type SubWorkflowCancelRequest 

type SubWorkflowCancelRequest struct {
	DAG        *core.DAG
	RootDAGRun exec.DAGRunRef
	RunID      string
	Intent     SubWorkflowCancelIntent
}

SubWorkflowCancelRequest describes a child workflow cancellation.

type SubWorkflowRequest 

type SubWorkflowRequest struct {
	DAG               *core.DAG
	ParentDAG         *core.DAG
	RootDAGRun        exec.DAGRunRef
	ParentDAGRun      exec.DAGRunRef
	RunID             string
	Params            string
	ProfileName       string
	TriggerActor      string
	WorkDir           string
	WorkerSelector    map[string]string
	ExternalStepRetry bool
	Reuse             bool
	RetryPath         exec.RetryPath
	Workspace         *SubWorkflowWorkspace
}

SubWorkflowRequest describes a child workflow invocation.

type SubWorkflowRetryRequest 

type SubWorkflowRetryRequest struct {
	SubWorkflowRequest
	StepName string
}

SubWorkflowRetryRequest describes a child workflow step retry.

type SubWorkflowRunner 

type SubWorkflowRunner interface {
	ShouldRun(ctx context.Context, req SubWorkflowRequest) bool
	Run(ctx context.Context, req SubWorkflowRequest) (*exec.RunStatus, error)
	Retry(ctx context.Context, req SubWorkflowRetryRequest) (*exec.RunStatus, error)
	Cancel(ctx context.Context, req SubWorkflowCancelRequest) error
}

SubWorkflowRunner runs child workflows behind a workflow-level interface.

func SubWorkflowRunnerFromContext 

func SubWorkflowRunnerFromContext(ctx context.Context) (SubWorkflowRunner, bool)

SubWorkflowRunnerFromContext returns the child workflow runner in ctx, if any.

type SubWorkflowWorkspace 

type SubWorkflowWorkspace struct {
	Descriptor workspacebundle.Descriptor
	Archive    []byte
}

SubWorkflowWorkspace carries an immutable child workflow workspace.

type TailWriter 

type TailWriter struct {
	// contains filtered or unexported fields
}

TailWriter forwards to an underlying writer and keeps a rolling tail of recent output up to `max` bytes. Safe for concurrent use.

func NewTailWriter 

func NewTailWriter(out io.Writer, max int) *TailWriter

NewTailWriter creates a tailWriter that keeps a rolling buffer of recent output with a maximum size of `max` bytes. If max <= 0, it falls back to defaultStderrTailLimit. If out is nil, it defaults to os.Stderr to preserve exec's behavior.

func NewTailWriterWithEncoding 

func NewTailWriterWithEncoding(out io.Writer, max int, encoding string) *TailWriter

NewTailWriterWithEncoding creates a TailWriter with character encoding support. The encoding parameter specifies the character encoding of the output (e.g., "utf-8", "shift_jis", "euc-jp"). If empty, UTF-8 is assumed.

func (*TailWriter) Tail 

func (t *TailWriter) Tail() string

Tail returns the rolling tail buffer (up to max bytes) as a decoded string. If an encoding was specified during creation, the buffer is decoded from that encoding to UTF-8. Otherwise, the raw bytes are returned as a string.

func (*TailWriter) Write 

func (t *TailWriter) Write(p []byte) (int, error)

type TaskOption 

type TaskOption func(*exec.DispatchTask)

TaskOption is a function that modifies a dispatch task.

func WithBaseConfig 

func WithBaseConfig(content string) TaskOption

WithBaseConfig sets the base config YAML content on the task. This allows workers to apply base config without needing local base config files.

func WithExternalStepRetry 

func WithExternalStepRetry(enabled bool) TaskOption

WithExternalStepRetry enables parent-managed step retries for the dispatched task.

func WithLabels 

func WithLabels(labels string) TaskOption

WithLabels sets additional labels (comma-separated) for the task.

func WithParentDagRun 

func WithParentDagRun(ref exec.DAGRunRef) TaskOption

WithParentDagRun sets the parent DAG run name and ID in the task.

func WithPreviousStatus 

func WithPreviousStatus(status *exec.DAGRunStatus) TaskOption

WithPreviousStatus sets the previous status for retry operations. When set, workers can retry without needing local DAGRunStore access.

func WithProfileName 

func WithProfileName(profileName string) TaskOption

WithProfileName sets the runtime profile name for a dispatched task.

func WithRetryPath 

func WithRetryPath(path exec.RetryPath) TaskOption

WithRetryPath sets the persisted child DAG path for a retry task.

func WithRootDagRun 

func WithRootDagRun(ref exec.DAGRunRef) TaskOption

WithRootDagRun sets the root DAG run name and ID in the task.

func WithScheduleTime 

func WithScheduleTime(scheduleTime string) TaskOption

WithScheduleTime sets the RFC 3339 timestamp of when the task was scheduled.

func WithSourceFile 

func WithSourceFile(sourceFile string) TaskOption

WithSourceFile sets the original DAG source file path for provenance-aware flows.

func WithStep 

func WithStep(step string) TaskOption

WithStep sets the step name for retry operations.

func WithTags 

func WithTags(tags string) TaskOption

WithTags sets additional labels (comma-separated) for the task. Deprecated: use WithLabels.

func WithTaskParams 

func WithTaskParams(params string) TaskOption

WithTaskParams sets the parameters for the task.

func WithTriggerActor 

func WithTriggerActor(actor string) TaskOption

WithTriggerActor sets the attributable trigger actor for a dispatched task.

func WithWorkerSelector 

func WithWorkerSelector(selector map[string]string) TaskOption

WithWorkerSelector sets the worker selector labels for the task.

func WithWorkspaceBundle 

func WithWorkspaceBundle(desc workspacebundle.Descriptor) TaskOption

WithWorkspaceBundle sets workspace bundle metadata for worker dispatch.

type ToolDefinitionProvider 

type ToolDefinitionProvider interface {
	GetToolDefinitions() []exec.ToolDefinition
}

ToolDefinitionProvider is an interface for executors that provide tool definitions. This is used by chat executors to report what tools were available to the LLM for debugging and visibility purposes.

type WorkspaceSeed 

type WorkspaceSeed struct {
	Descriptor workspacebundle.Descriptor
	Archive    []byte
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.