runtime

package
v2.23.0 Latest Latest
Warning

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

 Go to latest Published: Oct 26, 2026 License: GPL-3.0 Imports: 63 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ErrMsgOtherConditionNotMet = "other condition was not met"
	ErrMsgOtherConditionFailed = "other condition failed to evaluate"
)

Back-fill messages for conditions that passed while a sibling decided the outcome. The wording mirrors how the owner is reported: a not-met condition skips it, while an evaluation error fails it.

Variables

View Source 
var (
	// NewContext creates a new context with DAG execution metadata.
	NewContext = runctx.NewContext
	// LookupDAGContext returns the DAG execution metadata when it is present.
	LookupDAGContext = runctx.LookupContext
	// WithDAGLoader sets the DAG loader.
	WithDAGLoader = runctx.WithDAGLoader
	// WithRootDAGRun sets the root DAG run reference for sub-DAG execution.
	WithRootDAGRun = runctx.WithRootDAGRun
	// WithRetryPath sets a targeted child DAG retry path.
	WithRetryPath = runctx.WithRetryPath
	// WithIncludeDownstream records that a targeted step retry should also
	// reset reachable descendants.
	WithIncludeDownstream = runctx.WithIncludeDownstream
	// WithBypassPreconditions records that steps reset by a targeted step
	// retry skip step precondition evaluation.
	WithBypassPreconditions = runctx.WithBypassPreconditions
	// WithAttemptID sets the DAG-run attempt identifier.
	WithAttemptID = runctx.WithAttemptID
	// WithWorkerID sets the execution host identifier.
	WithWorkerID = runctx.WithWorkerID
	// WithTriggerType sets the DAG-run trigger type.
	WithTriggerType = runctx.WithTriggerType
	// WithTriggerActor sets the attributable trigger actor.
	WithTriggerActor = runctx.WithTriggerActor
	// WithRunStartedAt sets the recorded DAG-run start timestamp.
	WithRunStartedAt = runctx.WithRunStartedAt
	// WithScheduleTime sets the logical schedule time.
	WithScheduleTime = runctx.WithScheduleTime
	// WithParams sets runtime parameters.
	WithParams = runctx.WithParams
	// WithParamsJSON replaces the DAG's params JSON in the environment steps
	// see.
	WithParamsJSON = runctx.WithParamsJSON
	// WithDefaultEnvVars sets low-precedence inherited environment variables.
	WithDefaultEnvVars = runctx.WithDefaultEnvVars
	// WithEnvVars sets additional execution-scoped environment variables.
	WithEnvVars = runctx.WithEnvVars
	// WithInheritedEnvs sets environment entries a child run inherits from the
	// parent run scope.
	WithInheritedEnvs = runctx.WithInheritedEnvs
	// WithCoordinator sets the coordinator dispatcher for distributed execution.
	WithCoordinator = runctx.WithCoordinator
	// WithDefaultSecrets sets low-precedence inherited secret environment variables.
	WithDefaultSecrets = runctx.WithDefaultSecrets
	// WithSecrets sets secret environment variables.
	WithSecrets = runctx.WithSecrets
	// WithLogEncoding sets the log file character encoding.
	WithLogEncoding = runctx.WithLogEncoding
	// WithLogWriterFactory sets the log writer factory for remote log streaming.
	WithLogWriterFactory = runctx.WithLogWriterFactory
	// WithDefaultExecMode sets the server-level default execution mode.
	WithDefaultExecMode = runctx.WithDefaultExecMode
	// WithRunStateStore sets the execution-state store.
	WithRunStateStore = runctx.WithRunStateStore
	// WithStateStore sets the persistent DAG state store.
	WithStateStore = runctx.WithStateStore
	// WithMaterializationStore sets the build materialization store.
	WithMaterializationStore = runctx.WithMaterializationStore
	// WithNoReuse records that manifest hits are disabled for the run.
	WithNoReuse = runctx.WithNoReuse
	// WithDAGRunLogDir sets the base log directory for newly persisted DAG runs.
	WithDAGRunLogDir = runctx.WithDAGRunLogDir
	// WithDAGRunArtifactDir sets the base artifact directory for newly persisted DAG runs.
	WithDAGRunArtifactDir = runctx.WithDAGRunArtifactDir
	// WithWorkDir sets the per-DAG-run working directory path.
	WithWorkDir = runctx.WithWorkDir
	// WithArtifactDir sets the per-DAG-run artifact directory path.
	WithArtifactDir = runctx.WithArtifactDir
	// WithRuntimeProfile sets selected runtime profile metadata.
	WithRuntimeProfile = runctx.WithRuntimeProfile
	// WithRuntimeProfileValues sets the resolved runtime profile environment.
	WithRuntimeProfileValues = runctx.WithRuntimeProfileValues
)

