package plugin

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var (
	errExecutorAtCapacity = errors.New("executor is at capacity")
	errSchedulerShutdown  = errors.New("scheduler shutdown")
)

const (
	defaultSchedulerTick                       = 5 * time.Second
	defaultScheduledDetectionInterval          = 300 * time.Second
	defaultScheduledDetectionTimeout           = 45 * time.Second
	defaultScheduledExecutionTimeout           = 90 * time.Second
	defaultScheduledJobTypeMaxRuntime          = 30 * time.Minute
	defaultScheduledMaxResults           int32 = 1000
	defaultScheduledExecutionConcurrency       = 1
	defaultScheduledPerWorkerConcurrency       = 1
	// Hold a job on replica-host workers until their reservation load
	// reaches this; 0 disables the pin and spills immediately.
	defaultLocalQueuePin = 5
	// Wait this long after a home worker disappears before stealing
	// its queue; 0 steals immediately. Covers a reconnect blink.
	defaultOrphanRehomeGrace         = 30 * time.Second
	completedJobEchoGrace            = 30 * time.Second
	maxScheduledExecutionConcurrency = 128
	defaultScheduledRetryBackoff     = 5 * time.Second
	defaultClusterContextTimeout     = 10 * time.Second
	defaultWaitingBacklogFloor       = 8
	defaultWaitingBacklogMultiplier  = 4
	maxEstimatedRuntimeCap           = 8 * time.Hour
	// How long started jobs may drain past the run window close.
	scheduledExecutionDrainGrace = 30 * time.Minute
)

type schedulerPolicy struct {
	DetectionInterval      time.Duration
	DetectionTimeout       time.Duration
	ExecutionTimeout       time.Duration
	JobTypeMaxRuntime      time.Duration
	RetryBackoff           time.Duration
	MaxResults             int32
	ExecutionConcurrency   int
	PerWorkerConcurrency   int
	RetryLimit             int
	ExecutorReserveBackoff time.Duration
	// LocalQueuePin is K: do not spill off replica-host workers while
	// min(load(pref)) < K. Zero is the old immediate-spill path.
	LocalQueuePin int
	// OrphanRehomeGrace is how long a missing home's queue is held
	// before steal-without-K. Zero steals as soon as the worker is gone.
	OrphanRehomeGrace time.Duration
}

// laneSchedulerLoop is the main scheduling goroutine for a single lane.
// Each lane runs independently with its own timing, lock scope, and wake channel.
func (r *Plugin) laneSchedulerLoop(ls *schedulerLaneState) {
	defer r.wg.Done()
	for {
		select {
		case <-r.shutdownCh:
			return
		default:
		}

		hadJobs := r.runLaneSchedulerIteration(ls)
		r.recordLaneIterationComplete(ls, hadJobs)

		if hadJobs {
			continue
		}

		r.setLaneLoopState(ls, "", "sleeping")
		idleSleep := LaneIdleSleep(ls.lane)
		if nextRun := r.earliestLaneDetectionAt(ls.lane); !nextRun.IsZero() {
			if until := time.Until(nextRun); until <= 0 {
				idleSleep = 0
			} else if until < idleSleep {
				idleSleep = until
			}
		}
		if idleSleep <= 0 {
			continue
		}

		timer := time.NewTimer(idleSleep)
		select {
		case <-r.shutdownCh:
			timer.Stop()
			return
		case <-ls.wakeCh:
			if !timer.Stop() {
				<-timer.C
			}
			continue
		case <-timer.C:
		}
	}
}

// runLaneSchedulerIteration runs one scheduling pass for a single lane,
// processing only the job types assigned to that lane.
//
// For lanes that require a lock (e.g. LaneDefault), job types are processed
// sequentially because their volume management operations share global
// state, and the cluster admin lock is taken around each detection and each
// dispatched job rather than the whole pass, so manual shell operations
// sharing the same lock can interleave between jobs.
//
// For lanes that do not require a lock (e.g. LaneIceberg, LaneLifecycle),
// each job type runs independently in its own goroutine so they do not
// block each other.
func (r *Plugin) runLaneSchedulerIteration(ls *schedulerLaneState) bool {
	r.expireStaleJobs(time.Now().UTC())

	allJobTypes := r.registry.DetectableJobTypes()
	// Filter to only job types belonging to this lane.
	var jobTypes []string
	for _, jt := range allJobTypes {
		if JobTypeLane(jt) == ls.lane {
			jobTypes = append(jobTypes, jt)
		}
	}
	if len(jobTypes) == 0 {
		r.setLaneLoopState(ls, "", "idle")
		return false
	}

	if LaneRequiresLock(ls.lane) {
		return r.runLaneSchedulerIterationSequential(ls, jobTypes)
	}
	return r.runLaneSchedulerIterationConcurrent(ls, jobTypes)
}

// dueJobType pairs a job type with its resolved scheduling policy.
type dueJobType struct {
	jobType string
	policy  schedulerPolicy
}

// collectDueJobTypes loads policies for all job types in the lane and
// returns those whose detection interval has elapsed. It also returns
// the full set of active job type names for later pruning.
func (r *Plugin) collectDueJobTypes(ls *schedulerLaneState, jobTypes []string) (active map[string]struct{}, due []dueJobType) {
	active = make(map[string]struct{}, len(jobTypes))
	for _, jobType := range jobTypes {
		active[jobType] = struct{}{}

		policy, enabled, err := r.loadSchedulerPolicy(jobType)
		if err != nil {
			glog.Warningf("Plugin scheduler [%s] failed to load policy for %s: %v", ls.lane, jobType, err)
			continue
		}
		if !enabled {
			r.clearSchedulerJobType(jobType)
			continue
		}
		initialDelay := time.Duration(0)
		if runInfo := r.snapshotSchedulerRun(jobType); runInfo.lastRunStartedAt.IsZero() {
			initialDelay = 5 * time.Second
		}
		if !r.markDetectionDue(jobType, policy.DetectionInterval, initialDelay) {
			continue
		}
		due = append(due, dueJobType{jobType: jobType, policy: policy})
	}
	return active, due
}

// runLaneSchedulerIterationSequential processes job types one at a time.
// Used by the default lane where volume management operations must be
// serialised. The cluster admin lock is not held across the pass;
// runJobTypeIteration and dispatchScheduledProposals take it around each
// detection and each job.
func (r *Plugin) runLaneSchedulerIterationSequential(ls *schedulerLaneState, jobTypes []string) bool {
	r.setLaneLoopState(ls, "", "busy")

	active, due := r.collectDueJobTypes(ls, jobTypes)
	hadJobs := false
	for _, w := range due {
		if r.runJobTypeIteration(w.jobType, w.policy) {
			hadJobs = true
		}
	}

	r.pruneSchedulerState(ls.lane, active)
	r.pruneDetectorLeases(ls.lane, active)
	r.setLaneLoopState(ls, "", "idle")
	return hadJobs
}

// runLaneSchedulerIterationConcurrent processes each job type in its own
// goroutine so they run independently. Used by lanes (e.g. iceberg,
// lifecycle) whose job types do not share global state.
func (r *Plugin) runLaneSchedulerIterationConcurrent(ls *schedulerLaneState, jobTypes []string) bool {
	active, due := r.collectDueJobTypes(ls, jobTypes)

	r.setLaneLoopState(ls, "", "busy")

	var hadJobs atomic.Bool
	var wg sync.WaitGroup
	for _, w := range due {
		wg.Add(1)
		go func(jobType string, policy schedulerPolicy) {
			defer wg.Done()
			if r.runJobTypeIteration(jobType, policy) {
				hadJobs.Store(true)
			}
		}(w.jobType, w.policy)
	}
	wg.Wait()

	r.pruneSchedulerState(ls.lane, active)
	r.pruneDetectorLeases(ls.lane, active)
	r.setLaneLoopState(ls, "", "idle")
	return hadJobs.Load()
}

