runtime

package
v2.11.4 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const ErrMsgOtherConditionNotMet = "other condition was not met"

Error message for when not all conditions are met

Variables

View Source 
var (
	// NewContext creates a new context with DAG execution metadata.
	NewContext = exec.NewContext
	// LookupDAGContext returns the DAG execution metadata when it is present.
	LookupDAGContext = exec.LookupContext
	// WithDatabase sets the database interface.
	WithDatabase = exec.WithDatabase
	// WithRootDAGRun sets the root DAG run reference for sub-DAG execution.
	WithRootDAGRun = exec.WithRootDAGRun
	// WithRetryPath sets a targeted child DAG retry path.
	WithRetryPath = exec.WithRetryPath
	// WithAttemptID sets the DAG-run attempt identifier.
	WithAttemptID = exec.WithAttemptID
	// WithTriggerType sets the DAG-run trigger type.
	WithTriggerType = exec.WithTriggerType
	// WithTriggerActor sets the attributable trigger actor.
	WithTriggerActor = exec.WithTriggerActor
	// WithRunStartedAt sets the recorded DAG-run start timestamp.
	WithRunStartedAt = exec.WithRunStartedAt
	// WithScheduleTime sets the logical schedule time.
	WithScheduleTime = exec.WithScheduleTime
	// WithParams sets runtime parameters.
	WithParams = exec.WithParams
	// WithDefaultEnvVars sets low-precedence inherited environment variables.
	WithDefaultEnvVars = exec.WithDefaultEnvVars
	// WithEnvVars sets additional execution-scoped environment variables.
	WithEnvVars = exec.WithEnvVars
	// WithCoordinator sets the coordinator dispatcher for distributed execution.
	WithCoordinator = exec.WithCoordinator
	// WithDefaultSecrets sets low-precedence inherited secret environment variables.
	WithDefaultSecrets = exec.WithDefaultSecrets
	// WithSecrets sets secret environment variables.
	WithSecrets = exec.WithSecrets
	// WithLogEncoding sets the log file character encoding.
	WithLogEncoding = exec.WithLogEncoding
	// WithLogWriterFactory sets the log writer factory for remote log streaming.
	WithLogWriterFactory = exec.WithLogWriterFactory
	// WithDefaultExecMode sets the server-level default execution mode.
	WithDefaultExecMode = exec.WithDefaultExecMode
	// WithDAGRunStore sets the dag-run store.
	WithDAGRunStore = exec.WithDAGRunStore
	// WithQueueStore sets the queue store.
	WithQueueStore = exec.WithQueueStore
	// WithStateStore sets the persistent DAG state store.
	WithStateStore = exec.WithStateStore
	// WithDAGRunLogDir sets the base log directory for newly persisted DAG runs.
	WithDAGRunLogDir = exec.WithDAGRunLogDir
	// WithDAGRunArtifactDir sets the base artifact directory for newly persisted DAG runs.
	WithDAGRunArtifactDir = exec.WithDAGRunArtifactDir
	// WithWorkDir sets the per-DAG-run working directory path.
	WithWorkDir = exec.WithWorkDir
	// WithArtifactDir sets the per-DAG-run artifact directory path.
	WithArtifactDir = exec.WithArtifactDir
	// WithRuntimeProfile sets selected runtime profile metadata.
	WithRuntimeProfile = exec.WithRuntimeProfile
)

Re-export execution package functions for convenience.

View Source 
var (
	ErrCyclicPlan  = errors.New("cyclic plan detected")
	ErrMissingNode = errors.New("missing node in execution plan")
)
View Source 
var (
	ErrUpstreamFailed   = fmt.Errorf("upstream failed")
	ErrUpstreamSkipped  = fmt.Errorf("upstream skipped")
	ErrUpstreamRejected = fmt.Errorf("upstream rejected")
	ErrDeadlockDetected = errors.New("deadlock detected: no runnable nodes but DAG not finished")
)
View Source 
var (
	ErrConditionNotMet = fmt.Errorf("condition was not met")
)

Errors for condition evaluation

Functions

func AllEnvs 

func AllEnvs(ctx context.Context) []string

AllEnvs returns all environment variables that needs to be passed to the command. Each element is in the form of "key=value".

func AllEnvsMap 

func AllEnvsMap(ctx context.Context) map[string]string

AllEnvsMap builds a map of environment variables from the current Env. It returns the EnvScope's ToMap directly, avoiding the round-trip through string splitting.

func DAGShell 

func DAGShell(ctx context.Context) []string

DAGShell returns the evaluated shell command for DAG-level operations. This is used for preconditions and other operations that run before any steps. Unlike Env.Shell(), this doesn't require a step context.

func EffectiveLLMConfig 

func EffectiveLLMConfig(cfg *core.LLMConfig, model core.ModelEntry) *core.LLMConfig

EffectiveLLMConfig folds one model entry into the shared LLM config, so the entry's own provider, name, and overrides win where it sets them.

func EvalBool 

func EvalBool(ctx context.Context, value any) (bool, error)

EvalBool evaluates the given value with the variables within the execution context and parses it as a boolean.

func EvalCondition 

func EvalCondition(ctx context.Context, shell []string, c *core.Condition) error

EvalCondition evaluates the condition and returns the actual value. It returns an error if the evaluation failed or the condition is invalid. If c.Negate is true, the result is inverted: the condition passes when it would normally fail, and vice versa.

func EvalConditions 

