scheduler

type Alerter interface {
	// AlertRunFailed sends every on_failure rule and reports whether all of them
	// were delivered. The caller stamps delivery only on true, so a failed send
	// stays retryable instead of being marked and lost.
	AlertRunFailed(ctx context.Context, run RunState) bool
}

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

type AsyncDispatchStore interface {
	// DispatchAttempts returns the task's consecutive dispatch failures while it
	// is still scheduled or queued; active is false once it has moved on.
	DispatchAttempts(ctx context.Context, runID, taskID string) (attempts int, active bool, err error)
	// RequeueDispatch moves a scheduled or queued task back to scheduled, held
	// until nextAt, adding one dispatch attempt when countAttempt, and clearing
	// the start of an outage window (ADR 0093) when clearUnavailable.
	RequeueDispatch(ctx context.Context, runID, taskID string, countAttempt, clearUnavailable bool, nextAt time.Time) (bool, error)
	MarkTaskDispatchFailed(ctx context.Context, runID, taskID, reason string) error
	// FailDispatchRefused fails a refused task and spends its retry budget, as
	// on the sync path.
	FailDispatchRefused(ctx context.Context, runID, taskID, reason string) error
	// RecordDispatchUnavailable re-offers a task whose pod create met an
	// unavailable Kubernetes API without counting an attempt, as on the sync
	// path (ADR 0093).
	RecordDispatchUnavailable(ctx context.Context, runID, taskID string, now time.Time, minDelay, maxDelay time.Duration) (since, nextAt time.Time, applied bool, err error)
}

type Dispatcher interface {
	Dispatch(ctx context.Context, runID, dagID, dagVersionID string, task domain.TaskSpec) (executor.Disposition, error)
}

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

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

type PlannedTransition struct {
	TaskID string
	To     domain.TaskState
}

type PoolReservation struct {
	RunID  string
	TaskID string
}

type PoolWait struct {
	TaskID string
	Pool   string // budget key, PoolKey(tenant, pool)
	Slots  int
}

type Recorder interface {
	RecordSchedulerDecision(decisionType string)
	RecordTaskTransition(from, to, dagID string)
	// RecordUndispatchable counts tasks queued with no executor to launch them.
	RecordUndispatchable(reason string)
	// RecordSchedulerStepDown counts a leader step-down by reason, the rate
	// of this counter is the operator-facing SLI for leader-churn (#311). A
	// nil Recorder is tolerated by the scheduler so tests need not stub it.
	RecordSchedulerStepDown(reason string)
	// ObserveSchedulerReacquire records the wall-clock duration of a step-down
	// → re-acquire cycle (#311). Operators alert on P99 to spot churn that
	// starts to delay scheduling latency.
	ObserveSchedulerReacquire(d time.Duration)
	// RecordAlert counts one on-failure alert outcome by dag, channel type, and
	// result. The scheduler records result="dropped" when its dispatch semaphore
	// is saturated (#435); the dispatcher records "sent"/"failed" for deliveries.
	RecordAlert(dagID, channelType, result string)
	// ObserveSchedulerLoop records the wall time of one leader tick, whether
	// its step succeeded or not (#1186).
	ObserveSchedulerLoop(d time.Duration)
	// RecordTaskRetry counts one retry released for a task (#1186).
	RecordTaskRetry(dagID, taskID string)
	// RecordSchedulerWokenTick counts a tick started early by Wake
	// (scheduler.eager_promotion) instead of by the loop interval.
	RecordSchedulerWokenTick()
	// ObserveScheduleDelay records how late a scheduled run was created: the
	// time from its slot to its creation, for the latest slot a tick created
	// for a DAG (#2175).
	ObserveScheduleDelay(d time.Duration)
}

