exec

type ActiveDistributedRun struct {
	AttemptKey string      `json:"attemptKey"`
	DAGRun     DAGRunRef   `json:"dagRun"`
	Root       DAGRunRef   `json:"root,omitzero"`
	AttemptID  string      `json:"attemptId"`
	WorkerID   string      `json:"workerId"`
	Status     core.Status `json:"status"`
	UpdatedAt  int64       `json:"updatedAt"`
}

type ActiveDistributedRunStore interface {
	Upsert(ctx context.Context, record ActiveDistributedRun) error
	Delete(ctx context.Context, attemptKey string) error
	Get(ctx context.Context, attemptKey string) (*ActiveDistributedRun, error)
	ListAll(ctx context.Context) ([]ActiveDistributedRun, error)
}

type ClaimedDispatchTask struct {
	Task       *DispatchTask
	ClaimToken string
	ClaimedAt  time.Time
	WorkerID   string
	PollerID   string
	Owner      CoordinatorEndpoint
}

type CompareAndSwapStatusOption func(*CompareAndSwapStatusOptions)

type CompareAndSwapStatusOptions struct {
	RootDAGRun         DAGRunRef
	ExpectedAttemptKey string
}

type Context struct {
	DAGRunID           string
	RootDAGRun         DAGRunRef
	RetryPath          RetryPath
	AttemptID          string
	TriggerType        core.TriggerType
	TriggerActor       string
	RunStartedAt       string
	ScheduleTime       string
	DAG                *core.DAG
	DB                 Database
	BaseEnv            *config.BaseEnv
	EnvScope           *cmnvalue.EnvScope // Unified environment scope for runtime variables
	CoordinatorCli     Dispatcher
	DAGRunStore        DAGRunStore
	QueueStore         QueueStore
	StateStore         dagstate.Store
	DAGRunLogDir       string
	DAGRunArtifactDir  string
	ProfileName        string
	ProfileResolvedAt  string
	ProfileEntries     []RuntimeProfileEntry
	Shell              string               // Default shell for this DAG (from DAG.Shell)
	LogEncodingCharset string               // Character encoding for log files (e.g., "utf-8", "shift_jis", "euc-jp")
	LogWriterFactory   LogWriterFactory     // For remote log streaming (nil = use local files)
	DefaultExecMode    config.ExecutionMode // Server-level default execution mode (local or distributed)
}

type ContextOption func(*contextOptions)

type CoordinatorEndpoint struct {
	ID   string `json:"id,omitempty"`
	Host string `json:"host,omitempty"`
	Port int    `json:"port,omitempty"`
}

type CursorResult[T any] struct {
	Items      []T
	HasMore    bool
	NextCursor string
}

type DAGRunAttempt interface {
	// ID returns the identifier for the attempt that is unique within the dag-run.
	ID() string
	// Open prepares the attempt for writing status updates
	Open(ctx context.Context) error
	// Write updates the status of the attempt
	Write(ctx context.Context, status DAGRunStatus) error
	// Close finalizes writing to the attempt
	Close(ctx context.Context) error
	// ReadStatus retrieves the current status of the attempt
	ReadStatus(ctx context.Context) (*DAGRunStatus, error)
	// ReadDAG reads the DAG associated with this run attempt
	ReadDAG(ctx context.Context) (*core.DAG, error)
	// SetDAG sets the DAG for this attempt (must be called before Open for DAG to be persisted)
	SetDAG(dag *core.DAG)
	// Abort requests aborting the attempt
	Abort(ctx context.Context) error
	// IsAborting checks if an abort has been requested for the attempt
	IsAborting(ctx context.Context) (bool, error)
	// Hide marks the attempt as hidden from normal operations.
	// This is useful for preserving previous state visibility when dequeuing.
	Hide(ctx context.Context) error
	// Hidden returns true if the attempt is hidden from normal operations.
	Hidden() bool
	// WriteOutputs writes the collected step outputs for the dag-run.
	// Does nothing if outputs is nil or has no output entries.
	WriteOutputs(ctx context.Context, outputs *DAGRunOutputs) error
	// ReadOutputs reads the collected step outputs for the dag-run.
	// Returns nil if no outputs file exists or if the file is in v1 format.
	ReadOutputs(ctx context.Context) (*DAGRunOutputs, error)
	// WriteStepMessages writes LLM messages for a single step.
	WriteStepMessages(ctx context.Context, stepName string, messages []LLMMessage) error
	// ReadStepMessages reads LLM messages for a single step.
	// Returns nil if no messages exist for the step.
	ReadStepMessages(ctx context.Context, stepName string) ([]LLMMessage, error)
	// WorkDir returns the path to the per-DAG-run working directory.
	// Returns "" if the attempt does not support local storage.
	WorkDir() string
}

type DAGRunCondition struct {
	Type      string `json:"type"`
	Status    string `json:"status"`
	Reason    string `json:"reason"`
	Message   string `json:"message"`
	CheckedAt string `json:"checkedAt"`
}