func EvalConditions(ctx context.Context, shell []string, cond []*core.Condition) error

EvalConditions evaluates a list of conditions and checks the results. It returns an error if any of the conditions were not met.

func EvalObject 

func EvalObject[T any](ctx context.Context, obj T) (T, error)

EvalObject recursively evaluates the string fields of the given object with the variables within the execution context.

func GenerateSubDAGRunID 

func GenerateSubDAGRunID(ctx context.Context, params string, repeated bool) string

GenerateSubDAGRunID generates a unique run ID based on the current DAG run ID, step name, and parameters.

func GenerateSubDAGRunIDForTarget 

func GenerateSubDAGRunIDForTarget(ctx context.Context, dagName, params string, repeated bool) string

GenerateSubDAGRunIDForTarget generates a unique run ID for a sub-DAG target. Including the target keeps deterministic IDs stable for retries while avoiding collisions when one parent step dispatches different child DAGs with identical params.

func MaskSecretsForProvider 

func MaskSecretsForProvider(ctx context.Context, msgs []exec.LLMMessage) []exec.LLMMessage

MaskSecretsForProvider replaces secret values in messages with a mask before they leave for an external model. Fields such as an LLM system prompt are resolved against the runtime scope, so a reference to a secret becomes the secret itself; only the copy sent to the provider is masked, and the run's own transcript keeps the resolved text.

func NewContextForTest 

func NewContextForTest(ctx context.Context, dag *core.DAG, dagRunID, logFile string) context.Context

NewContextForTest creates a minimal context for testing purposes. This is useful when you need a context with just basic DAG metadata.

func NewDAGRunRef 

func NewDAGRunRef(name, runID string) exec.DAGRunRef

NewDAGRunRef is a convenience wrapper for execution.NewDAGRunRef.

func NewLLMProvider 

func NewLLMProvider(ctx context.Context, cfg *core.LLMConfig) (llmpkg.Provider, error)

NewLLMProvider builds an LLM provider from a resolved DAG or step LLM config. The API key and base URL are evaluated against the current runtime env.

func NormalizeEnvVarExpr 

func NormalizeEnvVarExpr(expr string) string

NormalizeEnvVarExpr converts an environment variable reference to ${VAR} form, accepting VAR, $VAR, and ${VAR}.

func OutputValuesFromExecNodes 

func OutputValuesFromExecNodes(nodes []*exec.Node) map[string]any

OutputValuesFromExecNodes extracts typed DAG/action outputs from persisted nodes.

func OutputValuesFromNodes 

func OutputValuesFromNodes(nodes []NodeData) map[string]any

OutputValuesFromNodes extracts typed DAG/action outputs from runtime nodes.

func RepairStaleLocalRun 

func RepairStaleLocalRun(
	ctx context.Context,
	attempt exec.DAGRunAttempt,
	dag *core.DAG,
) (*exec.DAGRunStatus, bool, error)

RepairStaleLocalRun marks an active local run as failed after liveness checks have confirmed the local proc file is stale or missing.

func RepairStaleRemoteRun 

func RepairStaleRemoteRun(
	ctx context.Context,
	cfg StaleRunRepairConfig,
	status *exec.DAGRunStatus,
	fallbackAttemptID string,
	fallbackWorkerID string,
) (*exec.DAGRunStatus, bool, error)

RepairStaleRemoteRun marks an active remote run failed only when both the claim lease and worker evidence confirm that its execution claim is gone.

func ResolveDAGShell 

func ResolveDAGShell(ctx context.Context) ([]string, error)

ResolveDAGShell returns the evaluated shell command for DAG-level operations.

func ResolveModels 

func ResolveModels(ctx context.Context, models []core.ModelEntry) ([]core.ModelEntry, error)

ResolveModels evaluates variable substitution in the provider and model name of each entry. base_url is resolved later, when the provider is built, once shared config has been merged in; api_key_name is not value-resolved at all, since it names an environment variable the provider construction reads.

func ResolveString 

func ResolveString(ctx context.Context, raw string, field cmnvalue.Field) (string, error)

ResolveString resolves raw with the semantic field in the runtime environment.

func ValueResolver 

func ValueResolver(ctx context.Context) cmnvalue.Resolver

ValueResolver returns a semantic value resolver for the runtime environment in ctx.

func ValueResolverWithScope 

func ValueResolverWithScope(ctx context.Context, scope *cmnvalue.EnvScope) cmnvalue.Resolver

ValueResolverWithScope returns a semantic value resolver using scope as the runtime env scope.

func WithDAGContext 

func WithDAGContext(ctx context.Context, rCtx Context) context.Context

WithDAGContext returns a new context with the given DAGContext. This is a convenience wrapper for execution.WithContext.

func WithEnv 

func WithEnv(ctx context.Context, e Env) context.Context

WithEnv returns a new context with the given execution context.

Types

type ArtifactFinalizer 

type ArtifactFinalizer interface {
	Finalize(ctx context.Context, attemptID, dir string) error
}

ArtifactFinalizer persists artifacts before terminal status is reported.

type AttemptRejected 

type AttemptRejected interface {
	error
	AttemptRejectedReason() string
}

AttemptRejected marks a status push failure caused by a non-authoritative attempt.

type ChatMessagesHandler 

type ChatMessagesHandler interface {
	// WriteStepMessages writes messages for a single step.
	WriteStepMessages(ctx context.Context, stepName string, messages []exec.LLMMessage) error
	// ReadStepMessages reads messages for a single step.
	ReadStepMessages(ctx context.Context, stepName string) ([]exec.LLMMessage, error)
}