// wakeAllLanes wakes all lane scheduler goroutines.
func (r *Plugin) wakeAllLanes() {
	if r == nil {
		return
	}
	for _, ls := range r.lanes {
		select {
		case ls.wakeCh <- struct{}{}:
		default:
		}
	}
}

// wakeScheduler wakes the lane that owns the given job type, or all lanes
// if no job type is specified. Kept for backward compatibility.
func (r *Plugin) wakeScheduler() {
	r.wakeAllLanes()
}

func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) bool {
	r.recordSchedulerRunStart(jobType)
	r.clearWaitingJobQueue(jobType)
	r.setSchedulerLoopStateForJobType(jobType, "detecting")
	r.markJobTypeInFlight(jobType)
	defer r.finishDetection(jobType)

	start := time.Now().UTC()
	r.appendActivity(JobActivity{
		JobType:    jobType,
		Source:     "admin_scheduler",
		Message:    "scheduled detection started",
		Stage:      "detecting",
		OccurredAt: timeToPtr(start),
	})

	if skip, waitingCount, waitingThreshold := r.shouldSkipDetectionForWaitingJobs(jobType, policy); skip {
		r.recordSchedulerDetectionSkip(jobType, fmt.Sprintf("waiting backlog %d reached threshold %d", waitingCount, waitingThreshold))
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    fmt.Sprintf("scheduled detection skipped: waiting backlog %d reached threshold %d", waitingCount, waitingThreshold),
			Stage:      "skipped_waiting_backlog",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
		r.recordSchedulerRunComplete(jobType, "skipped")
		return false
	}

	maxRuntime := policy.JobTypeMaxRuntime
	if maxRuntime <= 0 {
		maxRuntime = defaultScheduledJobTypeMaxRuntime
	}
	jobCtx, cancel := context.WithTimeout(context.Background(), maxRuntime)
	defer cancel()

	clusterContext, err := r.loadSchedulerClusterContext(jobCtx)
	if err != nil {
		r.recordSchedulerDetectionError(jobType, err)
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    fmt.Sprintf("scheduled detection aborted: %v", err),
			Stage:      "failed",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
		r.recordSchedulerRunComplete(jobType, "error")
		return false
	}

	// Detection plans against the live topology, so it runs under the shared
	// cluster admin lock; the lock is released as soon as detection returns.
	releaseDetectionLock := func() {}
	if LaneRequiresLock(JobTypeLane(jobType)) {
		r.setSchedulerLoopStateForJobType(jobType, "waiting_for_lock")
		release, lockErr := r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s detection", jobType))
		if lockErr != nil {
			r.recordSchedulerDetectionError(jobType, lockErr)
			r.appendActivity(JobActivity{
				JobType:    jobType,
				Source:     "admin_scheduler",
				Message:    fmt.Sprintf("scheduled detection aborted: %v", lockErr),
				Stage:      "failed",
				OccurredAt: timeToPtr(time.Now().UTC()),
			})
			r.recordSchedulerRunComplete(jobType, "error")
			return false
		}
		var releaseOnce sync.Once
		releaseDetectionLock = func() { releaseOnce.Do(release) }
		r.setSchedulerLoopStateForJobType(jobType, "detecting")
	}
	defer releaseDetectionLock()

	remaining := time.Until(start.Add(maxRuntime))
	if remaining <= 0 {
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    "scheduled run timed out before detection",
			Stage:      "timeout",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
		r.recordSchedulerRunComplete(jobType, "timeout")
		return false
	}
	detect := func(avoid *streamSession) (*DetectionReport, error) {
		detectionTimeout := policy.DetectionTimeout
		if detectionTimeout <= 0 {
			detectionTimeout = defaultScheduledDetectionTimeout
		}
		if remaining := time.Until(start.Add(maxRuntime)); detectionTimeout > remaining {
			detectionTimeout = remaining
		}
		detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout)
		defer cancelDetect()
		return r.runDetectionWithReport(detectCtx, jobType, clusterContext, policy.MaxResults, avoid)
	}

	report, err := detect(nil)
	// A detector that dropped mid-request gets one replacement pick, steered
	// off the stream that dropped, in this iteration while the runtime budget
	// allows; anything else waits for the next cycle, which re-detects anyway.
	var lost *detectionWorkerDisconnectedError
	if errors.As(err, &lost) && time.Until(start.Add(maxRuntime)) > 0 {
		r.recordSchedulerDetectionError(jobType, err)
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    fmt.Sprintf("scheduled detection retrying on another worker: %v", err),
			Stage:      "retrying",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
		report, err = detect(lost.session)
	}
	releaseDetectionLock()
	if err != nil {
		r.recordSchedulerDetectionError(jobType, err)
		stage := "failed"
		status := "error"
		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
			stage = "timeout"
			status = "timeout"
		}
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    fmt.Sprintf("scheduled detection failed: %v", err),
			Stage:      stage,
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
		r.recordSchedulerRunComplete(jobType, status)
		return false
	}

	if report.Complete.GetSkippedReason() != "" {
		r.recordSchedulerDetectionSkip(jobType, report.Complete.GetSkippedReason())
		r.recordSchedulerRunComplete(jobType, "skipped")
		return false
	}
	proposals := report.Proposals
	r.appendActivity(JobActivity{
		JobType:    jobType,
		Source:     "admin_scheduler",
		Message:    fmt.Sprintf("scheduled detection completed: %d proposal(s)", len(proposals)),
		Stage:      "detected",
		OccurredAt: timeToPtr(time.Now().UTC()),
	})
	r.recordSchedulerDetectionSuccess(jobType, len(proposals))

	detected := len(proposals) > 0

	filteredByActive, skippedActive := r.filterProposalsWithActiveJobs(jobType, proposals)
	if skippedActive > 0 {
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    fmt.Sprintf("scheduled detection skipped %d proposal(s) due to active assigned/running jobs", skippedActive),
			Stage:      "deduped_active_jobs",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
	}

	if len(filteredByActive) == 0 {
		r.recordSchedulerRunComplete(jobType, "success")
		return detected
	}

	filtered := r.filterScheduledProposals(filteredByActive)
	if len(filtered) != len(filteredByActive) {
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    fmt.Sprintf("scheduled detection deduped %d proposal(s) within this run", len(filteredByActive)-len(filtered)),
			Stage:      "deduped",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
	}

	if len(filtered) == 0 {
		r.recordSchedulerRunComplete(jobType, "success")
		return detected
	}

	r.setSchedulerLoopStateForJobType(jobType, "executing")

	remaining = time.Until(start.Add(maxRuntime))
	if remaining <= 0 {
		r.appendActivity(JobActivity{
			JobType:    jobType,
			Source:     "admin_scheduler",
			Message:    "scheduled execution skipped: job type max runtime reached",
			Stage:      "timeout",
			OccurredAt: timeToPtr(time.Now().UTC()),
		})
		r.recordSchedulerRunComplete(jobType, "timeout")
		return detected
	}

	execPolicy := policy
	if execPolicy.ExecutionTimeout <= 0 {
		execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout
	}

	// jobCtx bounds how long this run keeps STARTING jobs; started jobs
	// drain past the window close (see scheduledAttemptContext), while
	// still-queued jobs are canceled and re-proposed by a later detection.
	successCount, errorCount, canceledCount := r.dispatchScheduledProposals(jobCtx, jobType, filtered, clusterContext, execPolicy)

	status := "success"
	if jobCtx.Err() != nil {
		status = "timeout"
	} else if errorCount > 0 || canceledCount > 0 {
		status = "error"
	}

	r.appendActivity(JobActivity{
		JobType:    jobType,
		Source:     "admin_scheduler",
		Message:    fmt.Sprintf("scheduled execution finished: success=%d error=%d canceled=%d", successCount, errorCount, canceledCount),
		Stage:      "executed",
		OccurredAt: timeToPtr(time.Now().UTC()),
	})
	r.recordSchedulerRunComplete(jobType, status)
	return detected
}