type DAGRunLease struct {
	AttemptKey      string              `json:"attemptKey"`
	DAGRun          DAGRunRef           `json:"dagRun"`
	Root            DAGRunRef           `json:"root,omitzero"`
	AttemptID       string              `json:"attemptId"`
	QueueName       string              `json:"queueName"`
	WorkerID        string              `json:"workerId"`
	Owner           CoordinatorEndpoint `json:"owner"`
	ClaimedAt       int64               `json:"claimedAt"`
	LastHeartbeatAt int64               `json:"lastHeartbeatAt"`
}

type DAGRunLeaseStore interface {
	Upsert(ctx context.Context, lease DAGRunLease) error
	Touch(ctx context.Context, attemptKey string, observedAt time.Time) error
	Delete(ctx context.Context, attemptKey string) error
	Get(ctx context.Context, attemptKey string) (*DAGRunLease, error)
	ListByQueue(ctx context.Context, queueName string) ([]DAGRunLease, error)
	ListAll(ctx context.Context) ([]DAGRunLease, error)
}

type DAGRunNotQueuedError struct {
	Status    core.Status
	HasStatus bool
}

type DAGRunOutputs struct {
	Metadata OutputsMetadata   `json:"metadata"`
	Outputs  map[string]string `json:"outputs"`
}

type DAGRunRef struct {
	Name string `json:"name,omitempty"`
	ID   string `json:"id,omitempty"`
}

type DAGRunStatus struct {
	Root           DAGRunRef         `json:"root,omitzero"`
	Parent         DAGRunRef         `json:"parent,omitzero"`
	Name           string            `json:"name"`
	DAGRunID       string            `json:"dagRunId"`
	AttemptID      string            `json:"attemptId"`
	AttemptKey     string            `json:"attemptKey,omitempty"` // Globally unique attempt identifier
	ClaimKey       string            `json:"claimKey,omitempty"`   // Worker claim that executes this attempt
	Status         core.Status       `json:"status"`
	Conditions     []DAGRunCondition `json:"conditions,omitempty"`
	TriggerType    core.TriggerType  `json:"triggerType,omitempty"`
	TriggerActor   string            `json:"triggerActor,omitempty"`
	WorkerID       string            `json:"workerId,omitempty"`
	PID            PID               `json:"pid,omitempty"`
	PIDStartedAt   int64             `json:"pidStartedAt,omitempty"`
	Nodes          []*Node           `json:"nodes,omitempty"`
	OnInit         *Node             `json:"onInit,omitempty"`
	OnExit         *Node             `json:"onExit,omitempty"`
	OnSuccess      *Node             `json:"onSuccess,omitempty"`
	OnFailure      *Node             `json:"onFailure,omitempty"`
	OnAbort        *Node             `json:"onAbort,omitempty"`
	OnWait         *Node             `json:"onWait,omitempty"`
	CreatedAt      int64             `json:"createdAt,omitempty"`
	QueuedAt       string            `json:"queuedAt,omitempty"`
	ScheduleTime   string            `json:"scheduleTime,omitempty"`
	StartedAt      string            `json:"startedAt,omitempty"`
	FinishedAt     string            `json:"finishedAt,omitempty"`
	AutoRetryCount int               `json:"autoRetryCount,omitempty"`
	AutoRetryLimit int               `json:"autoRetryLimit,omitempty"`
	// AutoRetryInterval is stored as a duration snapshot for retry scanner decisions.
	AutoRetryInterval time.Duration `json:"autoRetryInterval,omitempty"`
	AutoRetryBackoff  float64       `json:"autoRetryBackoff,omitempty"`
	// AutoRetryMaxInterval is stored as a duration snapshot for retry scanner decisions.
	AutoRetryMaxInterval time.Duration         `json:"autoRetryMaxInterval,omitempty"`
	ProcGroup            string                `json:"procGroup,omitempty"`
	SuspendFlagName      string                `json:"suspendFlagName,omitempty"`
	Log                  string                `json:"log,omitempty"`
	WorkingDir           string                `json:"workingDir,omitempty"`
	ArchiveDir           string                `json:"archiveDir,omitempty"`
	Error                string                `json:"error,omitempty"`
	Params               string                `json:"params,omitempty"`
	ParamsList           []string              `json:"paramsList,omitempty"`
	ProfileName          string                `json:"profileName,omitempty"`
	ProfileResolvedAt    string                `json:"profileResolvedAt,omitempty"`
	ProfileEntries       []RuntimeProfileEntry `json:"profileEntries,omitempty"`
	PendingStepRetries   []PendingStepRetry    `json:"pendingStepRetries"`
	Preconditions        []*core.Condition     `json:"preconditions,omitempty"`
	Labels               []string              `json:"labels,omitempty"`
	LeaseAt              int64                 `json:"leaseAt,omitempty"` // Unix millis; stamped by coordinator on observed run liveness
}

type DAGRunStatusPage struct {
	Items      []*DAGRunStatus
	NextCursor string
}

type DAGRunStatusResult struct {
	Found  bool
	Status *DAGRunStatus
}