ChatMessagesHandler handles chat session messages for persistence.

type Config 

type Config struct {
	LogDir          string
	MaxActiveSteps  int
	Timeout         time.Duration
	Delay           time.Duration
	Dry             bool
	OnInit          *core.Step
	OnExit          *core.Step
	OnSuccess       *core.Step
	OnFailure       *core.Step
	OnAbort         *core.Step
	DAGRunID        string
	MessagesHandler ChatMessagesHandler
	OnWait          *core.Step
	ForcedStatus    *core.Status

	DAGRunAutoRetryCount int
	DAGRunAutoRetryLimit int
	DAGRunIsRoot         bool
}

type Context 

type Context = exec.Context

Context is an alias for execution.Context

func GetDAGContext 

func GetDAGContext(ctx context.Context) Context

GetDAGContext retrieves the DAGContext from the context. This is a convenience wrapper for execution.GetContext.

type ContextOption 

type ContextOption = exec.ContextOption

ContextOption is an alias for execution.ContextOption

type Data 

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

Data is a thread-safe wrapper around NodeData.

func (*Data) AddSubRunsRepeated 

func (d *Data) AddSubRunsRepeated(subRun ...SubDAGRun)

AddSubRunsRepeated appends repeated sub DAG runs to the node.

func (*Data) Args 

func (d *Data) Args() []string

func (*Data) ClearState 

func (d *Data) ClearState(s core.Step)

func (*Data) ClearVariable 

func (d *Data) ClearVariable(key string)

func (*Data) CompleteHumanTaskDryRun 

func (d *Data) CompleteHumanTaskDryRun(prompt string)

CompleteHumanTaskDryRun records the resolved prompt and completes a dry-run task.

func (*Data) ContinueOn 

func (d *Data) ContinueOn() core.ContinueOn

func (*Data) Data 

func (s *Data) Data() NodeData

func (*Data) Error 

func (d *Data) Error() error

func (*Data) Finish 

func (d *Data) Finish()

func (*Data) GetApprovalInputs 

func (d *Data) GetApprovalInputs() map[string]string

GetApprovalInputs returns a copy of the approval inputs map.

func (*Data) GetChatMessages 

func (d *Data) GetChatMessages() []exec.LLMMessage

GetChatMessages returns the chat session messages for the node.

func (*Data) GetDoneCount 

func (d *Data) GetDoneCount() int

func (*Data) GetExitCode 

func (d *Data) GetExitCode() int

func (*Data) GetRetryCount 

func (d *Data) GetRetryCount() int

func (*Data) GetStderr 

func (d *Data) GetStderr() string

func (*Data) GetStdout 

func (d *Data) GetStdout() string

func (*Data) GetToolDefinitions 

func (d *Data) GetToolDefinitions() []exec.ToolDefinition

GetToolDefinitions returns the tool definitions that were available to the LLM.

func (*Data) IncDoneCount 

func (d *Data) IncDoneCount()

func (*Data) IncRetryCount 

func (d *Data) IncRetryCount()

func (*Data) IsRepeated 

func (d *Data) IsRepeated() bool

func (*Data) MarkError 

func (d *Data) MarkError(err error)

func (*Data) MatchExitCode 

func (d *Data) MatchExitCode(exitCodes []int) bool

func (*Data) Name 

func (d *Data) Name() string

func (*Data) OpenHumanTask 

func (d *Data) OpenHumanTask(prompt string, startedAt time.Time)

OpenHumanTask records the resolved prompt and transitions the node to waiting.

func (*Data) ResetError 

func (d *Data) ResetError()

func (*Data) SetApprovalInputs 

func (d *Data) SetApprovalInputs(inputs map[string]string)

SetApprovalInputs sets the approval inputs map.

func (*Data) SetArgs 

func (d *Data) SetArgs(args []string)

func (*Data) SetChatMessages 

func (d *Data) SetChatMessages(messages []exec.LLMMessage)

SetChatMessages sets the chat session messages for the node.

func (*Data) SetControllerState 

func (d *Data) SetControllerState(raw json.RawMessage)

SetControllerState stores the controller's goal progress on the node.

func (*Data) SetError 

func (d *Data) SetError(err error)

func (*Data) SetExecutorConfig 

func (d *Data) SetExecutorConfig(cfg core.ExecutorConfig)

func (*Data) SetExitCode 

func (d *Data) SetExitCode(exitCode int)

func (*Data) SetRepeated 

func (d *Data) SetRepeated(repeated bool)

func (*Data) SetRetriedAt 

func (d *Data) SetRetriedAt(retriedAt time.Time)

func (*Data) SetRetryCount 

func (d *Data) SetRetryCount(retryCount int)

func (*Data) SetScript 

func (d *Data) SetScript(script string)

func (*Data) SetStatus 

func (d *Data) SetStatus(s core.NodeStatus)

func (*Data) SetStep 

func (s *Data) SetStep(step core.Step)

func (*Data) SetSubDAG 

func (d *Data) SetSubDAG(subDAG core.SubDAG)

func (*Data) SetSubRuns 

func (d *Data) SetSubRuns(subRuns []SubDAGRun)

SetSubRuns replaces the sub DAG runs associated with the node.

func (*Data) SetToolDefinitions 

func (d *Data) SetToolDefinitions(tools []exec.ToolDefinition)

SetToolDefinitions sets the tool definitions that were available to the LLM.