func (r *Plugin) loadSchedulerPolicy(jobType string) (schedulerPolicy, bool, error) {
	cfg, err := r.store.LoadJobTypeConfig(jobType)
	if err != nil {
		return schedulerPolicy{}, false, err
	}
	descriptor, err := r.store.LoadDescriptor(jobType)
	if err != nil {
		return schedulerPolicy{}, false, err
	}

	adminRuntime := deriveSchedulerAdminRuntime(cfg, descriptor)
	if adminRuntime == nil {
		return schedulerPolicy{}, false, nil
	}
	if !adminRuntime.Enabled {
		return schedulerPolicy{}, false, nil
	}

	policy := schedulerPolicy{
		DetectionInterval:      durationFromMinutes(adminRuntime.DetectionIntervalMinutes, defaultScheduledDetectionInterval),
		DetectionTimeout:       durationFromSeconds(adminRuntime.DetectionTimeoutSeconds, defaultScheduledDetectionTimeout),
		ExecutionTimeout:       durationFromSeconds(adminRuntime.ExecutionTimeoutSeconds, defaultScheduledExecutionTimeout),
		JobTypeMaxRuntime:      durationFromSeconds(adminRuntime.JobTypeMaxRuntimeSeconds, defaultScheduledJobTypeMaxRuntime),
		RetryBackoff:           durationFromSeconds(adminRuntime.RetryBackoffSeconds, defaultScheduledRetryBackoff),
		MaxResults:             adminRuntime.MaxJobsPerDetection,
		ExecutionConcurrency:   int(adminRuntime.GlobalExecutionConcurrency),
		PerWorkerConcurrency:   int(adminRuntime.PerWorkerExecutionConcurrency),
		RetryLimit:             int(adminRuntime.RetryLimit),
		ExecutorReserveBackoff: 200 * time.Millisecond,
		LocalQueuePin:          defaultLocalQueuePin,
		OrphanRehomeGrace:      defaultOrphanRehomeGrace,
	}

	if policy.DetectionInterval < r.schedulerTick {
		policy.DetectionInterval = r.schedulerTick
	}
	if policy.MaxResults <= 0 {
		policy.MaxResults = defaultScheduledMaxResults
	}
	if policy.ExecutionConcurrency <= 0 {
		policy.ExecutionConcurrency = defaultScheduledExecutionConcurrency
	}
	if policy.ExecutionConcurrency > maxScheduledExecutionConcurrency {
		policy.ExecutionConcurrency = maxScheduledExecutionConcurrency
	}
	if policy.PerWorkerConcurrency <= 0 {
		policy.PerWorkerConcurrency = defaultScheduledPerWorkerConcurrency
	}
	if policy.PerWorkerConcurrency > policy.ExecutionConcurrency {
		policy.PerWorkerConcurrency = policy.ExecutionConcurrency
	}
	if policy.RetryLimit < 0 {
		policy.RetryLimit = 0
	}
	if policy.JobTypeMaxRuntime <= 0 {
		policy.JobTypeMaxRuntime = defaultScheduledJobTypeMaxRuntime
	}

	if policy.ExecutionTimeout < defaultScheduledExecutionTimeout {
		policy.ExecutionTimeout = defaultScheduledExecutionTimeout
	}
	if policy.LocalQueuePin < 0 {
		policy.LocalQueuePin = 0
	}

	return policy, true, nil
}

func (r *Plugin) ListSchedulerStates() ([]SchedulerJobTypeState, error) {
	jobTypes, err := r.ListKnownJobTypes()
	if err != nil {
		return nil, err
	}

	r.schedulerMu.Lock()
	nextDetectionAt := make(map[string]time.Time, len(r.nextDetectionAt))
	for jobType, nextRun := range r.nextDetectionAt {
		nextDetectionAt[jobType] = nextRun
	}
	detectionInFlight := make(map[string]bool, len(r.detectionInFlight))
	for jobType, inFlight := range r.detectionInFlight {
		detectionInFlight[jobType] = inFlight
	}
	r.schedulerMu.Unlock()

	states := make([]SchedulerJobTypeState, 0, len(jobTypes))
	for _, jobTypeInfo := range jobTypes {
		jobType := jobTypeInfo.JobType
		state := SchedulerJobTypeState{
			JobType:           jobType,
			Lane:              string(JobTypeLane(jobType)),
			DetectionInFlight: detectionInFlight[jobType],
		}

		if nextRun, ok := nextDetectionAt[jobType]; ok && !nextRun.IsZero() {
			nextRunUTC := nextRun.UTC()
			state.NextDetectionAt = &nextRunUTC
		}

		policy, enabled, loadErr := r.loadSchedulerPolicy(jobType)

		if loadErr != nil {
			state.PolicyError = loadErr.Error()
		} else {
			state.Enabled = enabled
			if enabled {
				state.DetectionIntervalMinutes = minutesFromDuration(policy.DetectionInterval)
				state.DetectionTimeoutSeconds = secondsFromDuration(policy.DetectionTimeout)
				state.ExecutionTimeoutSeconds = secondsFromDuration(policy.ExecutionTimeout)
				state.JobTypeMaxRuntimeSeconds = secondsFromDuration(policy.JobTypeMaxRuntime)
				state.MaxJobsPerDetection = policy.MaxResults
				state.GlobalExecutionConcurrency = policy.ExecutionConcurrency
				state.PerWorkerExecutionConcurrency = policy.PerWorkerConcurrency
				state.RetryLimit = policy.RetryLimit
				state.RetryBackoffSeconds = secondsFromDuration(policy.RetryBackoff)
			}
		}

		runInfo := r.snapshotSchedulerRun(jobType)
		if !runInfo.lastRunStartedAt.IsZero() {
			at := runInfo.lastRunStartedAt
			state.LastRunStartedAt = &at
		}
		if !runInfo.lastRunCompletedAt.IsZero() {
			at := runInfo.lastRunCompletedAt
			state.LastRunCompletedAt = &at
		}
		if runInfo.lastRunStatus != "" {
			state.LastRunStatus = runInfo.lastRunStatus
		}

		leasedWorkerID := r.getDetectorLease(jobType)
		if leasedWorkerID != "" {
			state.DetectorWorkerID = leasedWorkerID
			if worker, ok := r.registry.Get(leasedWorkerID); ok {
				if capability := worker.Capabilities[jobType]; capability != nil && capability.CanDetect {
					state.DetectorAvailable = true
				}
			}
		}
		if state.DetectorWorkerID == "" {
			detector, detectorErr := r.registry.PickDetector(jobType)
			if detectorErr == nil && detector != nil {
				state.DetectorAvailable = true
				state.DetectorWorkerID = detector.WorkerID
			}
		}

		executors, executorErr := r.registry.ListExecutors(jobType)
		if executorErr == nil {
			state.ExecutorWorkerCount = len(executors)
		}

		states = append(states, state)
	}

	return states, nil
}