type DAGRunStore interface {
	// CreateAttempt creates a new execution record for a dag-run.
	CreateAttempt(ctx context.Context, dag *core.DAG, ts time.Time, dagRunID string, opts NewDAGRunAttemptOptions) (DAGRunAttempt, error)
	// RecentAttempts returns the most recent dag-run's attempt for the DAG name, limited by itemLimit
	RecentAttempts(ctx context.Context, name string, itemLimit int) []DAGRunAttempt
	// LatestAttempt returns the most recent dag-run's attempt for the DAG name.
	LatestAttempt(ctx context.Context, name string) (DAGRunAttempt, error)
	// ListStatuses returns a list of statuses.
	ListStatuses(ctx context.Context, opts ...ListDAGRunStatusesOption) ([]*DAGRunStatus, error)
	// ListStatusesPage returns one forward-only page of statuses in canonical list order.
	ListStatusesPage(ctx context.Context, opts ...ListDAGRunStatusesOption) (DAGRunStatusPage, error)
	// CompareAndSwapLatestAttemptStatus atomically updates the latest attempt status
	// when both the latest attempt ID and status still match the expected values.
	CompareAndSwapLatestAttemptStatus(
		ctx context.Context,
		dagRun DAGRunRef,
		expectedAttemptID string,
		expectedStatus core.Status,
		mutate func(*DAGRunStatus) error,
		opts ...CompareAndSwapStatusOption,
	) (*DAGRunStatus, bool, error)
	// FindAttempt finds the latest attempt for the dag-run.
	FindAttempt(ctx context.Context, dagRun DAGRunRef) (DAGRunAttempt, error)
	// FindSubAttempt finds a sub dag-run record by dag-run ID.
	FindSubAttempt(ctx context.Context, dagRun DAGRunRef, subDAGRunID string) (DAGRunAttempt, error)
	// CreateSubAttempt creates a new sub dag-run attempt under the root dag-run.
	// This is used for distributed sub-DAG execution where the coordinator needs
	// to create the attempt directory before the worker reports status.
	CreateSubAttempt(ctx context.Context, rootRef DAGRunRef, subDAGRunID string) (DAGRunAttempt, error)
	// RemoveOldDAGRuns deletes dag-run records older than retentionDays, by absolute
	// cutoff (WithOlderThan), or by run count (WithRetentionRuns).
	// If retentionDays is negative and OlderThan is not set, it won't delete any records.
	// If retentionDays is zero, it will delete all records for the DAG name.
	// But it will not delete the records with non-final statuses (e.g., running, queued).
	// Returns a list of dag-run IDs that were removed (or would be removed in dry-run mode).
	RemoveOldDAGRuns(ctx context.Context, name string, retentionDays int, opts ...RemoveOldDAGRunsOption) ([]string, error)
	// RenameDAGRuns renames all run data from oldName to newName
	// The name means the DAG name, renaming it will allow user to manage those runs
	// with the new DAG name.
	RenameDAGRuns(ctx context.Context, oldName, newName string) error
	// RemoveDAGRun removes a dag-run record by its reference.
	RemoveDAGRun(ctx context.Context, dagRun DAGRunRef, opts ...RemoveDAGRunOption) error
}

type DAGStore interface {
	// Create stores a new DAG definition with the given name and returns its file name
	Create(ctx context.Context, fileName string, spec []byte) error
	// Delete removes a DAG definition by name
	Delete(ctx context.Context, fileName string) error
	// List returns a paginated list of DAG definitions with filtering options
	List(ctx context.Context, params ListDAGsOptions) (PaginatedResult[*core.DAG], []string, error)
	// GetMetadata retrieves only the metadata of a DAG definition (faster than full load)
	GetMetadata(ctx context.Context, fileName string) (*core.DAG, error)
	// GetDetails retrieves the complete DAG definition including all fields
	GetDetails(ctx context.Context, fileName string, opts ...spec.LoadOption) (*core.DAG, error)
	// Grep searches for a pattern in all DAG definitions and returns matching results
	Grep(ctx context.Context, pattern string) (ret []*GrepDAGsResult, errs []string, err error)
	// SearchCursor returns lightweight, cursor-based search hits for DAG definitions.
	SearchCursor(ctx context.Context, opts SearchDAGsOptions) (*CursorResult[SearchDAGResult], []string, error)
	// SearchMatches returns cursor-based match snippets for a specific DAG definition.
	SearchMatches(ctx context.Context, fileName string, opts SearchDAGMatchesOptions) (*CursorResult[*Match], error)
	// Rename changes a DAG's identifier from oldID to newID
	Rename(ctx context.Context, oldID, newID string) error
	// GetSpec retrieves the raw YAML specification of a DAG
	GetSpec(ctx context.Context, fileName string) (string, error)
	// UpdateSpec modifies the specification of an existing DAG
	UpdateSpec(ctx context.Context, fileName string, spec []byte) error
	// LoadSpec loads a DAG from a YAML file and returns the DAG object
	LoadSpec(ctx context.Context, spec []byte, opts ...spec.LoadOption) (*core.DAG, error)
	// LabelList returns all unique labels across all DAGs with any errors encountered
	LabelList(ctx context.Context) ([]string, []string, error)
	// ToggleSuspend changes the suspension state of a DAG by ID
	ToggleSuspend(ctx context.Context, fileName string, suspend bool) error
	// IsSuspended checks if a DAG is currently suspended
	IsSuspended(ctx context.Context, fileName string) bool
}