Re-export execution package functions for convenience.

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

Errors for condition evaluation

Functions

func AllEnvs 

func AllEnvs(ctx context.Context) []string

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

func AllEnvsMap 

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

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

func DAGShell 

func DAGShell(ctx context.Context) []string

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

func EffectiveLLMConfig 

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

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

func EvalBool 

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

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

func EvalCondition 

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

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

func EvalObject 

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

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

func EvaluateConditions added in v2.13.0 

func EvaluateConditions(ctx context.Context, shell []string, conditions []*ir.Condition) ([]ir.ConditionResult, error)

EvaluateConditions evaluates conditions and returns their runtime results.

func GenerateSubDAGRunID 

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

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

func GenerateSubDAGRunIDForTarget 

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

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

func MaskSecretsForProvider 

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

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

func NewContextForTest 

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

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

func NewDAGRunRef 

func NewDAGRunRef(name, runID string) ir.DAGRunRef

NewDAGRunRef is a convenience wrapper for execution.NewDAGRunRef.

func NewLLMProvider 

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

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

func NormalizeEnvVarExpr 

func NormalizeEnvVarExpr(expr string) string

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

func OutputValuesFromExecNodes 

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

OutputValuesFromExecNodes extracts typed DAG/action outputs from persisted nodes. Payloads that are not objects are skipped; see NodeData.OutputsValueMap.

func OutputValuesFromNodes 

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

OutputValuesFromNodes extracts typed DAG/action outputs from runtime nodes. Payloads that are not objects are skipped; see NodeData.OutputsValueMap.

func RepairStaleLocalRun 

func RepairStaleLocalRun(
	ctx context.Context,
	attempt dagrun.Attempt,
	dag *ir.DAG,
) (*ir.DAGRunStatus, bool, error)

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

func RepairStaleRemoteRun 

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

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

func ResolveDAGShell 

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

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

func ResolveModels 

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

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

func ResolveNumericComparison added in v2.17.0 

func ResolveNumericComparison(ctx context.Context, pattern, fieldPath string) (stringutil.NumericComparison, error)

ResolveNumericComparison parses a numeric-comparison pattern, resolving a value reference in its threshold first. fieldPath names the field for notices.

The whole pattern is resolved rather than just the threshold, which is equivalent because the numeric prefix and the ordering operators contain no dollar sign. A reference that cannot be resolved is preserved as its own text, so it reaches the parser as a non-number and fails there.

No returned error quotes a resolved threshold, which may hold a secret, so a caller can wrap the error while reporting the pattern as authored. Every surface that compares a numeric pattern must go through here, so that gating and routing cannot disagree about what a threshold means.

func ResolveString 

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

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

func ValueResolver 

func ValueResolver(ctx context.Context) cmnvalue.Resolver

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

func ValueResolverWithScope 

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

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

func WithDAGContext 

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

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

func WithEnv 

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

WithEnv returns a new context with the given execution context.

Types

type ArtifactFinalizer 

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

ArtifactFinalizer persists artifacts before terminal status is reported.

type AttemptRejected 

type AttemptRejected interface {
	error
	AttemptRejectedReason() string
}

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

type ChatMessagesHandler 

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

ChatMessagesHandler handles chat session messages for persistence.

type Config 

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

	DAGRunAutoRetryCount int
	DAGRunAutoRetryLimit int
	DAGRunIsRoot         bool
}

type Context 

type Context = runctx.Context

Context is an alias for execution.Context

func GetDAGContext 

func GetDAGContext(ctx context.Context) Context

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

func MustDAGContext added in v2.12.0 

func MustDAGContext(ctx context.Context) Context