func deriveSchedulerAdminRuntime(
	cfg *plugin_pb.PersistedJobTypeConfig,
	descriptor *plugin_pb.JobTypeDescriptor,
) *plugin_pb.AdminRuntimeConfig {
	if cfg != nil && cfg.AdminRuntime != nil {
		adminConfig := proto.Clone(cfg.AdminRuntime).(*plugin_pb.AdminRuntimeConfig)
		// Overlay descriptor defaults for any zero numeric fields. Persisted
		// configs from older versions have no execution_timeout_seconds, and
		// without this overlay the scheduler would fall back to the 90s
		// default instead of the handler's declared baseline.
		if descriptor != nil && descriptor.AdminRuntimeDefaults != nil {
			defaults := descriptor.AdminRuntimeDefaults
			if adminConfig.DetectionIntervalMinutes <= 0 {
				adminConfig.DetectionIntervalMinutes = defaults.DetectionIntervalMinutes
			}
			if adminConfig.DetectionTimeoutSeconds <= 0 {
				adminConfig.DetectionTimeoutSeconds = defaults.DetectionTimeoutSeconds
			}
			if adminConfig.MaxJobsPerDetection <= 0 {
				adminConfig.MaxJobsPerDetection = defaults.MaxJobsPerDetection
			}
			if adminConfig.GlobalExecutionConcurrency <= 0 {
				adminConfig.GlobalExecutionConcurrency = defaults.GlobalExecutionConcurrency
			}
			if adminConfig.PerWorkerExecutionConcurrency <= 0 {
				adminConfig.PerWorkerExecutionConcurrency = defaults.PerWorkerExecutionConcurrency
			}
			if adminConfig.RetryBackoffSeconds <= 0 {
				adminConfig.RetryBackoffSeconds = defaults.RetryBackoffSeconds
			}
			if adminConfig.JobTypeMaxRuntimeSeconds <= 0 {
				adminConfig.JobTypeMaxRuntimeSeconds = defaults.JobTypeMaxRuntimeSeconds
			}
			if adminConfig.ExecutionTimeoutSeconds <= 0 {
				adminConfig.ExecutionTimeoutSeconds = defaults.ExecutionTimeoutSeconds
			}
		}
		return adminConfig
	}

	if descriptor == nil || descriptor.AdminRuntimeDefaults == nil {
		return nil
	}

	defaults := descriptor.AdminRuntimeDefaults
	return &plugin_pb.AdminRuntimeConfig{
		Enabled:                       defaults.Enabled,
		DetectionIntervalMinutes:      defaults.DetectionIntervalMinutes,
		DetectionTimeoutSeconds:       defaults.DetectionTimeoutSeconds,
		MaxJobsPerDetection:           defaults.MaxJobsPerDetection,
		GlobalExecutionConcurrency:    defaults.GlobalExecutionConcurrency,
		PerWorkerExecutionConcurrency: defaults.PerWorkerExecutionConcurrency,
		RetryLimit:                    defaults.RetryLimit,
		RetryBackoffSeconds:           defaults.RetryBackoffSeconds,
		JobTypeMaxRuntimeSeconds:      defaults.JobTypeMaxRuntimeSeconds,
		ExecutionTimeoutSeconds:       defaults.ExecutionTimeoutSeconds,
	}
}

func (r *Plugin) markDetectionDue(jobType string, interval, initialDelay time.Duration) bool {
	now := time.Now().UTC()

	r.schedulerMu.Lock()
	defer r.schedulerMu.Unlock()

	if r.detectionInFlight[jobType] {
		return false
	}

	nextRun, exists := r.nextDetectionAt[jobType]
	if exists && now.Before(nextRun) {
		return false
	}
	if !exists && initialDelay > 0 {
		r.nextDetectionAt[jobType] = now.Add(initialDelay)
		return false
	}

	r.nextDetectionAt[jobType] = now.Add(interval)
	r.detectionInFlight[jobType] = true
	return true
}

// earliestLaneDetectionAt returns the earliest next detection time among
// job types that belong to the given lane.
func (r *Plugin) earliestLaneDetectionAt(lane SchedulerLane) time.Time {
	if r == nil {
		return time.Time{}
	}

	r.schedulerMu.Lock()
	defer r.schedulerMu.Unlock()

	var earliest time.Time
	for jobType, nextRun := range r.nextDetectionAt {
		if JobTypeLane(jobType) != lane {
			continue
		}
		if nextRun.IsZero() {
			continue
		}
		if earliest.IsZero() || nextRun.Before(earliest) {
			earliest = nextRun
		}
	}

	return earliest
}

// earliestNextDetectionAt returns the earliest next detection time across
// all job types regardless of lane. Kept for backward compatibility and
// the global scheduler status API.
func (r *Plugin) earliestNextDetectionAt() time.Time {
	if r == nil {
		return time.Time{}
	}

	r.schedulerMu.Lock()
	defer r.schedulerMu.Unlock()

	var earliest time.Time
	for _, nextRun := range r.nextDetectionAt {
		if nextRun.IsZero() {
			continue
		}
		if earliest.IsZero() || nextRun.Before(earliest) {
			earliest = nextRun
		}
	}

	return earliest
}

func (r *Plugin) markJobTypeInFlight(jobType string) {
	r.schedulerMu.Lock()
	r.detectionInFlight[jobType] = true
	r.schedulerMu.Unlock()
}

func (r *Plugin) finishDetection(jobType string) {
	r.schedulerMu.Lock()
	delete(r.detectionInFlight, jobType)
	r.schedulerMu.Unlock()
}

// pruneSchedulerState removes next-detection state for inactive job types
// in lane only. Prune must not touch other lanes' entries in the global map.
func (r *Plugin) pruneSchedulerState(lane SchedulerLane, activeJobTypes map[string]struct{}) {
	r.schedulerMu.Lock()
	defer r.schedulerMu.Unlock()

	for jobType := range r.nextDetectionAt {
		if JobTypeLane(jobType) != lane {
			continue
		}
		if _, ok := activeJobTypes[jobType]; !ok {
			delete(r.nextDetectionAt, jobType)
			delete(r.detectionInFlight, jobType)
		}
	}
}

func (r *Plugin) clearSchedulerJobType(jobType string) {
	r.schedulerMu.Lock()
	delete(r.nextDetectionAt, jobType)
	delete(r.detectionInFlight, jobType)
	r.schedulerMu.Unlock()
	r.clearDetectorLease(jobType, "")
}

// pruneDetectorLeases removes detector leases for inactive job types in lane only.
func (r *Plugin) pruneDetectorLeases(lane SchedulerLane, activeJobTypes map[string]struct{}) {
	r.detectorLeaseMu.Lock()
	defer r.detectorLeaseMu.Unlock()

	for jobType := range r.detectorLeases {
		if JobTypeLane(jobType) != lane {
			continue
		}
		if _, ok := activeJobTypes[jobType]; !ok {
			delete(r.detectorLeases, jobType)
		}
	}
}

func (r *Plugin) loadSchedulerClusterContext(ctx context.Context) (*plugin_pb.ClusterContext, error) {
	if r.clusterContextProvider == nil {
		return nil, fmt.Errorf("cluster context provider is not configured")
	}

	if ctx == nil {
		ctx = context.Background()
	}
	clusterCtx, cancel := context.WithTimeout(ctx, defaultClusterContextTimeout)
	defer cancel()

	clusterContext, err := r.clusterContextProvider(clusterCtx)
	if err != nil {
		return nil, err
	}
	if clusterContext == nil {
		return nil, fmt.Errorf("cluster context provider returned nil")
	}
	return clusterContext, nil
}