type Database interface {
	// GetDAG retrieves a DAG by its name.
	GetDAG(ctx context.Context, name string) (*core.DAG, error)
	// GetSubDAGRunStatus retrieves the status of a sub dag-run by its ID and the root dag-run reference.
	GetSubDAGRunStatus(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) (*RunStatus, error)
	// IsSubDAGRunCompleted checks if a sub dag-run has completed.
	IsSubDAGRunCompleted(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) (bool, error)
	// RequestChildCancel requests cancellation of a sub dag-run.
	RequestChildCancel(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) error
}

type DispatchAdmissionBindRequest struct {
	ReservationToken string
	Task             *DispatchTask
}

type DispatchAdmissionDecision struct {
	Reserved         bool
	Reason           DispatchAdmissionRejectReason
	ReservationToken string
}

type DispatchAdmissionRejectReason string

type DispatchAdmissionRequest struct {
	QueueName             string
	MaxConcurrency        int
	NonAdmissionOccupancy int
	AttemptKey            string
	AttemptID             string
	DAGRun                DAGRunRef
	StaleThreshold        time.Duration
}

type DispatchAdmissionStore interface {
	ReserveAdmission(ctx context.Context, req DispatchAdmissionRequest) (*DispatchAdmissionDecision, error)
	BindAdmission(ctx context.Context, req DispatchAdmissionBindRequest) error
	ReleaseAdmissionToken(ctx context.Context, reservationToken string) error
	FinalizeAdmissionAttempt(ctx context.Context, attemptKey string) error
	CleanupAdmissions(ctx context.Context, staleThreshold time.Duration) error
}

type DispatchOperation int32

type DispatchRequest struct {
	Task                      *DispatchTask
	AdmissionReservationToken string
}

type DispatchTask struct {
	RootDAGRunName string
	RootDAGRunID   string

	ParentDAGRunName string
	ParentDAGRunID   string

	Operation    DispatchOperation
	DAGRunID     string
	Target       string
	Definition   string
	AttemptID    string
	AttemptKey   string
	Step         string
	Params       string
	QueueName    string
	WorkerID     string
	ProfileName  string
	TriggerActor string

	PreviousStatus *DAGRunStatus

	BaseConfig   string
	Labels       string
	ScheduleTime string
	SourceFile   string

	WorkerSelector map[string]string

	ExternalStepRetry bool
	RetryPath         string

	WorkspaceBundleDigest      string
	WorkspaceBundleSize        int64
	WorkspaceBundleDAGPath     string
	WorkspaceBundleOriginalRef string
	WorkspaceBundleResolvedRef string

	Owner      CoordinatorEndpoint
	ClaimToken string
}

type DispatchTaskClaim struct {
	WorkerID     string
	PollerID     string
	Labels       map[string]string
	Owner        CoordinatorEndpoint
	ClaimTimeout time.Duration
}

type DispatchTaskStore interface {
	Enqueue(ctx context.Context, task *DispatchTask) error
	ClaimNext(ctx context.Context, claim DispatchTaskClaim) (*ClaimedDispatchTask, error)
	GetClaim(ctx context.Context, claimToken string) (*ClaimedDispatchTask, error)
	ReleaseClaim(ctx context.Context, claimToken string) error
	DeleteClaim(ctx context.Context, claimToken string) error
	CountOutstandingByQueue(ctx context.Context, queueName string, claimTimeout time.Duration) (int, error)
	HasOutstandingAttempt(ctx context.Context, attemptKey string, claimTimeout time.Duration) (bool, error)
}

type Dispatcher interface {
	Dispatch(ctx context.Context, req DispatchRequest) error
	Cleanup(ctx context.Context) error
	GetDAGRunStatus(ctx context.Context, dagName, dagRunID string, rootRef *DAGRunRef) (*DAGRunStatusResult, error)
	RequestCancel(ctx context.Context, dagName, dagRunID string, rootRef *DAGRunRef) error
}

type EnqueueRetryOptions struct {
	// AutoRetry marks scheduler-issued DAG auto-retries. These consume the
	// DAG-level retry budget at enqueue time.
	AutoRetry bool
	// TriggerActor replaces the attributable actor for a user-issued retry.
	// Nil preserves the actor already recorded on the run.
	TriggerActor *string
}

type FailedAutoRetryCancelEligibility int

type FailedAutoRetryCancelStateChangedError struct {
	CurrentStatus *DAGRunStatus
}

type GrepDAGsResult struct {
	Name    string    // Name of the DAG
	DAG     *core.DAG // The DAG object
	Matches []*Match  // Matching lines and their context
}