MustDAGContext retrieves the DAGContext from the context and panics when one is not present. This is a convenience wrapper for execution.MustContext.

type ContextOption 

type ContextOption = runctx.ContextOption

ContextOption is an alias for execution.ContextOption

type DAGLoader added in v2.14.0 

type DAGLoader = runctx.DAGLoader

DAGLoader is an alias for runctx.DAGLoader.

type Data 

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

Data is a thread-safe wrapper around NodeData.

func (*Data) AddSubRunsRepeated 

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

AddSubRunsRepeated appends repeated sub DAG runs to the node.

func (*Data) Args 

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

func (*Data) ClearState 

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

ClearState resets the step for another execution. It keeps what the step's earlier attempts spent on models, which the run still spent.

func (*Data) ClearVariable 

func (d *Data) ClearVariable(key string)

func (*Data) CompleteHumanTaskDryRun 

func (d *Data) CompleteHumanTaskDryRun(prompt string, artifacts []string)

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

func (*Data) ContinueOn 

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

func (*Data) Data 

func (s *Data) Data() NodeData

func (*Data) Error 

func (d *Data) Error() error

func (*Data) Finish 

func (d *Data) Finish()

func (*Data) GetAgentSession added in v2.14.0 

func (d *Data) GetAgentSession() *ir.AgentSession

GetAgentSession returns a detached managed coding-agent state snapshot.

func (*Data) GetApprovalInputs 

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

GetApprovalInputs returns a copy of the approval inputs map.

func (*Data) GetChatMessages 

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

GetChatMessages returns the chat session messages for the node.

func (*Data) GetDoneCount 

func (d *Data) GetDoneCount() int

func (*Data) GetExitCode 

func (d *Data) GetExitCode() int

func (*Data) GetRetryCount 

func (d *Data) GetRetryCount() int

func (*Data) GetStderr 

func (d *Data) GetStderr() string

func (*Data) GetStdout 

func (d *Data) GetStdout() string

func (*Data) GetToolDefinitions 

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

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

func (*Data) IncDoneCount 

func (d *Data) IncDoneCount()

func (*Data) IncRetryCount 

func (d *Data) IncRetryCount()

func (*Data) IsRepeated 

func (d *Data) IsRepeated() bool

func (*Data) MarkError 

func (d *Data) MarkError(err error)

func (*Data) MatchExitCode 

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

func (*Data) ModelUsage 

func (d *Data) ModelUsage() *ir.StepModelUsage

ModelUsage returns a detached copy of what the step's calls to models used.

func (*Data) Name 

func (d *Data) Name() string

func (*Data) OpenHumanTask 

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

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

func (*Data) ResetError 

func (d *Data) ResetError()

func (*Data) SetAgentSession added in v2.14.0 

func (d *Data) SetAgentSession(session *ir.AgentSession)

SetAgentSession stores managed coding-agent state on the node.

func (*Data) SetAgentState added in v2.14.0 

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

SetAgentState stores the agent's goal progress on the node.

func (*Data) SetApprovalInputs 

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

SetApprovalInputs sets the approval inputs map.

func (*Data) SetArgs 

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

func (*Data) SetChatMessages 

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

SetChatMessages sets the chat session messages for the node.

func (*Data) SetError 

func (d *Data) SetError(err error)

func (*Data) SetExecutorConfig 

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

func (*Data) SetExitCode 

func (d *Data) SetExitCode(exitCode int)

func (*Data) SetModelUsage 

func (d *Data) SetModelUsage(usage *ir.StepModelUsage)

SetModelUsage records what the step's calls to models used.

func (*Data) SetPreconditionResults added in v2.13.0 

func (d *Data) SetPreconditionResults(results []ir.ConditionResult)

SetPreconditionResults replaces the latest precondition evaluation results.

func (*Data) SetRepeated 

func (d *Data) SetRepeated(repeated bool)

func (*Data) SetRetriedAt 

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

func (*Data) SetRetryCount 

func (d *Data) SetRetryCount(retryCount int)

func (*Data) SetScript 

func (d *Data) SetScript(script string)

func (*Data) SetSkippedByRetry added in v2.15.0 

func (d *Data) SetSkippedByRetry(skipped bool)

func (*Data) SetStatus 

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

func (*Data) SetStatusDetails added in v2.12.0 

