Documentation
¶
Index ¶
- Constants
- Variables
- func BuildWebhookRuntimeParams(payload, headers string, extras map[string]string) string
- func BuiltinCLIHarnessProviderNames() []string
- func BuiltinHarnessProviderNames() []string
- func CalculateBackoffInterval(interval time.Duration, backoff float64, maxInterval time.Duration, ...) time.Duration
- func InitializeDefaults(d *DAG)
- func IsBuiltinCLIHarnessProvider(name string) bool
- func IsBuiltinHarnessProvider(name string) bool
- func IsDeniedWebhookForwardHeader(name string) bool
- func IsSynthesizedControllerStep(name string) bool
- func IsValidWebhookHeaderToken(name string) bool
- func NewValidationError(field string, value any, err error) error
- func NormalizeBuiltinHarnessFlagKeys(cfg map[string]any) map[string]any
- func NormalizeWebhookForwardHeader(name string) string
- func ParseCPULimit(value string) (int64, error)
- func ParseDuration(s string) (time.Duration, error)
- func ParseMemoryLimit(value string) (int64, error)
- func RegisterExecutorCapabilities(executorType string, caps ExecutorCapabilities)
- func RegisterExecutorConfigSchema(executorType string, schema *jsonschema.Schema)
- func RegisterStepValidator(executorType string, validator StepValidator)
- func ReportValueReferenceNotices(dag *DAG, sink cmnvalue.ValueReferenceNoticeSink)
- func SockAddr(name, dagRunID string) string
- func SupportsCommand(executorType string) bool
- func SupportsContainer(executorType string) bool
- func SupportsLLM(executorType string) bool
- func SupportsMultipleCommands(executorType string) bool
- func SupportsScript(executorType string) bool
- func SupportsShell(executorType string) bool
- func SupportsSubDAG(executorType string) bool
- func SupportsWorkerSelector(executorType string) bool
- func UnregisterExecutorCapabilities(executorType string)
- func UnregisterStepValidator(executorType string)
- func ValidateController(d *DAG) error
- func ValidateDAGName(name string) error
- func ValidateExecutorConfig(executorType string, config map[string]any) error
- func ValidateLabel(t Label) error
- func ValidateLabels(labels Labels) error
- func ValidateStartArgs(hasDash bool, args []string) error
- func ValidateStartParams(defaultParams string, input StartParamInput) error
- func ValidateSteps(dag *DAG) error
- func ValidateTag(label Label) errordeprecated
- func ValidateTags(labels Labels) errordeprecated
- type ApprovalConfig
- type ArtifactsConfig
- type AuthConfig
- type BastionConfig
- type CommandEntry
- type Condition
- type Container
- type ContainerRuntime
- type ContainerStartup
- type ContainerWaitFor
- type ContinueOn
- type ControllerTask
- type DAG
- func (d *DAG) ArtifactsEnabled() bool
- func (d *DAG) Clone() *DAG
- func (d *DAG) ControllerMaxContextTokens() int
- func (d *DAG) ControllerMaxIterations() int
- func (d *DAG) ControllerObservationKeepRecent() int
- func (d *DAG) ControllerObservationMaxBytes() int
- func (d *DAG) ControllerStep() *Step
- func (d *DAG) FileName() string
- func (d *DAG) GetName() string
- func (d *DAG) HasApprovalSteps() bool
- func (d *DAG) HasHumanTaskSteps() bool
- func (d *DAG) HasLabel(label string) bool
- func (d *DAG) HasTag(tag string) bool
- func (d *DAG) IsController() bool
- func (d *DAG) LoadDotEnv(ctx context.Context)
- func (d *DAG) NextRun(now time.Time) time.Time
- func (d *DAG) ParamDeclarations() cmnvalue.Values
- func (d *DAG) ParamValues() cmnvalue.Values
- func (d *DAG) ParamsMap() map[string]string
- func (d *DAG) ProcGroup() string
- func (d *DAG) SockAddr(dagRunID string) string
- func (d *DAG) SockAddrForSubDAGRun(dagRunID string) string
- func (d *DAG) String() string
- func (d *DAG) SuspendFlagName() string
- func (d *DAG) UnmarshalJSON(data []byte) error
- func (d *DAG) Validate() error
- type DAGRetryPolicy
- type ErrorList
- type ExecutorCapabilities
- type ExecutorConfig
- type ForeachConfig
- type HandlerOn
- type HandlerType
- type HarnessConfig
- type HarnessDefinition
- type HarnessDefinitions
- type HarnessFlagStyle
- type HarnessPromptMode
- type HarnessPromptPosition
- type Healthcheck
- type HumanTaskConfig
- type KubernetesConfig
- type LLMConfig
- type LLMMessage
- type LLMRole
- type Label
- type LabelFilter
- type LabelFilterType
- type Labels
- func (t Labels) Get(key string) []string
- func (t Labels) HasKey(key string) bool
- func (t Labels) Keys() []string
- func (t Labels) MarshalJSON() ([]byte, error)
- func (t Labels) MatchesFilters(filters []LabelFilter) bool
- func (t Labels) Strings() []string
- func (t *Labels) UnmarshalJSON(data []byte) error
- type LogOutputMode
- type MailConfig
- type MailOn
- type ModelEntry
- type NodeStatus
- type OTelConfig
- type OverlapPolicy
- type ParallelConfig
- type ParallelItem
- type ParamDef
- type ParamType
- type Params
- type PullPolicy
- type RedisConfig
- type ReferenceField
- type RepeatMode
- type RepeatPolicy
- type ResourceLimits
- type Resources
- type RetryPolicy
- type RouteEntry
- type RouterConfig
- type RunConfig
- type S3Config
- type SMTPConfig
- type SSHConfig
- type Schedule
- func (s Schedule) DisplayValue() string
- func (s Schedule) Fingerprint() string
- func (s Schedule) GetKind() ScheduleKind
- func (s Schedule) IsCron() bool
- func (s Schedule) IsOneOff() bool
- func (s Schedule) MarshalJSON() ([]byte, error)
- func (s Schedule) Next(now time.Time) time.Time
- func (s Schedule) OneOffTime() (time.Time, bool)
- func (s *Schedule) UnmarshalJSON(data []byte) error
- type ScheduleKind
- type ScheduleParseOptions
- type SecretRef
- type StartParamInput
- type Status
- type Step
- func (s Step) CommandResolution(ctx context.Context) cmnvalue.CommandContext
- func (s Step) HasDeclaredOutputs() bool
- func (s *Step) HasMultipleCommands() bool
- func (s Step) HasOutputSchema() bool
- func (s Step) HasStdoutOutputs() bool
- func (s Step) HasStructuredOutput() bool
- func (s Step) ScriptResolution(ctx context.Context) cmnvalue.CommandContext
- func (s *Step) String() string
- func (s *Step) UnmarshalJSON(data []byte) error
- func (s Step) UsesStructuredOutputSource(source string) bool
- type StepOutputDeclaration
- type StepOutputEntry
- type StepOutputsConfig
- type StepValidator
- type SubDAG
- type Tagdeprecated
- type TagFilterdeprecated
- type TagFilterTypedeprecated
- type Tagsdeprecated
- type ThinkingConfig
- type ThinkingEffort
- type ToolCall
- type ToolConfig
- type ToolPackage
- type ToolRegistry
- type ToolResult
- type TriggerType
- type ValidationError
- type WebSearchConfig
- type WebSearchUserLocation
- type WebhookConfig
Constants ¶
const ( // ControllerStepName is the reserved name of the synthesized step that drives // a controller DAG. It cannot be used as a step name or ID. ControllerStepName = "__controller__" // AskUserStepName is the reserved name and ID of the synthesized human task // a controller opens when it needs to ask a question no declared step // covers. It cannot be used as a step name or ID. // // The name is plain rather than underscore-wrapped because it is what an // operator types to answer: // // dagu human-task complete <dag> --step ask_user --inputs-json '{"answer":"..."}' AskUserStepName = "ask_user" // AskUserAnswerField is the single form field an ask_user question collects. AskUserAnswerField = "answer" // DefaultControllerMaxIterations bounds the number of controller turns when // llm.max_tool_iterations is not set. DefaultControllerMaxIterations = 50 // DefaultControllerMaxStepRuns caps how many times the controller may run a // single declared step within one DAG run. DefaultControllerMaxStepRuns = 5 // DefaultControllerMaxQuestions caps how many questions a controller may put // to a person in one run. Each one suspends the run, so an unbounded // controller could pester someone indefinitely. DefaultControllerMaxQuestions = 5 // DefaultControllerMaxContextTokens is the prompt size at which observation // aging starts. DefaultControllerMaxContextTokens = 200_000 // DefaultControllerObservationMaxBytes bounds each tool result added to a // controller conversation. DefaultControllerObservationMaxBytes = 512 * 1024 // DefaultControllerObservationKeepRecent is the number of tool results that // remain complete after observation aging starts. DefaultControllerObservationKeepRecent = 20 )
const ( // TypeGraph runs dependency-aware steps in parallel where possible. TypeGraph = "graph" // TypeChain runs steps strictly in declaration order. TypeChain = "chain" // TypeController lets an LLM choose which step runs next until every task is complete. TypeController = "controller" )
Supported DAG execution types.
const ( ParamDefTypeString = "string" ParamDefTypeInteger = "integer" ParamDefTypeNumber = "number" ParamDefTypeBoolean = "boolean" )
const ( // MaxLabelKeyLength is the maximum allowed length for label keys (63 chars). MaxLabelKeyLength = 63 // MaxLabelValueLength is the maximum allowed length for label values (255 chars). MaxLabelValueLength = 255 )
Label validation constants.
const ( // LabelKeyPatternStr is the regex pattern for valid label keys. // Allows alphanumeric, dash, underscore, dot. Must start with letter/number. LabelKeyPatternStr = `^[a-zA-Z0-9][a-zA-Z0-9_.-]*$` // LabelValuePatternStr is the regex pattern for valid label values. // Allows alphanumeric, dash, underscore, dot, slash. Must start with letter/number. LabelValuePatternStr = `^[a-zA-Z0-9][a-zA-Z0-9_./-]*$` )
Label validation pattern strings (exported for use by other packages).
const ( // Deprecated: use MaxLabelKeyLength instead. MaxTagKeyLength = MaxLabelKeyLength // Deprecated: use MaxLabelValueLength instead. MaxTagValueLength = MaxLabelValueLength // Deprecated: use LabelKeyPatternStr instead. TagKeyPatternStr = LabelKeyPatternStr // Deprecated: use LabelValuePatternStr instead. TagValuePatternStr = LabelValuePatternStr // Deprecated: use LabelFilterTypeKeyOnly instead. TagFilterTypeKeyOnly = LabelFilterTypeKeyOnly // Deprecated: use LabelFilterTypeExact instead. TagFilterTypeExact = LabelFilterTypeExact // Deprecated: use LabelFilterTypeNegation instead. TagFilterTypeNegation = LabelFilterTypeNegation // Deprecated: use LabelFilterTypeWildcard instead. TagFilterTypeWildcard = LabelFilterTypeWildcard )
const ( StepOutputSourceStdout = "stdout" StepOutputSourceStderr = "stderr" StepOutputSourceFile = "file" StepOutputDecodeText = "text" StepOutputDecodeJSON = "json" StepOutputDecodeYAML = "yaml" StepDeclaredOutputTypeString = "string" StepDeclaredOutputTypeJSON = "json" )
const ( // ExecutorTypeDAG is the executor type for a sub DAG. ExecutorTypeDAG = "dag" // ExecutorTypeDAGEnqueue is the executor type for asynchronously queueing a sub DAG. ExecutorTypeDAGEnqueue = "dag_enqueue" // ExecutorTypeParallel is the executor type for parallel steps. ExecutorTypeParallel = "parallel" // ExecutorTypeForeach is the executor type for foreach steps. ExecutorTypeForeach = "foreach" // ExecutorTypeRouter is the executor type for router steps. ExecutorTypeRouter = "router" // ExecutorTypeController is the executor type for the synthesized step that // drives a controller DAG. ExecutorTypeController = "controller" // ExecutorTypeAction is the executor type for external Dagu actions. ExecutorTypeAction = "action" )
const (
DAGNameMaxLen = 40
)
Constants for validation limits.
const ( // DefaultAquaStandardRegistryRef is the last-resort aqua standard registry // commit, used only when the latest registry release cannot be resolved // and no previously resolved ref is cached on disk. DefaultAquaStandardRegistryRef = "080d723b75cd0ea7c2b2059bf6266d3ab39aa792" )
const DefaultMaxConcurrent = 10
DefaultMaxConcurrent is the default maximum concurrent executions for parallel steps.
const DefaultMaxOutputSize = 1024 * 1024
DefaultMaxOutputSize is the default maximum captured step output size in bytes.
const ExecutorTypeChat = "chat"
ExecutorTypeChat is the executor type for chat steps.
const MaxExpansionConcurrency = 1000
MaxExpansionConcurrency is the maximum concurrency accepted by expansion constructs.
const ParallelItemVariable = "ITEM"
ParallelItemVariable is the special variable name that represents the current item in parallel execution
Variables ¶
var ( ErrNameTooLong = errors.New("name must be less than 40 characters") ErrNameInvalidChars = errors.New("name must only contain alphanumeric characters, dashes, dots, and underscores") ErrInvalidSchedule = errors.New("invalid schedule") ErrScheduleMustBeStringOrArray = errors.New("schedule must be a string or an array of strings") ErrInvalidScheduleType = errors.New("invalid schedule type") ErrInvalidKeyType = errors.New("invalid key type") ErrExecutorConfigMustBeString = errors.New("with/config key must be string") ErrDuplicateFunction = errors.New("duplicate function") ErrFuncParamsMismatch = errors.New("func params and args given to func command do not match") ErrInvalidStepData = errors.New("invalid step data") ErrStepNameRequired = errors.New("step name must be specified") ErrStepNameDuplicate = errors.New("step name must be unique") ErrStepNameTooLong = errors.New("step name must be at most 255 characters") ErrStepIDTooLong = errors.New("step ID must be at most 40 characters") ErrStepCommandIsRequired = errors.New("step command is required") ErrStepCommandIsEmpty = errors.New("step command is empty") ErrStepCommandMustBeArrayOrString = errors.New("step command must be an array of strings or a string") ErrInvalidParamValue = errors.New("invalid parameter value") ErrCallFunctionNotFound = errors.New("call must specify a functions that exists") ErrNumberOfParamsMismatch = errors.New("the number of parameters defined in the function does not match the number of parameters given") ErrRequiredParameterNotFound = errors.New("required parameter not found") ErrScheduleKeyMustBeString = errors.New("schedule key must be a string") ErrInvalidSignal = errors.New("invalid signal") ErrInvalidEnvValue = errors.New("env config should be map of strings or array of key=value formatted string") ErrArgsMustBeConvertibleToIntOrString = errors.New("args must be convertible to either int or string") ErrExecutorTypeMustBeString = errors.New("type value must be string") ErrExecutorConfigValueMustBeMap = errors.New("with/config value must be a map") ErrExecutorHasInvalidKey = errors.New("action config has invalid key") ErrExecutorConfigMustBeStringOrMap = errors.New("action config must be string or map") ErrDotEnvMustBeStringOrArray = errors.New("dotenv must be a string or an array of strings") ErrPreconditionMustBeArrayOrString = errors.New("precondition must be a string or an array of strings") ErrPreconditionValueMustBeString = errors.New("precondition value must be a string") ErrPreconditionHasInvalidKey = errors.New("precondition has invalid key") ErrContinueOnOutputMustBeStringOrArray = errors.New("continueOn.Output must be a string or an array of strings") ErrContinueOnExitCodeMustBeIntOrArray = errors.New("continueOn.ExitCode must be an int or an array of ints") ErrDependsMustBeStringOrArray = errors.New("depends must be a string or an array of strings") ErrDependsNotAllowedInChainType = errors.New("depends field is not allowed for DAGs with type 'chain'; use type 'graph' for explicit dependencies") ErrStepsMustBeArrayOrMap = errors.New("steps must be an array or a map") )
errors on building a DAG.
var ( // ValidLabelKeyPattern validates label keys. ValidLabelKeyPattern = regexp.MustCompile(LabelKeyPatternStr) // ValidLabelValuePattern validates label values. ValidLabelValuePattern = regexp.MustCompile(LabelValuePatternStr) )
Label validation patterns (compiled regex).
var ( // Deprecated: use ValidLabelKeyPattern instead. ValidTagKeyPattern = ValidLabelKeyPattern // Deprecated: use ValidLabelValuePattern instead. ValidTagValuePattern = ValidLabelValuePattern )
Functions ¶
func BuildWebhookRuntimeParams ¶
BuildWebhookRuntimeParams formats webhook payload, headers, and optional trigger-specific metadata into the inline runtime-param syntax Dagu already uses for queued and direct webhook runs.
func BuiltinCLIHarnessProviderNames ¶
func BuiltinCLIHarnessProviderNames() []string
BuiltinCLIHarnessProviderNames returns the built-in CLI harness provider names.
func BuiltinHarnessProviderNames ¶
func BuiltinHarnessProviderNames() []string
BuiltinHarnessProviderNames returns the built-in harness provider names.
func CalculateBackoffInterval ¶
func CalculateBackoffInterval(interval time.Duration, backoff float64, maxInterval time.Duration, attemptCount int) time.Duration
CalculateBackoffInterval returns the delay for the given retry attempt. A non-positive backoff keeps the interval fixed.
func InitializeDefaults ¶
func InitializeDefaults(d *DAG)
InitializeDefaults exposes initializeDefaults for packages that prepare DAGs before execution.
func IsBuiltinCLIHarnessProvider ¶
IsBuiltinCLIHarnessProvider reports whether name selects a built-in CLI harness provider.
func IsBuiltinHarnessProvider ¶
IsBuiltinHarnessProvider reports whether name is a built-in harness provider.
func IsDeniedWebhookForwardHeader ¶
IsDeniedWebhookForwardHeader reports whether a webhook header must never be forwarded into DAG runtime params.
func IsSynthesizedControllerStep ¶
IsSynthesizedControllerStep reports whether a step name belongs to the scaffolding a controller DAG is built with rather than to a declared action.
func IsValidWebhookHeaderToken ¶
IsValidWebhookHeaderToken reports whether name matches the RFC 9110 token grammar used by HTTP header field names.
func NewValidationError ¶
NewValidationError wraps an error with field context so other packages can build LoadError instances.
func NormalizeBuiltinHarnessFlagKeys ¶
NormalizeBuiltinHarnessFlagKeys clones cfg and canonicalizes builtin harness flag aliases to kebab-case so equivalent keys merge predictably.
func NormalizeWebhookForwardHeader ¶
NormalizeWebhookForwardHeader canonicalizes a webhook header name for config validation and runtime matching.
func ParseCPULimit ¶
ParseCPULimit parses Kubernetes-style CPU quantities into millicores.
func ParseDuration ¶
ParseDuration extends time.ParseDuration with support for "d" (days = 24h). Rejects empty strings and zero/negative durations.
func ParseMemoryLimit ¶
ParseMemoryLimit parses Kubernetes-style memory quantities into bytes.
func RegisterExecutorCapabilities ¶
func RegisterExecutorCapabilities(executorType string, caps ExecutorCapabilities)
RegisterExecutorCapabilities registers capabilities for an executor type.
func RegisterExecutorConfigSchema ¶
func RegisterExecutorConfigSchema(executorType string, schema *jsonschema.Schema)
RegisterExecutorConfigSchema registers a JSON schema for an executor config. The schema is used to validate executor config at DAG parse time.
func RegisterStepValidator ¶
func RegisterStepValidator(executorType string, validator StepValidator)
RegisterStepValidator registers a validator for a specific executor type.
func ReportValueReferenceNotices ¶
func ReportValueReferenceNotices(dag *DAG, sink cmnvalue.ValueReferenceNoticeSink)
ReportValueReferenceNotices reports passive notices for value references in dag.
func SockAddr ¶
SockAddr returns the unix socket address for the DAG. The address is used to communicate with the agent process.
func SupportsCommand ¶
SupportsCommand returns whether the executor type supports the command field.
func SupportsContainer ¶
SupportsContainer returns whether the executor type supports step-level container config.
func SupportsLLM ¶
SupportsLLM returns whether the executor type supports the llm field.
func SupportsMultipleCommands ¶
SupportsMultipleCommands returns whether the executor type supports multiple commands.
func SupportsScript ¶
SupportsScript returns whether the executor type supports the script field.
func SupportsShell ¶
SupportsShell returns whether the executor type uses shell configuration.
func SupportsSubDAG ¶
SupportsSubDAG returns whether the executor type can execute sub-DAGs.
func SupportsWorkerSelector ¶
SupportsWorkerSelector returns whether the executor type supports worker selection.
func UnregisterExecutorCapabilities ¶
func UnregisterExecutorCapabilities(executorType string)
UnregisterExecutorCapabilities removes capabilities for an executor type.
func UnregisterStepValidator ¶
func UnregisterStepValidator(executorType string)
UnregisterStepValidator removes a validator for a specific executor type.
func ValidateController ¶
ValidateController checks the DAG-level invariants of a controller DAG: an LLM must be configured, at least one uniquely named task must be declared, and the declared steps must form a tool catalog rather than a dependency graph.
func ValidateDAGName ¶
ValidateDAGName validates a DAG name according to shared rules. Empty name is allowed (caller may provide one via context or filename). Non-empty name must satisfy length and allowed character constraints.
func ValidateExecutorConfig ¶
ValidateExecutorConfig validates config against the registered schema. Returns nil if no schema is registered (backward compatible).
func ValidateLabel ¶
ValidateLabel validates a label's key and value format.
func ValidateLabels ¶
ValidateLabels validates all labels in the collection.
func ValidateStartArgs ¶
func ValidateStartParams ¶
func ValidateStartParams(defaultParams string, input StartParamInput) error
ValidateStartParams validates positional params against declared defaults. Rule: allow 0..expected positional params; reject only when positional params exceed expected.
func ValidateSteps ¶
ValidateSteps validates all steps in a DAG, collecting all validation errors.
func ValidateTag
deprecated
func ValidateTags
deprecated
Types ¶
type ApprovalConfig ¶
type ApprovalConfig struct {
// Prompt is the message displayed to the approver.
Prompt string `json:"prompt,omitempty"`
// Input is the list of expected input field names from the approver.
Input []string `json:"input,omitempty"`
// Required is the subset of Input fields that must be provided.
Required []string `json:"required,omitempty"`
// RewindTo is the step name or ID to restart from on push-back.
// When empty, push-back re-executes the approval step itself.
RewindTo string `json:"rewindTo,omitempty"`
}
ApprovalConfig configures the approval gate for a step. When a step has an ApprovalConfig, it pauses in Waiting state after execution completes, allowing a human to approve, push back (re-run with feedback), or reject.
type ArtifactsConfig ¶
ArtifactsConfig controls DAG run artifact storage.
type AuthConfig ¶
type AuthConfig struct {
// Username for registry authentication
Username string `json:"username,omitempty"`
// Password for registry authentication
Password string `json:"password,omitempty"`
// Auth can be used instead of username/password for pre-encoded credentials
// This should be base64(username:password)
Auth string `json:"auth,omitempty"`
}
AuthConfig represents Docker registry authentication configuration. This is a simplified structure for user convenience that will be converted to Docker's registry.AuthConfig format when needed.
type BastionConfig ¶
type BastionConfig struct {
// Host is the bastion host address.
Host string `json:"host,omitempty"`
// Port is the bastion SSH port. Default is "22".
Port string `json:"port,omitempty"`
// User is the bastion SSH user.
User string `json:"user,omitempty"`
// Key is the path to the SSH private key for the bastion.
Key string `json:"key,omitempty"`
// Password is the SSH password for the bastion.
Password string `json:"password,omitempty"`
}
BastionConfig contains the configuration for a bastion/jump host.
type CommandEntry ¶
type CommandEntry struct {
// Command is the executable name or path.
Command string `json:"command"`
// Args contains the arguments for the command.
Args []string `json:"args,omitempty"`
// CmdWithArgs is the original command string for display purposes.
CmdWithArgs string `json:"cmdWithArgs,omitempty"`
}
CommandEntry represents a single command in a multi-command step. Each entry contains a parsed command with its arguments.
func (CommandEntry) String ¶
func (c CommandEntry) String() string
String returns a display string for the command entry.
type Condition ¶
type Condition struct {
Condition string // Condition to evaluate
Eval string // Dynamic value to evaluate
Expected string // Expected value
Negate bool // Negate the condition result (run when condition does NOT match)
// contains filtered or unexported fields
}
Condition describes a precondition command check or value match.
func (*Condition) GetErrorMessage ¶
func (*Condition) MarshalJSON ¶
func (*Condition) SetErrorMessage ¶
func (*Condition) UnmarshalJSON ¶
type Container ¶
type Container struct {
// Exec specifies an existing container name to exec into.
// Mutually exclusive with Image.
Exec string `yaml:"exec,omitempty"`
// Name is the container name to use. If empty, Docker generates a random name.
Name string `yaml:"name,omitempty"`
// Image is the container image to use.
Image string `yaml:"image,omitempty"`
// PullPolicy is the policy to pull the image (e.g., "Always", "IfNotPresent").
PullPolicy PullPolicy `yaml:"pull_policy,omitempty"`
// Env specifies environment variables for the container.
// Serialized to JSON so it can be restored when ReadDAG deserializes from
// the stored DAGDefinition file.
Env []string `yaml:"env,omitempty" json:"env,omitempty"` // List of environment variables in "key=value" format
// Volumes specifies the volumes to mount in the container.
Volumes []string `yaml:"volumes,omitempty"` // Map of volume names to volume definitions
// User is the user to run the container as.
User string `yaml:"user,omitempty"` // User to run the container as
// WorkingDir is the working directory inside the container.
WorkingDir string `yaml:"working_dir,omitempty"` // Working directory inside the container
// Platform specifies the platform for the container (e.g., "linux/amd64").
Platform string `yaml:"platform,omitempty"` // Platform for the container
// Ports specifies the ports to expose from the container.
Ports []string `yaml:"ports,omitempty"` // List of ports to expose
// Network is the network configuration for the container.
Network string `yaml:"network,omitempty"` // Network configuration for the container
// KeepContainer is the flag to keep the container after the DAG run.
KeepContainer bool `yaml:"keep_container,omitempty"` // Keep the container after the DAG run
// Startup determines how the DAG-level container starts up.
// One of: "keepalive" (default), "entrypoint", "command".
Startup ContainerStartup `yaml:"startup,omitempty"`
// Command is used when Startup == "command".
Command []string `yaml:"command,omitempty"`
// WaitFor determines readiness gate before steps run: "running" (default) or "healthy".
WaitFor ContainerWaitFor `yaml:"wait_for,omitempty"`
// LogPattern optionally waits for a regex to appear in container logs before proceeding.
LogPattern string `yaml:"log_pattern,omitempty"`
// RestartPolicy applies Docker restart policy for long-running containers ("no", "always", or "unless-stopped").
RestartPolicy string `yaml:"restart_policy,omitempty"`
// Healthcheck specifies a custom healthcheck for the container.
// If specified with waitFor: healthy, this healthcheck is used instead of
// relying on the image's built-in healthcheck.
Healthcheck *Healthcheck `yaml:"healthcheck,omitempty"`
// Shell specifies the shell wrapper for executing step commands.
// When specified, all step commands are wrapped with this shell.
// Format: ["/bin/bash", "-o", "errexit", "-o", "xtrace", "-c"]
// The step command will be appended as the final argument.
// Works in both exec mode and image mode.
Shell []string `yaml:"shell,omitempty"`
}
Container defines the container configuration for the DAG.
func (Container) GetWorkingDir ¶
GetWorkingDir returns the working directory inside the container
func (Container) IsExecMode ¶
IsExecMode returns true if this container is configured to exec into an existing container
type ContainerRuntime ¶
type ContainerRuntime string
ContainerRuntime is the runtime used to launch or exec into a container.
const ( ContainerRuntimeDocker ContainerRuntime = "docker" ContainerRuntimePodman ContainerRuntime = "podman" )
func ParseContainerRuntime ¶
func ParseContainerRuntime(raw string) (ContainerRuntime, error)
ParseContainerRuntime parses a container runtime from a raw string.
type ContainerStartup ¶
type ContainerStartup string
ContainerStartup is an enum for DAG-level container startup modes.
const ( StartupKeepalive ContainerStartup = "keepalive" StartupEntrypoint ContainerStartup = "entrypoint" StartupCommand ContainerStartup = "command" )
type ContainerWaitFor ¶
type ContainerWaitFor string
ContainerWaitFor is an enum for container readiness conditions.
const ( WaitForRunning ContainerWaitFor = "running" WaitForHealthy ContainerWaitFor = "healthy" )
type ContinueOn ¶
type ContinueOn struct {
Failure bool `json:"failure,omitempty"` // Failure is the flag to continue to the next step on failure.
Skipped bool `json:"skipped,omitempty"` // Skipped is the flag to continue to the next step on skipped.
ExitCode []int `json:"exitCode,omitempty"` // ExitCode is the list of exit codes to continue to the next step.
Output []string `json:"output,omitempty"` // Output is the list of output (stdout/stderr) to continue to the next step.
MarkSuccess bool `json:"markSuccess,omitempty"` // MarkSuccess is the flag to mark the step as success when the condition is met.
}
ContinueOn contains the conditions to continue on failure or skipped. Failure is the flag to continue to the next step on failure. Skipped is the flag to continue to the next step on skipped. A step can be skipped when the preconditions are not met. Then if the ContinueOn.Skip is set, the step will continue to the next step.
type ControllerTask ¶
type ControllerTask struct {
// Name identifies the task. It is unique within the DAG.
Name string `json:"name"`
// Description states the completion criteria in natural language.
Description string `json:"description,omitempty"`
}
ControllerTask is a goal the controller must satisfy. A controller DAG run concludes successfully once every task has been marked complete.
type DAG ¶
type DAG struct {
// WorkingDir is the working directory to run the DAG.
// Default value is the directory of DAG file.
// Relative paths are resolved at build time; variables are expanded at runtime.
// Supports environment variable templates (e.g., ${MY_DIR}).
WorkingDir string `json:"workingDir,omitempty"`
// WorkingDirExplicit is true when WorkingDir was explicitly set in YAML,
// base config, or via DefaultWorkingDir option. When false, WorkingDir
// was auto-defaulted by the loader to the DAG file's parent directory.
// Not serialized — runtime-only flag.
WorkingDirExplicit bool `json:"-"`
// Location is the absolute path to the DAG file.
// It is used to generate unix socket name and can be blank
Location string `json:"location,omitempty"`
// SourceFile is the original DAG file path this run was loaded from.
// Unlike Location, it is provenance-only and is preserved even when queued
// execution clears or rewrites runtime locations.
SourceFile string `json:"sourceFile,omitempty"`
// Group is the group name of the DAG. This is optional.
Group string `json:"group,omitempty"`
// Name is the name of the DAG. The default is the filename without the extension.
Name string `json:"name,omitempty"`
// Type is the execution type (graph, chain, or controller). Default is graph.
Type string `json:"type,omitempty"`
// Tasks are the goals a controller DAG must satisfy before it concludes.
// Only meaningful when Type is TypeController.
Tasks []ControllerTask `json:"tasks,omitempty"`
// Shell is the default shell to use for all steps in this DAG.
// If not specified, the system default shell is used.
// Can be overridden at the step level.
// Supports environment variable templates (e.g., ${MY_SHELL}).
Shell string `json:"shell,omitempty"`
// ShellArgs contains additional arguments to pass to the shell.
// These are populated when Shell is specified as a string with arguments
// (e.g., "bash -e") or as an array (e.g., ["bash", "-e"]).
// Supports environment variable templates.
ShellArgs []string `json:"shellArgs,omitempty"`
// Dotenv is the path to the dotenv file. This is optional.
Dotenv []string `json:"dotenv,omitempty"`
// Labels contains the list of labels for the DAG. This is optional.
Labels Labels `json:"labels,omitempty"`
// Description is the description of the DAG. This is optional.
Description string `json:"description,omitempty"`
// Schedule configuration for starting, stopping, and restarting the DAG.
Schedule []Schedule `json:"schedule,omitempty"`
// StopSchedule contains the cron expressions for stopping the DAG.
StopSchedule []Schedule `json:"stopSchedule,omitempty"`
// RestartSchedule contains the cron expressions for restarting the DAG.
RestartSchedule []Schedule `json:"restartSchedule,omitempty"`
// SkipIfSuccessful indicates whether to skip the DAG if it was successful previously.
// E.g., when the DAG has already been executed manually before the scheduled time.
SkipIfSuccessful bool `json:"skipIfSuccessful,omitempty"`
// CatchupWindow is the lookback horizon for missed cron intervals.
// If set, enables catch-up on scheduler restart. If omitted, no catch-up.
CatchupWindow time.Duration `json:"catchupWindow,omitempty"`
// OverlapPolicy controls behavior when a new run is triggered while one is active.
// Defaults to "skip". See OverlapPolicy constants for options.
OverlapPolicy OverlapPolicy `json:"overlapPolicy,omitempty"`
// Env contains a list of environment variables to be set before running the DAG.
// Note: This field is evaluated at build time and may contain secrets.
// It is excluded from JSON serialization to prevent secret leakage.
Env []string `json:"-"`
// Consts contains immutable values resolved while loading the DAG.
Consts map[string]any `json:"consts,omitempty"`
// EnvEvaluated reports whether Env is safe to reuse as resolved build env.
EnvEvaluated bool `json:"-"`
// PresolvedBuildEnv stores resolved DAG/base-config env entries needed to
// rebuild the DAG from persisted YAML during retry/restart paths.
// It is serialized with dag.json because direct retry/restart cannot rely on
// parent-process transport once the original process has exited.
PresolvedBuildEnv map[string]string `json:"presolvedBuildEnv,omitempty"`
// LogDir is the directory where the logs are stored.
LogDir string `json:"logDir,omitempty"`
// Artifacts config controls optional DAG run artifact storage.
Artifacts *ArtifactsConfig `json:"artifacts,omitempty"`
// LogOutput specifies how stdout and stderr are handled in log files.
// Can be "separate" (default) for separate .out and .err files,
// or "merged" for a single combined .log file.
LogOutput LogOutputMode `json:"logOutput,omitempty"`
// DefaultParams contains the default parameters to be passed to the DAG.
DefaultParams string `json:"defaultParams,omitempty"`
// ParamDefs contains ordered parameter metadata derived from DAG params.
// It is exposed to the API for typed UI rendering and validation hints.
ParamDefs []ParamDef `json:"paramDefs,omitempty"`
// ParamSchema contains the resolved JSON Schema for schema-backed DAG params
// when that schema is safe for direct UI form rendering.
ParamSchema json.RawMessage `json:"paramSchema,omitempty"`
// Params contains the list of parameters to be passed to the DAG.
// Note: This field is evaluated at build time and may contain secrets.
// It is excluded from JSON serialization to prevent secret leakage.
Params []string `json:"-"`
// ParamsJSON contains the JSON representation of the resolved parameters.
// When params were supplied as JSON, the original payload is preserved.
// Steps can consume this via the DAG_PARAMS_JSON environment variable.
// Note: This field is evaluated at build time and may contain secrets.
// It is excluded from JSON serialization to prevent secret leakage.
ParamsJSON string `json:"-"`
// Steps contains the list of steps in the DAG.
Steps []Step `json:"steps,omitempty"`
// HandlerOn contains the steps to be executed on different events.
HandlerOn HandlerOn `json:"handlerOn,omitzero"`
// Preconditions contains the conditions to be met before running the DAG.
Preconditions []*Condition `json:"preconditions,omitempty"`
// SMTP contains the SMTP configuration.
// Excluded from JSON: may contain password.
SMTP *SMTPConfig `json:"-"`
// ErrorMail contains the mail configuration for errors.
ErrorMail *MailConfig `json:"errorMail,omitempty"`
// InfoMail contains the mail configuration for informational messages.
InfoMail *MailConfig `json:"infoMail,omitempty"`
// WaitMail contains the mail configuration for wait status.
WaitMail *MailConfig `json:"waitMail,omitempty"`
// MailOn contains the conditions to send mail.
MailOn *MailOn `json:"mailOn,omitempty"`
// Timeout specifies the maximum execution time of the DAG task.
Timeout time.Duration `json:"timeout,omitempty"`
// Delay is the delay before starting the DAG.
Delay time.Duration `json:"delay,omitempty"`
// RestartWait is the time to wait before restarting the DAG.
RestartWait time.Duration `json:"restartWait,omitempty"`
// MaxActiveSteps specifies the maximum concurrent steps to run in an execution.
MaxActiveSteps int `json:"maxActiveSteps,omitempty"`
// MaxActiveRuns specifies the maximum number of concurrent dag-runs.
// DEPRECATED: This field is ignored for local (DAG-based) queues.
// For concurrency control, define a global queue in config and use the 'queue' field.
MaxActiveRuns int `json:"maxActiveRuns,omitempty"`
// MaxCleanUpTime is the maximum time to wait for cleanup when the DAG is stopped.
MaxCleanUpTime time.Duration `json:"maxCleanUpTime,omitempty"`
// HistRetentionDays is the number of days to keep the history of dag-runs.
HistRetentionDays int `json:"histRetentionDays,omitempty"`
// HistRetentionRuns is the number of dag-runs to keep in history.
HistRetentionRuns int `json:"histRetentionRuns,omitempty"`
// Queue is the name of the queue to assign this DAG to.
Queue string `json:"queue,omitempty"`
// RetryPolicy controls automatic DAG-level retry behavior for failed runs.
RetryPolicy *DAGRetryPolicy `json:"retryPolicy,omitempty"`
// WorkerSelector defines labels required for worker selection in distributed execution.
// If specified, the DAG will only run on workers with matching labels.
WorkerSelector map[string]string `json:"workerSelector,omitempty"`
// ForceLocal forces the DAG to run locally even when the server default is distributed.
// Set by worker_selector: local in the DAG spec.
ForceLocal bool `json:"forceLocal,omitempty"`
// MaxOutputSize is the maximum size of step output to capture in bytes.
// Default is 1MB. Output exceeding this will return an error.
MaxOutputSize int `json:"maxOutputSize,omitempty"`
// OTel contains the OpenTelemetry configuration for the DAG.
OTel *OTelConfig `json:"otel,omitempty"`
// BuildErrors contains any errors encountered while building the DAG.
BuildErrors []error `json:"-"`
// BuildWarnings contains non-fatal warnings detected while building the DAG.
BuildWarnings []string `json:"-"`
// LocalDAGs contains DAGs defined in the same file, keyed by DAG name
LocalDAGs map[string]*DAG `json:"localDAGs,omitempty"`
// YamlData contains the raw YAML data of the DAG.
YamlData []byte `json:"yamlData,omitempty"`
// BaseConfigData contains the raw base config YAML content.
// This is used to propagate base config through distributed execution
// and sub-DAG chains, so workers don't need local base config files.
BaseConfigData []byte `json:"baseConfigData,omitempty"`
// Container contains the container definition for the DAG.
Container *Container `json:"container,omitempty"`
// RunConfig contains configuration for controlling user interactions during DAG runs.
RunConfig *RunConfig `json:"runConfig,omitempty"`
// Resources contains CPU and memory limits requested for this DAG run.
Resources *Resources `json:"resources,omitempty"`
// Webhook contains DAG-level webhook trigger behavior configuration.
Webhook *WebhookConfig `json:"webhook,omitempty"`
// RegistryAuths maps registry hostnames to authentication configs.
// Optional: If not specified, falls back to DOCKER_AUTH_CONFIG or docker config.
// Credentials are evaluated at runtime. Excluded from JSON: may contain passwords.
RegistryAuths map[string]*AuthConfig `json:"-"`
// SSH contains the default SSH configuration for the DAG.
// Excluded from JSON: may contain password.
SSH *SSHConfig `json:"-"`
// S3 contains the default S3 configuration for the DAG.
// Excluded from JSON: may contain credentials.
S3 *S3Config `json:"-"`
// LLM contains the default LLM configuration for the DAG.
// Steps with type: chat inherit this configuration if they don't specify their own llm field.
LLM *LLMConfig `json:"llm,omitempty"`
// Redis contains the default Redis configuration for the DAG.
// Steps with type: redis inherit this configuration.
// Excluded from JSON: may contain password.
Redis *RedisConfig `json:"-"`
// Harness contains the default harness executor configuration for the DAG.
// Steps with type: harness inherit this configuration.
// Excluded from JSON: may contain API key references or provider-specific secrets.
Harness *HarnessConfig `json:"-"`
// Harnesses contains reusable custom harness definitions available to harness steps.
// Excluded from JSON: derived from DAG/base config and rebuilt from YAML when needed.
Harnesses HarnessDefinitions `json:"-"`
// Kubernetes contains the default Kubernetes executor configuration for the DAG.
// Steps with type: k8s or type: kubernetes inherit this configuration.
// Excluded from JSON: may contain secret references.
Kubernetes KubernetesConfig `json:"-"`
// Secrets contains references to external secrets to be resolved at runtime.
Secrets []SecretRef `json:"secrets,omitempty"`
// Tools declares external CLI tools that must be installed before the DAG runs.
Tools *ToolConfig `json:"tools,omitempty"`
// contains filtered or unexported fields
}
DAG contains all information about a DAG.
func (*DAG) ArtifactsEnabled ¶
ArtifactsEnabled reports whether the DAG has artifact storage enabled.
func (*DAG) Clone ¶
Clone creates a shallow copy of the DAG. The sync.Once field is reset to zero value, allowing LoadDotEnv to be called independently on the clone. This is safe for read-only field modifications like changing Location.
func (*DAG) ControllerMaxContextTokens ¶ added in v2.11.3
ControllerMaxContextTokens returns the prompt size at which observation aging starts. Zero disables proactive aging.
func (*DAG) ControllerMaxIterations ¶
ControllerMaxIterations returns the upper bound on controller turns for a single run.
func (*DAG) ControllerObservationKeepRecent ¶ added in v2.11.3
ControllerObservationKeepRecent returns how many recent observations remain complete after aging starts. Zero disables observation aging.
func (*DAG) ControllerObservationMaxBytes ¶ added in v2.11.3
ControllerObservationMaxBytes returns the maximum controller-facing size of one observation. Zero disables the size limit.
func (*DAG) ControllerStep ¶
ControllerStep returns the synthesized controller step, or nil when the DAG is not a controller DAG.
func (*DAG) GetName ¶
GetName returns the name of the DAG. If the name is not set, it returns the default name (filename without extension).
func (*DAG) HasApprovalSteps ¶
HasApprovalSteps returns true if the DAG contains any steps that require human approval. DAGs with approval steps cannot be dispatched to workers because approval steps require local storage access.
func (*DAG) HasHumanTaskSteps ¶
HasHumanTaskSteps reports whether the DAG declares a human task. A controller DAG's synthesized ask_user task does not count: it is scaffolding every controller carries, and the controller declines to use it outside a root run, so counting it here would bar controllers from being composed as child DAGs.
func (*DAG) HasLabel ¶
HasLabel checks if the DAG has a label matching the given filter. Supports both simple labels ("production") and key-value filters ("env=prod").
func (*DAG) HasTag ¶
HasTag checks if the DAG has a tag matching the given filter. Deprecated: use HasLabel.
func (*DAG) IsController ¶
IsController reports whether the DAG is driven by an LLM controller instead of a static dependency graph.
func (*DAG) LoadDotEnv ¶
LoadDotEnv loads all dotenv files in order, with later files overriding earlier ones. This method is thread-safe and idempotent - concurrent calls will only load once.
func (*DAG) ParamDeclarations ¶
ParamDeclarations returns named parameters that can be referenced through ${params.name}.
func (*DAG) ParamValues ¶
ParamValues returns named runtime parameter values for ${params.name}.
func (*DAG) ProcGroup ¶
ProcGroup returns the name of the process group for this DAG. The process group name is used to identify and manage related DAG executions.
Returns:
- If Queue is set: returns the Queue value
- If Queue is empty: returns the DAG name as the default
The process group name is used by the process store to:
- Manage heartbeat files for active DAG runs
- Enforce concurrency limits (max concurrent runs) across DAGs in the same group
This allows the scheduler to control how many DAGs can run simultaneously within the same process group.
func (*DAG) SockAddr ¶
SockAddr returns the unix socket address for the DAG. The address is used to communicate with the agent process.
func (*DAG) SockAddrForSubDAGRun ¶
SockAddrForSubDAGRun returns the unix socket address for a specific dag-run ID. This is used to control sub dag-runs.
func (*DAG) SuspendFlagName ¶
SuspendFlagName returns the filename stem used by the file-based suspend flag system. This intentionally follows DAG file naming, not dag.Name.
func (*DAG) UnmarshalJSON ¶
UnmarshalJSON deserializes DAGs written by both the canonical labels field and the deprecated tags field used by older persisted dag.json files.
type DAGRetryPolicy ¶
type DAGRetryPolicy struct {
// Limit is the maximum number of retry attempts allowed.
Limit int `json:"limit,omitempty"`
// Interval is the base delay before retrying.
Interval time.Duration `json:"interval,omitempty"`
// IntervalSecStr preserves the original interval string representation.
IntervalSecStr string `json:"intervalSecStr,omitempty"`
// Backoff is the retry delay multiplier. Zero keeps a fixed interval.
Backoff float64 `json:"backoff,omitempty"`
// MaxInterval caps the computed retry delay.
MaxInterval time.Duration `json:"maxInterval,omitempty"`
}
DAGRetryPolicy contains the retry policy for a DAG run.
type ErrorList ¶
type ErrorList []error
ErrorList is just a list of errors. It is used to collect multiple errors in building a DAG.
func (ErrorList) Dedupe ¶ added in v2.11.4
Dedupe returns the list with repeated occurrences of the same error value removed, keeping first positions. Distinct errors with equal messages are preserved; only identical instances collapse. Values of non-comparable dynamic types are kept as-is.
func (ErrorList) Error ¶
Error implements the error interface. It returns a string with all the errors separated by a semicolon.
func (*ErrorList) ToStringList ¶
ToStringList returns the list of errors as a slice of strings.
type ExecutorCapabilities ¶
type ExecutorCapabilities struct {
// Command indicates whether the executor supports the command field.
Command bool
// MultipleCommands indicates whether the executor supports multiple commands.
MultipleCommands bool
// Script indicates whether the executor supports the script field.
Script bool
// Shell indicates whether the executor uses shell/shellArgs/shellPackages.
Shell bool
// Container indicates whether the executor supports step-level container config.
Container bool
// SubDAG indicates whether the executor can execute sub-DAGs.
SubDAG bool
// WorkerSelector indicates whether the executor supports worker selection.
WorkerSelector bool
// LLM indicates whether the executor supports the llm field.
LLM bool
// CommandContext returns command execution facts for command field resolution.
CommandContext func(ctx context.Context, step Step) cmnvalue.CommandContext
// ScriptContext returns command execution facts for script field resolution.
ScriptContext func(ctx context.Context, step Step) cmnvalue.CommandContext
}
ExecutorCapabilities defines what an executor can do.
type ExecutorConfig ¶
type ExecutorConfig struct {
// Type represents one of the registered executors.
// See `executor.Register` in `internal/executor/executor.go`.
Type string `json:"type,omitempty"`
Config map[string]any `json:"config,omitempty"` // Config contains executor-specific configuration.
// Metadata contains additional metadata for the executor that is not passed to the executor itself.
// This is used internally for optimization purposes.
Metadata map[string]any `json:"metadata,omitempty"`
}
ExecutorConfig contains the configuration for the executor.
func (ExecutorConfig) IsCommand ¶
func (e ExecutorConfig) IsCommand() bool
IsCommand returns true if the executor is a command
type ForeachConfig ¶
type ForeachConfig struct {
// Items is the static list of items to iterate.
Items []any `json:"items,omitempty"`
// ItemsExpr is a value-resolved expression that must produce a JSON array.
ItemsExpr string `json:"itemsExpr,omitempty"`
// As is the item alias exposed under the foreach namespace.
As string `json:"as,omitempty"`
// Key is an optional item key expression.
Key string `json:"key,omitempty"`
// MaxConcurrent is the maximum number of item bodies running at once.
MaxConcurrent int `json:"maxConcurrent,omitempty"`
// Steps is the item body graph.
Steps []Step `json:"steps,omitempty"`
// Collect maps output names to value-resolved expressions.
Collect map[string]string `json:"collect,omitempty"`
}
ForeachConfig contains the configuration for inline item-body iteration.
type HandlerOn ¶
type HandlerOn struct {
Init *Step `json:"init,omitempty"`
Failure *Step `json:"failure,omitempty"`
Success *Step `json:"success,omitempty"`
Abort *Step `json:"abort,omitempty"`
Exit *Step `json:"exit,omitempty"`
Wait *Step `json:"wait,omitempty"`
}
HandlerOn contains the steps to be executed on different events in the DAG.
type HandlerType ¶
type HandlerType string
HandlerType is the type of the handler.
const ( HandlerOnInit HandlerType = "onInit" HandlerOnSuccess HandlerType = "onSuccess" HandlerOnFailure HandlerType = "onFailure" HandlerOnAbort HandlerType = "onAbort" HandlerOnExit HandlerType = "onExit" HandlerOnWait HandlerType = "onWait" )
func (HandlerType) String ¶
func (h HandlerType) String() string
type HarnessConfig ¶
type HarnessConfig struct {
// Config contains the primary provider selection and CLI flags.
Config map[string]any `json:"-"`
// Fallback contains ordered alternative provider configs tried on failure.
Fallback []map[string]any `json:"-"`
}
HarnessConfig contains the default harness executor configuration for the DAG. Steps with type: harness inherit Config as their primary attempt and Fallback as ordered alternative provider configs.
type HarnessDefinition ¶
type HarnessDefinition struct {
Binary string `json:"binary,omitempty"`
PrefixArgs []string `json:"prefixArgs,omitempty"`
PromptMode HarnessPromptMode `json:"promptMode,omitempty"`
PromptFlag string `json:"promptFlag,omitempty"`
PromptPosition HarnessPromptPosition `json:"promptPosition,omitempty"`
FlagStyle HarnessFlagStyle `json:"flagStyle,omitempty"`
OptionFlags map[string]string `json:"optionFlags,omitempty"`
}
HarnessDefinition describes how to invoke a named harness CLI.
type HarnessDefinitions ¶
type HarnessDefinitions map[string]*HarnessDefinition
HarnessDefinitions contains named reusable harness definitions. Nil values are used internally during base-config merge to delete inherited entries.
type HarnessFlagStyle ¶
type HarnessFlagStyle string
const ( HarnessFlagStyleGNULong HarnessFlagStyle = "gnu_long" HarnessFlagStyleSingleDash HarnessFlagStyle = "single_dash" )
type HarnessPromptMode ¶
type HarnessPromptMode string
const ( HarnessPromptModeArg HarnessPromptMode = "arg" HarnessPromptModeFlag HarnessPromptMode = "flag" HarnessPromptModeStdin HarnessPromptMode = "stdin" )
type HarnessPromptPosition ¶
type HarnessPromptPosition string
const ( HarnessPromptPositionBeforeFlags HarnessPromptPosition = "before_flags" HarnessPromptPositionAfterFlags HarnessPromptPosition = "after_flags" )
type Healthcheck ¶
type Healthcheck struct {
// Test is the command to run to check health. Must start with:
// - ["NONE"] - disable healthcheck
// - ["CMD", "command", "arg1", ...] - run command directly
// - ["CMD-SHELL", "command"] - run command in shell
Test []string `yaml:"test,omitempty"`
// Interval is the time between health checks (e.g., "5s", "1m").
Interval time.Duration `yaml:"interval,omitempty"`
// Timeout is how long to wait for the health check to complete (e.g., "3s").
Timeout time.Duration `yaml:"timeout,omitempty"`
// StartPeriod is the grace period for the container to initialize (e.g., "10s").
StartPeriod time.Duration `yaml:"start_period,omitempty"`
// Retries is the number of consecutive failures needed to consider unhealthy.
Retries int `yaml:"retries,omitempty"`
}
Healthcheck defines a custom health check for a container. This allows waitFor: healthy to work with images that don't have built-in healthchecks.
type HumanTaskConfig ¶
type HumanTaskConfig struct {
Prompt string `json:"prompt,omitempty"`
Form json.RawMessage `json:"form,omitempty"`
}
HumanTaskConfig defines the prompt and input form for a human task step.
type KubernetesConfig ¶
KubernetesConfig contains default Kubernetes executor configuration for the DAG. It stores the raw executor config map so step-level overrides can be merged using executor-specific semantics during DAG build.
type LLMConfig ¶
type LLMConfig struct {
// Provider is the LLM provider (openai, anthropic, gemini, openrouter, local).
// Used for single model config (backward compatible).
Provider string `json:"provider,omitempty"`
// Model is the model to use (e.g., gpt-4o, claude-sonnet-4-6).
// Used for single model config (backward compatible).
Model string `json:"model,omitempty"`
// Models is an array of models for fallback support.
// First model is primary, rest are tried in order if primary fails.
// When set, Provider/Model fields are ignored.
Models []ModelEntry `json:"models,omitempty"`
// System is the default system prompt for sessions.
System string `json:"system,omitempty"`
// Temperature controls randomness (0.0-2.0).
Temperature *float64 `json:"temperature,omitempty"`
// MaxTokens is the maximum number of tokens to generate.
MaxTokens *int `json:"maxTokens,omitempty"`
// TopP is the nucleus sampling parameter.
TopP *float64 `json:"topP,omitempty"`
// BaseURL is a custom API endpoint.
BaseURL string `json:"baseURL,omitempty"`
// APIKeyName is the name of the environment variable containing the API key.
// If not specified, the default environment variable for the provider is used.
APIKeyName string `json:"apiKeyName,omitempty"`
// Stream enables or disables streaming output.
// Default is true.
Stream *bool `json:"stream,omitempty"`
// Thinking enables extended thinking/reasoning mode.
// Provider-specific: Anthropic uses budget_tokens, OpenAI uses reasoning.effort,
// Gemini uses thinkingLevel/thinkingBudget, OpenRouter normalizes across providers.
Thinking *ThinkingConfig `json:"thinking,omitempty"`
// Tools is a list of DAG names to use as callable tools.
// Tool names, descriptions, and parameters are auto-discovered from DAG definitions.
// Example: ["search-tool", "analyzer-tool"]
Tools []string `json:"tools,omitempty"`
// MaxToolIterations limits the number of tool calling rounds.
// Default is 10 if not specified.
MaxToolIterations *int `json:"maxToolIterations,omitempty"`
// MaxContextTokens activates controller observation aging once a decision
// prompt reaches this many tokens. Zero disables proactive aging.
MaxContextTokens *int `json:"maxContextTokens,omitempty"`
// ObservationMaxBytes limits the size of each controller observation. Zero
// disables this limit.
ObservationMaxBytes *int `json:"observationMaxBytes,omitempty"`
// ObservationKeepRecent is the number of recent controller observations kept
// in full after observation aging starts. Zero disables observation aging.
ObservationKeepRecent *int `json:"observationKeepRecent,omitempty"`
// WebSearch configures provider-native web search.
WebSearch *WebSearchConfig `json:"webSearch,omitempty"`
}
LLMConfig contains the configuration for LLM-based executors.
func (*LLMConfig) GetMaxToolIterations ¶
GetMaxToolIterations returns the maximum number of tool calling iterations. Default is 10 if not specified.
func (*LLMConfig) GetModels ¶
func (c *LLMConfig) GetModels() []ModelEntry
GetModels returns the ordered list of models to try. If Models array is set, returns it. Otherwise, creates single-entry list from Provider/Model.
func (*LLMConfig) HasFallback ¶
HasFallback returns true if there are multiple models configured.
func (*LLMConfig) StreamEnabled ¶
StreamEnabled returns true if streaming is enabled. Default is true if Stream is nil.
type LLMMessage ¶
type LLMMessage struct {
// Role is the message role (system, user, assistant, tool).
Role LLMRole `json:"role,omitempty"`
// Content is the message content. Supports variable substitution with ${VAR}.
Content string `json:"content,omitempty"`
}
LLMMessage represents a message in the LLM session.
type LLMRole ¶
type LLMRole string
LLMRole represents the role of a message sender in a session.
const ( LLMRoleSystem LLMRole = "system" LLMRoleUser LLMRole = "user" LLMRoleAssistant LLMRole = "assistant" LLMRoleTool LLMRole = "tool" )
LLM message role constants.
func ParseLLMRole ¶
ParseLLMRole validates and returns an LLMRole from a string. Returns error for invalid or empty role values.
type Label ¶
Label represents a key-value label with optional value. For backward compatibility, key-only labels have an empty Value.
func ParseLabel ¶
ParseLabel parses a string into a Label. Supports "key=value" and "key" (key-only) formats. Both key and value are normalized to lowercase.
type LabelFilter ¶
type LabelFilter struct {
Type LabelFilterType
Key string
Value string
}
LabelFilter represents a parsed filter condition.
func ParseLabelFilter ¶
func ParseLabelFilter(s string) LabelFilter
ParseLabelFilter parses a filter string into LabelFilter. Formats:
- "key" - matches any label with that key (KeyOnly)
- "key=value" - matches exact key=value (Exact)
- "!key" - matches if key does NOT exist (Negation)
- "key*" or "key=value*" - matches using glob patterns (Wildcard)
func ParseTagFilter
deprecated
func ParseTagFilter(s string) LabelFilter
Deprecated: use ParseLabelFilter instead.
func (LabelFilter) MatchesLabels ¶
func (f LabelFilter) MatchesLabels(labels Labels) bool
MatchesLabels checks if a label collection matches this filter.
func (LabelFilter) MatchesTags
deprecated
func (f LabelFilter) MatchesTags(labels Labels) bool
Deprecated: use MatchesLabels instead.
type LabelFilterType ¶
type LabelFilterType int
LabelFilterType represents the type of label filter.
const ( // LabelFilterTypeKeyOnly matches any label with the specified key (regardless of value). LabelFilterTypeKeyOnly LabelFilterType = iota // LabelFilterTypeExact matches labels with exact key=value. LabelFilterTypeExact // LabelFilterTypeNegation matches if the key does NOT exist. LabelFilterTypeNegation // LabelFilterTypeWildcard matches labels using glob patterns (* and ?). LabelFilterTypeWildcard )
type Labels ¶
type Labels []Label
Labels represents a collection of labels.
func (Labels) MarshalJSON ¶
MarshalJSON serializes Labels as an array of strings for backward compatibility.
func (Labels) MatchesFilters ¶
func (t Labels) MatchesFilters(filters []LabelFilter) bool
MatchesFilters checks if labels match all filters (AND logic).
func (*Labels) UnmarshalJSON ¶
UnmarshalJSON deserializes Labels from an array of strings.
type LogOutputMode ¶
type LogOutputMode string
LogOutputMode represents the mode for log output handling. It determines how stdout and stderr are written to log files.
const ( // LogOutputSeparate keeps stdout and stderr in separate files (.out and .err). // This is the default behavior for backward compatibility. LogOutputSeparate LogOutputMode = "separate" // LogOutputMerged combines stdout and stderr into a single log file (.log). // Both streams are interleaved in the order they are written. LogOutputMerged LogOutputMode = "merged" )
func EffectiveLogOutput ¶
func EffectiveLogOutput(dag *DAG, step *Step) LogOutputMode
EffectiveLogOutput returns the effective log output mode for a step. Priority: step-level > DAG-level > default (LogOutputSeparate).
type MailConfig ¶
type MailConfig struct {
From string `json:"from,omitempty"`
To []string `json:"to,omitempty"`
Prefix string `json:"prefix,omitempty"`
AttachLogs bool `json:"attachLogs,omitempty"`
}
MailConfig contains the mail configuration.
type MailOn ¶
type MailOn struct {
Failure bool `json:"failure,omitempty"`
Success bool `json:"success,omitempty"`
Wait bool `json:"wait,omitempty"`
}
MailOn contains the conditions to send mail.
type ModelEntry ¶
type ModelEntry struct {
// Provider is the LLM provider for this model.
Provider string `json:"provider"`
// Name is the model name (e.g., gpt-4o, claude-sonnet-4-6).
Name string `json:"name"`
// Temperature overrides the shared temperature for this model.
Temperature *float64 `json:"temperature,omitempty"`
// MaxTokens overrides the shared maxTokens for this model.
MaxTokens *int `json:"maxTokens,omitempty"`
// TopP overrides the shared topP for this model.
TopP *float64 `json:"topP,omitempty"`
// BaseURL is a custom API endpoint for this model.
BaseURL string `json:"baseURL,omitempty"`
// APIKeyName overrides the API key environment variable for this model.
APIKeyName string `json:"apiKeyName,omitempty"`
}
ModelEntry represents a single model in the model array for fallback support. When multiple models are specified, they are tried in order until one succeeds.
type NodeStatus ¶
type NodeStatus int
NodeStatus represents the canonical lifecycle phases for an individual node.
const ( // Keep numeric values stable because NodeStatus is persisted in run status // snapshots and older files/tests depend on the historical encoding. NodeNotStarted NodeStatus = 0 NodeRunning NodeStatus = 1 NodeFailed NodeStatus = 2 NodeAborted NodeStatus = 3 NodeSucceeded NodeStatus = 4 NodeSkipped NodeStatus = 5 NodePartiallySucceeded NodeStatus = 6 NodeWaiting NodeStatus = 7 NodeRejected NodeStatus = 8 NodeRetrying NodeStatus = 9 )
func (NodeStatus) IsDone ¶
func (s NodeStatus) IsDone() bool
IsDone checks if the node has completed (success, failure, skipped, aborted, rejected, or partially succeeded).
func (NodeStatus) IsSuccess ¶
func (s NodeStatus) IsSuccess() bool
IsSuccess checks if the node status indicates a successful execution.
func (NodeStatus) IsWaiting ¶
func (s NodeStatus) IsWaiting() bool
IsWaiting checks if the node status requires manual action before it can continue.
func (NodeStatus) String ¶
func (s NodeStatus) String() string
String returns the canonical lowercase token for the node lifecycle phase.
type OTelConfig ¶
type OTelConfig struct {
Enabled bool `json:"enabled,omitempty"`
Endpoint string `json:"endpoint,omitempty"`
Headers map[string]string `json:"headers,omitempty"`
Insecure bool `json:"insecure,omitempty"`
Timeout time.Duration `json:"timeout,omitempty"`
Resource map[string]any `json:"resource,omitempty"`
}
OTelConfig contains the OpenTelemetry configuration.
type OverlapPolicy ¶
type OverlapPolicy string
OverlapPolicy controls behavior when a new run is triggered while a previous run is still active.
const ( // OverlapPolicySkip skips a new run if the previous is still running. OverlapPolicySkip OverlapPolicy = "skip" // OverlapPolicyAll queues all runs and executes them sequentially in chronological order. OverlapPolicyAll OverlapPolicy = "all" // OverlapPolicyLatest discards all but the most recent missed interval. OverlapPolicyLatest OverlapPolicy = "latest" )
func ParseOverlapPolicy ¶
func ParseOverlapPolicy(s string) (OverlapPolicy, error)
ParseOverlapPolicy parses a string into an OverlapPolicy. Empty string defaults to OverlapPolicySkip.
type ParallelConfig ¶
type ParallelConfig struct {
// Variable is the name of a variable that contains the json array of items to process in parallel.
Variable string `json:"variable,omitempty"`
// Items is the array of items to process in parallel.
// Can be a direct array or a reference to a variable containing an array.
Items []ParallelItem `json:"items,omitempty"`
// MaxConcurrent is the maximum number of parallel executions.
// Default is 10 if not specified.
MaxConcurrent int `json:"max_concurrent,omitempty"`
}
ParallelConfig contains the configuration for parallel execution of a step. MVP version supports basic parallel execution with max_concurrent control.
type ParallelItem ¶
type ParallelItem struct {
// Value is used for simple string items or variable references
// E.g. "item1", "item2", "${ITEM_VAR}"
Value string `json:"value,omitempty"`
// Params is used for key-value pairs that will be passed as parameters
// E.g. {"SOURCE": "s3://customers", "TYPE": "csv"}
// Uses DeterministicMap to ensure consistent JSON marshaling for hashing
Params collections.DeterministicMap `json:"params,omitempty"`
}
ParallelItem represents a single item to be processed in parallel. It can be either a simple value or a set of parameters.
type ParamDef ¶
type ParamDef struct {
Name string `json:"name,omitempty"`
Type string `json:"type,omitempty"`
Default any `json:"default,omitempty"`
Description string `json:"description,omitempty"`
Required bool `json:"required,omitempty"`
Enum []any `json:"enum,omitempty"`
Minimum *float64 `json:"minimum,omitempty"`
Maximum *float64 `json:"maximum,omitempty"`
MinLength *int `json:"minLength,omitempty"`
MaxLength *int `json:"maxLength,omitempty"`
Pattern *string `json:"pattern,omitempty"`
}
ParamDef describes a single DAG parameter for API/UI consumers. Name is empty for positional parameters.
type Params ¶
type Params struct {
// Simple params (backward compatible)
Simple map[string]string `json:"simple,omitempty"`
// Rich params with type preservation
Rich map[string]any `json:"rich,omitempty"`
// Raw params for lazy parsing
Raw json.RawMessage `json:"raw,omitempty"`
}
Params holds parameter data in one of several formats Only one of Simple, Rich, or Raw should be non-nil
func NewRawParams ¶
func NewRawParams(raw json.RawMessage) Params
NewRawParams creates params from json.RawMessage
func NewRichParams ¶
NewRichParams creates params from map[string]any
func NewSimpleParams ¶
NewSimpleParams creates params from map[string]string
func ParseParams ¶
ParseParams creates Params from any input type
func (*Params) AsStringMap ¶
AsStringMap returns all parameters as map[string]string Non-string values are converted using fmt.Sprintf
func (*Params) MarshalJSON ¶
MarshalJSON implements json.Marshaler
func (*Params) UnmarshalJSON ¶
UnmarshalJSON implements json.Unmarshaler
type PullPolicy ¶
type PullPolicy int
PullPolicy defines image pull policy for a container execution
const ( PullPolicyAlways PullPolicy = iota PullPolicyNever PullPolicyMissing PullPolicyFallback )
func ParsePullPolicy ¶
func ParsePullPolicy(raw any) (PullPolicy, error)
ParsePullPolicy parses a pull policy from a raw value.
func (PullPolicy) String ¶
func (p PullPolicy) String() string
type RedisConfig ¶
type RedisConfig struct {
// URL is the Redis connection URL (redis://user:pass@host:port/db).
URL string `json:"url,omitempty"`
// Host is the Redis host (alternative to URL).
Host string `json:"host,omitempty"`
// Port is the Redis port (default: 6379).
Port int `json:"port,omitempty"`
// Password is the authentication password.
Password string `json:"password,omitempty"`
// Username is the ACL username (Redis 6+).
Username string `json:"username,omitempty"`
// DB is the database number (0-15).
DB int `json:"db,omitempty"`
// TLS enables TLS connection.
TLS bool `json:"tls,omitempty"`
// TLSSkipVerify skips TLS certificate verification.
TLSSkipVerify bool `json:"tlsSkipVerify,omitempty"`
// Mode is the connection mode (standalone, sentinel, cluster).
Mode string `json:"mode,omitempty"`
// SentinelMaster is the sentinel master name.
SentinelMaster string `json:"sentinelMaster,omitempty"`
// SentinelAddrs is the list of sentinel addresses.
SentinelAddrs []string `json:"sentinelAddrs,omitempty"`
// ClusterAddrs is the list of cluster node addresses.
ClusterAddrs []string `json:"clusterAddrs,omitempty"`
// MaxRetries is the maximum number of retries.
MaxRetries int `json:"maxRetries,omitempty"`
}
RedisConfig contains the default Redis configuration for the DAG. Steps with type: redis inherit this configuration.
type ReferenceField ¶
type ReferenceField struct {
Path string
Value string
OwnerStepName string
OwnerStepID string
// OwnerStepPath is the spec path of the owning step, such as "steps[0]" or
// "handler_on.exit". It is empty for DAG-level fields.
OwnerStepPath string
Field cmnvalue.Field
// contains filtered or unexported fields
}
func ReferenceFields ¶
func ReferenceFields(dag *DAG) []ReferenceField
type RepeatMode ¶
type RepeatMode string
RepeatMode is the type for the repeat mode.
const ( // RepeatModeWhile repeats the step while the condition is met. RepeatModeWhile RepeatMode = "while" // RepeatModeUntil repeats the step until the condition is met. RepeatModeUntil RepeatMode = "until" )
type RepeatPolicy ¶
type RepeatPolicy struct {
// RepeatMode determines if and how the step should be repeated.
// It can be 'while' or 'until'.
RepeatMode RepeatMode `json:"repeatMode,omitempty"`
// Interval is the time to wait between repeats.
Interval time.Duration `json:"interval,omitempty"`
// IntervalStr is the string representation of interval_sec for deferred evaluation.
IntervalStr string `json:"intervalStr,omitempty"`
// Limit is the maximum number of times to repeat the step.
Limit int `json:"limit,omitempty"`
// LimitStr is the string representation of the limit for deferred evaluation.
LimitStr string `json:"limitStr,omitempty"`
// Backoff is the exponential backoff multiplier (e.g., 2.0 for doubling).
Backoff float64 `json:"backoff,omitempty"`
// MaxInterval is the maximum interval cap for exponential backoff.
MaxInterval time.Duration `json:"maxInterval,omitempty"`
// MaxIntervalStr is the string representation of max_interval_sec for deferred evaluation.
MaxIntervalStr string `json:"maxIntervalStr,omitempty"`
// Condition is the condition object to be met for the repeat.
Condition *Condition `json:"condition,omitempty"`
// ExitCode is the list of exit codes that should trigger a repeat.
ExitCode []int `json:"exitCode,omitempty"`
}
RepeatPolicy contains the repeat policy for a step.
func (*RepeatPolicy) UnmarshalJSON ¶
func (r *RepeatPolicy) UnmarshalJSON(data []byte) error
UnmarshalJSON implements the json.Unmarshaler interface for RepeatPolicy. It handles the legacy boolean repeat field and the new string repeat modes.
type ResourceLimits ¶
type ResourceLimits struct {
CPU string `json:"cpu,omitempty"`
Memory string `json:"memory,omitempty"`
CPUMillis int64 `json:"-"`
MemoryBytes int64 `json:"-"`
}
ResourceLimits contains CPU and memory limits requested for a DAG run.
func NewResourceLimits ¶
func NewResourceLimits(cpu, memory string) (*ResourceLimits, error)
NewResourceLimits validates authored resource limits and returns a normalized copy.
func (*ResourceLimits) UnmarshalJSON ¶
func (r *ResourceLimits) UnmarshalJSON(data []byte) error
UnmarshalJSON decodes authored limits and restores their normalized values.
type Resources ¶
type Resources struct {
Limits *ResourceLimits `json:"limits,omitempty"`
}
Resources contains resource requests for a DAG run.
type RetryPolicy ¶
type RetryPolicy struct {
// Limit is the number of retries allowed.
Limit int `json:"limit,omitempty"`
// Interval is the time to wait between retries.
Interval time.Duration `json:"interval,omitempty"`
// LimitStr is the string representation of the limit.
LimitStr string `json:"limitStr,omitempty"`
// IntervalSecStr is the string representation of the interval.
IntervalSecStr string `json:"intervalSecStr,omitempty"`
// ExitCodes is the list of exit codes that should trigger a retry.
ExitCodes []int `json:"exitCode,omitempty"`
// Backoff is the exponential backoff multiplier (e.g., 2.0 for doubling).
Backoff float64 `json:"backoff,omitempty"`
// MaxInterval is the maximum interval cap for exponential backoff.
MaxInterval time.Duration `json:"maxInterval,omitempty"`
}
RetryPolicy contains the retry policy for a step.
type RouteEntry ¶
type RouteEntry struct {
Pattern string `json:"pattern"` // Match pattern (exact or "re:regex")
Targets []string `json:"targets"` // Step names to route to when pattern matches
}
RouteEntry represents a single routing rule.
type RouterConfig ¶
type RouterConfig struct {
Value string `json:"value"` // Value expression to evaluate (e.g., "${STATUS}")
Routes []RouteEntry `json:"routes"` // Ordered list of pattern → targets
}
RouterConfig contains routing configuration for router-type steps.
type RunConfig ¶
type RunConfig struct {
// DisableParamEdit when set to true, prevents users from editing parameters when starting a DAG.
DisableParamEdit bool `json:"disableParamEdit,omitempty"`
// DisableRunIdEdit when set to true, prevents users from specifying custom run IDs.
DisableRunIdEdit bool `json:"disableRunIdEdit,omitempty"`
}
RunConfig contains configuration for controlling user interactions during DAG runs.
type S3Config ¶
type S3Config struct {
// Region is the AWS region (e.g., us-east-1).
Region string `json:"region,omitempty"`
// Endpoint is a custom S3-compatible endpoint URL.
// Use this for S3-compatible services like MinIO, LocalStack, etc.
Endpoint string `json:"endpoint,omitempty"`
// AccessKeyID is the AWS access key ID.
AccessKeyID string `json:"accessKeyId,omitempty"`
// SecretAccessKey is the AWS secret access key.
SecretAccessKey string `json:"secretAccessKey,omitempty"`
// SessionToken is the AWS session token (for temporary credentials).
SessionToken string `json:"sessionToken,omitempty"`
// Profile is the AWS credentials profile name.
Profile string `json:"profile,omitempty"`
// ForcePathStyle enables path-style addressing (required for S3-compatible services).
ForcePathStyle bool `json:"forcePathStyle,omitempty"`
// DisableSSL disables SSL for the connection (for local testing only).
DisableSSL bool `json:"disableSSL,omitempty"`
// Bucket is the default S3 bucket name.
// Can be overridden at the step level.
Bucket string `json:"bucket,omitempty"`
}
S3Config contains the default S3 configuration for the DAG. This allows steps to inherit S3 settings without specifying them individually.
type SMTPConfig ¶
type SMTPConfig struct {
Host string `json:"host,omitempty"`
Port string `json:"port,omitempty"`
Username string `json:"username,omitempty"`
Password string `json:"password,omitempty"`
}
SMTPConfig contains the SMTP configuration.
type SSHConfig ¶
type SSHConfig struct {
// User is the SSH user.
User string `json:"user,omitempty"`
// Host is the SSH host.
Host string `json:"host,omitempty"`
// Port is the SSH port. Default is "22".
Port string `json:"port,omitempty"`
// Key is the path to the SSH private key.
Key string `json:"key,omitempty"`
// Password is the SSH password.
Password string `json:"password,omitempty"`
// StrictHostKey enables strict host key checking. Defaults to true.
StrictHostKey bool `json:"strictHostKey,omitempty"`
// KnownHostFile is the path to the known_hosts file. Defaults to ~/.ssh/known_hosts.
KnownHostFile string `json:"knownHostFile,omitempty"`
// Shell is the shell to use for remote command execution.
// If not specified, commands are executed directly without shell wrapping.
Shell string `json:"shell,omitempty"`
// ShellArgs contains additional arguments that should be passed to the shell executable.
ShellArgs []string `json:"shellArgs,omitempty"`
// Timeout is the connection timeout duration (e.g., "30s", "1m"). Defaults to 30s.
Timeout string `json:"timeout,omitempty"`
// Bastion is the jump host / bastion server configuration for connecting to the target host.
Bastion *BastionConfig `json:"bastion,omitempty"`
}
SSHConfig contains the SSH configuration for the DAG.
type Schedule ¶
type Schedule struct {
// Kind identifies the schedule type.
Kind ScheduleKind `json:"kind,omitempty"`
// Expression is the cron expression.
Expression string `json:"expression,omitempty"`
// At is the canonical RFC 3339 timestamp for one-off schedules.
At string `json:"at,omitempty"`
// Profile is the runtime profile name that activates this schedule.
Profile string `json:"profile,omitempty"`
// Parsed is the parsed cron schedule.
Parsed cron.Schedule `json:"-"`
// AtTime is the parsed one-off schedule time.
AtTime time.Time `json:"-"`
// Warnings contains non-fatal schedule warnings.
Warnings []string `json:"warnings,omitempty"`
}
Schedule contains the cron expression and the parsed cron schedule.
func NewCronSchedule ¶
NewCronSchedule parses a cron schedule into its canonical representation.
func NewOneOffSchedule ¶
NewOneOffSchedule parses a one-off timestamp into its canonical representation.
func ParseScheduleValue ¶
func ParseScheduleValue(v any, opts ScheduleParseOptions) (Schedule, error)
ParseScheduleValue parses a YAML/JSON schedule entry into the canonical model.
func (Schedule) DisplayValue ¶
DisplayValue returns the user-facing schedule value.
func (Schedule) Fingerprint ¶
Fingerprint returns the canonical schedule fingerprint used for durable state.
func (Schedule) GetKind ¶
func (s Schedule) GetKind() ScheduleKind
GetKind returns the normalized schedule kind.
func (Schedule) MarshalJSON ¶
MarshalJSON implements the json.Marshaler interface.
func (Schedule) OneOffTime ¶
OneOffTime returns the parsed one-off schedule time, if any.
func (*Schedule) UnmarshalJSON ¶
UnmarshalJSON implements the json.Unmarshaler interface. It also parses the cron expression to populate the Parsed field.
type ScheduleKind ¶
type ScheduleKind string
ScheduleKind identifies the supported schedule types.
const ( ScheduleKindCron ScheduleKind = "cron" ScheduleKindAt ScheduleKind = "at" )
type ScheduleParseOptions ¶
type ScheduleParseOptions struct {
AllowAt bool
}
ScheduleParseOptions controls which schedule kinds are accepted.
type SecretRef ¶
type SecretRef struct {
// Name is the environment variable name to set (required).
Name string `json:"name"`
// Ref is the workspace-local registry reference for a team-managed secret.
Ref string `json:"ref,omitempty"`
// Provider specifies the secret backend (e.g., "env", "file", "vault", "kubernetes").
Provider string `json:"provider,omitempty"`
// Key is the provider-specific identifier for a direct provider reference.
Key string `json:"key,omitempty"`
// Options contains provider-specific configuration (optional).
Options map[string]string `json:"options,omitempty"`
}
SecretRef represents a reference to an external secret. Secrets are resolved at DAG execution time and never persisted to disk.
type StartParamInput ¶
StartParamInput describes start params regardless of caller (CLI/API). Use DashArgs for params passed after "--", or RawParams for --params style input.
type Status ¶
type Status int
Status represents the canonical lifecycle phases for a DAG execution.
func (Status) IsActive ¶
IsActive checks if the status is active (not yet completed). This includes running, queued, and waiting executions.
type Step ¶
type Step struct {
// ID is the optional unique identifier for the step.
ID string `json:"id,omitempty"`
// Name is the name of the step.
Name string `json:"name"`
// Description is the description of the step. This is optional.
Description string `json:"description,omitempty"`
// Shell is the shell program to execute the command. This is optional.
Shell string `json:"shell,omitempty"`
// ShellPackages is the list of packages to install. This is used only when the shell is `nix-shell`.
ShellPackages []string `json:"shellPackages,omitempty"`
// ShellArgs is the list of arguments for the shell program.
ShellArgs []string `json:"shellArgs,omitempty"`
// Dir is the working directory for the step.
Dir string `json:"dir,omitempty"`
// ExecutorConfig contains the configuration for the executor.
ExecutorConfig ExecutorConfig `json:"executorConfig,omitzero"`
// CmdWithArgs is the command with arguments for display purposes.
// Deprecated: Use Commands[0].CmdWithArgs instead. Kept for JSON backward compatibility.
CmdWithArgs string `json:"cmdWithArgs,omitempty"`
// CmdArgsSys is the command with arguments for the system.
// Deprecated: Kept for JSON backward compatibility.
CmdArgsSys string `json:"cmdArgsSys,omitempty"`
// Command specifies only the command without arguments.
// Deprecated: Use Commands field instead. Kept for JSON backward compatibility.
Command string `json:"command,omitempty"`
// ShellCmdArgs is the shell command with arguments.
ShellCmdArgs string `json:"shellCmdArgs,omitempty"`
// Script is the script to be executed.
Script string `json:"script,omitempty"`
// Args contains the arguments for the command.
// Deprecated: Use Commands field instead. Kept for JSON backward compatibility.
Args []string `json:"args,omitempty"`
// Commands is the source of truth for commands to execute.
// Each entry represents a command to be executed sequentially.
// For single commands, this will contain exactly one entry.
Commands []CommandEntry `json:"commands,omitempty"`
// Stdout is the file to store the standard output.
Stdout string `json:"stdout,omitempty"`
// StdoutArtifact is the artifact-relative file path to store standard output.
StdoutArtifact string `json:"stdoutArtifact,omitempty"`
// StdoutOutputs maps standard output into the DAG/action outputs object.
StdoutOutputs *StepOutputsConfig `json:"stdoutOutputs,omitempty"`
// Stderr is the file to store the standard error.
Stderr string `json:"stderr,omitempty"`
// StderrArtifact is the artifact-relative file path to store standard error.
StderrArtifact string `json:"stderrArtifact,omitempty"`
// LogOutput specifies how stdout and stderr are handled in log files for this step.
// Overrides the DAG-level LogOutput setting. Empty string means inherit from DAG.
LogOutput LogOutputMode `json:"logOutput,omitempty"`
// Output is the variable name to store captured stdout.
Output string `json:"output,omitempty"`
// StructuredOutput publishes post-processed step-scoped outputs for ${step.output.*} access.
StructuredOutput map[string]StepOutputEntry `json:"structuredOutput,omitempty"`
// OutputSchema validates stdout JSON before publishing step-scoped output.
OutputSchema map[string]any `json:"outputSchema,omitzero"`
// Outputs declares file-based step outputs published through DAGU_OUTPUT_FILE.
Outputs []StepOutputDeclaration `json:"outputs,omitempty"`
// Depends contains the list of step names to depend on.
Depends []string `json:"depends,omitempty"`
// ExplicitlyNoDeps indicates the depends field was explicitly set to empty
ExplicitlyNoDeps bool `json:"-"`
// ContinueOn contains the conditions to continue on failure or skipped.
ContinueOn ContinueOn `json:"continueOn,omitzero"`
// RetryPolicy contains the retry policy for the step.
RetryPolicy RetryPolicy `json:"retryPolicy,omitzero"`
// RepeatPolicy contains the repeat policy for the step.
RepeatPolicy RepeatPolicy `json:"repeatPolicy,omitzero"`
// MailOnError is the flag to send mail on error.
MailOnError bool `json:"mailOnError,omitempty"`
// Preconditions contains the conditions to be met before running the step.
Preconditions []*Condition `json:"preconditions,omitempty"`
// SignalOnStop is the signal to send on stop.
SignalOnStop string `json:"signalOnStop,omitempty"`
// SubDAG contains the information about a sub DAG to be executed.
SubDAG *SubDAG `json:"childDag,omitempty"`
// WorkerSelector specifies required worker labels for execution.
WorkerSelector map[string]string `json:"workerSelector,omitempty"`
// Parallel contains the configuration for parallel execution.
Parallel *ParallelConfig `json:"parallel,omitempty"`
// Foreach contains the configuration for inline item-body iteration.
Foreach *ForeachConfig `json:"foreach,omitempty"`
// Env contains environment variables for the step.
Env []string `json:"env,omitempty"`
// Params contains parameters/inputs for the step.
Params Params `json:"params,omitzero"`
// Timeout specifies the maximum execution time for the step.
// If set, this timeout takes precedence over the DAG-level timeout for this step.
Timeout time.Duration `json:"timeout,omitempty"`
// Container specifies the container configuration for this step.
// If set, the step runs in its own container instead of the DAG-level container.
// This uses the same configuration format as the DAG-level container field.
Container *Container `json:"container,omitempty"`
// LLM contains the configuration for LLM-based executors.
// Used with explicit type: chat.
LLM *LLMConfig `json:"llm,omitempty"`
// Messages contains the session messages for chat executor.
// Only used when type is "chat".
Messages []LLMMessage `json:"messages,omitempty"`
// Router contains the routing configuration for router-type steps.
// Only used when type is "router".
Router *RouterConfig `json:"router,omitempty"`
// Approval configures a human approval gate after step execution.
// When set, the step pauses in Waiting state after execution completes.
Approval *ApprovalConfig `json:"approval,omitempty"`
// HumanTask configures a processless step completed by a local operator.
HumanTask *HumanTaskConfig `json:"humanTask,omitempty"`
}
Step contains the runtime information for a step in a DAG. A step is created from parsing a DAG file written in YAML. It marshals/unmarshals to/from JSON when it is saved in the execution history.
func NewControllerStep ¶
NewControllerStep builds the step that carries the controller's LLM config and task list. It is appended to a controller DAG at build time and is the node the runner drives the decision loop from.
func (Step) CommandResolution ¶
func (s Step) CommandResolution(ctx context.Context) cmnvalue.CommandContext
CommandResolution returns command execution facts for command field resolution.
func (Step) HasDeclaredOutputs ¶
HasDeclaredOutputs reports whether the step declares file-based outputs.
func (*Step) HasMultipleCommands ¶
HasMultipleCommands returns true if the step has multiple commands to execute.
func (Step) HasOutputSchema ¶
HasOutputSchema reports whether the step validates stdout JSON with an output schema.
func (Step) HasStdoutOutputs ¶
HasStdoutOutputs reports whether stdout should publish DAG/action outputs.
func (Step) HasStructuredOutput ¶
HasStructuredOutput reports whether the step publishes object-form output.
func (Step) ScriptResolution ¶
func (s Step) ScriptResolution(ctx context.Context) cmnvalue.CommandContext
ScriptResolution returns command execution facts for script field resolution.
func (*Step) UnmarshalJSON ¶
UnmarshalJSON implements json.Unmarshaler for backward compatibility. It handles old JSON format where command/args fields were used instead of commands.
func (Step) UsesStructuredOutputSource ¶
UsesStructuredOutputSource reports whether any structured output entry reads from source.
type StepOutputDeclaration ¶
StepOutputDeclaration defines one top-level file-based step output.
type StepOutputEntry ¶
type StepOutputEntry struct {
// HasValue distinguishes literal/null values from source-based outputs.
HasValue bool `json:"hasValue,omitempty"`
// Value is the literal value to publish when HasValue is true.
Value any `json:"value"`
// From selects a runtime source to read from: stdout, stderr, or file.
From string `json:"from,omitempty"`
// Path is the file path used when From is file.
Path string `json:"path,omitempty"`
// Decode controls how the source content is decoded before selection.
Decode string `json:"decode,omitempty"`
// Select is an optional jq/gojq path applied after decode.
Select string `json:"select,omitempty"`
}
StepOutputEntry defines one structured object-form output entry.
type StepOutputsConfig ¶
type StepOutputsConfig struct {
// Field writes the decoded stdout value to a single outputs field.
Field string `json:"field,omitempty"`
// Decode controls how stdout is decoded before writing outputs.
Decode string `json:"decode,omitempty"`
// Select is an optional jq/gojq path applied after decode.
Select string `json:"select,omitempty"`
// Fields maps individual outputs fields from stdout or literal values.
Fields map[string]StepOutputEntry `json:"fields,omitempty"`
}
StepOutputsConfig defines how stdout is mapped into the DAG/action outputs object.
type StepValidator ¶
StepValidator is a function type for validating step configurations.
type TagFilter
deprecated
type TagFilter = LabelFilter
Deprecated: use LabelFilter instead.
type TagFilterType
deprecated
type TagFilterType = LabelFilterType
Deprecated: use LabelFilterType instead.
type ThinkingConfig ¶
type ThinkingConfig struct {
// Enabled activates thinking mode for supported models.
Enabled bool `json:"enabled,omitempty"`
// Effort controls reasoning depth: low, medium, high, xhigh.
// Maps to provider-specific parameters.
Effort ThinkingEffort `json:"effort,omitempty"`
// BudgetTokens sets explicit token budget (provider-specific).
// For Anthropic: minimum 1024, max 128K.
// For Gemini 2.5: range 128-32768.
BudgetTokens *int `json:"budgetTokens,omitempty"`
// IncludeInOutput includes thinking blocks in stdout.
// Default is false for consistency across providers.
IncludeInOutput bool `json:"includeInOutput,omitempty"`
}
ThinkingConfig contains configuration for extended thinking/reasoning.
type ThinkingEffort ¶
type ThinkingEffort string
ThinkingEffort represents the reasoning depth level for thinking mode.
const ( ThinkingEffortLow ThinkingEffort = "low" ThinkingEffortMedium ThinkingEffort = "medium" ThinkingEffortHigh ThinkingEffort = "high" ThinkingEffortXHigh ThinkingEffort = "xhigh" )
ThinkingEffort constants for reasoning/thinking depth.
func ParseThinkingEffort ¶
func ParseThinkingEffort(s string) (ThinkingEffort, error)
ParseThinkingEffort validates and returns a ThinkingEffort from a string. Returns empty string for empty input (no effort specified). Returns error for invalid effort values.
type ToolCall ¶
type ToolCall struct {
// ID is the unique identifier for this tool call.
ID string `json:"id"`
// Name is the name of the tool to invoke (matches DAG name field).
Name string `json:"name"`
// Arguments contains the tool arguments as key-value pairs.
Arguments map[string]any `json:"arguments"`
}
ToolCall represents an LLM's request to invoke a tool.
type ToolConfig ¶
type ToolConfig struct {
Provider string `json:"provider,omitempty"`
Registry *ToolRegistry `json:"registry,omitempty"`
Packages []ToolPackage `json:"packages,omitempty"`
}
ToolConfig declares external CLI tools required by a DAG run.
type ToolPackage ¶
type ToolPackage struct {
Name string `json:"name,omitempty"`
Package string `json:"package"`
Version string `json:"version"`
Commands []string `json:"commands,omitempty"`
Registry string `json:"registry,omitempty"`
// Digest pins the sha256 of the downloaded artifact for the platform the
// DAG runs on, in "sha256:<64 hex>" form. The install fails when the
// artifact hash recorded for this package does not match.
Digest string `json:"digest,omitempty"`
}
ToolPackage declares one aqua package and optional command names Dagu should expose.
type ToolRegistry ¶
type ToolRegistry struct {
Name string `json:"name,omitempty"`
Type string `json:"type,omitempty"`
RepoOwner string `json:"repoOwner,omitempty"`
RepoName string `json:"repoName,omitempty"`
Ref string `json:"ref,omitempty"`
Path string `json:"path,omitempty"`
}
ToolRegistry identifies the aqua registry used to resolve tool packages.
type ToolResult ¶
type ToolResult struct {
// ToolCallID is the ID of the tool call this result corresponds to.
ToolCallID string `json:"tool_call_id"`
// Name is the name of the tool that was executed.
Name string `json:"name"`
// Content is the result content from the tool execution.
Content string `json:"content"`
// Error contains any error message if the tool execution failed.
Error string `json:"error,omitempty"`
}
ToolResult represents the result of a tool execution.
type TriggerType ¶
type TriggerType int
TriggerType represents how a DAG run was initiated.
const ( TriggerTypeUnknown TriggerType = iota TriggerTypeScheduler TriggerTypeManual TriggerTypeWebhook TriggerTypeSubDAG TriggerTypeRetry TriggerTypeCatchUp )
func ParseTriggerType ¶
func ParseTriggerType(s string) TriggerType
ParseTriggerType parses a string into a TriggerType.
func (TriggerType) String ¶
func (t TriggerType) String() string
String returns the canonical lowercase token for the trigger type.
type ValidationError ¶
ValidationError represents an error in a specific field of the configuration
func (*ValidationError) Error ¶
func (e *ValidationError) Error() string
func (*ValidationError) Unwrap ¶
func (e *ValidationError) Unwrap() error
type WebSearchConfig ¶
type WebSearchConfig struct {
// Enabled activates provider-native web search.
Enabled bool `json:"enabled,omitempty"`
// MaxUses limits search invocations per request.
MaxUses *int `json:"maxUses,omitempty"`
// AllowedDomains restricts results to these domains (Anthropic only).
AllowedDomains []string `json:"allowedDomains,omitempty"`
// BlockedDomains excludes results from these domains (Anthropic only).
BlockedDomains []string `json:"blockedDomains,omitempty"`
// UserLocation localizes search results.
UserLocation *WebSearchUserLocation `json:"userLocation,omitempty"`
}
WebSearchConfig contains configuration for provider-native web search.
type WebSearchUserLocation ¶
type WebSearchUserLocation struct {
City string `json:"city,omitempty"`
Region string `json:"region,omitempty"`
Country string `json:"country,omitempty"`
Timezone string `json:"timezone,omitempty"`
}
WebSearchUserLocation provides approximate location for search localization.
type WebhookConfig ¶
type WebhookConfig struct {
// ForwardHeaders is the allowlist of request headers to expose to
// webhook-triggered DAG runs via the WEBHOOK_HEADERS runtime variable.
ForwardHeaders []string `json:"forwardHeaders,omitempty"`
}
WebhookConfig contains DAG-level webhook trigger behavior.
Source Files
¶
- capabilities.go
- condition.go
- container.go
- controller.go
- dag.go
- duration.go
- errors.go
- executor_schema.go
- foreach.go
- harness.go
- label.go
- llm.go
- overlap_policy.go
- parallel.go
- params.go
- resources.go
- retry.go
- schedule.go
- startparams.go
- status.go
- step.go
- tools.go
- validator.go
- value_fields.go
- value_notices.go
- webhook.go
- webhook_params.go