type HostInfo struct {
	// ID is a unique identifier for the host
	ID string
	// Host is the hostname or IP address
	Host string
	// Port is the port number (0 if not applicable)
	Port int
	// Status is the operational status of the service instance
	Status ServiceStatus
	// StartedAt is when the service instance was started
	StartedAt time.Time
}

type LLMMessage struct {
	// Role is the message role (system, user, assistant, tool).
	Role core.LLMRole `json:"role"`
	// Content is the message content.
	Content string `json:"content"`
	// ToolCallID is the ID of the tool call this message is responding to.
	// Only set when Role is "tool".
	ToolCallID string `json:"tool_call_id,omitempty"`
	// ToolCalls contains tool calls made by the assistant.
	// Only set when Role is "assistant" and the model requested tool calls.
	ToolCalls []ToolCall `json:"tool_calls,omitempty"`
	// Metadata contains API call metadata (only set for assistant responses).
	Metadata *LLMMessageMetadata `json:"metadata,omitempty"`
}

type LLMMessageMetadata struct {
	// Provider is the LLM provider used (openai, anthropic, etc.).
	Provider string `json:"provider,omitempty"`
	// Model is the model identifier used.
	Model string `json:"model,omitempty"`
	// PromptTokens is the number of tokens in the prompt.
	PromptTokens int `json:"promptTokens,omitempty"`
	// CompletionTokens is the number of tokens in the completion.
	CompletionTokens int `json:"completionTokens,omitempty"`
	// TotalTokens is the sum of prompt and completion tokens.
	TotalTokens int `json:"totalTokens,omitempty"`
	// Cost is the estimated USD cost for this API call.
	Cost float64 `json:"cost,omitempty"`
}

type ListDAGRunStatusesOption func(*ListDAGRunStatusesOptions)

type ListDAGRunStatusesOptions struct {
	DAGRunID        string
	Name            string
	ExactName       string
	From            TimeInUTC
	To              TimeInUTC
	Statuses        []core.Status
	Limit           int
	Cursor          string
	Labels          []string // Filter by DAG labels (AND logic - all labels must match)
	WorkspaceFilter *WorkspaceFilter
	Unlimited       bool
	AllHistory      bool
}

type ListDAGsOptions struct {
	Paginator         *Paginator
	Name              string                               // Optional search filter for DAG name or file name
	Labels            []string                             // Optional labels filter (AND logic - all labels must match)
	Sort              string                               // Optional sort field (name, updated_at, created_at, nextRun)
	Order             string                               // Optional sort order (asc, desc)
	Time              *time.Time                           // Optional reference time for nextRun sorting/projection (defaults to time.Now())
	NextRunProjection func(*core.DAG, time.Time) time.Time // Optional scheduler-aware nextRun projector used when Sort == "nextRun"
	WorkspaceFilter   *WorkspaceFilter                     // Optional workspace visibility filter
}

type ListDAGsResult struct {
	DAGs   []*core.DAG // The list of DAGs for the current page
	Count  int         // Total count of DAGs matching the filter
	Errors []string    // Any errors encountered during listing
}

type LogWriterFactory interface {
	// NewStepWriter creates a writer for a step's log output.
	// stepName identifies the step, streamType should be StreamTypeStdout or StreamTypeStderr.
	NewStepWriter(ctx context.Context, stepName string, streamType int) io.WriteCloser
}

type Match struct {
	Line       string
	LineNumber int
	StartLine  int
}

type MockDAGRunAttempt struct {
	mock.Mock
	// Status can be set for tests that need to return a specific status without mock setup
	Status *DAGRunStatus
}

type MockQueueStore struct {
	mock.Mock
}

type NewDAGRunAttemptOptions struct {
	// RootDAGRun is the root dag-run reference for this attempt.
	RootDAGRun *DAGRunRef
	// Retry indicates whether this is a retry of a previous run.
	Retry bool
	// AttemptID is an optional attempt ID. If set, this ID is used instead of generating a new one.
	// This is used when the coordinator has already created an attempt and wants the worker
	// to use the same ID for consistency.
	AttemptID string
}