func (r *Plugin) dispatchScheduledProposals(
	ctx context.Context,
	jobType string,
	proposals []*plugin_pb.JobProposal,
	clusterContext *plugin_pb.ClusterContext,
	policy schedulerPolicy,
) (int, int, int) {
	if ctx == nil {
		ctx = context.Background()
	}

	// Volume management jobs mutate volume placement, so each one runs under
	// the shared cluster admin lock. Taking it per job instead of around the
	// whole dispatch lets manual shell operations interleave between jobs.
	jobsRequireLock := LaneRequiresLock(JobTypeLane(jobType))

	jobs := make([]*plugin_pb.JobSpec, 0, len(proposals))
	for index, proposal := range proposals {
		select {
		case <-r.shutdownCh:
			for _, job := range jobs {
				r.cancelQueuedJob(job, errSchedulerShutdown)
			}
			return 0, 0, len(jobs)
		default:
		}
		job := buildScheduledJobSpec(jobType, proposal, index)
		r.trackExecutionQueued(job)
		jobs = append(jobs, job)
	}

	queues := r.assignJobsToWorkerQueues(jobType, jobs, policy)

	var wg sync.WaitGroup
	var statsMu sync.Mutex
	successCount := 0
	errorCount := 0
	canceledCount := 0

	workerCount := policy.ExecutionConcurrency
	if workerCount < 1 {
		workerCount = 1
	}

	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			var blockedSince time.Time
			for {
				select {
				case <-r.shutdownCh:
					return
				default:
				}
				if ctx.Err() != nil {
					return
				}
				if queues.isEmpty() {
					return
				}

				item, targetID := r.takeQueuedJob(queues, jobType, policy)
				if item == nil {
					if queues.isEmpty() {
						return
					}
					if blockedSince.IsZero() {
						blockedSince = time.Now()
					}
					if wait, reason := r.executorCapacityOutlook(jobType, policy, time.Since(blockedSince)); !wait {
						given := r.giveUpQueuedJobs(queues, jobType, reason)
						statsMu.Lock()
						canceledCount += given
						statsMu.Unlock()
						return
					}
					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
						return
					}
					continue
				}
				if ctx.Err() != nil {
					queues.pushFront(item)
					return
				}

				executor, release, reserveErr := r.reserveExecutorByID(jobType, policy, targetID, item.job.JobId)
				if reserveErr != nil {
					queues.pushFront(item)
					if ctx.Err() != nil {
						return
					}
					if blockedSince.IsZero() {
						blockedSince = time.Now()
					}
					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
						return
					}
					continue
				}
				blockedSince = time.Time{}

				var releaseJobLock func()
				if jobsRequireLock {
					var lockErr error
					releaseJobLock, lockErr = r.acquireAdminLock(fmt.Sprintf("plugin scheduler %s job %s", jobType, item.job.JobId))
					if lockErr != nil {
						release()
						statsMu.Lock()
						errorCount++
						statsMu.Unlock()
						r.appendActivity(JobActivity{
							JobID:      item.job.JobId,
							JobType:    item.job.JobType,
							Source:     "admin_scheduler",
							Message:    fmt.Sprintf("scheduled execution lock failed: %v", lockErr),
							Stage:      "failed",
							OccurredAt: timeToPtr(time.Now().UTC()),
						})
						continue
					}
					if ctx.Err() != nil {
						releaseJobLock()
						release()
						queues.pushFront(item)
						return
					}
				}

				err := r.executeScheduledJobWithExecutor(ctx, executor, item.job, clusterContext, policy)
				if releaseJobLock != nil {
					releaseJobLock()
				}
				release()
				if errors.Is(err, errExecutorAtCapacity) {
					queues.pushFront(item)
					if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
						return
					}
					continue
				}
				if err != nil {
					if ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
						r.cancelQueuedJob(item.job, err)
						statsMu.Lock()
						canceledCount++
						statsMu.Unlock()
						continue
					}
					statsMu.Lock()
					errorCount++
					statsMu.Unlock()
					r.appendActivity(JobActivity{
						JobID:      item.job.JobId,
						JobType:    item.job.JobType,
						Source:     "admin_scheduler",
						Message:    fmt.Sprintf("scheduled execution failed: %v", err),
						Stage:      "failed",
						OccurredAt: timeToPtr(time.Now().UTC()),
					})
					continue
				}

				statsMu.Lock()
				successCount++
				statsMu.Unlock()
			}
		}()
	}

	wg.Wait()

	drainErr := ctx.Err()
	if drainErr == nil {
		drainErr = errSchedulerShutdown
	}
	for _, item := range queues.drain() {
		r.cancelQueuedJob(item.job, drainErr)
		canceledCount++
	}

	return successCount, errorCount, canceledCount
}

// executorCapacityOutlook decides whether a dispatch goroutine that found
// no executor for jobType should keep waiting. Capacity held by this lane's
// own jobs comes back when they complete, so that is worth the wait.
// Capacity held by other lanes or by work this admin never dispatched
// clears on a schedule the lane has no view of, and a lane that serialises
// its job types would idle every other type in it for the whole runtime
// budget; those jobs are given up now, with the reason, and re-proposed by
// the next detection. A missing worker, or a free one that the queue's pin
// keeps off a busy home, is waited for only as long as the orphan grace.
func (r *Plugin) executorCapacityOutlook(jobType string, policy schedulerPolicy, blockedFor time.Duration) (bool, string) {
	executors, err := r.registry.ListExecutors(jobType)
	if err != nil || len(executors) == 0 {
		if blockedFor < policy.OrphanRehomeGrace {
			return true, ""
		}
		return false, fmt.Sprintf("no executor worker available for %s", jobType)
	}

	lane := JobTypeLane(jobType)
	held := make([]string, 0, len(executors))
	free := false
	r.execResMu.Lock()
	defer r.execResMu.Unlock()
	for _, executor := range executors {
		if executor == nil {
			continue
		}
		fits, blocking, limit := r.executorCapacityLocked(executor, jobType, policy)
		if fits {
			free = true
			continue
		}
		if blocking.reservedLanes[lane] > 0 {
			return true, ""
		}
		if limit <= 0 {
			held = append(held, fmt.Sprintf("worker %s accepts no %s jobs", executor.WorkerID, jobType))
			continue
		}
		held = append(held, fmt.Sprintf("worker %s has %d/%d slots used by %s",
			executor.WorkerID, blocking.total(), limit, blocking.holders()))
	}
	if free && (len(held) == 0 || blockedFor < policy.OrphanRehomeGrace) {
		return true, ""
	}
	return false, fmt.Sprintf("no executor capacity for %s: %s", jobType, strings.Join(held, "; "))
}

// giveUpQueuedJobs cancels whatever is still queued for jobType with the
// reason no executor can take it, so the lane moves on and the next
// detection proposes the work again. Returns how many jobs it canceled.
func (r *Plugin) giveUpQueuedJobs(queues *workerJobQueues, jobType, reason string) int {
	items := queues.drain()
	if len(items) == 0 {
		return 0
	}
	r.appendActivity(JobActivity{
		JobType:    jobType,
		Source:     "admin_scheduler",
		Message:    reason,
		Stage:      "skipped_no_executor",
		OccurredAt: timeToPtr(time.Now().UTC()),
	})
	cause := errors.New(reason)
	for _, item := range items {
		r.cancelQueuedJob(item.job, cause)
	}
	return len(items)
}

func (r *Plugin) reserveScheduledExecutor(
	ctx context.Context,
	jobType string,
	policy schedulerPolicy,
	preferredHosts []string,
) (*WorkerSession, func(), error) {
	if ctx == nil {
		ctx = context.Background()
	}

	deadline := time.Now().Add(policy.ExecutionTimeout)
	if policy.ExecutionTimeout <= 0 {
		deadline = time.Now().Add(10 * time.Minute) // Default cap
	}
	if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) {
		deadline = ctxDeadline
	}

	for {
		select {
		case <-r.shutdownCh:
			return nil, nil, fmt.Errorf("plugin is shutting down")
		default:
		}
		if ctx.Err() != nil {
			return nil, nil, ctx.Err()
		}

		if time.Now().After(deadline) {
			return nil, nil, fmt.Errorf("timed out waiting for executor capacity for %s", jobType)
		}

		executors, err := r.registry.ListExecutors(jobType, preferredHosts...)
		if err != nil {
			if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
				if ctx.Err() != nil {
					return nil, nil, ctx.Err()
				}
				return nil, nil, fmt.Errorf("plugin is shutting down")
			}
			continue
		}

		pref, rest := partitionByPreferredHost(executors, preferredHosts)

		if len(pref) > 0 {
			sortExecutorsByLoad(pref, r.executorWorkerLoad)
			if executor, release, ok := r.tryReserveFirst(pref, jobType, "", policy); ok {
				return executor, release, nil
			}
		}

		sortExecutorsByLoad(rest, r.executorWorkerLoad)
		if executor, release, ok := r.tryReserveFirst(rest, jobType, "", policy); ok {
			return executor, release, nil
		}

		if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) {
			if ctx.Err() != nil {
				return nil, nil, ctx.Err()
			}
			return nil, nil, fmt.Errorf("plugin is shutting down")
		}
	}
}

func (r *Plugin) tryReserveFirst(executors []*WorkerSession, jobType, jobID string, policy schedulerPolicy) (*WorkerSession, func(), bool) {
	for _, executor := range executors {
		release, ok := r.reserveExecutorSlot(executor, jobType, jobID, policy)
		if ok {
			return executor, release, true
		}
	}
	return nil, nil, false
}