func (*Data) SetWorkingDir 

func (d *Data) SetWorkingDir(workingDir string)

SetWorkingDir records the effective working directory for the node execution.

func (*Data) Setup 

func (d *Data) Setup(ctx context.Context, logFile string, startedAt time.Time) error

func (*Data) SignalOnStop 

func (d *Data) SignalOnStop() string

func (*Data) State 

func (d *Data) State() NodeState

func (*Data) Status 

func (d *Data) Status() core.NodeStatus

func (*Data) Step 

func (d *Data) Step() core.Step

func (*Data) StepInfo 

func (d *Data) StepInfo() cmnvalue.StepInfo

type Database 

type Database = exec.Database

Database is an alias for execution.Database

type Dispatcher 

type Dispatcher = exec.Dispatcher

Dispatcher is an alias for execution.Dispatcher

type Env 

type Env struct {
	// Embedded execution metadata from parent DAG run containing DAGRunID,
	// RootDAGRun reference, DAG configuration, database interface, and
	// coordinator dispatcher
	Context

	// Unified scope chain for environment variable lookups.
	// This scope is the source for $VAR and ${VAR} expansion.
	// Layers (highest to lowest precedence): StepEnv > Outputs > Secrets > DAGEnv > OS
	Scope *cmnvalue.EnvScope

	// The current step being executed within this environment context
	Step core.Step

	// Maps step IDs to their execution information (stdout, stderr, exitCode)
	// allowing steps to reference outputs from other steps using expressions
	// like ${stepID.stdout} or ${stepID.exitCode} in their configurations.
	// Step references are resolved separately from environment variables.
	StepMap map[string]cmnvalue.StepInfo

	// Foreach contains the current item scope for foreach body evaluation.
	Foreach cmnvalue.Values

	// Resolved absolute path for the step's working directory, determined by:
	// 1. Step's Dir field if specified (resolved to absolute path)
	// 2. Current working directory if Dir is not specified
	// This path is also set as the PWD environment variable
	WorkingDir string
}

Env holds information about the DAG and the current step to execute including the variables (environment variables and DAG variables) that are available to the step.

func GetEnv 

func GetEnv(ctx context.Context) Env

GetEnv returns the execution context from the given context.

func LookupEnv 

func LookupEnv(ctx context.Context) (Env, bool)

LookupEnv returns the execution environment when one is present in ctx.

func NewEnv 

func NewEnv(ctx context.Context, step core.Step) Env

NewEnv creates a new Env configured for executing the provided step. It resolves the step's working directory and sets initial per-step environment variables: PWD to the resolved working directory and the DAG run step name.

func NewEnvWithError 

func NewEnvWithError(ctx context.Context, step core.Step) (Env, error)

NewEnvWithError creates an Env and returns working directory resolution errors.

func NewPlanEnv 

func NewPlanEnv(ctx context.Context, step core.Step, plan *Plan) Env

func NewPlanEnvForNode 

func NewPlanEnvForNode(ctx context.Context, node *Node, plan *Plan) Env

func NewPlanEnvForNodeWithError 

func NewPlanEnvForNodeWithError(ctx context.Context, node *Node, plan *Plan) (Env, error)

func NewPlanEnvWithError 

func NewPlanEnvWithError(ctx context.Context, step core.Step, plan *Plan) (Env, error)

func (Env) AllEnvs 

func (e Env) AllEnvs() []string

AllEnvs returns all environment variables that needs to be passed to the command. Uses EnvScope as the source of environment variables.

func (Env) DAGRunRef 

func (e Env) DAGRunRef() exec.DAGRunRef

DAGRunRef returns the DAGRunRef for the current execution context.

func (Env) EvalBool 

func (e Env) EvalBool(ctx context.Context, value any) (bool, error)

EvalBool evaluates the given value with the variables within the execution context

func (Env) MailerConfig 

func (e Env) MailerConfig(ctx context.Context) (mailer.Config, error)

MailerConfig returns the SMTP mailer configuration with variables evaluated.

func (Env) ResolveShell 

func (e Env) ResolveShell(ctx context.Context) ([]string, error)

ResolveShell returns the shell command to use for this execution context.

func (Env) Shell 

func (e Env) Shell(ctx context.Context) []string

Shell returns the shell command to use for this execution context.

func (Env) UserEnvsMap 

func (e Env) UserEnvsMap() map[string]string

UserEnvsMap returns user-defined environment variables as a map, excluding OS environment (BaseEnv). Use this for isolated execution environments. Uses EnvScope as the source of environment variables.

func (Env) WithEnvVars 

func (e Env) WithEnvVars(envs ...string) Env

WithEnvVars returns a new Env with the given environment variable(s) added to the Scope.

type LogWriterFactory 

type LogWriterFactory = exec.LogWriterFactory

LogWriterFactory is re-exported from execution package

type Manager 

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

Manager provides methods to interact with DAGs, including starting, stopping, restarting, and retrieving status information. It communicates with the DAG through a socket interface and manages dag-run data.

func NewManager 

func NewManager(drs exec.DAGRunStore, ps exec.ProcStore, cfg *config.Config, opts ...ManagerOption) Manager

NewManager creates a new Manager instance. The Manager is used to interact with the DAG.

func (*Manager) FindSubDAGRunStatus 

func (m *Manager) FindSubDAGRunStatus(ctx context.Context, rootDAGRun exec.DAGRunRef, subRunID string) (*exec.DAGRunStatus, error)