type Node struct {
	Step             core.Step            `json:"step,omitzero"`
	Stdout           string               `json:"stdout"` // standard output log file path
	Stderr           string               `json:"stderr"` // standard error log file path
	WorkingDir       string               `json:"workingDir,omitempty"`
	StartedAt        string               `json:"startedAt"`
	FinishedAt       string               `json:"finishedAt"`
	Status           core.NodeStatus      `json:"status"`
	RetriedAt        string               `json:"retriedAt,omitempty"`
	RetryCount       int                  `json:"retryCount,omitempty"`
	DoneCount        int                  `json:"doneCount,omitempty"`
	Repeated         bool                 `json:"repeated,omitempty"` // indicates if the node has been repeated
	SkippedByRetry   bool                 `json:"skippedByRetry,omitempty"`
	Error            string               `json:"error,omitempty"`
	SubRuns          []SubDAGRun          `json:"children,omitempty"`
	SubRunsRepeated  []SubDAGRun          `json:"childrenRepeated,omitempty"` // repeated sub DAG runs
	OutputVariables  *collections.SyncMap `json:"outputVariables,omitempty"`
	OutputValue      *string              `json:"outputValue,omitempty"`
	OutputsValue     *string              `json:"outputsValue,omitempty"`
	StepOutputsValue *string              `json:"stepOutputsValue,omitempty"`
	HumanTaskInput   json.RawMessage      `json:"humanTaskInput,omitempty"`
	// ControllerState stores the goal progress of a controller DAG's controller
	// step, so a suspended run resumes with its task list intact.
	ControllerState json.RawMessage `json:"controllerState,omitempty"`
	// HumanTaskCompletedBy records who completed this human task.
	HumanTaskCompletedBy string `json:"humanTaskCompletedBy,omitempty"`
	// HumanTaskCompletedByID records the subject ID that completed this human task.
	HumanTaskCompletedByID string `json:"humanTaskCompletedById,omitempty"`
	// ApprovedAt records when this wait step was approved
	ApprovedAt string `json:"approvedAt,omitempty"`
	// ApprovalInputs stores key-value parameters provided during approval
	ApprovalInputs map[string]string `json:"approvalInputs,omitempty"`
	// ApprovedBy records who approved this wait step (username)
	ApprovedBy string `json:"approvedBy,omitempty"`
	// ApprovedByID records the subject ID that approved this wait step.
	ApprovedByID string `json:"approvedById,omitempty"`
	// RejectedAt records when this wait step was rejected
	RejectedAt string `json:"rejectedAt,omitempty"`
	// RejectedBy records who rejected this wait step (username)
	RejectedBy string `json:"rejectedBy,omitempty"`
	// RejectedByID records the subject ID that rejected this wait step.
	RejectedByID string `json:"rejectedById,omitempty"`
	// RejectionReason stores the optional reason for rejection
	RejectionReason string `json:"rejectionReason,omitempty"`
	// ApprovalIteration tracks how many times this step has been pushed back.
	ApprovalIteration int `json:"approvalIteration,omitempty"`
	// PushBackInputs stores the inputs from the last push-back.
	// These are injected as environment variables when the step re-executes.
	PushBackInputs map[string]string `json:"pushBackInputs,omitempty"`
	// PushBackHistory stores the chronological push-back inputs for this step.
	PushBackHistory []PushBackEntry `json:"pushBackHistory,omitempty"`
	// PushBackPreviousStdout stores the stdout log path from the execution that
	// was reset by the latest push-back.
	PushBackPreviousStdout string `json:"pushBackPreviousStdout,omitempty"`
	// ChatMessages stores the session messages for chat/LLM steps.
	// This field is populated during execution and synced via status updates
	// from workers.
	ChatMessages []LLMMessage `json:"chatMessages,omitempty"`
	// ToolDefinitions stores the tool definitions that were available to the LLM.
	// This enables debugging visibility into what tools and schemas were sent.
	ToolDefinitions []ToolDefinition `json:"toolDefinitions,omitempty"`
}

type OutputsMetadata struct {
	DAGName     string `json:"dagName"`
	DAGRunID    string `json:"dagRunId"`
	AttemptID   string `json:"attemptId"`
	Status      string `json:"status"`
	CompletedAt string `json:"completedAt"`
	Params      string `json:"params,omitempty"` // JSON-serialized parameters
}

type PID int

type PageRange struct {
	Range     []int
	SkipFirst bool
	SkipLast  bool
}

type PaginatedResult[T any] struct {
	Items       []T
	CurrentPage int
	TotalPages  int
	TotalCount  int
	Offset      int
	HasNextPage bool
	HasPrevPage bool
	NextPage    int
	PrevPage    int
}

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

type PendingStepRetry struct {
	StepName string        `json:"stepName"`
	Interval time.Duration `json:"interval"`
}

type ProcEntry struct {
	GroupName       string
	Identity        ProcEntryID
	Meta            ProcMeta
	LastHeartbeatAt int64
	Fresh           bool
}

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

type ProcHandle interface {
	// Stop stops the heartbeat for the process.
	Stop(ctx context.Context) error
	// GetMeta retrieves the metadata for the process.
	GetMeta() ProcMeta
}

type ProcHeartbeat struct {
	GroupName       string
	DAGRun          DAGRunRef
	AttemptID       string
	StartedAt       int64
	LastHeartbeatAt int64
	ObservedAt      time.Time
	Fresh           bool
}

type ProcMeta struct {
	StartedAt    int64
	Name         string
	DAGRunID     string
	AttemptID    string
	RootName     string
	RootDAGRunID string
}