type RunState struct {
	// RunID is the dag_runs primary key (a UUID). It is what dispatch, the agent
	// token and the log sink key on, so it must not be swapped for the
	// user-facing id.
	RunID string
	// DisplayRunID is the dag_runs.run_id column, the identifier an operator
	// sees in the UI and passes to the API ("manual__2026-07-30T12:00:00+00:00").
	// Alerts render this one: a UUID is not something anyone can act on.
	DisplayRunID string
	// LogicalDate is the run's logical date in RFC3339, empty for an unscheduled
	// run. Carried here so an alert can say which interval failed without a
	// second query.
	LogicalDate string
	DagID       string
	// DagVersionID is the dag_versions row this run pins (dag_runs.dag_version_id).
	// It names the warm-worker pool a task is placed onto (ADR 0058 N1b1-place):
	// dispatch threads it to the placer so an attempt reuses a worker of the exact
	// version it compiled against. Empty for runs loaded by paths that do not set
	// it; the dedicated pod path ignores it.
	DagVersionID string
	TenantID     string
	State        domain.DagRunState
	// Held reports that the run's DAG is paused. A held run that is still
	// queued waits: the tick does not start it, materialize its task instances
	// or dispatch anything for it until the DAG is unpaused, as Airflow 3 does
	// (#1705). A held run that is already running advances as usual, so the
	// tasks it started finish. Either way a held run stays in the tick's
	// snapshot: it counts toward max_active_runs, and its queued and running
	// task instances count toward max_active_tasks and the pool caps (ADR
	// 0053, ADR 0066). A clear on a running run puts it back in queued while
	// other tasks of it may still run, so a queued held run can hold slots.
	Held bool
	// Suspension is the suspension mode of the run's tenant, empty when it is
	// not suspended (#2181). Under drain or stop the tick dispatches nothing
	// of the run and does not start it while it is queued; under stop it also
	// fails the run's queued and running task instances (stopSuspended).
	Suspension domain.SuspensionMode
	// MaxActiveRuns is the DAG's max_active_runs, read from its dags row (the
	// same value run creation caps on). The tick starts a queued run only
	// while fewer than this many runs of the DAG are running, oldest logical
	// date first, as Airflow's _start_queued_dagruns does (#1766). Zero or
	// negative means unlimited.
	MaxActiveRuns int
	Tasks         []domain.TaskSpec
	// Graph is the task index for Tasks, built once per dag_version by the store
	// and shared read-only by every run of that version. Nil means "build it
	// when needed", so callers that construct a RunState by hand need not set it.
	Graph  *TaskGraph
	States map[string]domain.TaskState
	// Tries and MaxTries hold the current and maximum attempt counts per task,
	// driving retry decisions. Absent entries mean no retry budget.
	Tries    map[string]int
	MaxTries map[string]int
	// EndedAt holds the failure timestamp per task (only meaningful for tasks
	// currently in `up_for_retry`). Combined with RetryDelaySeconds + Now, it
	// gates the `up_for_retry → none` transition so retries honor user-
	// declared backoff (issue #201). Absent entries mean no cooldown applies.
	EndedAt           map[string]*time.Time
	RetryDelaySeconds map[string]int
	// RescheduleAt holds the next-poke time per task (only meaningful for tasks in
	// up_for_reschedule). The planner re-dispatches such a task once Now >=
	// reschedule_at, without consuming retry budget (#380). Absent/zero entries
	// re-dispatch immediately (preserves the test seam, like EndedAt).
	RescheduleAt map[string]*time.Time
	// NextDispatchAt holds the earliest time a `scheduled` task may be dispatched
	// again after a synchronous dispatch failure (ADR 0031 Amendment A). The
	// planner does not promote scheduled→queued while Now < next_dispatch_at.
	// Absent/zero entries dispatch immediately (the common case: never failed).
	NextDispatchAt map[string]*time.Time
	// DispatchAttempts counts consecutive synchronous dispatch failures per task.
	// It is a separate counter from try_number, a dispatch failure is infra, not
	// a task failure, and drives the dispatch_failed give-up at
	// dispatchMaxAttempts. Absent entries mean zero.
	DispatchAttempts map[string]int
	// InfraFailed marks a `failed` task whose failure was infra-caused (agent/pod/
	// dispatch lost), from the durable last_failure_kind signal. Such a task
	// re-places without consuming the task's retry budget (ADR 0051 Phase 1), an
	// infrastructure fault is not the user's task failing.
	InfraFailed map[string]bool
	// InfraProvisional marks an InfraFailed task whose mark the pod reconciler
	// has not confirmed yet (infra_confirmed_at NULL, ADR 0052 amendment). Until
	// it is confirmed, or InfraConfirmMaxWait has passed since ended_at, the
	// task is active: no re-place and no downstream condemnation, so a durable
	// SUCCESS record can still settle it. Absent entries mean confirmed.
	InfraProvisional map[string]bool
	// InfraAttempts counts asynchronous infra re-placements per task, the
	// try_number-free analog of DispatchAttempts for agent/pod/dispatch-lost faults.
	// It bounds the re-place at infraMaxAttempts so a poison placement cannot loop
	// forever. Absent entries mean zero.
	InfraAttempts map[string]int
	// Now is the wall-clock value the planner compares against EndedAt. Zero
	// means "skip the cooldown gate" so legacy callers + tests that don't set
	// retry_delay get the previous (immediate-retry) behavior.
	Now time.Time
	// Alerts carries the DAG's native on-failure alerting config (#424), loaded
	// from the dag.json alongside Tasks. nil means no alerting; the scheduler
	// dispatches these rules once the run finalizes as failed.
	Alerts *domain.AlertsConfig
	// MaxActiveTasks is the DAG's per-DAG task-concurrency cap (Airflow's
	// max_active_tasks, ADR 0053 Stage 1), loaded from the spec. Zero or negative
	// means unlimited, the admission gate is a no-op, so a DAG that never sets it
	// (and all of Lite) plans exactly as before.
	MaxActiveTasks int
	// ActiveTaskCount is the DAG-wide number of currently non-terminal (queued or
	// running) task instances across every active run of this DAG, plus any this
	// tick already admitted for the same DAG in earlier sibling runs. PlanRun
	// subtracts it from MaxActiveTasks to derive this run's admission headroom, so
	// the cap holds per-DAG across runs and one tick cannot breach it. The Step
	// loop populates it; it defaults to zero, leaving the gate unlimited when
	// MaxActiveTasks is unset.
	ActiveTaskCount int
	// PoolsEnabled turns on the named-pool slot gate (ADR 0053 Stage 3). The
	// Step loop sets it from the scheduler's pools flag (scheduler.pools.enabled).
	// Left false, the pool gate is a no-op and planning is byte-identical to the
	// max_active_tasks-only path.
	PoolsEnabled bool
	// ConfineUndefinedPools makes a task that names a pool its tenant has not
	// defined draw on default_pool instead of running unlimited. The Step loop
	// sets it when tenant pool writes are locked (server.pools_read_only): a
	// tenant cannot create pools then, so an undefined name would otherwise be a
	// way around the default_pool budget (#646).
	ConfineUndefinedPools bool
	// PoolBudgets is the per-pool slot cap keyed by PoolKey(TenantID, pool). A pool
	// with a non-positive or absent budget is unlimited (fail open, never
	// deadlock). The Step loop sets it from the once-per-tick PoolBudgets snapshot;
	// nil in Lite. Shared read-only across sibling runs.
	PoolBudgets map[string]int
	// PoolActive is the cross-DAG number of slots held by currently non-terminal
	// (queued+running) task instances per pool (each takes its pool_slots, ADR
	// 0066), keyed like PoolBudgets, plus any this tick already admitted into the
	// same pool by earlier runs. PlanRun adds its own within-call
	// promotions on top. The Step loop sets it and folds each run's admissions back
	// in so a single tick cannot breach a pool across runs; nil in Lite.
	PoolActive map[string]int
	// PoolReservations holds, per pool key, the starved task the pool is
	// reserved for (ADR 0066 §4): while set, the pool admits only that task.
	// The Step loop sets it from the leader's in-memory reservations; nil
	// means no pool is reserved. Shared read-only across sibling runs.
	PoolReservations map[string]PoolReservation
}

