Versions in this module Expand all Collapse all v2 v2.12.0 Aug 6, 2026 Changes in this version + const EnvKeyDAGDocsDir + func DAGRunListReadBatchID(ctx context.Context) (uint64, bool) + func WithDAGRunListReadBatch(ctx context.Context, batch *DAGRunListReadBatch) context.Context type Context + func MustContext(ctx context.Context) Context + type DAGRunListReadBatch struct + func NewDAGRunListReadBatch() *DAGRunListReadBatch type ListDAGsOptions + ActiveOnly bool type Node + StatusDetails []NodeStatusDetail + type NodeStatusDetail struct + Label string + Status core.NodeStatus v2.11.4 Aug 3, 2026 v2.11.3 Aug 2, 2026 Changes in this version type RemoveOldDAGRunsOption + func WithOlderThan(t time.Time) RemoveOldDAGRunsOption type RemoveOldDAGRunsOptions + OlderThan *time.Time v2.11.2 Jul 31, 2026 Changes in this version + const DefaultStaleLeaseThreshold + const EnvKeyDAGName + const EnvKeyDAGParamsJSON + const EnvKeyDAGParamsJSONCompat + const EnvKeyDAGPushBack + const EnvKeyDAGPushBackIteration + const EnvKeyDAGPushBackPreviousStdoutFile + const EnvKeyDAGRunArtifactsDir + const EnvKeyDAGRunID + const EnvKeyDAGRunLogFile + const EnvKeyDAGRunStatus + const EnvKeyDAGRunStepName + const EnvKeyDAGRunStepStderrFile + const EnvKeyDAGRunStepStdoutFile + const EnvKeyDAGRunWorkDir + const EnvKeyDAGUOutputFile + const EnvKeyDAGWaitingSteps + const EnvKeyExternalStepRetry + const EnvKeyQueueDispatchRetry + const RoleAssistant + const RoleSystem + const RoleTool + const RoleUser + const StreamTypeStderr + const StreamTypeStdout + const WorkspaceLabelKey + var ErrActiveRunNotFound = errors.New("active distributed run not found") + var ErrCorruptedStatusFile = errors.New("corrupted status file") + var ErrDAGAlreadyExists = errors.New("DAG already exists") + var ErrDAGNotFound = errors.New("DAG is not found") + var ErrDAGRunActive = errors.New("dag-run is active") + var ErrDAGRunAlreadyExists = errors.New("dag-run already exists") + var ErrDAGRunIDNotFound = errors.New("dag-run ID not found") + var ErrDAGRunLeaseNotFound = errors.New("dag-run lease not found") + var ErrDispatchAdmissionConflict = errors.New("dispatch admission conflict") + var ErrDispatchAdmissionLivenessNotConfigured = errors.New("dispatch admission liveness not configured") + var ErrDispatchAdmissionNotFound = errors.New("dispatch admission not found") + var ErrDispatchTaskNotFound = errors.New("dispatch task claim not found") + var ErrFailedAutoRetryCancelStateChanged = errors.New("dag-run state changed before failed auto-retry cancel could be applied") + var ErrInvalidCursor = errors.New("invalid cursor") + var ErrInvalidQueryCursor = errors.New("dagrun: invalid query cursor") + var ErrInvalidRetryPath = errors.New("retry path is invalid") + var ErrInvalidRunRefFormat = errors.New("invalid dag-run reference format") + var ErrNoStatusData = errors.New("no status data") + var ErrNoopAttemptNotSupported = errors.New("operation not supported by no-op DAG run attempt") + var ErrQueueEmpty = errors.New("queue is empty") + var ErrQueueItemNotFound = errors.New("queue item not found") + var ErrRepeatingStepTarget = errors.New("child DAG runs of a repeating step cannot be retried individually") + var ErrRetryStaleLatest = errors.New("retry target is no longer the latest attempt") + var ErrRetryStepNotFound = errors.New("retry step not found") + var ErrWorkerHeartbeatNotFound = errors.New("worker heartbeat not found") + func AbortQueuedDAGRun(ctx context.Context, dagRunStore DAGRunStore, dagRun DAGRunRef) error + func AttemptKeyForStatus(status *DAGRunStatus, fallbackAttemptID string) string + func CanCancelFailedAutoRetryPendingRun(status *DAGRunStatus) bool + func CancelFailedAutoRetryPendingRun(ctx context.Context, dagRunStore latestAttemptStatusSwapper, ...) error + func DecodeSearchCursor(raw string, dest any) error + func DistributedLeaseExpiredReason(workerID string) string + func EncodeSearchCursor(payload any) string + func EnqueueRetry(ctx context.Context, dagRunStore DAGRunStore, queueStore QueueStore, ...) (bool, error) + func FilterPushBackInputs(allowed []string, inputs map[string]string) map[string]string + func FormatTime(val time.Time) string + func GenerateAttemptKey(rootName, rootID, dagName, dagRunID, attemptID string) string + func IsLeaseActive(status *DAGRunStatus, staleThreshold time.Duration) bool + func IsQueuedCatchup(status *DAGRunStatus) bool + func IsRemoteWorkerID(workerID string) bool + func LeaseIdentityMatchesStatus(lease *DAGRunLease, status *DAGRunStatus, fallbackAttemptID string) bool + func LeaseMatchesStatus(lease *DAGRunLease, status *DAGRunStatus, fallbackAttemptID string, ...) bool + func NewContext(ctx context.Context, dag *core.DAG, dagRunID string, logFile string, ...) context.Context + func NewDAGRunID() (string, error) + func NormalizeDAGRunConditions(status *DAGRunStatus) + func PreservedQueueTriggerType(status *DAGRunStatus) core.TriggerType + func ResolveRetryPath(ctx context.Context, store DAGRunStore, root DAGRunRef, targetRunID string, ...) (RetryPath, *DAGRunStatus, error) + func ValidateDAGRunID(dagRunID string) error + func WithContext(ctx context.Context, rCtx Context) context.Context + func WorkspaceNameFromLabels(labels core.Labels) (string, bool) + type ActiveDistributedRun struct + AttemptID string + AttemptKey string + DAGRun DAGRunRef + Root DAGRunRef + Status core.Status + UpdatedAt int64 + WorkerID string + type ActiveDistributedRunStore interface + Delete func(ctx context.Context, attemptKey string) error + Get func(ctx context.Context, attemptKey string) (*ActiveDistributedRun, error) + ListAll func(ctx context.Context) ([]ActiveDistributedRun, error) + Upsert func(ctx context.Context, record ActiveDistributedRun) error + type ClaimedDispatchTask struct + ClaimToken string + ClaimedAt time.Time + Owner CoordinatorEndpoint + PollerID string + Task *DispatchTask + WorkerID string + type CompareAndSwapStatusOption func(*CompareAndSwapStatusOptions) + func WithCompareAndSwapExpectedAttemptKey(attemptKey string) CompareAndSwapStatusOption + func WithCompareAndSwapRootDAGRun(root DAGRunRef) CompareAndSwapStatusOption + type CompareAndSwapStatusOptions struct + ExpectedAttemptKey string + RootDAGRun DAGRunRef + func NewCompareAndSwapStatusOptions(opts ...CompareAndSwapStatusOption) CompareAndSwapStatusOptions + type Context struct + AttemptID string + BaseEnv *config.BaseEnv + CoordinatorCli Dispatcher + DAG *core.DAG + DAGRunArtifactDir string + DAGRunID string + DAGRunLogDir string + DAGRunStore DAGRunStore + DB Database + DefaultExecMode config.ExecutionMode + EnvScope *cmnvalue.EnvScope + LogEncodingCharset string + LogWriterFactory LogWriterFactory + ProfileEntries []RuntimeProfileEntry + ProfileName string + ProfileResolvedAt string + QueueStore QueueStore + RetryPath RetryPath + RootDAGRun DAGRunRef + RunStartedAt string + ScheduleTime string + Shell string + StateStore dagstate.Store + TriggerActor string + TriggerType core.TriggerType + func GetContext(ctx context.Context) Context + func LookupContext(ctx context.Context) (Context, bool) + func (e Context) AllEnvs() []string + func (e Context) DAGRunRef() DAGRunRef + func (e Context) UserEnvsMap() map[string]string + type ContextOption func(*contextOptions) + func WithArtifactDir(dir string) ContextOption + func WithAttemptID(attemptID string) ContextOption + func WithCoordinator(cli Dispatcher) ContextOption + func WithDAGRunArtifactDir(dir string) ContextOption + func WithDAGRunLogDir(dir string) ContextOption + func WithDAGRunStore(store DAGRunStore) ContextOption + func WithDatabase(db Database) ContextOption + func WithDefaultEnvVars(envs ...string) ContextOption + func WithDefaultExecMode(mode config.ExecutionMode) ContextOption + func WithDefaultSecrets(secrets []string) ContextOption + func WithEnvVars(envs ...string) ContextOption + func WithLogEncoding(charset string) ContextOption + func WithLogWriterFactory(factory LogWriterFactory) ContextOption + func WithParams(params []string) ContextOption + func WithQueueStore(store QueueStore) ContextOption + func WithRetryPath(path RetryPath) ContextOption + func WithRootDAGRun(ref DAGRunRef) ContextOption + func WithRunStartedAt(startedAt string) ContextOption + func WithRuntimeProfile(name, resolvedAt string, entries []RuntimeProfileEntry) ContextOption + func WithScheduleTime(scheduleTime string) ContextOption + func WithSecrets(secrets []string) ContextOption + func WithStateStore(store dagstate.Store) ContextOption + func WithTriggerActor(actor string) ContextOption + func WithTriggerType(triggerType core.TriggerType) ContextOption + func WithWorkDir(dir string) ContextOption + type CoordinatorEndpoint struct + Host string + ID string + Port int + func CoordinatorEndpointFromHostInfo(info HostInfo) CoordinatorEndpoint + func (e CoordinatorEndpoint) HostInfo() HostInfo + type CursorResult struct + HasMore bool + Items []T + NextCursor string + type DAGRunAttempt interface + Abort func(ctx context.Context) error + Close func(ctx context.Context) error + Hidden func() bool + Hide func(ctx context.Context) error + ID func() string + IsAborting func(ctx context.Context) (bool, error) + Open func(ctx context.Context) error + ReadDAG func(ctx context.Context) (*core.DAG, error) + ReadOutputs func(ctx context.Context) (*DAGRunOutputs, error) + ReadStatus func(ctx context.Context) (*DAGRunStatus, error) + ReadStepMessages func(ctx context.Context, stepName string) ([]LLMMessage, error) + SetDAG func(dag *core.DAG) + WorkDir func() string + Write func(ctx context.Context, status DAGRunStatus) error + WriteOutputs func(ctx context.Context, outputs *DAGRunOutputs) error + WriteStepMessages func(ctx context.Context, stepName string, messages []LLMMessage) error + func NewNoopDAGRunAttempt(id string, dag *core.DAG) DAGRunAttempt + type DAGRunCondition struct + CheckedAt string + Message string + Reason string + Status string + Type string + func MergeDAGRunConditions(conditions []DAGRunCondition, observations ...DAGRunCondition) []DAGRunCondition + func NewDAGRunCondition(conditionType, status, reason, message string, checkedAt time.Time) DAGRunCondition + func UpsertDAGRunCondition(conditions []DAGRunCondition, condition DAGRunCondition) []DAGRunCondition + type DAGRunLease struct + AttemptID string + AttemptKey string + ClaimedAt int64 + DAGRun DAGRunRef + LastHeartbeatAt int64 + Owner CoordinatorEndpoint + QueueName string + Root DAGRunRef + WorkerID string + func (l *DAGRunLease) MatchesClaim(claimKey, workerID string) bool + func (l DAGRunLease) ClaimedTime() time.Time + func (l DAGRunLease) IsFresh(now time.Time, staleThreshold time.Duration) bool + func (l DAGRunLease) LastHeartbeatTime() time.Time + type DAGRunLeaseStore interface + Delete func(ctx context.Context, attemptKey string) error + Get func(ctx context.Context, attemptKey string) (*DAGRunLease, error) + ListAll func(ctx context.Context) ([]DAGRunLease, error) + ListByQueue func(ctx context.Context, queueName string) ([]DAGRunLease, error) + Touch func(ctx context.Context, attemptKey string, observedAt time.Time) error + Upsert func(ctx context.Context, lease DAGRunLease) error + type DAGRunNotQueuedError struct + HasStatus bool + Status core.Status + func (e *DAGRunNotQueuedError) Error() string + type DAGRunOutputs struct + Metadata OutputsMetadata + Outputs map[string]string + type DAGRunRef struct + ID string + Name string + func NewDAGRunRef(name, runID string) DAGRunRef + func ParseDAGRunRef(s string) (DAGRunRef, error) + func (e DAGRunRef) String() string + func (e DAGRunRef) Zero() bool + type DAGRunStatus struct + ArchiveDir string + AttemptID string + AttemptKey string + AutoRetryBackoff float64 + AutoRetryCount int + AutoRetryInterval time.Duration + AutoRetryLimit int + AutoRetryMaxInterval time.Duration + ClaimKey string + Conditions []DAGRunCondition + CreatedAt int64 + DAGRunID string + Error string + FinishedAt string + Labels []string + LeaseAt int64 + Log string + Name string + Nodes []*Node + OnAbort *Node + OnExit *Node + OnFailure *Node + OnInit *Node + OnSuccess *Node + OnWait *Node + PID PID + PIDStartedAt int64 + Params string + ParamsList []string + Parent DAGRunRef + PendingStepRetries []PendingStepRetry + Preconditions []*core.Condition + ProcGroup string + ProfileEntries []RuntimeProfileEntry + ProfileName string + ProfileResolvedAt string + QueuedAt string + Root DAGRunRef + ScheduleTime string + StartedAt string + Status core.Status + SuspendFlagName string + TriggerActor string + TriggerType core.TriggerType + WorkerID string + WorkingDir string + func InitialStatus(dag *core.DAG) DAGRunStatus + func StatusFromJSON(s string) (*DAGRunStatus, error) + func (s DAGRunStatus) EffectiveClaimKey() string + func (s DAGRunStatus) Tags() []string + func (st *DAGRunStatus) DAGRun() DAGRunRef + func (st *DAGRunStatus) Errors() []error + func (st *DAGRunStatus) NodeByName(name string) (*Node, error) + func (st *DAGRunStatus) NodesInRunOrder() []*Node + func (st *DAGRunStatus) UnmarshalJSON(data []byte) error + type DAGRunStatusPage struct + Items []*DAGRunStatus + NextCursor string + type DAGRunStatusResult struct + Found bool + Status *DAGRunStatus + type DAGRunStore interface + CompareAndSwapLatestAttemptStatus func(ctx context.Context, dagRun DAGRunRef, expectedAttemptID string, ...) (*DAGRunStatus, bool, error) + CreateAttempt func(ctx context.Context, dag *core.DAG, ts time.Time, dagRunID string, ...) (DAGRunAttempt, error) + CreateSubAttempt func(ctx context.Context, rootRef DAGRunRef, subDAGRunID string) (DAGRunAttempt, error) + FindAttempt func(ctx context.Context, dagRun DAGRunRef) (DAGRunAttempt, error) + FindSubAttempt func(ctx context.Context, dagRun DAGRunRef, subDAGRunID string) (DAGRunAttempt, error) + LatestAttempt func(ctx context.Context, name string) (DAGRunAttempt, error) + ListStatuses func(ctx context.Context, opts ...ListDAGRunStatusesOption) ([]*DAGRunStatus, error) + ListStatusesPage func(ctx context.Context, opts ...ListDAGRunStatusesOption) (DAGRunStatusPage, error) + RecentAttempts func(ctx context.Context, name string, itemLimit int) []DAGRunAttempt + RemoveDAGRun func(ctx context.Context, dagRun DAGRunRef, opts ...RemoveDAGRunOption) error + RemoveOldDAGRuns func(ctx context.Context, name string, retentionDays int, ...) ([]string, error) + RenameDAGRuns func(ctx context.Context, oldName, newName string) error + type DAGStore interface + Create func(ctx context.Context, fileName string, spec []byte) error + Delete func(ctx context.Context, fileName string) error + GetDetails func(ctx context.Context, fileName string, opts ...spec.LoadOption) (*core.DAG, error) + GetMetadata func(ctx context.Context, fileName string) (*core.DAG, error) + GetSpec func(ctx context.Context, fileName string) (string, error) + Grep func(ctx context.Context, pattern string) (ret []*GrepDAGsResult, errs []string, err error) + IsSuspended func(ctx context.Context, fileName string) bool + LabelList func(ctx context.Context) ([]string, []string, error) + List func(ctx context.Context, params ListDAGsOptions) (PaginatedResult[*core.DAG], []string, error) + LoadSpec func(ctx context.Context, spec []byte, opts ...spec.LoadOption) (*core.DAG, error) + Rename func(ctx context.Context, oldID, newID string) error + SearchCursor func(ctx context.Context, opts SearchDAGsOptions) (*CursorResult[SearchDAGResult], []string, error) + SearchMatches func(ctx context.Context, fileName string, opts SearchDAGMatchesOptions) (*CursorResult[*Match], error) + ToggleSuspend func(ctx context.Context, fileName string, suspend bool) error + UpdateSpec func(ctx context.Context, fileName string, spec []byte) error + type Database interface + GetDAG func(ctx context.Context, name string) (*core.DAG, error) + GetSubDAGRunStatus func(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) (*RunStatus, error) + IsSubDAGRunCompleted func(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) (bool, error) + RequestChildCancel func(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) error + type DispatchAdmissionBindRequest struct + ReservationToken string + Task *DispatchTask + type DispatchAdmissionDecision struct + Reason DispatchAdmissionRejectReason + ReservationToken string + Reserved bool + type DispatchAdmissionRejectReason string + const DispatchAdmissionRejectedDuplicate + const DispatchAdmissionRejectedNoCapacity + type DispatchAdmissionRequest struct + AttemptID string + AttemptKey string + DAGRun DAGRunRef + MaxConcurrency int + NonAdmissionOccupancy int + QueueName string + StaleThreshold time.Duration + type DispatchAdmissionStore interface + BindAdmission func(ctx context.Context, req DispatchAdmissionBindRequest) error + CleanupAdmissions func(ctx context.Context, staleThreshold time.Duration) error + FinalizeAdmissionAttempt func(ctx context.Context, attemptKey string) error + ReleaseAdmissionToken func(ctx context.Context, reservationToken string) error + ReserveAdmission func(ctx context.Context, req DispatchAdmissionRequest) (*DispatchAdmissionDecision, error) + type DispatchOperation int32 + const DispatchOperationRetry + const DispatchOperationStart + const DispatchOperationUnspecified + func (o DispatchOperation) String() string + type DispatchRequest struct + AdmissionReservationToken string + Task *DispatchTask + type DispatchTask struct + AttemptID string + AttemptKey string + BaseConfig string + ClaimToken string + DAGRunID string + Definition string + ExternalStepRetry bool + Labels string + Operation DispatchOperation + Owner CoordinatorEndpoint + Params string + ParentDAGRunID string + ParentDAGRunName string + PreviousStatus *DAGRunStatus + ProfileName string + QueueName string + RetryPath string + RootDAGRunID string + RootDAGRunName string + ScheduleTime string + SourceFile string + Step string + Target string + TriggerActor string + WorkerID string + WorkerSelector map[string]string + WorkspaceBundleDAGPath string + WorkspaceBundleDigest string + WorkspaceBundleOriginalRef string + WorkspaceBundleResolvedRef string + WorkspaceBundleSize int64 + type DispatchTaskClaim struct + ClaimTimeout time.Duration + Labels map[string]string + Owner CoordinatorEndpoint + PollerID string + WorkerID string + type DispatchTaskStore interface + ClaimNext func(ctx context.Context, claim DispatchTaskClaim) (*ClaimedDispatchTask, error) + CountOutstandingByQueue func(ctx context.Context, queueName string, claimTimeout time.Duration) (int, error) + DeleteClaim func(ctx context.Context, claimToken string) error + Enqueue func(ctx context.Context, task *DispatchTask) error + GetClaim func(ctx context.Context, claimToken string) (*ClaimedDispatchTask, error) + HasOutstandingAttempt func(ctx context.Context, attemptKey string, claimTimeout time.Duration) (bool, error) + ReleaseClaim func(ctx context.Context, claimToken string) error + type Dispatcher interface + Cleanup func(ctx context.Context) error + Dispatch func(ctx context.Context, req DispatchRequest) error + GetDAGRunStatus func(ctx context.Context, dagName, dagRunID string, rootRef *DAGRunRef) (*DAGRunStatusResult, error) + RequestCancel func(ctx context.Context, dagName, dagRunID string, rootRef *DAGRunRef) error + type EnqueueRetryOptions struct + AutoRetry bool + TriggerActor *string + type FailedAutoRetryCancelEligibility int + const FailedAutoRetryCancelEligible + const FailedAutoRetryCancelMissingStatus + const FailedAutoRetryCancelNotPending + const FailedAutoRetryCancelNotRoot + func FailedAutoRetryCancelEligibilityOf(status *DAGRunStatus) FailedAutoRetryCancelEligibility + type FailedAutoRetryCancelStateChangedError struct + CurrentStatus *DAGRunStatus + func (e *FailedAutoRetryCancelStateChangedError) Error() string + func (e *FailedAutoRetryCancelStateChangedError) Unwrap() error + type GrepDAGsResult struct + DAG *core.DAG + Matches []*Match + Name string + type HostInfo struct + Host string + ID string + Port int + StartedAt time.Time + Status ServiceStatus + type LLMMessage struct + Content string + Metadata *LLMMessageMetadata + Role core.LLMRole + ToolCallID string + ToolCalls []ToolCall + func DeduplicateSystemMessages(messages []LLMMessage) []LLMMessage + type LLMMessageMetadata struct + CompletionTokens int + Cost float64 + Model string + PromptTokens int + Provider string + TotalTokens int + type ListDAGRunStatusesOption func(*ListDAGRunStatusesOptions) + func WithAllHistory() ListDAGRunStatusesOption + func WithCursor(cursor string) ListDAGRunStatusesOption + func WithDAGRunID(dagRunID string) ListDAGRunStatusesOption + func WithExactName(name string) ListDAGRunStatusesOption + func WithFrom(from TimeInUTC) ListDAGRunStatusesOption + func WithLabels(labels []string) ListDAGRunStatusesOption + func WithLimit(limit int) ListDAGRunStatusesOption + func WithName(name string) ListDAGRunStatusesOption + func WithStatuses(statuses []core.Status) ListDAGRunStatusesOption + func WithTags(tags []string) ListDAGRunStatusesOption + func WithTo(to TimeInUTC) ListDAGRunStatusesOption + func WithWorkspaceFilter(filter *WorkspaceFilter) ListDAGRunStatusesOption + func WithoutLimit() ListDAGRunStatusesOption + type ListDAGRunStatusesOptions struct + AllHistory bool + Cursor string + DAGRunID string + ExactName string + From TimeInUTC + Labels []string + Limit int + Name string + Statuses []core.Status + To TimeInUTC + Unlimited bool + WorkspaceFilter *WorkspaceFilter + type ListDAGsOptions struct + Labels []string + Name string + NextRunProjection func(*core.DAG, time.Time) time.Time + Order string + Paginator *Paginator + Sort string + Time *time.Time + WorkspaceFilter *WorkspaceFilter + type ListDAGsResult struct + Count int + DAGs []*core.DAG + Errors []string + type LogWriterFactory interface + NewStepWriter func(ctx context.Context, stepName string, streamType int) io.WriteCloser + type Match struct + Line string + LineNumber int + StartLine int + type MockDAGRunAttempt struct + Status *DAGRunStatus + func (m *MockDAGRunAttempt) Abort(ctx context.Context) error + func (m *MockDAGRunAttempt) Close(ctx context.Context) error + func (m *MockDAGRunAttempt) Hidden() bool + func (m *MockDAGRunAttempt) Hide(ctx context.Context) error + func (m *MockDAGRunAttempt) ID() string + func (m *MockDAGRunAttempt) IsAborting(ctx context.Context) (bool, error) + func (m *MockDAGRunAttempt) Open(ctx context.Context) error + func (m *MockDAGRunAttempt) ReadDAG(ctx context.Context) (*core.DAG, error) + func (m *MockDAGRunAttempt) ReadOutputs(ctx context.Context) (*DAGRunOutputs, error) + func (m *MockDAGRunAttempt) ReadStatus(ctx context.Context) (*DAGRunStatus, error) + func (m *MockDAGRunAttempt) ReadStepMessages(ctx context.Context, stepName string) ([]LLMMessage, error) + func (m *MockDAGRunAttempt) SetDAG(dag *core.DAG) + func (m *MockDAGRunAttempt) WorkDir() string + func (m *MockDAGRunAttempt) Write(ctx context.Context, status DAGRunStatus) error + func (m *MockDAGRunAttempt) WriteOutputs(ctx context.Context, outputs *DAGRunOutputs) error + func (m *MockDAGRunAttempt) WriteStepMessages(ctx context.Context, stepName string, messages []LLMMessage) error + type MockQueueStore struct + func (m *MockQueueStore) All(ctx context.Context) ([]QueuedItemData, error) + func (m *MockQueueStore) DeleteByItemIDs(ctx context.Context, name string, itemIDs []string) (int, error) + func (m *MockQueueStore) DequeueByDAGRunID(ctx context.Context, name string, dagRun DAGRunRef) ([]QueuedItemData, error) + func (m *MockQueueStore) DequeueByName(ctx context.Context, name string) (QueuedItemData, error) + func (m *MockQueueStore) Enqueue(ctx context.Context, name string, priority QueuePriority, dagRun DAGRunRef) error + func (m *MockQueueStore) Len(ctx context.Context, name string) (int, error) + func (m *MockQueueStore) List(ctx context.Context, name string) ([]QueuedItemData, error) + func (m *MockQueueStore) ListByDAGName(ctx context.Context, name, dagName string) ([]QueuedItemData, error) + func (m *MockQueueStore) ListCursor(ctx context.Context, name, cursor string, limit int) (CursorResult[QueuedItemData], error) + func (m *MockQueueStore) QueueList(ctx context.Context) ([]string, error) + func (m *MockQueueStore) QueueWatcher(ctx context.Context) QueueWatcher + type NewDAGRunAttemptOptions struct + AttemptID string + Retry bool + RootDAGRun *DAGRunRef + type Node struct + ApprovalInputs map[string]string + ApprovalIteration int + ApprovedAt string + ApprovedBy string + ApprovedByID string + ChatMessages []LLMMessage + ControllerState json.RawMessage + DoneCount int + Error string + FinishedAt string + HumanTaskCompletedBy string + HumanTaskCompletedByID string + HumanTaskInput json.RawMessage + OutputValue *string + OutputVariables *collections.SyncMap + OutputsValue *string + PushBackHistory []PushBackEntry + PushBackInputs map[string]string + PushBackPreviousStdout string + RejectedAt string + RejectedBy string + RejectedByID string + RejectionReason string + Repeated bool + RetriedAt string + RetryCount int + SkippedByRetry bool + StartedAt string + Status core.NodeStatus + Stderr string + Stdout string + Step core.Step + StepOutputsValue *string + SubRuns []SubDAGRun + SubRunsRepeated []SubDAGRun + ToolDefinitions []ToolDefinition + WorkingDir string + func NewNodeFromStep(step core.Step) *Node + func NewNodeOrNil(s *core.Step) *Node + func NewNodesFromSteps(steps []core.Step) []*Node + type OutputsMetadata struct + AttemptID string + CompletedAt string + DAGName string + DAGRunID string + Params string + Status string + type PID int + func (p PID) String() string + type PageRange struct + Range []int + SkipFirst bool + SkipLast bool + type PaginatedResult struct + CurrentPage int + HasNextPage bool + HasPrevPage bool + Items []T + NextPage int + Offset int + PrevPage int + TotalCount int + TotalPages int + func NewPaginatedResult[T any](items []T, total int, pg Paginator) PaginatedResult[T] + func (r PaginatedResult[T]) Data() []T + func (r PaginatedResult[T]) PageRange(size int) PageRange + func (r PaginatedResult[T]) RangeEnd() int + func (r PaginatedResult[T]) RangeStart() int + type Paginator struct + func DefaultPaginator() Paginator + func NewPaginator(page, perPage int) Paginator + func (pg *Paginator) Limit() int + func (pg *Paginator) Offset() int + type PendingStepRetry struct + Interval time.Duration + StepName string + func PendingStepRetriesFromNodes(nodes []*Node) []PendingStepRetry + func PendingStepRetriesFromStatus(status *DAGRunStatus) []PendingStepRetry + func (p *PendingStepRetry) UnmarshalJSON(data []byte) error + func (p PendingStepRetry) MarshalJSON() ([]byte, error) + type ProcEntry struct + Fresh bool + GroupName string + Identity ProcEntryID + LastHeartbeatAt int64 + Meta ProcMeta + func (e ProcEntry) AttemptKey() string + func (e ProcEntry) DAGRun() DAGRunRef + func (e ProcEntry) IsRoot() bool + func (e ProcEntry) RunScopeKey() string + type ProcEntryID struct + func NewProcEntryID(token string) ProcEntryID + func (id ProcEntryID) IsZero() bool + func (id ProcEntryID) String() string + type ProcHandle interface + GetMeta func() ProcMeta + Stop func(ctx context.Context) error + type ProcHeartbeat struct + AttemptID string + DAGRun DAGRunRef + Fresh bool + GroupName string + LastHeartbeatAt int64 + ObservedAt time.Time + StartedAt int64 + func (h ProcHeartbeat) AdvancedSince(previous ProcHeartbeat) bool + type ProcMeta struct + AttemptID string + DAGRunID string + Name string + RootDAGRunID string + RootName string + StartedAt int64 + func (m ProcMeta) DAGRun() DAGRunRef + func (m ProcMeta) Root() DAGRunRef + type ProcStore interface + Acquire func(ctx context.Context, groupName string, meta ProcMeta) (ProcHandle, error) + CountAlive func(ctx context.Context, groupName string) (int, error) + CountAliveByDAGName func(ctx context.Context, groupName, dagName string) (int, error) + IsAttemptAlive func(ctx context.Context, groupName string, dagRun DAGRunRef, attemptID string) (bool, error) + IsRunAlive func(ctx context.Context, groupName string, dagRun DAGRunRef) (bool, error) + LatestFreshEntryByDAGName func(ctx context.Context, groupName, dagName string) (*ProcEntry, error) + LatestHeartbeat func(ctx context.Context, groupName string, dagRun DAGRunRef) (*ProcHeartbeat, error) + ListAlive func(ctx context.Context, groupName string) ([]DAGRunRef, error) + ListAllAlive func(ctx context.Context) (map[string][]DAGRunRef, error) + ListAllEntries func(ctx context.Context) ([]ProcEntry, error) + ListEntries func(ctx context.Context, groupName string) ([]ProcEntry, error) + Lock func(ctx context.Context, groupName string) error + RemoveIfStale func(ctx context.Context, entry ProcEntry) error + Unlock func(ctx context.Context, groupName string) + type PushBackEntry struct + At string + By string + ByID string + Inputs map[string]string + Iteration int + func ClonePushBackHistory(src []PushBackEntry) []PushBackEntry + func NormalizePushBackHistory(allowed []string, iteration int, latestInputs map[string]string, ...) []PushBackEntry + type QueuePriority int + const QueuePriorityHigh + const QueuePriorityLow + type QueueStore interface + All func(ctx context.Context) ([]QueuedItemData, error) + DeleteByItemIDs func(ctx context.Context, name string, itemIDs []string) (int, error) + DequeueByDAGRunID func(ctx context.Context, name string, dagRun DAGRunRef) ([]QueuedItemData, error) + DequeueByName func(ctx context.Context, name string) (QueuedItemData, error) + Enqueue func(ctx context.Context, name string, priority QueuePriority, dagRun DAGRunRef) error + Len func(ctx context.Context, name string) (int, error) + List func(ctx context.Context, name string) ([]QueuedItemData, error) + ListByDAGName func(ctx context.Context, name, dagName string) ([]QueuedItemData, error) + ListCursor func(ctx context.Context, name, cursor string, limit int) (CursorResult[QueuedItemData], error) + QueueList func(ctx context.Context) ([]string, error) + QueueWatcher func(ctx context.Context) QueueWatcher + type QueueWatcher interface + Start func(ctx context.Context) (<-chan struct{}, error) + Stop func(ctx context.Context) + type QueuedItem struct + func NewQueuedItem(data QueuedItemData) *QueuedItem + type QueuedItemData interface + Data func() (*DAGRunRef, error) + ID func() string + type RemoveDAGRunOption func(*RemoveDAGRunOptions) + func WithRejectActiveDAGRun() RemoveDAGRunOption + type RemoveDAGRunOptions struct + RejectActive bool + type RemoveOldDAGRunsOption func(*RemoveOldDAGRunsOptions) + func WithDryRun() RemoveOldDAGRunsOption + func WithRetentionRuns(runs int) RemoveOldDAGRunsOption + type RemoveOldDAGRunsOptions struct + DryRun bool + RetentionRuns *int + type RetryHop struct + RunID string + Step string + type RetryPath struct + Hops []RetryHop + Step string + func ParseRetryPath(value string) (RetryPath, error) + func (p RetryPath) Advance() RetryPath + func (p RetryPath) Current() (RetryHop, bool) + func (p RetryPath) Encode() string + func (p RetryPath) NextStep() string + func (p RetryPath) RootStep() string + type RunStatus struct + DAGRunID string + Name string + OutputValues map[string]any + Outputs map[string]string + Params string + PendingStepRetries []PendingStepRetry + Status core.Status + func (r *RunStatus) MarshalJSON() ([]byte, error) + type RunningTask struct + AttemptKey string + DAGName string + DAGRunID string + ParentDAGRunID string + ParentDAGRunName string + RootDAGRunID string + RootDAGRunName string + StartedAt int64 + type RuntimeProfileEntry struct + Key string + Kind string + type SearchDAGMatchesOptions struct + Cursor string + Labels []string + Limit int + Query string + WorkspaceFilter *WorkspaceFilter + type SearchDAGResult struct + FileName string + HasMoreMatches bool + Matches []*Match + Name string + NextMatchesCursor string + Workspace string + type SearchDAGsOptions struct + Cursor string + Labels []string + Limit int + MatchLimit int + Query string + WorkspaceFilter *WorkspaceFilter + type ServiceName string + const ServiceNameCoordinator + const ServiceNameScheduler + type ServiceRegistry interface + GetServiceMembers func(ctx context.Context, serviceName ServiceName) ([]HostInfo, error) + Register func(ctx context.Context, serviceName ServiceName, hostInfo HostInfo) error + Unregister func(ctx context.Context) + UpdateStatus func(ctx context.Context, serviceName ServiceName, status ServiceStatus) error + type ServiceStatus int + const ServiceStatusActive + const ServiceStatusInactive + const ServiceStatusUnknown + func (s ServiceStatus) String() string + type StaleQueueDispatchError struct + Reason string + func ParseStaleQueueDispatchError(msg string) (*StaleQueueDispatchError, bool) + func (e *StaleQueueDispatchError) Error() string + type SubDAGRun struct + DAGName string + DAGRunID string + Params string + type TimeInUTC struct + func NewUTC(t time.Time) TimeInUTC + type ToolCall struct + Function ToolCallFunction + ID string + Type string + type ToolCallFunction struct + Arguments string + Name string + type ToolDefinition struct + Description string + Name string + Parameters map[string]any + type WorkerHeartbeatRecord struct + Labels map[string]string + LastHeartbeatAt int64 + Stats *WorkerStats + WorkerID string + func (r WorkerHeartbeatRecord) LastHeartbeatTime() time.Time + type WorkerHeartbeatStore interface + DeleteStale func(ctx context.Context, before time.Time) (int, error) + Get func(ctx context.Context, workerID string) (*WorkerHeartbeatRecord, error) + List func(ctx context.Context) ([]WorkerHeartbeatRecord, error) + Upsert func(ctx context.Context, record WorkerHeartbeatRecord) error + type WorkerStats struct + BusyPollers int32 + RunningTasks []*RunningTask + TotalPollers int32 + type WorkspaceFilter struct + Enabled bool + IncludeUnlabelled bool + Workspaces []string + func (f *WorkspaceFilter) MatchesLabels(labels core.Labels) bool + type WorkspaceLabelState int + const WorkspaceLabelInvalid + const WorkspaceLabelMissing + const WorkspaceLabelValid + func WorkspaceLabelFromLabels(labels core.Labels) (string, WorkspaceLabelState)
Go to main page