type ProcStore interface {
	// Lock try to lock process group return error if it's held by another process
	Lock(ctx context.Context, groupName string) error
	// UnLock unlocks process group
	Unlock(ctx context.Context, groupName string)
	// Acquire creates a new process entry for a given group name and execution metadata.
	// It automatically starts the heartbeat for the process.
	Acquire(ctx context.Context, groupName string, meta ProcMeta) (ProcHandle, error)
	// CountAlive retrieves the number of processes associated with a group name.
	CountAlive(ctx context.Context, groupName string) (int, error)
	// CountAlive retrieves the number of processes associated with a group name.
	CountAliveByDAGName(ctx context.Context, groupName, dagName string) (int, error)
	// IsRunAlive checks if a specific DAG run is currently alive.
	IsRunAlive(ctx context.Context, groupName string, dagRun DAGRunRef) (bool, error)
	// IsAttemptAlive checks if a specific DAG-run attempt is currently alive.
	IsAttemptAlive(ctx context.Context, groupName string, dagRun DAGRunRef, attemptID string) (bool, error)
	// ListAlive returns list of running DAG runs by the group name.
	ListAlive(ctx context.Context, groupName string) ([]DAGRunRef, error)
	// ListAllAlive returns all running DAG runs across all groups.
	// Returns a map where key is the group name and value is list of DAG runs.
	ListAllAlive(ctx context.Context) (map[string][]DAGRunRef, error)
	// ListEntries returns all proc entries for a group, including stale entries.
	ListEntries(ctx context.Context, groupName string) ([]ProcEntry, error)
	// LatestFreshEntryByDAGName returns the freshest proc entry for the DAG in the group.
	LatestFreshEntryByDAGName(ctx context.Context, groupName, dagName string) (*ProcEntry, error)
	// LatestHeartbeat returns the latest heartbeat observation for a DAG-run in the group.
	LatestHeartbeat(ctx context.Context, groupName string, dagRun DAGRunRef) (*ProcHeartbeat, error)
	// ListAllEntries returns all proc entries across all groups, including stale entries.
	ListAllEntries(ctx context.Context) ([]ProcEntry, error)
	// RemoveIfStale removes the exact proc entry if it is still stale and unchanged.
	RemoveIfStale(ctx context.Context, entry ProcEntry) error
}

type PushBackEntry struct {
	Iteration int               `json:"iteration"`
	By        string            `json:"by,omitempty"`
	ByID      string            `json:"byId,omitempty"`
	At        string            `json:"at,omitempty"`
	Inputs    map[string]string `json:"inputs,omitempty"`
}

type QueuePriority int

type QueueStore interface {
	// Enqueue adds an item to the queue
	Enqueue(ctx context.Context, name string, priority QueuePriority, dagRun DAGRunRef) error
	// DequeueByName retrieves an item from the queue and removes it
	DequeueByName(ctx context.Context, name string) (QueuedItemData, error)
	// DequeueByDAGRunID retrieves items from the queue by dag-run reference and removes them
	DequeueByDAGRunID(ctx context.Context, name string, dagRun DAGRunRef) ([]QueuedItemData, error)
	// DeleteByItemIDs removes the exact queued items identified by their queue item IDs.
	DeleteByItemIDs(ctx context.Context, name string, itemIDs []string) (int, error)
	// Len returns the number of items in the queue
	Len(ctx context.Context, name string) (int, error)
	// List returns all items in the queue with the given name
	List(ctx context.Context, name string) ([]QueuedItemData, error)
	// ListCursor returns one forward-only page of queued items for a specific queue.
	ListCursor(ctx context.Context, name, cursor string, limit int) (CursorResult[QueuedItemData], error)
	// All returns all items in the queue
	All(ctx context.Context) ([]QueuedItemData, error)
	// ListByDAGName returns all items that has a specific DAG name
	ListByDAGName(ctx context.Context, name, dagName string) ([]QueuedItemData, error)
	// QueueList lists all queue names that have at least one item in the queue
	QueueList(ctx context.Context) ([]string, error)
	// Watcher returns a QueueWatcher for the queue data
	QueueWatcher(ctx context.Context) QueueWatcher
}

type QueueWatcher interface {
	// Start start swatching queue data and signal when a queue state changed
	Start(ctx context.Context) (<-chan struct{}, error)
	// Stop stops watching queue data
	Stop(ctx context.Context)
}

type QueuedItem struct {
	QueuedItemData
}

type QueuedItemData interface {
	// ID returns the ID of the queued item
	ID() string
	// Data returns the data of the queued item
	Data() (*DAGRunRef, error)
}

type RemoveDAGRunOption func(*RemoveDAGRunOptions)

type RemoveDAGRunOptions struct {
	// RejectActive if true, refuses to remove dag-runs with an active status.
	RejectActive bool
}

type RemoveOldDAGRunsOption func(*RemoveOldDAGRunsOptions)

type RemoveOldDAGRunsOptions struct {
	// DryRun if true, only returns the paths that would be removed without actually deleting
	DryRun bool
	// RetentionRuns keeps the most recent number of dag-runs when set.
	RetentionRuns *int
	// OlderThan when set, deletes dag-runs whose recorded time is strictly before this
	// cutoff. When set, the retentionDays argument is ignored.
	OlderThan *time.Time
}