type ScheduledDAG struct {
	// TenantID is the UUID of the tenant that owns the DAG. A dag_id is unique
	// only within its tenant, so every per-DAG decision keys on the pair (#209).
	TenantID    string
	DagID       string
	Schedule    string
	LastLogical *time.Time
	StartDate   *time.Time
	Catchup     bool
	// MaxActiveRuns caps how many runs of this DAG may be active (queued or
	// running) at once. Zero means "unlimited" (Airflow-faithful: a missing
	// limit is unbounded). The scheduler enforces the cap in createDueRuns:
	// once the active-run count reaches this value, additional due slots are
	// skipped on this tick (issue #200). Starting a queued run is capped
	// separately, on the running runs alone (RunState.MaxActiveRuns, #1766).
	MaxActiveRuns int
	// NextRunAt is the next due time the scheduler stored for the DAG
	// (SetNextRunAt, #2175), nil when none is stored. ScheduleEpoch counts the
	// changes that invalidate it (a new current version, a deleted run); a
	// write conditioned on the epoch the list returned is dropped when the DAG
	// changed in between.
	NextRunAt     *time.Time
	ScheduleEpoch int64
}

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

type Store interface {
	ActiveRuns(ctx context.Context) ([]RunState, error)
	// PoolBudgets returns every named pool's slot cap keyed by PoolKey(tenant,
	// pool), the cross-DAG admission budget the pool gate enforces (ADR 0053
	// Stage 3). Called once per tick, and ONLY on the Pro path (poolsEnabled);
	// Lite never invokes it.
	PoolBudgets(ctx context.Context) (map[string]int, error)
	MaterializeTasks(ctx context.Context, runID string, tasks []domain.TaskSpec) error
	ApplyTransition(ctx context.Context, runID, taskID string, to domain.TaskState) error
	// MarkQueued is the scheduled -> queued write after an accepted dispatch,
	// guarded to the slot the tick planned: still scheduled, with the
	// next_dispatch_at the tick read. It reports false when the row moved on (a
	// buffered dispatch worker already failed or re-offered the task, or the
	// agent already reported), so that newer outcome is never overwritten.
	MarkQueued(ctx context.Context, runID, taskID string, expectNextDispatchAt *time.Time) (bool, error)
	// ApplyTransitions moves every listed task of a run that is still in the
	// from state to the SAME target state in one statement. The scheduler groups
	// a tick's plain state-set transitions (scheduled/skipped/upstream_failed/
	// up_for_retry) by source and target state and flushes each group here, so R
	// updates collapse to one per group. from is the state the tick's snapshot
	// read: a row a concurrent writer moved since (a user's mark or clear, an
	// agent's report under an overlapping leader) is left alone (#1605). It
	// returns the task ids that actually moved; the guarded none-rail resets stay
	// on their own methods.
	ApplyTransitions(ctx context.Context, runID string, taskIDs []string, from, to domain.TaskState) ([]string, error)
	// ResetForRetry returns a task to 'none' and increments its try number so a
	// retry re-evaluates and re-runs it. It is guarded to the up_for_retry source
	// state; the bool reports whether the reset actually fired (false = the TI was
	// no longer up_for_retry, nothing reset) so the caller does not record a
	// phantom retry on a lost race.
	ResetForRetry(ctx context.Context, runID, taskID string) (bool, error)
	// ResetForInfraReplace returns a task a reaper failed as infra to 'none' for a
	// re-run, PRESERVING try_number and bumping infra_attempts instead, an infra
	// fault must not consume the retry budget (ADR 0051 Phase 1). Guarded to the
	// failed+infra source; the bool reports whether the reset fired (false = no
	// longer a failed-infra candidate) so the caller does not record a phantom
	// re-placement on a lost race.
	ResetForInfraReplace(ctx context.Context, runID, taskID string) (bool, error)
	// RedispatchReschedule returns a task parked in up_for_reschedule to 'none' for
	// re-dispatch, PRESERVING try_number (reschedule is not a retry; #380).
	RedispatchReschedule(ctx context.Context, runID, taskID string) error
	// RecordDispatchFailure backs off a scheduled task after a synchronous dispatch
	// failure: it increments dispatch_attempts and sets next_dispatch_at so the
	// planner does not re-attempt until the backoff elapses (ADR 0031 Amendment A).
	RecordDispatchFailure(ctx context.Context, runID, taskID string, nextAt time.Time) error
	// RecordDispatchBackpressure backs off a scheduled task after a retriable-
	// forever cluster-backpressure dispatch failure (quota 403 / APF 429): it sets
	// next_dispatch_at WITHOUT incrementing dispatch_attempts, so the task is
	// re-offered once the backoff elapses but never accumulates toward the
	// dispatch_failed cap (ADR 0053).
	RecordDispatchBackpressure(ctx context.Context, runID, taskID string, nextAt time.Time) error
	// RecordDispatchUnavailable re-offers a task whose pod create met an
	// unavailable Kubernetes API, WITHOUT incrementing dispatch_attempts, held
	// for the time already spent in the outage clamped to [minDelay, maxDelay].
	// It returns when the task's outage window started and the next attempt;
	// applied is false when the task is no longer scheduled or queued (ADR 0093).
	RecordDispatchUnavailable(ctx context.Context, runID, taskID string, now time.Time, minDelay, maxDelay time.Duration) (since, nextAt time.Time, applied bool, err error)
	// FailDispatchExhausted fails a scheduled task as dispatch_failed once its
	// dispatch-attempt budget is spent, so the run finalizes instead of looping.
	FailDispatchExhausted(ctx context.Context, runID, taskID, reason string) error
	// FailDispatchRefused fails a task whose dispatch the executor refused (ADR
	// 0066 section 3) and spends its retry budget: the refusal is permanent, so
	// the planner must not retry it.
	FailDispatchRefused(ctx context.Context, runID, taskID, reason string) error
	// SetRunState moves a run from the state the tick's snapshot read to
	// another one, and reports whether it moved. False means the run moved on
	// after the tick read it (a clear re-opened it, a user marked it), so the
	// tick leaves it alone (#2047).
	SetRunState(ctx context.Context, runID string, from, to domain.DagRunState) (bool, error)
	// StartQueuedRun moves a queued run to running, guarded on the run still
	// being queued, and reports whether it moved. False means the run moved on
	// after the tick read it (a user marked or deleted it), so the tick leaves
	// it alone (#1766).
	StartQueuedRun(ctx context.Context, runID, tenantID string) (bool, error)
	// ClaimAlertAttempt atomically claims ONE on-failure send attempt, reporting
	// true iff this call won it. It refuses when the episode was already
	// delivered, when the attempt budget is spent, or when the backoff has not
	// elapsed, so a dead endpoint stops being retried instead of being hit once
	// per tick for the life of the run (#431, and the delivery split below).
	// It returns the attempt number won, or 0 when the claim was refused.
	ClaimAlertAttempt(ctx context.Context, runID string, maxAttempts int, backoff time.Duration) (int, error)
	// MarkRunAlertDelivered stamps the episode as paged, for the attempt the
	// caller won. Claiming and stamping were once the same operation, which meant
	// every failed send was a page lost with no retry. The attempt is part of the
	// call because the send is detached from the tick: an operator clear can start
	// a new episode mid-send, and a stamp from the superseded one must not land.
	MarkRunAlertDelivered(ctx context.Context, runID string, attempt int) error
	// ScheduledDAGs lists the scheduled DAGs the tick must look at: those with
	// no stored next due time and those whose time is at or before now, which
	// is the scheduler's clock so the store's own clock never delays a run.
	ScheduledDAGs(ctx context.Context, now time.Time) ([]ScheduledDAG, error)
	// SetNextRunAt stores when the DAG is next due, unless its schedule epoch
	// is no longer epoch.
	SetNextRunAt(ctx context.Context, tenantID, dagID string, epoch int64, at time.Time) error
	CreateScheduledRun(ctx context.Context, tenantID, dagID string, logical time.Time) error
	// SetTaskNote attaches operational context to a task instance (shown in the
	// UI), e.g. why it is queued but not running.
	SetTaskNote(ctx context.Context, runID, taskID, note string) error
}

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

type TenantStopper interface {
	FailStoppedTenantTaskInstances(ctx context.Context, tenantID string) (int64, error)
}

type TriggerDecision int