func (d *Data) SetStatusDetails(details []ir.NodeStatusDetail)

SetStatusDetails replaces the independently tracked execution statuses.

func (*Data) SetStdin added in v2.17.0 

func (d *Data) SetStdin(stdin string)

func (*Data) SetStep 

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

func (*Data) SetSubDAG 

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

func (*Data) SetSubRuns 

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

SetSubRuns replaces the sub DAG runs associated with the node.

func (*Data) SetToolDefinitions 

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

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

func (*Data) SetWorkingDir 

func (d *Data) SetWorkingDir(workingDir string)

SetWorkingDir records the effective working directory for the node execution.

func (*Data) Setup 

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

func (*Data) SignalOnStop 

func (d *Data) SignalOnStop() string

func (*Data) State 

func (d *Data) State() NodeState

func (*Data) Status 

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

func (*Data) Step 

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

func (*Data) StepInfo 

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

type Dispatcher 

type Dispatcher = dispatch.Dispatcher

Dispatcher is an alias for execution.Dispatcher

type DryRunCheck 

type DryRunCheck struct {
	Step    string `json:"step"`
	Field   string `json:"field"`
	Message string `json:"message"`
}

DryRunCheck is one problem a dry run can show for a step. Field is the step field the problem concerns, such as "with.sheet" or "run", and Message is the text after that field label.

type Env 