type RetryHop struct {
	Step  string `json:"step"`
	RunID string `json:"runId"`
}

type RetryPath struct {
	Hops []RetryHop `json:"hops"`
	Step string     `json:"step"`
}

type RunStatus struct {
	// Name represents the name of the executed DAG.
	Name string
	// DAGRunID is the ID of the dag-run.
	DAGRunID string
	// Params is the parameters of the DAG.
	Params string
	// Outputs is the outputs of the dag-run.
	Outputs map[string]string
	// OutputValues contains typed outputs published through stdout.outputs or outputs.write.
	OutputValues map[string]any
	// Status is the execution status of the dag-run.
	Status core.Status
	// PendingStepRetries contains any step retries that are waiting to be scheduled
	// by the parent executor.
	PendingStepRetries []PendingStepRetry
}

type RunningTask struct {
	DAGRunID         string `json:"dagRunId,omitempty"`
	DAGName          string `json:"dagName,omitempty"`
	StartedAt        int64  `json:"startedAt,omitempty"`
	RootDAGRunName   string `json:"rootDagRunName,omitempty"`
	RootDAGRunID     string `json:"rootDagRunId,omitempty"`
	ParentDAGRunName string `json:"parentDagRunName,omitempty"`
	ParentDAGRunID   string `json:"parentDagRunId,omitempty"`
	AttemptKey       string `json:"attemptKey,omitempty"`
}

type RuntimeProfileEntry struct {
	// Key is the injected environment variable name.
	Key string `json:"key"`
	// Kind is the profile entry type, such as variable or secret.
	Kind string `json:"kind"`
}

type SearchDAGMatchesOptions struct {
	Cursor          string
	Limit           int
	Query           string
	Labels          []string
	WorkspaceFilter *WorkspaceFilter
}

type SearchDAGResult struct {
	Name              string
	FileName          string
	Workspace         string
	Matches           []*Match
	HasMoreMatches    bool
	NextMatchesCursor string
}

type SearchDAGsOptions struct {
	Cursor          string
	Limit           int
	Query           string
	MatchLimit      int
	Labels          []string
	WorkspaceFilter *WorkspaceFilter
}

type ServiceName string

type ServiceRegistry interface {
	// Register registers services for the given service name and host info.
	// It returns an error if the registry failed to start heartbeat.
	Register(ctx context.Context, serviceName ServiceName, hostInfo HostInfo) error

	// Unregister un-registers current service.
	Unregister(ctx context.Context)

	// GetServiceMembers returns the list of active hosts for the given service.
	// This method combines service resolution and member lookup.
	GetServiceMembers(ctx context.Context, serviceName ServiceName) ([]HostInfo, error)

	// UpdateStatus updates the status of the current registered instance
	UpdateStatus(ctx context.Context, serviceName ServiceName, status ServiceStatus) error
}

type ServiceStatus int

type StaleQueueDispatchError struct {
	Reason string
}

type SubDAGRun struct {
	DAGRunID string `json:"dagRunId,omitempty"`
	Params   string `json:"params,omitempty"`
	// 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 `json:"dagName,omitempty"`
}

type TimeInUTC struct{ time.Time }

type ToolCall struct {
	// ID is a unique identifier for this tool call.
	ID string `json:"id"`
	// Type is always "function" for function calls.
	Type string `json:"type"`
	// Function contains the function call details.
	Function ToolCallFunction `json:"function"`
}

type ToolCallFunction struct {
	// Name is the name of the function to call.
	Name string `json:"name"`
	// Arguments is a JSON string containing the function arguments.
	Arguments string `json:"arguments"`
}

type ToolDefinition struct {
	// Name is the tool/function name as presented to the LLM.
	Name string `json:"name"`
	// Description describes what the tool does.
	Description string `json:"description,omitempty"`
	// Parameters is the JSON Schema describing the tool's parameters.
	Parameters map[string]any `json:"parameters,omitempty"`
}

type WorkerHeartbeatRecord struct {
	WorkerID        string            `json:"workerId"`
	Labels          map[string]string `json:"labels,omitempty"`
	Stats           *WorkerStats      `json:"stats,omitempty"`
	LastHeartbeatAt int64             `json:"lastHeartbeatAt"`
}

type WorkerHeartbeatStore interface {
	Upsert(ctx context.Context, record WorkerHeartbeatRecord) error
	Get(ctx context.Context, workerID string) (*WorkerHeartbeatRecord, error)
	List(ctx context.Context) ([]WorkerHeartbeatRecord, error)
	DeleteStale(ctx context.Context, before time.Time) (int, error)
}

type WorkerStats struct {
	TotalPollers int32          `json:"totalPollers,omitempty"`
	BusyPollers  int32          `json:"busyPollers,omitempty"`
	RunningTasks []*RunningTask `json:"runningTasks,omitempty"`
}

type WorkspaceFilter struct {
	Enabled           bool
	Workspaces        []string
	IncludeUnlabelled bool
}

type WorkspaceLabelState int