func (r *Plugin) reserveExecutorByID(jobType string, policy schedulerPolicy, workerID, jobID string) (*WorkerSession, func(), error) {
	workerID = strings.TrimSpace(workerID)
	if workerID == "" {
		return nil, nil, fmt.Errorf("executor id is empty")
	}
	executors, err := r.registry.ListExecutors(jobType)
	if err != nil {
		return nil, nil, err
	}
	for _, executor := range executors {
		if executor == nil || executor.WorkerID != workerID {
			continue
		}
		release, ok := r.reserveExecutorSlot(executor, jobType, jobID, policy)
		if !ok {
			return nil, nil, errExecutorAtCapacity
		}
		return executor, release, nil
	}
	return nil, nil, fmt.Errorf("executor %s is not available for %s", workerID, jobType)
}

func sortExecutorsByLoad(executors []*WorkerSession, load func(*WorkerSession) int) {
	sort.SliceStable(executors, func(i, j int) bool {
		if executors[i] == nil {
			return false
		}
		if executors[j] == nil {
			return true
		}
		li, lj := load(executors[i]), load(executors[j])
		if li != lj {
			return li < lj
		}
		return executors[i].WorkerID < executors[j].WorkerID
	})
}

// execReservation is an execution slot this admin holds on a worker for one
// job, from the moment the job is picked until its completion is handled.
// The job id is what tells a slot the admin already counts from one only the
// worker's heartbeat knows about. The worker reports the completion before
// the heartbeat that frees the slot, so a released reservation is kept for
// completedJobEchoGrace to claim that echo.
type execReservation struct {
	workerID   string
	jobType    string
	jobID      string
	reservedAt time.Time
	releasedAt time.Time
}

// executorLoad is how many execution slots on one worker the admin must
// treat as taken. reserved is this admin's own reservations, including the
// echo of a job it just completed; foreign is work the worker reports that
// no reservation covers: manual runs and work still running from before a
// reconnect.
type executorLoad struct {
	reserved   int
	foreign    int
	unitemized int
	// The lane of each reservation and the job types behind the whole load,
	// so a blocked dispatch can tell whether the load is its own lane's and
	// name what holds the worker when it is not.
	reservedLanes map[SchedulerLane]int
	byType        map[string]int
}

func (l executorLoad) total() int {
	return l.reserved + l.foreign
}

// holders names the job types behind the load, busiest first.
func (l executorLoad) holders() string {
	if len(l.byType) == 0 {
		return "work the worker did not itemize"
	}
	types := make([]string, 0, len(l.byType))
	for jobType := range l.byType {
		types = append(types, jobType)
	}
	sort.Slice(types, func(i, j int) bool {
		if l.byType[types[i]] != l.byType[types[j]] {
			return l.byType[types[i]] > l.byType[types[j]]
		}
		return types[i] < types[j]
	})
	parts := make([]string, 0, len(types)+1)
	add := func(holder string, n int) {
		if n > 1 {
			holder = fmt.Sprintf("%s x%d", holder, n)
		}
		parts = append(parts, holder)
	}
	for _, jobType := range types {
		add(jobType, l.byType[jobType])
	}
	if l.unitemized > 0 {
		add("unitemized work", l.unitemized)
	}
	out := strings.Join(parts, ", ")
	if l.reserved == 0 {
		out += " this admin did not dispatch"
	}
	return out
}

// executorLoadLocked tallies the worker's slots as the admin sees them:
// jobType "" counts work of every type, otherwise only that type. A
// heartbeat entry whose job id matches a reservation is the same slot seen
// twice and counts once. Slots the worker counts without itemizing them go
// first to this admin's jobs it does not list, and only from a heartbeat
// that came after the job's reservation: a pending job's from then on, a
// completed job's until the worker's next heartbeat. The rest are of unknown
// type and count against every type. Caller holds r.execResMu.
func (r *Plugin) executorLoadLocked(executor *WorkerSession, jobType string) executorLoad {
	load := executorLoad{
		reservedLanes: make(map[SchedulerLane]int),
		byType:        make(map[string]int),
	}
	if executor == nil {
		return load
	}
	countReserved := func(res execReservation) {
		if jobType != "" && res.jobType != jobType {
			return
		}
		load.reserved++
		load.reservedLanes[JobTypeLane(res.jobType)]++
		load.byType[res.jobType]++
	}
	running := executor.Heartbeat.GetRunningWork()
	listed := make(map[string]struct{}, len(running))
	unitemized := int(executor.Heartbeat.GetExecutionSlotsUsed())
	for _, work := range running {
		if work.GetKind() == plugin_pb.WorkKind_WORK_KIND_EXECUTION {
			listed[work.GetWorkId()] = struct{}{}
			unitemized--
		}
	}
	workerID := strings.TrimSpace(executor.WorkerID)
	claimed := make(map[string]struct{})
	var unlistedEchoes []execReservation
	now := time.Now()
	for token, res := range r.execRes {
		released := !res.releasedAt.IsZero()
		if released && now.Sub(res.releasedAt) > completedJobEchoGrace {
			delete(r.execRes, token)
			continue
		}
		if res.workerID != workerID {
			continue
		}
		isListed := false
		if res.jobID != "" {
			claimed[res.jobID] = struct{}{}
			_, isListed = listed[res.jobID]
		}
		switch {
		case !released:
			if !isListed && !executor.heartbeatAt.Before(res.reservedAt) {
				unitemized--
			}
			countReserved(res)
		case isListed:
			countReserved(res)
		case !executor.heartbeatAt.Before(res.reservedAt) && executor.heartbeatAt.Before(res.releasedAt):
			unlistedEchoes = append(unlistedEchoes, res)
		}
	}
	sort.Slice(unlistedEchoes, func(i, j int) bool {
		return unlistedEchoes[i].releasedAt.After(unlistedEchoes[j].releasedAt)
	})
	for _, res := range unlistedEchoes {
		if unitemized <= 0 {
			break
		}
		unitemized--
		countReserved(res)
	}
	if unitemized > 0 {
		load.foreign += unitemized
		load.unitemized = unitemized
	}
	for _, work := range running {
		if work.GetKind() != plugin_pb.WorkKind_WORK_KIND_EXECUTION {
			continue
		}
		if _, ok := claimed[work.GetWorkId()]; ok {
			continue
		}
		if jobType != "" && work.GetJobType() != jobType {
			continue
		}
		load.foreign++
		load.byType[work.GetJobType()]++
	}
	return load
}

// executorCapacityLocked reports whether one more jobType job fits on the
// worker. per_worker_execution_concurrency is a per-job-type limit, so it is
// checked against jobType work alone; the worker's advertised slot total is
// checked against work of every type. When the job does not fit, the load
// and the limit that refused it come back for the caller to act on or
// report. Caller holds r.execResMu.
func (r *Plugin) executorCapacityLocked(executor *WorkerSession, jobType string, policy schedulerPolicy) (bool, executorLoad, int) {
	limit := schedulerWorkerExecutionLimit(executor, jobType, policy)
	if limit <= 0 {
		return false, executorLoad{}, 0
	}
	if load := r.executorLoadLocked(executor, jobType); load.total() >= limit {
		return false, load, limit
	}
	if total := int(executor.Heartbeat.GetExecutionSlotsTotal()); total > 0 {
		if load := r.executorLoadLocked(executor, ""); load.total() >= total {
			return false, load, total
		}
	}
	return true, executorLoad{}, limit
}

func (r *Plugin) executorHasCapacity(executor *WorkerSession, jobType string, policy schedulerPolicy) bool {
	if executor == nil || strings.TrimSpace(executor.WorkerID) == "" {
		return false
	}
	r.execResMu.Lock()
	defer r.execResMu.Unlock()
	fits, _, _ := r.executorCapacityLocked(executor, jobType, policy)
	return fits
}