FindSubDAGRunStatus retrieves the status of a sub dag-run by its ID. It repairs stale local child runs before returning their status.

func (*Manager) GenDAGRunID 

func (m *Manager) GenDAGRunID(_ context.Context) (string, error)

GenDAGRunID generates a unique ID for a dag-run.

func (*Manager) GetCurrentStatus 

func (m *Manager) GetCurrentStatus(ctx context.Context, dag *core.DAG, dagRunID string) (*exec.DAGRunStatus, error)

GetCurrentStatus retrieves the current status of a dag-run by its run ID. If the run ID is empty, it resolves the latest run first. If the dag-run is running, it queries the socket for the current status. If the socket doesn't exist or times out, it falls back to stored status or creates an initial status.

func (*Manager) GetLatestStatus 

func (m *Manager) GetLatestStatus(ctx context.Context, dag *core.DAG) (exec.DAGRunStatus, error)

GetLatestStatus retrieves the latest status of a DAG. If the DAG is running, it attempts to get the current status from the socket. If that fails and the local proc is dead, it repairs the stale run before returning it.

func (*Manager) GetSavedStatus 

func (m *Manager) GetSavedStatus(ctx context.Context, dagRun exec.DAGRunRef) (*exec.DAGRunStatus, error)

GetSavedStatus retrieves the saved status of a dag-run by its core.DAGRun reference. For stale local runs, it repairs the persisted status before returning it.

func (*Manager) IsRunning 

func (m *Manager) IsRunning(ctx context.Context, dag *core.DAG, dagRunID string) bool

IsRunning checks if a dag-run is currently running. It prefers the live socket status and falls back to a fresh proc heartbeat plus persisted running status.

func (*Manager) ListRecentStatus 

func (m *Manager) ListRecentStatus(ctx context.Context, name string, n int) []exec.DAGRunStatus

ListRecentStatus retrieves the n most recent statuses for a DAG by name. It returns a slice of Status objects, filtering out any that cannot be read.

func (*Manager) Stop 

func (m *Manager) Stop(ctx context.Context, dag *core.DAG, dagRunID string) error

Stop stops running DAG-runs and can cancel an explicit failed DAG-run that is still pending DAG-level auto-retry when dagRunID is provided.

func (*Manager) UpdateStatus 

func (m *Manager) UpdateStatus(ctx context.Context, rootDAGRun exec.DAGRunRef, newStatus exec.DAGRunStatus) error

UpdateStatus updates the status of a dag-run.

type ManagerOption 

type ManagerOption func(*Manager)

ManagerOption configures a Manager.

func WithManagerClock 

func WithManagerClock(now func() time.Time) ManagerOption

WithManagerClock overrides the Manager clock.

type Node 

type Node struct {
	Data
	// contains filtered or unexported fields
}

Node is a node in a DAG. It executes a command.

func NewNode 

func NewNode(step core.Step, state NodeState) *Node

func NodeWithData 

func NodeWithData(data NodeData) *Node

func (*Node) BuildSubDAGRuns 

func (n *Node) BuildSubDAGRuns(ctx context.Context, subDAG *core.SubDAG) ([]SubDAGRun, error)

BuildSubDAGRuns constructs the sub DAG runs based on parallel configuration.

func (*Node) Cancel 

func (n *Node) Cancel()

func (*Node) Execute 

func (n *Node) Execute(ctx context.Context, onSetup ...func()) error

func (*Node) ID 

func (n *Node) ID() int

func (*Node) Init 

func (n *Node) Init()

func (*Node) ItemToParam 

func (n *Node) ItemToParam(item any) (string, error)

ItemToParam converts a parallel item to a parameter string

func (*Node) LogContainsPattern 

func (n *Node) LogContainsPattern(ctx context.Context, patterns []string) (bool, error)

LogContainsPattern checks if any of the given patterns exist in the node's log file. If a pattern starts with "regexp:", it will be treated as a regular expression. Returns false if no log file exists or no pattern is found. Returns error if there are issues reading the file or invalid regex pattern.

func (*Node) NodeData 

func (n *Node) NodeData() NodeData

func (*Node) OutputVariablesMap 

func (n *Node) OutputVariablesMap() map[string]string

OutputVariablesMap returns output variables as key->value map. This is used to build the EnvScope chain with predecessor outputs.

func (*Node) Prepare 

func (n *Node) Prepare(ctx context.Context, logDir string, dagRunID string) error

func (*Node) ResetForRerun 

func (n *Node) ResetForRerun(step core.Step)

ResetForRerun returns the node to its declared definition so it can execute again. It clears the command-evaluation cache along with the run state, since arguments holding runtime references must be resolved against current values rather than those captured on the first attempt.

func (*Node) SetupEnv 

func (n *Node) SetupEnv(ctx context.Context) context.Context

func (*Node) ShouldContinue 

func (n *Node) ShouldContinue(ctx context.Context) bool

func (*Node) ShouldMarkSuccess 

func (n *Node) ShouldMarkSuccess(ctx context.Context) bool

func (*Node) Signal 

func (n *Node) Signal(ctx context.Context, sig os.Signal, allowOverride bool)

func (*Node) StdoutFile 

func (n *Node) StdoutFile() string

func (*Node) Stop 

func (n *Node) Stop(ctx context.Context, intent cmdutil.TerminationIntent, allowOverride bool)

Stop requests that the node's executor stop according to lifecycle intent.

func (*Node) Teardown 