type Env struct {
	// Embedded execution metadata from the parent DAG run.
	Context

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

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

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

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

	// Inputs contains final materialization paths scoped to the current step.
	Inputs cmnvalue.Values
	// Outputs contains attempt staging paths scoped to the current step.
	Outputs cmnvalue.Values

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

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

func GetEnv 

func GetEnv(ctx context.Context) Env

GetEnv returns the execution context from the given context.

func LookupEnv 

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

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

func NewEnv 

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

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

func NewEnvWithError 

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

NewEnvWithError creates an Env and returns working directory resolution errors.

func NewPlanEnv 

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

func NewPlanEnvForNode 

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

func NewPlanEnvForNodeWithError 

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

func NewPlanEnvWithError 

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

func (Env) AllEnvs 

func (e Env) AllEnvs() []string

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

func (Env) DAGRunRef 

func (e Env) DAGRunRef() ir.DAGRunRef

DAGRunRef returns the DAGRunRef for the current execution context.

func (Env) EvalBool 

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

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

func (Env) MailAccount added in v2.18.0 

func (e Env) MailAccount(ctx context.Context, address string) (*ir.MailAccount, error)

MailAccount returns the mail account configured for address, with variables evaluated. Only that account is resolved.

func (Env) MailerConfig 

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

MailerConfig returns the SMTP mailer configuration with variables evaluated.

func (Env) ResolveShell 

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

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

func (Env) Shell 

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

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

func (Env) UserEnvsMap 

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

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

func (Env) WithEnvVars 

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

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

type LogWriterFactory 

type LogWriterFactory = runctx.LogWriterFactory

LogWriterFactory is re-exported from execution package

type Manager 

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

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

func NewManager 

func NewManager(dagRunRepository *persis.DAGRunRepository, processes processRepository, cfg *config.Config, opts ...ManagerOption) Manager

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

func (*Manager) FindSubDAGRunStatus 

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

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

func (*Manager) GetCurrentStatus 

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

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

func (*Manager) GetLatestStatus 

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

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

func (*Manager) GetSavedStatus 

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

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

func (*Manager) IsRunning 

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

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

func (*Manager) Stop 

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

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

func (*Manager) UpdateStatus 

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

UpdateStatus updates the status of a dag-run.

type ManagerOption 

type ManagerOption func(*Manager)

ManagerOption configures a Manager.

func WithLatestStatusAllHistory added in v2.14.0 

func WithLatestStatusAllHistory() ManagerOption

WithLatestStatusAllHistory makes latest-status lookups search all retained runs.

func WithManagerClock 

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

WithManagerClock overrides the Manager clock.

type Node 

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

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

func NewNode 

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

func NodeWithData 

func NodeWithData(data NodeData) *Node

func (*Node) BuildSubDAGRuns 

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

BuildSubDAGRuns constructs the sub DAG runs based on parallel configuration.

func (*Node) BypassPreconditions added in v2.17.0 

func (n *Node) BypassPreconditions() bool

BypassPreconditions reports whether the node skips step precondition evaluation.

func (*Node) Cancel 

func (n *Node) Cancel()

func (*Node) Execute 

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

func (*Node) ID 

func (n *Node) ID() int

func (*Node) Init 

func (n *Node) Init()

func (*Node) ItemToParam 

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

ItemToParam converts a parallel item to a parameter string

func (*Node) LogContainsPattern 

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

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

func (*Node) NodeData 

func (n *Node) NodeData() NodeData

func (*Node) OutputVariablesMap 

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

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

func (*Node) Prepare 

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

func (*Node) ResetForRerun 

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

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

func (*Node) SetBypassPreconditions added in v2.17.0 

func (n *Node) SetBypassPreconditions(bypass bool)

SetBypassPreconditions marks the node to skip step precondition evaluation when it executes. It is in-memory only and does not persist into node state.

func (*Node) SetupEnv 

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

func (*Node) ShouldContinue 

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

func (*Node) ShouldMarkSuccess 

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

func (*Node) Signal 

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

func (*Node) StdoutFile 

func (n *Node) StdoutFile() string

func (*Node) Stop 

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

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

func (*Node) Teardown 

func (n *Node) Teardown() error

type NodeData 

type NodeData struct {
	Step  ir.Step
	State NodeState
}

NodeData represents the data of a node.

func (NodeData) OutputsValueMap 

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

OutputsValueMap returns the step's published outputs as name and value pairs. A step that publishes a payload which is not an object, such as the array of per-child outputs a parallel step publishes, contributes nothing: such a payload has no names to merge under.

func (NodeData) OutputsValueStringMap 

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

func (NodeData) StepOutputsValueMap 

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

StepOutputsValueMap returns the step's published outputs as text. A published value may be a number, a boolean, or a nested object, so each is rendered the same way a strict step-output reference renders it.

func (NodeData) StringFormOutputValue 

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

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

type NodeState 

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

type OutputCoordinator 

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

func (*OutputCoordinator) StdoutFile 

func (oc *OutputCoordinator) StdoutFile() string

type Parallel 

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

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

type ParallelItem 

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

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

type Plan 

type Plan struct {

	// Adjacency lists (immutable after planning; exposed for unit tests)
	DependencyMap map[int][]int // node ID -> list of dependency node IDs (upstream)
	DependantMap  map[int][]int // node ID -> list of dependent node IDs (downstream)
	// contains filtered or unexported fields
}

Plan represents a plan of execution for a set of steps. Graph mutation is confined to planning before execution starts.

func CreateRetryPlan 

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

CreateRetryPlan creates a new execution plan for retrying specific nodes.

func CreateRetryPlanWithOptions 

func CreateRetryPlanWithOptions(ctx context.Context, dag *ir.DAG, opts RetryPlanOptions, nodes ...*Node) (*Plan, error)

CreateRetryPlanWithOptions creates a default retry plan with explicit checkpoint-resume behavior.

func CreateStepRetryPlan 

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

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

func CreateStepRetryPlanWithOptions added in v2.15.0 

func CreateStepRetryPlanWithOptions(dag *ir.DAG, nodes []*Node, stepName string, opts StepRetryPlanOptions) (*Plan, error)

CreateStepRetryPlanWithOptions creates a step retry plan with explicit options.

func NewPlan 

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

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

func NewPlanFromNodes 

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

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

func (*Plan) AddInferredDependency added in v2.13.0 

func (p *Plan) AddInferredDependency(producerName, consumerName string) error

AddInferredDependency adds one data-derived dependency, from a matching build path or a step-output reference, before execution starts.

func (*Plan) CheckFinished 

func (p *Plan) CheckFinished() bool

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

func (*Plan) Dependencies 

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

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

func (*Plan) Dependents 

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

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

func (*Plan) Duration 

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

func (*Plan) Finish 

func (p *Plan) Finish()

func (*Plan) FinishAt 

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

func (*Plan) GetNode 

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

GetNode returns the node with the given ID.

func (*Plan) GetNodeByName 

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

GetNodeByName returns the node with the given name.

func (*Plan) HasActiveNodes 

func (p *Plan) HasActiveNodes() bool

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

func (*Plan) IsAgent added in v2.14.0 

func (p *Plan) IsAgent() bool

IsAgent reports whether execution order is decided by an agent step rather than by dependency edges.

func (*Plan) IsFinished 

func (p *Plan) IsFinished() bool

func (*Plan) IsInferredDependency added in v2.13.0 

func (p *Plan) IsInferredDependency(producerID, consumerID int) bool

IsInferredDependency reports whether an edge was derived from data rather than from depends. It may be called concurrently after planning is complete.

func (*Plan) IsRunning 

func (p *Plan) IsRunning() bool

IsRunning checks if any node is currently running or pending.

func (*Plan) IsStarted 

func (p *Plan) IsStarted() bool

func (*Plan) NodeData 

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

NodeData returns a snapshot of data for all nodes.

func (*Plan) NodeStates 

func (p *Plan) NodeStates() PlanNodeStates

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

func (*Plan) Nodes 

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

Nodes returns a slice of all nodes in the plan.

func (*Plan) StartAt 

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

func (*Plan) WaitingStepNames 

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

WaitingStepNames returns the names of steps that require manual action.

type PlanNodeStates 

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

PlanNodeStates holds the state flags for nodes in a plan.

type ProgressUpdate added in v2.17.0 

type ProgressUpdate struct {
	Node *Node
	// contains filtered or unexported fields
}

ProgressUpdate carries a node state change to the agent that owns the progress channel.

When ack is non-nil the sender is waiting for the status snapshot covering this change to be persisted, so a node's state reaches durable storage before execution continues past it. Receivers release the sender by calling Ack with the outcome of that attempt; a zero value carries no ack and releases nobody.

func (ProgressUpdate) Ack added in v2.17.0 

func (u ProgressUpdate) Ack(err error)

Ack releases the sender waiting on this update. err reports whether the status snapshot was persisted. The first call decides the outcome; later calls, and calls on an update carrying no ack channel, are no-ops.

Ack never blocks, so a receiver cannot be wedged by a sender that stopped waiting.

type RetryPlanOptions 

type RetryPlanOptions struct {
	// Resume preserves partially succeeded nodes for a checkpoint resume.
	// Explicit retries leave this false and rerun partial work.
	Resume bool
}

RetryPlanOptions configures a default retry plan.

type RetryPolicy 

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

func (*RetryPolicy) ShouldRetry 

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

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

type RunStatus 

type RunStatus = ir.RunStatus

RunStatus is an alias for execution.RunStatus

type Runner 

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

Runner runs a plan of steps.

func New 

func New(cfg *Config) *Runner

func (*Runner) Cancel 

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

Cancel sends -1 signal to all nodes.

func (*Runner) DryRunChecks 

func (r *Runner) DryRunChecks(plan *Plan) []DryRunCheck

DryRunChecks returns the problems this dry run recorded for plan, in run order: the init handler, the plan's steps, then the other handlers. It is empty when every step passed.

func (*Runner) GetMetrics 

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

GetMetrics returns the current metrics for the runner

func (*Runner) HandlerNode 

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

HandlerNode returns the handler node with the given name.

func (*Runner) NodesInRunOrder 

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

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

func (*Runner) PreconditionResults added in v2.13.0 

func (r *Runner) PreconditionResults() []ir.ConditionResult

PreconditionResults returns the latest DAG-level precondition evaluation.

func (*Runner) Run 

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

Run runs the plan of steps.

func (*Runner) Signal 

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

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

func (*Runner) Status 

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

Status returns the status of the runner.

func (*Runner) Stop 

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

Stop requests workflow steps to stop according to lifecycle intent. Completion excludes lifecycle handlers, which retain their own timeouts.

type SchedulerLogStreamer 

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

SchedulerLogStreamer streams a completed scheduler log.

type StaleRunRepairConfig 

type StaleRunRepairConfig struct {
	DAGRunRepository              *persis.DAGRunRepository
	DAGRunLeaseStore              dispatch.DAGRunLeaseStore
	WorkerHeartbeatStore          dispatch.WorkerHeartbeatStore
	StaleLeaseThreshold           time.Duration
	StaleWorkerHeartbeatThreshold time.Duration
	Now                           func() time.Time
}

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

type StatusPusher 

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

StatusPusher reports DAG run status outside the current execution process.

type StepExecutor 

type StepExecutor struct{}

StepExecutor owns a single step execution attempt.

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

func NewStepExecutor 

func NewStepExecutor() *StepExecutor

NewStepExecutor creates a StepExecutor.

func (*StepExecutor) Execute 

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

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

func (*StepExecutor) ExecuteWithProgress added in v2.14.0 

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

ExecuteWithProgress runs a step and reports live executor side-channel updates.

type StepRetryPlanOptions added in v2.15.0 

type StepRetryPlanOptions struct {
	// IncludeDownstream resets the selected step and every reachable descendant.
	// Unrelated branches keep their existing status.
	IncludeDownstream bool
	// BypassPreconditions marks reset nodes to skip step precondition
	// evaluation when they execute.
	BypassPreconditions bool
}

StepRetryPlanOptions configures a targeted step retry plan.

type SubDAGRun 

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

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

Source Files

View all Source files

Directories

Path Synopsis
Package agentloop implements the decision layer of an agent DAG: the catalog of actions offered to the LLM, the goal state it works against, and the planner that turns a conversation into the next action.
 Package agentloop implements the decision layer of an agent DAG: the catalog of actions offered to the LLM, the goal state it works against, and the planner that turns a conversation into the next action.
action
agent
Package agent registers the executor identity of the synthesized step that drives an agent DAG.
 Package agent registers the executor identity of the synthesized step that drives an agent DAG.
archive
artifact
browser
Package browser implements the browser.extract and browser.run actions.
 Package browser implements the browser.extract and browser.run actions.
chat
Package chat provides an executor for chat (LLM-based session) steps.
 Package chat provides an executor for chat (LLM-based session) steps.
clickhouse
Package clickhouse provides the ClickHouse executor, which runs SQL against the ClickHouse HTTP interface.
 Package clickhouse provides the ClickHouse executor, which runs SQL against the ClickHouse HTTP interface.
command
computer
Package computer implements the computer.extract and computer.run actions, which operate the desktop of the worker's user session.
 Package computer implements the computer.extract and computer.run actions, which operate the desktop of the worker's user session.
dag
data
decision
docker
file
foreach
git
harness
http
internal/agentstep
Package agentstep holds what the built-in steps that drive a model themselves, such as browser, computer, and chat, share: with-block validation helpers, the respond tool, and lock upkeep.
 Package agentstep holds what the built-in steps that drive a model themselves, such as browser, computer, and chat, share: with-block validation helpers, the respond tool, and lock upkeep.
jq
js
Package js runs a sandboxed JavaScript function body as a workflow step.
 Package js runs a sandboxed JavaScript function body as a workflow step.
kubernetes
log
mail
noop
outputs
redis
Package redis provides Redis executor capabilities for Dagu workflows.
 Package redis provides Redis executor capabilities for Dagu workflows.
router
s3
sql
Package sql provides SQL executor capabilities for PostgreSQL and SQLite databases.
 Package sql provides SQL executor capabilities for PostgreSQL and SQLite databases.
sql/drivers/postgres
Package postgres provides the PostgreSQL driver for the SQL executor.
 Package postgres provides the PostgreSQL driver for the SQL executor.
sql/drivers/sqlite
Package sqlite provides the SQLite driver for the SQL executor.
 Package sqlite provides the SQLite driver for the SQL executor.
ssh
state
template
wait
xlsx
Package xlsx implements the xlsx.* built-in actions over internal/cmn/workbook: reading, inspecting, and writing .xlsx workbooks without a spreadsheet application.
 Package xlsx implements the xlsx.* built-in actions over internal/cmn/workbook: reading, inspecting, and writing .xlsx workbooks without a spreadsheet application.
Package modelusage keeps what a step's calls to models use while the step runs.
 Package modelusage keeps what a step's calls to models use while the step runs.
Package runstate defines the execution-state port used by the runtime.
 Package runstate defines the execution-state port used by the runtime.
memstore
Package memstore provides an in-memory runtime run-state store.
 Package memstore provides an in-memory runtime run-state store.

Jump to

Keyboard shortcuts

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