// executorWorkerLoad is the worker's load across every job type, the
// number to balance by.
func (r *Plugin) executorWorkerLoad(executor *WorkerSession) int {
	if executor == nil {
		return 0
	}
	r.execResMu.Lock()
	defer r.execResMu.Unlock()
	return r.executorLoadLocked(executor, "").total()
}

// reserveExecutorSlot holds a slot on the worker for jobID if one more
// jobType job fits. The release returned gives it back; the reservation
// is the admin's own count of the slot until then.
func (r *Plugin) reserveExecutorSlot(
	executor *WorkerSession,
	jobType string,
	jobID string,
	policy schedulerPolicy,
) (func(), bool) {
	if executor == nil || strings.TrimSpace(executor.WorkerID) == "" {
		return nil, false
	}

	r.execResMu.Lock()
	defer r.execResMu.Unlock()
	if fits, _, _ := r.executorCapacityLocked(executor, jobType, policy); !fits {
		return nil, false
	}
	jobID = strings.TrimSpace(jobID)
	if jobID != "" {
		for token, res := range r.execRes {
			if res.jobID == jobID && !res.releasedAt.IsZero() {
				delete(r.execRes, token)
			}
		}
	}
	r.execResSeq++
	token := r.execResSeq
	r.execRes[token] = execReservation{
		workerID:   strings.TrimSpace(executor.WorkerID),
		jobType:    jobType,
		jobID:      jobID,
		reservedAt: time.Now(),
	}
	release := func() {
		r.execResMu.Lock()
		defer r.execResMu.Unlock()
		res, ok := r.execRes[token]
		if !ok || !res.releasedAt.IsZero() {
			return
		}
		if res.jobID == "" {
			delete(r.execRes, token)
			return
		}
		res.releasedAt = time.Now()
		r.execRes[token] = res
	}
	return release, true
}

// schedulerWorkerExecutionLimit is the most jobType jobs one worker may run
// at once for this admin: per_worker_execution_concurrency, capped by what
// the worker declared for the type and by its slot total. It bounds jobType
// work alone; other job types on the same worker count only against the
// slot total (see executorCapacityLocked).
func schedulerWorkerExecutionLimit(executor *WorkerSession, jobType string, policy schedulerPolicy) int {
	limit := policy.PerWorkerConcurrency
	if limit <= 0 {
		limit = defaultScheduledPerWorkerConcurrency
	}

	if capability := executor.Capabilities[jobType]; capability != nil && capability.MaxExecutionConcurrency > 0 {
		capLimit := int(capability.MaxExecutionConcurrency)
		if capLimit < limit {
			limit = capLimit
		}
	}

	if executor.Heartbeat != nil && executor.Heartbeat.ExecutionSlotsTotal > 0 {
		heartbeatLimit := int(executor.Heartbeat.ExecutionSlotsTotal)
		if heartbeatLimit < limit {
			limit = heartbeatLimit
		}
	}

	if limit < 0 {
		return 0
	}
	return limit
}

func (r *Plugin) executeScheduledJobWithExecutor(
	ctx context.Context,
	executor *WorkerSession,
	job *plugin_pb.JobSpec,
	clusterContext *plugin_pb.ClusterContext,
	policy schedulerPolicy,
) error {
	maxAttempts := policy.RetryLimit + 1
	if maxAttempts < 1 {
		maxAttempts = 1
	}

	var lastErr error
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		select {
		case <-r.shutdownCh:
			return fmt.Errorf("plugin is shutting down")
		default:
		}
		if ctx != nil && ctx.Err() != nil {
			return ctx.Err()
		}

		// Use the job's estimated runtime if provided and larger than the
		// default execution timeout. This lets handlers like vacuum scale
		// the timeout based on volume size so large volumes are not killed.
		timeout := policy.ExecutionTimeout
		estimated := scheduledEstimatedRuntime(job.Parameters)
		if estimated > timeout {
			timeout = estimated
		}
		execCtx, cancel := scheduledAttemptContext(ctx, timeout, estimated)
		_, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt))
		cancel()
		if err == nil {
			return nil
		}
		if isExecutorAtCapacityError(err) {
			return errExecutorAtCapacity
		}
		lastErr = err

		if attempt < maxAttempts {
			r.appendActivity(JobActivity{
				JobID:      job.JobId,
				JobType:    job.JobType,
				Source:     "admin_scheduler",
				Message:    fmt.Sprintf("retrying job attempt %d/%d after error: %v", attempt, maxAttempts, err),
				Stage:      "retry",
				OccurredAt: timeToPtr(time.Now().UTC()),
			})
			if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.RetryBackoff) {
				if ctx != nil && ctx.Err() != nil {
					return ctx.Err()
				}
				return fmt.Errorf("plugin is shutting down")
			}
		}
	}

	if lastErr == nil {
		lastErr = fmt.Errorf("execution failed without an explicit error")
	}
	return lastErr
}

// scheduledEstimatedRuntime returns the job's declared estimated runtime,
// capped at maxEstimatedRuntimeCap. The seconds are capped before the
// Duration conversion so oversized values cannot overflow.
func scheduledEstimatedRuntime(parameters map[string]*plugin_pb.ConfigValue) time.Duration {
	est, ok := parameters["estimated_runtime_seconds"]
	if !ok {
		return 0
	}
	v := est.GetInt64Value()
	if v <= 0 {
		return 0
	}
	if v > int64(maxEstimatedRuntimeCap/time.Second) {
		v = int64(maxEstimatedRuntimeCap / time.Second)
	}
	return time.Duration(v) * time.Second
}

// scheduledAttemptContext bounds one execution attempt: its own timeout,
// capped at the window deadline plus scheduledExecutionDrainGrace so a
// draining job cannot hold the lane indefinitely, but never below the
// job's declared estimated runtime.
func scheduledAttemptContext(window context.Context, timeout, estimated time.Duration) (context.Context, context.CancelFunc) {
	deadline := time.Now().Add(timeout)
	if window != nil {
		if windowEnd, ok := window.Deadline(); ok {
			if drainDeadline := windowEnd.Add(scheduledExecutionDrainGrace); deadline.After(drainDeadline) {
				if floor := time.Now().Add(estimated); floor.After(drainDeadline) {
					drainDeadline = floor
				}
				deadline = drainDeadline
			}
		}
	}
	return context.WithDeadline(context.Background(), deadline)
}

func (r *Plugin) shouldSkipDetectionForWaitingJobs(jobType string, policy schedulerPolicy) (bool, int, int) {
	waitingCount := r.countWaitingTrackedJobs(jobType)
	threshold := waitingBacklogThreshold(policy)
	if threshold <= 0 {
		return false, waitingCount, threshold
	}
	return waitingCount >= threshold, waitingCount, threshold
}

func (r *Plugin) countWaitingTrackedJobs(jobType string) int {
	normalizedJobType := strings.TrimSpace(jobType)
	if normalizedJobType == "" {
		return 0
	}

	waiting := 0
	r.jobsMu.RLock()
	for _, job := range r.jobs {
		if job == nil {
			continue
		}
		if strings.TrimSpace(job.JobType) != normalizedJobType {
			continue
		}
		if !isWaitingTrackedJobState(job.State) {
			continue
		}
		waiting++
	}
	r.jobsMu.RUnlock()

	return waiting
}

func (r *Plugin) clearWaitingJobQueue(jobType string) int {
	normalizedJobType := strings.TrimSpace(jobType)
	if normalizedJobType == "" {
		return 0
	}

	jobIDs := make([]string, 0)
	seen := make(map[string]struct{})

	r.jobsMu.RLock()
	for _, job := range r.jobs {
		if job == nil {
			continue
		}
		if strings.TrimSpace(job.JobType) != normalizedJobType {
			continue
		}
		if !isWaitingTrackedJobState(job.State) {
			continue
		}
		jobID := strings.TrimSpace(job.JobID)
		if jobID == "" {
			continue
		}
		if _, ok := seen[jobID]; ok {
			continue
		}
		seen[jobID] = struct{}{}
		jobIDs = append(jobIDs, jobID)
	}
	r.jobsMu.RUnlock()

	if len(jobIDs) == 0 {
		return 0
	}

	reason := fmt.Sprintf("cleared queued job before %s run", normalizedJobType)
	for _, jobID := range jobIDs {
		r.markJobCanceled(&plugin_pb.JobSpec{
			JobId:   jobID,
			JobType: normalizedJobType,
		}, reason)
	}

	return len(jobIDs)
}