func (n *Node) Teardown() error

type NodeData 

type NodeData struct {
	Step  core.Step
	State NodeState
}

NodeData represents the data of a node.

func (NodeData) OutputsValueMap 

func (d NodeData) OutputsValueMap() map[string]any

func (NodeData) OutputsValueStringMap 

func (d NodeData) OutputsValueStringMap() map[string]string

func (NodeData) StepOutputsValueMap 

func (d NodeData) StepOutputsValueMap() map[string]string

func (NodeData) StringFormOutputValue 

func (d NodeData) StringFormOutputValue() (string, bool)

StringFormOutputValue returns the canonical captured output for string-form output: NAME steps. OutputValue is the source of truth for newly executed steps; OutputVariables remains as a backward-compatible fallback for previously persisted state.

type NodeState 

type NodeState struct {
	// Status represents the state of the node.
	Status core.NodeStatus
	// Stdout is the log file path from the node.
	Stdout string
	// Stderr is the log file path for the error log (stderr).
	Stderr string
	// WorkingDir is the effective working directory used for this node execution.
	WorkingDir string
	// StepOutputFile is the DAGU_OUTPUT_FILE path for the current step attempt.
	StepOutputFile string
	// StartedAt is the time when the node started.
	StartedAt time.Time
	// FinishedAt is the time when the node finished.
	FinishedAt time.Time
	// RetryCount is the number of retries happened based on the retry policy.
	RetryCount int
	// RetriedAt is the time when the node was retried last time.
	RetriedAt time.Time
	// DoneCount is the number of times the node was executed.
	DoneCount int
	// Repeated is true if the node is a repeated step.
	// This is used to generate unique run IDs for repeated steps in case the node
	// runs nested DAGs.
	Repeated bool
	// SkippedByRetry marks a node that was intentionally skipped by an edited
	// retry while preserving its output variables for downstream steps.
	SkippedByRetry bool
	// Error is the error that the executor encountered.
	Error error
	// ExitCode is the exit code that the command exited with.
	// It only makes sense when the node is a command executor.
	ExitCode int
	// Parallel contains the evaluated parallel execution state for the node.
	// This is populated when a step has parallel configuration and tracks
	// all the items that need to be executed in parallel.
	*Parallel
	// SubRuns stores the sub dag-runs.
	SubRuns []SubDAGRun
	// SubRunsRepeated stores the repeated sub dag-runs.
	SubRunsRepeated []SubDAGRun
	// OutputVariables stores the output variables for the following steps.
	// It only contains the local output variables.
	OutputVariables *collections.SyncMap
	// OutputValue stores the step-scoped output payload for ${step.output} references.
	// String-form output stores captured stdout; object-form output stores compact JSON.
	OutputValue *string
	// OutputsValue stores the legacy DAG/action outputs payload.
	OutputsValue *string
	// StepOutputsValue stores declared file-based outputs for ${steps.<id>.outputs.<name>} references.
	StepOutputsValue *string
	// HumanTaskInput stores the validated input submitted to complete a human task.
	HumanTaskInput json.RawMessage
	// ControllerState stores the goal progress of a controller DAG. It is carried
	// across attempts so a suspended controller resumes where it left off.
	ControllerState json.RawMessage
	// HumanTaskCompletedBy is the name of the subject that completed the human task.
	HumanTaskCompletedBy string
	// HumanTaskCompletedByID is the ID of the subject that completed the human task.
	HumanTaskCompletedByID string
	// ChatMessages stores the chat session messages for message passing between steps.
	ChatMessages []exec.LLMMessage
	// ToolDefinitions stores the tool definitions that were available to the LLM during execution.
	// This provides visibility into what tools/functions the LLM could call.
	ToolDefinitions []exec.ToolDefinition
	// ApprovalInputs stores key-value parameters provided during approval.
	// These are available as environment variables in subsequent steps.
	ApprovalInputs map[string]string
	// ApprovedAt is the time when the step was approved.
	ApprovedAt string
	// ApprovedBy is the username of the user who approved the step.
	ApprovedBy string
	// ApprovedByID is the ID of the subject that approved the step.
	ApprovedByID string
	// RejectedAt is the time when the step was rejected.
	RejectedAt string
	// RejectedBy is the username of the user who rejected the step.
	RejectedBy string
	// RejectedByID is the ID of the subject that rejected the step.
	RejectedByID string
	// RejectionReason stores the optional reason for rejection.
	RejectionReason string
	// ApprovalIteration tracks how many times this step has been pushed back.
	ApprovalIteration int
	// PushBackInputs stores inputs from the last push-back for env var injection.
	PushBackInputs map[string]string
	// PushBackHistory stores the chronological push-back feedback for this step.
	PushBackHistory []exec.PushBackEntry
	// PushBackPreviousStdout stores the stdout log path from the execution that
	// was reset by the latest push-back.
	PushBackPreviousStdout string
}

type OutputCoordinator 

type OutputCoordinator struct {
	StderrRedirectFile *os.File
	// contains filtered or unexported fields
}

func (*OutputCoordinator) StdoutFile 

func (oc *OutputCoordinator) StdoutFile() string

type Parallel 

type Parallel struct {
	// Items contains all the parallel items to be executed.
	// Each item will result in a separate sub DAG run.
	Items []ParallelItem
}

Parallel represents the evaluated parallel execution configuration for a node. It contains the expanded list of items to be processed in parallel.

type ParallelItem 