func waitingBacklogThreshold(policy schedulerPolicy) int {
	concurrency := policy.ExecutionConcurrency
	if concurrency <= 0 {
		concurrency = defaultScheduledExecutionConcurrency
	}
	threshold := concurrency * defaultWaitingBacklogMultiplier
	if threshold < defaultWaitingBacklogFloor {
		threshold = defaultWaitingBacklogFloor
	}
	if policy.MaxResults > 0 && threshold > int(policy.MaxResults) {
		threshold = int(policy.MaxResults)
	}
	return threshold
}

func isExecutorAtCapacityError(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, errExecutorAtCapacity) {
		return true
	}
	return strings.Contains(strings.ToLower(err.Error()), "executor is at capacity")
}

func buildScheduledJobSpec(jobType string, proposal *plugin_pb.JobProposal, index int) *plugin_pb.JobSpec {
	now := timestamppb.Now()

	jobID := fmt.Sprintf("%s-scheduled-%d-%d", jobType, now.AsTime().UnixNano(), index)

	job := &plugin_pb.JobSpec{
		JobId:       jobID,
		JobType:     jobType,
		Priority:    plugin_pb.JobPriority_JOB_PRIORITY_NORMAL,
		Parameters:  map[string]*plugin_pb.ConfigValue{},
		Labels:      map[string]string{},
		CreatedAt:   now,
		ScheduledAt: now,
	}

	if proposal == nil {
		return job
	}

	if proposal.JobType != "" {
		job.JobType = proposal.JobType
	}
	job.Summary = proposal.Summary
	job.Detail = proposal.Detail
	if proposal.Priority != plugin_pb.JobPriority_JOB_PRIORITY_UNSPECIFIED {
		job.Priority = proposal.Priority
	}
	job.DedupeKey = proposal.DedupeKey
	job.Parameters = CloneConfigValueMap(proposal.Parameters)
	if proposal.Labels != nil {
		job.Labels = make(map[string]string, len(proposal.Labels))
		for k, v := range proposal.Labels {
			job.Labels[k] = v
		}
	}
	if proposal.NotBefore != nil {
		job.ScheduledAt = proposal.NotBefore
	}

	return job
}

func durationFromSeconds(seconds int32, defaultValue time.Duration) time.Duration {
	if seconds <= 0 {
		return defaultValue
	}
	return time.Duration(seconds) * time.Second
}

func durationFromMinutes(minutes int32, defaultValue time.Duration) time.Duration {
	if minutes <= 0 {
		return defaultValue
	}
	return time.Duration(minutes) * time.Minute
}

func secondsFromDuration(duration time.Duration) int32 {
	if duration <= 0 {
		return 0
	}
	return int32(duration / time.Second)
}

func minutesFromDuration(duration time.Duration) int32 {
	if duration <= 0 {
		return 0
	}
	return int32(duration / time.Minute)
}

func waitForShutdownOrTimerWithContext(shutdown <-chan struct{}, ctx context.Context, duration time.Duration) bool {
	if duration <= 0 {
		return true
	}
	if ctx == nil {
		ctx = context.Background()
	}

	timer := time.NewTimer(duration)
	defer timer.Stop()

	select {
	case <-shutdown:
		return false
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

// filterProposalsWithActiveJobs removes proposals whose dedupe keys already have active jobs.
// It first expires stale tracked jobs via expireStaleJobs, which can mutate scheduler state,
// so callers should treat this method as a stateful operation.
func (r *Plugin) filterProposalsWithActiveJobs(jobType string, proposals []*plugin_pb.JobProposal) ([]*plugin_pb.JobProposal, int) {
	if len(proposals) == 0 {
		return proposals, 0
	}

	r.expireStaleJobs(time.Now().UTC())

	activeKeys := make(map[string]struct{})
	r.jobsMu.RLock()
	for _, job := range r.jobs {
		if job == nil {
			continue
		}
		if strings.TrimSpace(job.JobType) != strings.TrimSpace(jobType) {
			continue
		}
		if !isActiveTrackedJobState(job.State) {
			continue
		}

		key := strings.TrimSpace(job.DedupeKey)
		if key == "" {
			key = strings.TrimSpace(job.JobID)
		}
		if key == "" {
			continue
		}
		activeKeys[key] = struct{}{}
	}
	r.jobsMu.RUnlock()

	if len(activeKeys) == 0 {
		return proposals, 0
	}

	filtered := make([]*plugin_pb.JobProposal, 0, len(proposals))
	skipped := 0
	for _, proposal := range proposals {
		if proposal == nil {
			continue
		}
		key := proposalExecutionKey(proposal)
		if key != "" {
			if _, exists := activeKeys[key]; exists {
				skipped++
				continue
			}
		}
		filtered = append(filtered, proposal)
	}

	return filtered, skipped
}

func proposalExecutionKey(proposal *plugin_pb.JobProposal) string {
	if proposal == nil {
		return ""
	}
	key := strings.TrimSpace(proposal.DedupeKey)
	if key != "" {
		return key
	}
	return strings.TrimSpace(proposal.ProposalId)
}

func isActiveTrackedJobState(state string) bool {
	normalized := strings.ToLower(strings.TrimSpace(state))
	switch normalized {
	case "pending", "assigned", "running", "in_progress", "job_state_pending", "job_state_assigned", "job_state_running":
		return true
	default:
		return false
	}
}

func isWaitingTrackedJobState(state string) bool {
	normalized := strings.ToLower(strings.TrimSpace(state))
	return normalized == "pending" || normalized == "job_state_pending"
}

// DispatchProposals dispatches a batch of proposals using the same capacity-aware
// dispatch logic as the scheduler loop: concurrent execution, executor reservation
// with backoff, and per-job retry on transient errors. The scheduler policy is
// loaded from the persisted job type config; if the job type has no config or is
// disabled a sensible default policy is used so manual runs always work.
func (r *Plugin) DispatchProposals(
	ctx context.Context,
	jobType string,
	proposals []*plugin_pb.JobProposal,
	clusterContext *plugin_pb.ClusterContext,
) (successCount, errorCount, canceledCount int) {
	if len(proposals) == 0 {
		return 0, 0, 0
	}

	policy, enabled, err := r.loadSchedulerPolicy(jobType)
	if err != nil || !enabled {
		policy = schedulerPolicy{
			ExecutionConcurrency:   defaultScheduledExecutionConcurrency,
			PerWorkerConcurrency:   defaultScheduledPerWorkerConcurrency,
			ExecutionTimeout:       defaultScheduledExecutionTimeout,
			RetryBackoff:           defaultScheduledRetryBackoff,
			ExecutorReserveBackoff: 200 * time.Millisecond,
			LocalQueuePin:          defaultLocalQueuePin,
			OrphanRehomeGrace:      defaultOrphanRehomeGrace,
		}
	}

	return r.dispatchScheduledProposals(ctx, jobType, proposals, clusterContext, policy)
}

func (r *Plugin) filterScheduledProposals(proposals []*plugin_pb.JobProposal) []*plugin_pb.JobProposal {
	filtered := make([]*plugin_pb.JobProposal, 0, len(proposals))
	seenInRun := make(map[string]struct{}, len(proposals))

	for _, proposal := range proposals {
		if proposal == nil {
			continue
		}

		key := proposal.DedupeKey
		if key == "" {
			key = proposal.ProposalId
		}
		if key == "" {
			filtered = append(filtered, proposal)
			continue
		}

		if _, exists := seenInRun[key]; exists {
			continue
		}

		seenInRun[key] = struct{}{}
		filtered = append(filtered, proposal)
	}

	return filtered
}