type ParallelItem struct {
	// Item contains the actual data for this parallel execution.
	// It can be either a simple value or a map of parameters from core.ParallelItem.
	Item core.ParallelItem
}

ParallelItem represents a single item in a parallel execution. It combines the item data with a unique identifier for tracking.

type Plan 

type Plan struct {

	// Immutable adjacency lists (exposing for unit tests)
	DependencyMap map[int][]int // node ID -> list of dependency node IDs (upstream)
	DependantMap  map[int][]int // node ID -> list of dependent node IDs (downstream)
	// contains filtered or unexported fields
}

Plan represents a plan of execution for a set of steps. It encapsulates the graph structure and ensures thread-safe access.

func CreateRetryPlan 

func CreateRetryPlan(ctx context.Context, dag *core.DAG, nodes ...*Node) (*Plan, error)

CreateRetryPlan creates a new execution plan for retrying specific nodes.

func CreateStepRetryPlan 

func CreateStepRetryPlan(dag *core.DAG, nodes []*Node, stepName string) (*Plan, error)

CreateStepRetryPlan creates a new execution plan for retrying a specific step.

func NewPlan 

func NewPlan(steps ...core.Step) (*Plan, error)

NewPlan creates a new execution plan from the given steps. It builds the graph, validates it (checking for cycles), and returns the plan.

func NewPlanFromNodes 

func NewPlanFromNodes(nodes ...*Node) (*Plan, error)

NewPlanFromNodes creates a plan from existing nodes without modifying their states.

func (*Plan) CheckFinished 

func (p *Plan) CheckFinished() bool

CheckFinished checks if all nodes have completed (successfully or otherwise).

func (*Plan) Dependencies 

func (p *Plan) Dependencies(nodeID int) []int

Dependencies returns the IDs of the nodes that the given node depends on.

func (*Plan) Dependents 

func (p *Plan) Dependents(nodeID int) []int

Dependents returns the IDs of the nodes that depend on the given node.

func (*Plan) Duration 

func (p *Plan) Duration() time.Duration

func (*Plan) Finish 

func (p *Plan) Finish()

func (*Plan) FinishAt 

func (p *Plan) FinishAt() time.Time

func (*Plan) GetNode 

func (p *Plan) GetNode(id int) *Node

GetNode returns the node with the given ID.

func (*Plan) GetNodeByName 

func (p *Plan) GetNodeByName(name string) *Node

GetNodeByName returns the node with the given name.

func (*Plan) HasActiveNodes 

func (p *Plan) HasActiveNodes() bool

HasActiveNodes checks if any node is actively executing or waiting for a retry.

func (*Plan) IsController 

func (p *Plan) IsController() bool

IsController reports whether execution order is decided by a controller step rather than by dependency edges.

func (*Plan) IsFinished 

func (p *Plan) IsFinished() bool

func (*Plan) IsRunning 

func (p *Plan) IsRunning() bool

IsRunning checks if any node is currently running or pending.

func (*Plan) IsStarted 

func (p *Plan) IsStarted() bool

func (*Plan) NodeData 

func (p *Plan) NodeData() []NodeData

NodeData returns a snapshot of data for all nodes.

func (*Plan) NodeStates 

func (p *Plan) NodeStates() PlanNodeStates

NodeStates returns whether any nodes are running, waiting, not started, or rejected. Single pass, single lock for atomic read.

func (*Plan) Nodes 

func (p *Plan) Nodes() []*Node

Nodes returns a slice of all nodes in the plan.

func (*Plan) StartAt 

func (p *Plan) StartAt() time.Time

func (*Plan) WaitingStepNames 

func (p *Plan) WaitingStepNames() []string

WaitingStepNames returns the names of steps that require manual action.

type PlanNodeStates 

type PlanNodeStates struct {
	HasRunning    bool
	HasRetrying   bool
	HasWaiting    bool
	HasNotStarted bool
	HasRejected   bool
}

PlanNodeStates holds the state flags for nodes in a plan.

type RetryPolicy 

type RetryPolicy struct {
	Limit     int
	Interval  time.Duration
	ExitCodes []int
}

func (*RetryPolicy) ShouldRetry 

func (r *RetryPolicy) ShouldRetry(exitCode int) bool

ShouldRetry determines if a node should be retried based on the exit code and retry policy

type RunStatus 

type RunStatus = exec.RunStatus

RunStatus is an alias for execution.RunStatus

type Runner 

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

Runner runs a plan of steps.

func New 

func New(cfg *Config) *Runner

func (*Runner) Cancel 

func (r *Runner) Cancel(p *Plan)

Cancel sends -1 signal to all nodes.

func (*Runner) GetMetrics 

func (r *Runner) GetMetrics() map[string]any

GetMetrics returns the current metrics for the runner

func (*Runner) HandlerNode 

func (r *Runner) HandlerNode(name core.HandlerType) *Node

HandlerNode returns the handler node with the given name.

func (*Runner) NodesInRunOrder 

func (r *Runner) NodesInRunOrder(plan *Plan) []*Node

NodesInRunOrder returns the plan's step nodes together with the lifecycle handler nodes that were configured, ordered by when they run. Handlers that the DAG does not declare are omitted.

func (*Runner) Run 

func (r *Runner) Run(ctx context.Context, plan *Plan, progressCh chan *Node) error

Run runs the plan of steps.

func (*Runner) Signal 

func (r *Runner) Signal(
	ctx context.Context, plan *Plan, sig os.Signal, done chan bool, allowOverride bool,
)

Signal sends a signal to the runner. for a node with repeat policy, it does not stop the node and wait to finish current run.

func (*Runner) Status 

func (r *Runner) Status(ctx context.Context, p *Plan) core.Status

Status returns the status of the runner.

func (*Runner) Stop 

func (r *Runner) Stop(
	ctx context.Context, plan *Plan, intent cmdutil.TerminationIntent, done chan bool, allowOverride bool,
)

Stop requests that all active nodes stop according to lifecycle intent.

type SchedulerLogStreamer 

type SchedulerLogStreamer interface {
	exec.LogWriterFactory
	NewSchedulerLogWriter(ctx context.Context, localFile *os.File) io.WriteCloser
	StreamSchedulerLog(ctx context.Context, logFilePath string) error
}

SchedulerLogStreamer streams a completed scheduler log.

type StaleRunRepairConfig 

type StaleRunRepairConfig struct {
	DAGRunStore                   exec.DAGRunStore
	DAGRunLeaseStore              exec.DAGRunLeaseStore
	WorkerHeartbeatStore          exec.WorkerHeartbeatStore
	StaleLeaseThreshold           time.Duration
	StaleWorkerHeartbeatThreshold time.Duration
	Now                           func() time.Time
}

StaleRunRepairConfig provides the stores, thresholds, and clock used to confirm and repair stale remote runs.

type StatusPusher 

type StatusPusher interface {
	Push(ctx context.Context, status exec.DAGRunStatus) error
}

StatusPusher reports DAG run status outside the current execution process.

type StepExecutor 

type StepExecutor struct{}

StepExecutor owns a single step execution attempt.

The scheduler-facing Runner decides when a node should run. StepExecutor owns the executor-specific protocol for setting up one run, passing executor context, collecting executor side channels, and capturing outputs.

func NewStepExecutor 

func NewStepExecutor() *StepExecutor

NewStepExecutor creates a StepExecutor.

func (*StepExecutor) Execute 

func (e *StepExecutor) Execute(ctx context.Context, node *Node, onSetup ...func()) error

Execute runs one node execution attempt and stores executor side effects on the node. Runner owns scheduling, retries, repeats, and final DAG-state decisions; StepExecutor only preserves executor-provided status overrides.

type SubDAGRun 

type SubDAGRun struct {
	// DAGRunID is the unique identifier for the sub dag-run.
	// It is generated as a base58-encoded SHA-256 hash of the string:
	// "<parent-dag-run-id>:<step-name>:<deterministic-json-params>"
	//
	// This deterministic ID generation ensures:
	// - Same parameters always produce the same sub DAG run ID
	// - Retries reuse existing sub DAG runs instead of creating duplicates
	// - Each step's children are namespaced by step name to prevent collisions
	//
	// The params are encoded as deterministic JSON (sorted keys) before hashing.
	// Example input: "abc123:process-regions:{"REGION":"us-east-1","VERSION":"1.0.0"}"
	// Example output: "5Kd3NBUAdUnhyzenEwVLy9pBKxSwXvE9FMPyR4UKZvpe"
	DAGRunID string
	// Params contains the raw parameters passed to the sub DAG run.
	// This can be:
	// - A simple string: "param1 param2"
	// - Key-value pairs: "KEY1=value1 KEY2=value2"
	// - Raw JSON: '{"region": "us-east-1", "config": {"timeout": 30}}'
	// The exact format depends on how the DAG expects to receive parameters.
	Params string
	// DAGName is the name of the executed sub-DAG.
	// For chat tool calls, this is the tool DAG name.
	// This field enables UI drill-down when step.call is not set.
	DAGName string
}

SubDAGRun represents a sub DAG execution within a parent DAG. Each sub DAG run has a deterministic ID based on its parameters to ensure idempotency.

Source Files

View all Source files

Directories

Path Synopsis
action
archive
artifact
chat
Package chat provides an executor for chat (LLM-based session) steps.
 Package chat provides an executor for chat (LLM-based session) steps.
command
controller
Package controller registers the executor identity of the synthesized step that drives a controller DAG.
 Package controller registers the executor identity of the synthesized step that drives a controller DAG.
dag
data
docker
file
foreach
git
harness
http
jq
kubernetes
log
mail
noop
outputs
redis
Package redis provides Redis executor capabilities for Dagu workflows.
 Package redis provides Redis executor capabilities for Dagu workflows.
router
s3
sql
Package sql provides SQL executor capabilities for PostgreSQL and SQLite databases.
 Package sql provides SQL executor capabilities for PostgreSQL and SQLite databases.
sql/drivers/postgres
Package postgres provides the PostgreSQL driver for the SQL executor.
 Package postgres provides the PostgreSQL driver for the SQL executor.
sql/drivers/sqlite
Package sqlite provides the SQLite driver for the SQL executor.
 Package sqlite provides the SQLite driver for the SQL executor.
ssh
state
template
wait
Package controller implements the decision layer of a controller DAG: the catalog of actions offered to the LLM, the goal state it works against, and the planner that turns a conversation into the next action.
 Package controller implements the decision layer of a controller DAG: the catalog of actions offered to the LLM, the goal state it works against, and the planner that turns a conversation into the next action.
Package runstate defines the execution-state port used by the runtime.
 Package runstate defines the execution-state port used by the runtime.
memstore
Package memstore provides an in-memory runtime run-state store.
 Package memstore provides an in-memory runtime run-state store.

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.