package automation

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// SchedulerService handles scheduled flow execution
type SchedulerService struct {
	persistence                    PersistenceProvider
	query                          ports.QueryService
	metadata                       *metadata.MetadataService
	flowExecutor                   *FlowExecutor
	listScheduledFlowDefinitionsFn func(context.Context) []*models.FlowDefinition
	getFlowDefinitionByIDFn        func(context.Context, string) *models.FlowDefinition
	getFlowVersionByIDFn           func(context.Context, string) *models.FlowVersion
	loadScheduledInputsFn          func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error)
	executeFlowVersionFn           func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error)
	// resumeInterviewFn + listDuePausedInterviewsFn provide test seams for
	// the Wait-element resume tick. Production wiring defers to the
	// flowExecutor (see resumePausedInterviews).
	resumeInterviewFn              func(ctx context.Context, id string, user *models.UserSession) error
	listDuePausedInterviewsFn      func(ctx context.Context, now time.Time, limit int, user *models.UserSession) ([]string, error)
	reservationStoreFn             func() scheduledRunReservationStore
	fencedReservationStoreFn       func() scheduledRunFencedStore
	fencedExecutionAuthority       ScheduledFlowExecutionAuthority
	executePreparedScheduledFlowFn func(
		context.Context,
		ScheduledFlowExecutionPlan,
		map[string]interface{},
		*models.UserSession,
	) (*models.FlowExecutionResult, error)
	// nowFn is an optional clock seam for tests. Production leaves it nil
	// (defaults to time.Now). Set once at construction; readers do not lock.
	nowFn        func() time.Time
	retryState   map[string]*scheduledFlowRetryState
	retryStateMu sync.Mutex
	// scheduledFlowWorkers bounds execution without making the discovery loop
	// wait for a free slot. Workers acquire capacity after they are dispatched.
	scheduledFlowWorkers chan struct{}
	stopCh               chan struct{}
	controllerCtx        context.Context
	tickInterval         time.Duration
	wg                   sync.WaitGroup
	running              bool
	mu                   sync.Mutex
	workerObservation    schedulerWorkerObservation
}

var errScheduledFlowResultRollback = errors.New("scheduled flow result requires effect rollback")

// schedulerBackoffLadder is per-FlowDefinition self-protection: when a
// Scheduled flow keeps failing tick after tick, the wait between attempts
// grows so a single misconfigured flow (broken FlowDefinition.ActiveVersionId,
// stale FK, etc.) cannot self-DDoS the log + DB. Issue #2199.
var schedulerBackoffLadder = []time.Duration{
	1 * time.Minute,
	5 * time.Minute,
	15 * time.Minute,
	1 * time.Hour,
	24 * time.Hour,
}

// schedulerAlertStageIndex marks the ladder index past which scheduled-flow
// failures should be considered alert-worthy. Logs gain an "alert=true"
// marker for downstream log-pipeline scraping.
const schedulerAlertStageIndex = 3 // 1h and beyond

const (
	schedulerTickInterval    = time.Minute
	schedulerStartupLookback = time.Minute
	// Every scheduled effect holds the single work-family ownership row through
	// commit. Parallel lanes cannot commit concurrently and only age leases
	// while waiting for the same fence, so production execution is explicitly
	// serialized. Durable acceptance remains batched.
	scheduledFlowWorkerConcurrency = 1
)

// scheduledFlowRetryState tracks per-flow consecutive-failure backoff. State
// is process-local self-protection (not a parallel SSoT for SF's
// AsyncApexJob); it is intentionally not persisted across restart — a real
// broken flow re-enters the ladder naturally on the next boot, and the
// alert at stage schedulerAlertStageIndex (1h) usually fires well before any
// restart cadence.
type scheduledFlowRetryState struct {
	consecutiveFailures int
	nextRetryAt         time.Time
	lastLoggedStage     int // -1 = never logged
}

// tickPhase classifies the outcome of a single executeScheduledDefinition
// call so the retry-state writer can decide whether to advance the ladder
// (allFailed / loadFailed), reset it (allSucceeded / mixed — the conservative
// reset on partial success preserves liveness for partially-broken flows),
// or leave it untouched (inputsEmpty — nothing was attempted).
type tickPhase int

const (
	phaseLoadFailed tickPhase = iota
	phaseInputsEmpty
	phaseExecAllFailed
	phaseExecMixed
	phaseExecAllSucceeded
)

type scheduledExecutionOutcome string

const (
	scheduledExecutionOutcomeNoActiveVersion    scheduledExecutionOutcome = "no_active_version"
	scheduledExecutionOutcomeNotDue             scheduledExecutionOutcome = "not_due"
	scheduledExecutionOutcomeLoadFailed         scheduledExecutionOutcome = "load_failed"
	scheduledExecutionOutcomeScanBudgetExceeded scheduledExecutionOutcome = "scan_budget_exceeded"
	scheduledExecutionOutcomeNoMatchingRecords  scheduledExecutionOutcome = "no_matching_records"
	scheduledExecutionOutcomePartialFailure     scheduledExecutionOutcome = "partial_failure"
	scheduledExecutionOutcomeAllFailure         scheduledExecutionOutcome = "all_failure"
	scheduledExecutionOutcomeAllSucceeded       scheduledExecutionOutcome = "all_succeeded"
	scheduledExecutionOutcomeBackoffSkipped     scheduledExecutionOutcome = "backoff_skipped"
	// IMPORTANT: add new outcome values to scheduledExecutionOutcomeAllValues
	// in scheduler_service_test.go to satisfy the parity invariant.
)

type scheduledRunReservationStore interface {
	Reserve(ctx context.Context, key ScheduledRunReservationKey) (ScheduledRunReserveResult, error)
	Complete(ctx context.Context, reservationID string, key ScheduledRunReservationKey, status ScheduledRunExecutionStatus, lastError string) error
}

type scheduledRunFencedStore interface {
	MetadataBinding() string
	Claim(context.Context, ScheduledRunReservationKey, []byte) (ScheduledRunLease, error)
	ClaimBatch(context.Context, []ScheduledRunAcceptance) (ScheduledRunBatchClaim, error)
	ListRecoverable(context.Context, int) ([]ScheduledRunCandidate, error)
	Reclaim(context.Context, ScheduledRunReservationKey) (ScheduledRunLease, error)
	RefreshLease(context.Context, ScheduledRunLease) (ScheduledRunLease, error)
	RecordPreparationFailure(
		context.Context,
		ScheduledRunLease,
		ScheduledRunPreparationFailureClass,
		string,
	) error
	Execute(context.Context, ScheduledRunLease, func(context.Context) error) error
	Complete(context.Context, ScheduledRunLease, ScheduledRunExecutionStatus, string) error
}

type ScheduledRunPreparationFailureClass string

const (
	ScheduledRunPreparationFailureRetryable ScheduledRunPreparationFailureClass = "retryable"
	ScheduledRunPreparationFailurePermanent ScheduledRunPreparationFailureClass = "permanent"
)

// ScheduledFlowExecutionAuthority reopens only the immutable Runtime view named
// by the durable work binding and prepares the complete exact Flow closure.
// A repository/current-version selector does not satisfy this contract.
type ScheduledFlowExecutionAuthority interface {
	PrepareScheduledFlow(
		context.Context,
		string,
		string,
		string,
		string,
	) (PreparedScheduledFlowExecution, error)
}

// ScheduledFlowAuthorityFailureClassifier lets the production Catalog adapter
// distinguish unavailable content-addressed artifacts from an immutable
// binding that can never be executed. Authorities that do not implement this
// optional contract remain retryable by default.
type ScheduledFlowAuthorityFailureClassifier interface {
	IsPermanentScheduledFlowAuthorityFailure(error) bool
}

type acceptedFencedScheduledRun struct {
	lease  ScheduledRunLease
	inputs map[string]interface{}
}

type rawDBProvider interface {
	RawDB() *sql.DB
}

// NewSchedulerService creates a new SchedulerService
func NewSchedulerService(
	persistence PersistenceProvider,
	query ports.QueryService,
	metadata *metadata.MetadataService,
	flowExecutor *FlowExecutor,
) *SchedulerService {
	return &SchedulerService{
		persistence:  persistence,
		query:        query,
		metadata:     metadata,
		flowExecutor: flowExecutor,
		stopCh:       make(chan struct{}),
	}
}

// StartScheduler starts the background scheduler
func (s *SchedulerService) StartScheduler() {
	s.startScheduler(schedulerDetachedContext(), true, true)
}

// RunWorkerRole runs scheduled discovery under a Runtime Worker controller
// context. Paused-interview wakeup deliberately remains excluded: its existing
// claim is not fenced by Runtime work-family ownership, so starting that loop
// here would create a second unsafe owner.
func (s *SchedulerService) RunWorkerRole(ctx context.Context) error {
	if ctx == nil {
		return errors.New("scheduler worker role: controller context is required")
	}
	if s.fencedScheduledReservationStore() == nil {
		return errors.New("scheduler worker role: fenced reservation store is required")
	}
	if s.fencedExecutionAuthority == nil {
		return errors.New("scheduler worker role: immutable scheduled Flow authority is required")
	}
	// Initialize the role and its bounded worker lane without starting
	// discovery. Restart recovery owns the first pass so an aggressively short
	// Worker poll interval cannot make discovery contend with accepted work.
	if !s.startScheduler(ctx, false, false) {
		return errors.New("scheduler worker role: scheduler is already running")
	}
	// Restart recovery is already bounded by the worker lane and scan limit.
	// Run it once at role start instead of leaving durably accepted work idle
	// until the first minute ticker. This is not historical schedule catch-up:
	// only ACCEPTED due rows and expired RESERVED attempts are eligible.
	s.recoverExpiredScheduledRuns()
	s.wg.Add(1)
	go s.runSchedulerLoop()
	<-ctx.Done()
	s.StopScheduler()
	return nil
}

func (s *SchedulerService) startScheduler(ctx context.Context, scheduled, paused bool) bool {
	s.mu.Lock()
	if s.running {
		s.mu.Unlock()
		return false
	}
	s.running = true
	s.stopCh = make(chan struct{})
	s.scheduledFlowWorkers = make(chan struct{}, scheduledFlowWorkerConcurrency)
	s.controllerCtx = ctx
	s.mu.Unlock()

	// Drop any retry-state carried over from a previous Start/Stop cycle so
	// in-process restarts (tests, future hot-reload) start clean.
	s.retryStateMu.Lock()
	s.retryState = nil
	s.retryStateMu.Unlock()

	if scheduled {
		s.wg.Add(1)
		go s.runSchedulerLoop()
	}
	if paused {
		s.wg.Add(1)
		go s.runPausedInterviewLoop()
	}
	return true
}

// now returns the wall clock unless a test seam is wired. Always UTC: even
// when nowFn is set, the returned value is forced through .UTC() so a test
// that injects a non-UTC time.Time cannot silently regress the host-TZ-skew
// fix tracked by issue #227.
func (s *SchedulerService) now() time.Time {
	if s.nowFn != nil {
		return s.nowFn().UTC()
	}
	return time.Now().UTC()
}

// effectiveStateKey returns a stable per-flow-version key for retry-state
// lookup. Composite of FlowDefinition Id (or DeveloperName fallback for
// legacy rows seen on localhost — see issue #2199 Notes) and the active
// FlowVersion Id, so that an admin who fixes a broken Scheduled flow by
// activating a new version is not suppressed by the prior version's
// backoff window. Returns "" when neither def-key nor version-key can be
// derived so the caller skips retry tracking entirely.
func effectiveStateKey(def *models.FlowDefinition, version *models.FlowVersion) string {
	if def == nil {
		return ""
	}
	defPart := def.Id
	if defPart == "" {
		defPart = def.DeveloperName
	}
	if defPart == "" {
		return ""
	}
	if version != nil && version.Id != "" {
		return defPart + ":" + version.Id
	}
	return defPart
}

// shouldSkipForBackoff returns true when the flow is currently in a backoff
// window from prior consecutive failures. Returns false (i.e. proceed) when
// there is no key, no state, or the wait has elapsed.
func (s *SchedulerService) shouldSkipForBackoff(def *models.FlowDefinition, version *models.FlowVersion) bool {
	key := effectiveStateKey(def, version)
	if key == "" {
		return false
	}
	s.retryStateMu.Lock()
	defer s.retryStateMu.Unlock()
	state := s.retryState[key]
	if state == nil {
		return false
	}
	return s.now().Before(state.nextRetryAt)
}

// recordTickOutcome updates the per-flow-version retry state after one
// tick. The caller has already done all logging for this tick's per-input
// attempts; this helper only emits a single "stage transition" log line
// per ladder advance so the storm-of-stage-N-logs pattern can never recur.
func (s *SchedulerService) recordTickOutcome(def *models.FlowDefinition, version *models.FlowVersion, phase tickPhase) {
	key := effectiveStateKey(def, version)
	if key == "" {
		return
	}
	s.retryStateMu.Lock()
	defer s.retryStateMu.Unlock()
	if s.retryState == nil {
		s.retryState = make(map[string]*scheduledFlowRetryState)
	}
	state := s.retryState[key]
	if state == nil {
		state = &scheduledFlowRetryState{lastLoggedStage: -1}
		s.retryState[key] = state
	}

	switch phase {
	case phaseInputsEmpty:
		// Nothing was attempted — neutral.
		return
	case phaseExecAllSucceeded, phaseExecMixed:
		// Any progress resets the ladder so a flow that recovers after a
		// blip (or has one stubborn input among many healthy ones) doesn't
		// get throttled forever.
		if state.consecutiveFailures > 0 || state.lastLoggedStage >= 0 {
			log.Printf("Scheduler: flow '%s' recovered after %d consecutive failure(s); retry backoff cleared",
				def.DeveloperName, state.consecutiveFailures)
		}
		state.consecutiveFailures = 0
		state.nextRetryAt = time.Time{}
		state.lastLoggedStage = -1
		return
	case phaseLoadFailed, phaseExecAllFailed:
		state.consecutiveFailures++
		stage := state.consecutiveFailures - 1
		if stage >= len(schedulerBackoffLadder) {
			stage = len(schedulerBackoffLadder) - 1
		}
		state.nextRetryAt = s.now().Add(schedulerBackoffLadder[stage])
		if stage != state.lastLoggedStage {
			alert := ""
			if stage >= schedulerAlertStageIndex {
				alert = " alert=true"
			}
			log.Printf("Scheduler: flow '%s' has failed %d consecutive tick(s); next retry deferred by %s%s",
				def.DeveloperName, state.consecutiveFailures, schedulerBackoffLadder[stage], alert)
			state.lastLoggedStage = stage
		}
	}
}

// StopScheduler stops the background scheduler
func (s *SchedulerService) StopScheduler() {
	s.mu.Lock()
	if !s.running {
		s.mu.Unlock()
		return
	}
	s.running = false
	close(s.stopCh)
	s.mu.Unlock()

	s.wg.Wait()
	s.mu.Lock()
	s.controllerCtx = nil
	s.mu.Unlock()
}

func (s *SchedulerService) runSchedulerLoop() {
	defer s.wg.Done()

	interval := s.schedulerInterval()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	// The first scan has a fixed, finite restart lookback independent of the
	// controller poll interval. Digest identity makes replay safe, while the
	// bound deliberately does not promise unbounded downtime catch-up.
	lastScanAt := s.now().Add(-schedulerStartupLookback)
	acceptedOccurrences := make(map[string]struct{})

	for {
		select {
		case <-s.stopCh:
			log.Println("Scheduled Flow discovery stopped")
			return
		case <-ticker.C:
			through := s.now()
			s.recoverExpiredScheduledRuns()
			if s.dispatchScheduledFlowsBetweenWithProgress(lastScanAt, through, acceptedOccurrences) {
				lastScanAt = through
				acceptedOccurrences = make(map[string]struct{})
			}
		}
	}
}

func (s *SchedulerService) recoverExpiredScheduledRuns() {
	started := time.Now()
	defer func() {
		s.workerObservation.recoveryDurationNano.Add(time.Since(started).Nanoseconds())
	}()
	store := s.fencedScheduledReservationStore()
	if store == nil {
		return
	}
	workers := s.scheduledFlowWorkerPool()
	available := cap(workers) - len(workers)
	if available <= 0 {
		s.workerObservation.capacityDeferrals.Add(1)
		return
	}
	ctx := s.schedulerControllerContext()
	candidates, err := store.ListRecoverable(ctx, available)
	if err != nil {
		if ctx.Err() == nil {
			log.Printf("Scheduler: failed to list recoverable scheduled runs: %v", err)
		}
		return
	}
	s.workerObservation.recoveryScanned.Add(uint64(len(candidates)))
	systemUser := newSystemAutomationUser()
	for _, candidate := range candidates {
		if !s.tryDispatchRecoveredScheduledRun(ctx, store, candidate, systemUser) {
			return
		}
	}
}

func (s *SchedulerService) tryDispatchRecoveredScheduledRun(
	ctx context.Context,
	store scheduledRunFencedStore,
	candidate ScheduledRunCandidate,
	systemUser *models.UserSession,
) bool {
	key := candidate.Key
	workers := s.scheduledFlowWorkerPool()
	select {
	case workers <- struct{}{}:
	default:
		s.workerObservation.capacityDeferrals.Add(1)
		return false
	}
	select {
	case <-s.stopCh:
		<-workers
		return false
	default:
	}

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		defer func() { <-workers }()
		defer func() {
			if r := recover(); r != nil {
				log.Printf("Scheduler: panic recovering scheduled occurrence %s: %v", key.CanonicalID(), r)
			}
		}()

		lease, err := store.Reclaim(ctx, key)
		if err != nil {
			if ctx.Err() == nil {
				log.Printf("Scheduler: failed to reclaim scheduled occurrence %s: %v", key.CanonicalID(), err)
			}
			return
		}
		switch lease.Status {
		case ScheduledRunLeaseDuplicate:
			return
		case ScheduledRunLeaseClaimed:
			s.workerObservation.recoveryRecovered.Add(1)
		default:
			log.Printf("Scheduler: recovered occurrence %s returned invalid lease status %q", key.CanonicalID(), lease.Status)
			return
		}
		if lease.MetadataBinding != candidate.MetadataBinding ||
			lease.EntryElement != candidate.EntryElement {
			s.recordRecoveredPreparationFailure(
				ctx,
				store,
				lease,
				ScheduledRunPreparationFailurePermanent,
				errors.New("immutable binding or timed entry changed between recovery scan and claim"),
			)
			return
		}
		inputs, err := decodeScheduledRunInputs(lease.CanonicalInputs)
		if err != nil {
			s.recordRecoveredPreparationFailure(
				ctx,
				store,
				lease,
				ScheduledRunPreparationFailurePermanent,
				fmt.Errorf("invalid canonical inputs: %w", err),
			)
			return
		}
		plan, err := s.prepareFencedScheduledFlow(
			ctx,
			lease.MetadataBinding,
			key.FlowDefinitionID,
			key.FlowVersionID,
			lease.EntryElement,
		)
		if err != nil {
			failureClass := ScheduledRunPreparationFailureRetryable
			if classifier, ok := s.fencedExecutionAuthority.(ScheduledFlowAuthorityFailureClassifier); ok &&
				classifier.IsPermanentScheduledFlowAuthorityFailure(err) {
				failureClass = ScheduledRunPreparationFailurePermanent
			}
			s.recordRecoveredPreparationFailure(
				ctx,
				store,
				lease,
				failureClass,
				fmt.Errorf("cannot reopen immutable Flow authority: %w", err),
			)
			return
		}
		lease, err = store.RefreshLease(ctx, lease)
		if err != nil {
			if ctx.Err() == nil {
				log.Printf(
					"Scheduler: failed to refresh recovered occurrence %s before execution: %v",
					key.CanonicalID(),
					err,
				)
			}
			return
		}
		result, execErr := s.executeFencedScheduledVersion(
			ctx, store, lease, plan, inputs, systemUser,
		)
		if execErr != nil {
			log.Printf("Scheduler: failed to execute recovered flow '%s': %v", plan.Definition.DeveloperName, execErr)
			return
		}
		if result != nil && !result.IsSuccess {
			log.Printf("Scheduler: recovered flow '%s' completed with a failed result", plan.Definition.DeveloperName)
			return
		}
		log.Printf("Scheduler: recovered flow '%s' executed successfully", plan.Definition.DeveloperName)
	}()
	return true
}

func (s *SchedulerService) recordRecoveredPreparationFailure(
	ctx context.Context,
	store scheduledRunFencedStore,
	lease ScheduledRunLease,
	failureClass ScheduledRunPreparationFailureClass,
	cause error,
) {
	if cause == nil {
		cause = errors.New("unknown scheduled Flow preparation failure")
	}
	if err := store.RecordPreparationFailure(
		ctx,
		lease,
		failureClass,
		cause.Error(),
	); err != nil {
		if ctx.Err() == nil {
			log.Printf(
				"Scheduler: failed to record %s preparation failure for occurrence %s: %v (cause: %v)",
				failureClass,
				lease.Key.CanonicalID(),
				err,
				cause,
			)
		}
		return
	}
	log.Printf(
		"Scheduler: recorded %s preparation failure for occurrence %s: %v",
		failureClass,
		lease.Key.CanonicalID(),
		cause,
	)
}

func (s *SchedulerService) prepareFencedScheduledFlow(
	ctx context.Context,
	binding string,
	definitionID string,
	versionID string,
	entryElement string,
) (PreparedScheduledFlowExecution, error) {
	if s == nil || s.fencedExecutionAuthority == nil {
		return PreparedScheduledFlowExecution{}, errors.New(
			"scheduler worker role: immutable scheduled Flow authority is unavailable",
		)
	}
	if strings.TrimSpace(binding) == "" {
		return PreparedScheduledFlowExecution{}, errors.New(
			"scheduler worker role: immutable scheduled Flow binding is absent",
		)
	}
	prepared, err := s.fencedExecutionAuthority.PrepareScheduledFlow(
		ctx,
		binding,
		definitionID,
		versionID,
		entryElement,
	)
	if err != nil {
		return PreparedScheduledFlowExecution{}, err
	}
	if err := prepared.Validate(); err != nil {
		return PreparedScheduledFlowExecution{}, err
	}
	return prepared, nil
}

func (s *SchedulerService) runPausedInterviewLoop() {
	defer s.wg.Done()

	ticker := time.NewTicker(s.schedulerInterval())
	defer ticker.Stop()

	for {
		select {
		case <-s.stopCh:
			log.Println("Paused Flow interview scheduler stopped")
			return
		case <-ticker.C:
			s.resumePausedInterviews()
		}
	}
}

// resumePausedInterviewsBatchSize bounds how many due interviews one tick
// will process. Keeps a single stuck scheduler tick from blocking for the
// whole table if catch-up after downtime produces a very large backlog.
const resumePausedInterviewsBatchSize = 200

// resumePausedInterviews loads all Paused FlowInterview rows whose
// PausedUntil has elapsed (catch-up safe: past-dated rows resume
// immediately) and dispatches each to the executor. Each resume runs in
// its own goroutine with panic recovery so one bad flow cannot stall the
// rest of the tick.
func (s *SchedulerService) resumePausedInterviews() {
	listFn := s.listDuePausedInterviewsFn
	resumeFn := s.resumeInterviewFn
	if listFn == nil || resumeFn == nil {
		// Default wiring (production): defer to flowExecutor + interview service.
		if s.flowExecutor == nil || s.flowExecutor.interviews == nil {
			return
		}
		if listFn == nil {
			listFn = s.flowExecutor.interviews.ListDuePausedInterviews
		}
		if resumeFn == nil {
			resumeFn = s.flowExecutor.ResumeInterview
		}
	}

	ctx := constants.WithSecurityContext(s.schedulerControllerContext(), constants.SystemContext, "flow interview resume")
	systemUser := newSystemAutomationUser()

	ids, err := listFn(ctx, s.now(), resumePausedInterviewsBatchSize, systemUser)
	if err != nil {
		log.Printf("Scheduler: failed to list due paused interviews: %v", err)
		return
	}
	if len(ids) == 0 {
		return
	}

	var wg sync.WaitGroup
	sem := make(chan struct{}, 10) // cap concurrent resumes
	for _, id := range ids {
		wg.Add(1)
		sem <- struct{}{}
		go func(interviewID string) {
			defer wg.Done()
			defer func() { <-sem }()
			defer func() {
				if r := recover(); r != nil {
					log.Printf("Scheduler: panic resuming interview %s: %v", interviewID, r)
				}
			}()
			if err := resumeFn(ctx, interviewID, systemUser); err != nil {
				log.Printf("Scheduler: failed to resume interview %s: %v", interviewID, err)
			}
		}(id)
	}
	wg.Wait()
}

func (s *SchedulerService) dispatchScheduledFlowsBetween(after, through time.Time) bool {
	return s.dispatchScheduledFlowsBetweenWithProgress(after, through, make(map[string]struct{}))
}

func (s *SchedulerService) dispatchScheduledFlowsBetweenWithProgress(
	after, through time.Time,
	acceptedOccurrences map[string]struct{},
) (complete bool) {
	started := time.Now()
	acceptedBefore := s.workerObservation.occurrencesAccepted.Load()
	s.workerObservation.discoveryPasses.Add(1)
	defer func() {
		s.workerObservation.discoveryDurationNano.Add(time.Since(started).Nanoseconds())
		if !complete {
			s.workerObservation.discoveryFailures.Add(1)
		} else if s.workerObservation.occurrencesAccepted.Load() == acceptedBefore {
			s.workerObservation.discoveryIdle.Add(1)
		}
	}()
	ctx := s.schedulerControllerContext()
	after = after.UTC()
	through = through.UTC()

	// Get all scheduled flow definitions
	flowDefs, err := s.getScheduledFlowDefinitions(ctx)
	if err != nil {
		log.Printf("Scheduler: scheduled Flow discovery authority is unavailable: %v", err)
		return false
	}
	s.workerObservation.definitionsConsidered.Add(uint64(len(flowDefs)))
	if len(flowDefs) == 0 {
		return true
	}

	// Scheduled flows run as trusted system automation, not as an end user.
	systemUser := newSystemAutomationUser()
	complete = true

	for _, def := range flowDefs {
		defCopy := *def
		def = &defCopy
		if def.ActiveVersionId == "" {
			s.emitScheduledExecutionSummary(def, nil, through, scheduledExecutionOutcomeNoActiveVersion, 0, 0, 0)
			continue
		}

		// Get the flow version to check schedule
		version := s.getFlowVersionByID(ctx, def.ActiveVersionId)
		if version == nil {
			s.emitScheduledExecutionSummary(def, nil, through, scheduledExecutionOutcomeNoActiveVersion, 0, 0, 0)
			continue
		}
		versionCopy := *version
		version = &versionCopy
		applyFlowDefinitionRuntimeFields(version, def)

		occurrence, due := scheduledOccurrenceBetween(version, after, through)
		if !due {
			s.emitScheduledExecutionSummary(def, version, through, scheduledExecutionOutcomeNotDue, 0, 0, 0)
			continue
		}
		progressKey := scheduledFlowDispatchProgressKey(def.Id, occurrence.start)
		if _, accepted := acceptedOccurrences[progressKey]; accepted {
			continue
		}

		log.Printf("Scheduler: executing scheduled flow '%s'", def.DeveloperName)
		if !s.tryDispatchScheduledDefinition(ctx, def.Id, version.Id, def.DeveloperName, systemUser, occurrence.start) {
			complete = false
			log.Printf("Scheduler: scheduled flow worker capacity is full; retaining occurrence %s for flow '%s' for the next discovery tick",
				occurrence.start.UTC().Format(time.RFC3339), def.DeveloperName)
			continue
		}
		acceptedOccurrences[progressKey] = struct{}{}
	}
	return complete
}

func scheduledFlowDispatchProgressKey(definitionID string, occurrence time.Time) string {
	return definitionID + ":" + occurrence.UTC().Format(time.RFC3339Nano)
}

func (s *SchedulerService) tryDispatchScheduledDefinition(ctx context.Context, definitionID, expectedVersionID, developerName string, systemUser *models.UserSession, occurrence time.Time) bool {
	if fencedStore := s.fencedScheduledReservationStore(); fencedStore != nil {
		return s.acceptAndDispatchFencedScheduledDefinition(
			ctx, fencedStore,
			definitionID, expectedVersionID, developerName, systemUser, occurrence,
		)
	}

	workers := s.scheduledFlowWorkerPool()
	select {
	case workers <- struct{}{}:
	default:
		s.workerObservation.capacityDeferrals.Add(1)
		return false
	}

	select {
	case <-s.stopCh:
		<-workers
		return false
	default:
	}

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		defer func() { <-workers }()
		defer func() {
			if r := recover(); r != nil {
				log.Printf("Scheduler: panic executing scheduled flow '%s': %v", developerName, r)
			}
		}()

		def := s.getFlowDefinitionByID(ctx, definitionID)
		if def == nil || def.ActiveVersionId == "" {
			log.Printf("Scheduler: skipping scheduled flow '%s' because it is no longer active", developerName)
			s.emitScheduledExecutionSummary(def, nil, occurrence, scheduledExecutionOutcomeNoActiveVersion, 0, 0, 0)
			return
		}
		if def.ActiveVersionId != expectedVersionID {
			log.Printf("Scheduler: skipping scheduled flow '%s' because active version changed from %s to %s",
				developerName, expectedVersionID, def.ActiveVersionId)
			s.emitScheduledExecutionSummary(def, nil, occurrence, scheduledExecutionOutcomeNotDue, 0, 0, 0)
			return
		}

		version := s.getFlowVersionByID(ctx, expectedVersionID)
		if version == nil {
			s.emitScheduledExecutionSummary(def, nil, occurrence, scheduledExecutionOutcomeNoActiveVersion, 0, 0, 0)
			return
		}
		defCopy := *def
		versionCopy := *version
		def = &defCopy
		version = &versionCopy
		applyFlowDefinitionRuntimeFields(version, def)
		currentOccurrence, due := scheduledOccurrenceFor(version, occurrence)
		if !due || !currentOccurrence.start.Equal(occurrence) {
			log.Printf("Scheduler: skipping scheduled flow '%s' because its schedule changed before execution", developerName)
			s.emitScheduledExecutionSummary(def, version, occurrence, scheduledExecutionOutcomeNotDue, 0, 0, 0)
			return
		}

		s.executeScheduledDefinitionAt(ctx, def, version, systemUser, occurrence)
	}()
	return true
}

// acceptAndDispatchFencedScheduledDefinition closes the crash window between
// advancing the discovery cursor and creating durable work. It prepares and
// claims every input occurrence before returning true to the caller; only then
// may dispatchScheduledFlowsBetweenWithProgress advance past this schedule
// occurrence. Runtime effects still execute asynchronously under the bounded
// worker lane.
func (s *SchedulerService) acceptAndDispatchFencedScheduledDefinition(
	ctx context.Context,
	store scheduledRunFencedStore,
	definitionID, expectedVersionID, developerName string,
	systemUser *models.UserSession,
	occurrence time.Time,
) bool {
	plan, err := s.prepareFencedScheduledFlow(
		ctx,
		store.MetadataBinding(),
		definitionID,
		expectedVersionID,
		"",
	)
	if err != nil {
		permanent := false
		if classifier, ok := s.fencedExecutionAuthority.(ScheduledFlowAuthorityFailureClassifier); ok {
			permanent = classifier.IsPermanentScheduledFlowAuthorityFailure(err)
		}
		log.Printf(
			"Scheduler: scheduled flow '%s' cannot reopen its immutable execution authority: %v",
			developerName,
			err,
		)
		if permanent {
			// This occurrence has no executable launch shape, so it is not
			// durable work and must not pin the global discovery cursor.
			// Retryable artifact availability failures still retain the cursor.
			log.Printf(
				"Scheduler: permanently rejecting ineligible occurrence %s for flow '%s'",
				occurrence.UTC().Format(time.RFC3339),
				developerName,
			)
			return true
		}
		return false
	}
	def := plan.Definition
	version := plan.Root
	currentOccurrence, due := scheduledOccurrenceFor(version, occurrence)
	if !due || !currentOccurrence.start.Equal(occurrence) {
		log.Printf("Scheduler: skipping scheduled flow '%s' because its schedule changed before acceptance", developerName)
		s.emitScheduledExecutionSummary(def, version, occurrence, scheduledExecutionOutcomeNotDue, 0, 0, 0)
		return false
	}
	if s.shouldSkipForBackoff(def, version) {
		s.emitScheduledExecutionSummary(def, version, occurrence, scheduledExecutionOutcomeBackoffSkipped, 0, 0, 0)
		return false
	}

	executionCtx := constants.WithSecurityContext(ctx, constants.SystemContext, "scheduled flow acceptance")
	inputSets, err := s.loadScheduledInputs(executionCtx, version, systemUser)
	if err != nil {
		log.Printf("Scheduler: failed to load scheduled inputs for flow '%s': %v", def.DeveloperName, err)
		phase, _ := classifyScheduledExecutionPhase(0, 0, 0, err)
		s.recordTickOutcome(def, version, phase)
		s.emitScheduledExecutionSummary(
			def, version, occurrence, classifyScheduledInputLoadOutcome(err), 0, 0, 0,
		)
		return false
	}
	if len(inputSets) == 0 {
		phase, outcome := classifyScheduledExecutionPhase(0, 0, 0, nil)
		s.recordTickOutcome(def, version, phase)
		s.emitScheduledExecutionSummary(def, version, occurrence, outcome, 0, 0, 0)
		return true
	}
	windowKey, err := BuildScheduleWindowKey(version, occurrence)
	if err != nil {
		log.Printf("Scheduler: failed to resolve schedule window for flow '%s': %v", def.DeveloperName, err)
		s.recordTickOutcome(def, version, phaseExecAllFailed)
		s.emitScheduledExecutionSummary(
			def, version, occurrence, scheduledExecutionOutcomeAllFailure,
			len(inputSets), 0, len(inputSets),
		)
		return false
	}

	requests := make([]ScheduledRunAcceptance, 0, len(inputSets))
	for _, inputs := range inputSets {
		key, keyErr := NewScheduledRunReservationKey(
			def.Id, version.Id, windowKey, scheduledInputRecordID(inputs),
		)
		if keyErr != nil {
			log.Printf("Scheduler: failed to build reservation key for flow '%s': %v", def.DeveloperName, keyErr)
			return false
		}
		canonicalInputs, encodeErr := encodeScheduledRunInputs(inputs)
		if encodeErr != nil {
			log.Printf("Scheduler: failed to persist canonical inputs for flow '%s': %v", def.DeveloperName, encodeErr)
			return false
		}
		requests = append(requests, ScheduledRunAcceptance{
			Key: key, CanonicalInputs: canonicalInputs,
		})
	}
	acceptanceStarted := time.Now()
	s.workerObservation.acceptanceAttempts.Add(1)
	batchClaim, claimErr := store.ClaimBatch(executionCtx, requests)
	s.workerObservation.acceptanceDurationNano.Add(time.Since(acceptanceStarted).Nanoseconds())
	if claimErr != nil {
		log.Printf("Scheduler: failed to atomically accept scheduled occurrence for flow '%s': %v", def.DeveloperName, claimErr)
		return false
	}
	if batchClaim.AlreadyAccepted {
		s.workerObservation.acceptanceEmpty.Add(1)
		return true
	}
	leases := batchClaim.Leases
	if len(leases) != len(inputSets) {
		log.Printf("Scheduler: atomic acceptance for flow '%s' returned %d leases for %d inputs",
			def.DeveloperName, len(leases), len(inputSets))
		return false
	}

	accepted := make([]acceptedFencedScheduledRun, 0, len(inputSets))
	acceptanceFailures := 0
	claimedCount := 0
	for _, lease := range leases {
		switch lease.Status {
		case ScheduledRunLeaseDuplicate:
			continue
		case ScheduledRunLeaseClaimed:
			claimedCount++
		default:
			acceptanceFailures++
			log.Printf("Scheduler: scheduled run for flow '%s' returned invalid lease status %q", def.DeveloperName, lease.Status)
			continue
		}
		persistedInputs, decodeErr := decodeScheduledRunInputs(lease.CanonicalInputs)
		if decodeErr != nil {
			acceptanceFailures++
			log.Printf("Scheduler: canonical scheduled inputs for flow '%s' are invalid: %v", def.DeveloperName, decodeErr)
			continue
		}
		accepted = append(accepted, acceptedFencedScheduledRun{
			lease: lease, inputs: persistedInputs,
		})
	}
	if claimedCount > 0 {
		s.workerObservation.occurrencesAccepted.Add(uint64(claimedCount))
	}

	if len(accepted) == 0 {
		if claimedCount == 0 {
			s.workerObservation.acceptanceEmpty.Add(1)
		}
		phase, outcome := classifyScheduledExecutionPhase(
			len(inputSets), 0, acceptanceFailures, nil,
		)
		s.recordTickOutcome(def, version, phase)
		s.emitScheduledExecutionSummary(
			def, version, occurrence, outcome,
			len(inputSets), 0, acceptanceFailures,
		)
		return true
	}
	// Durable acceptance is not gated by execution capacity. If the lane is
	// full, these leases expire without effects and the bounded recovery scan
	// executes them on a later tick; the discovery cursor can still advance
	// because no accepted work exists only in memory.
	workers := s.scheduledFlowWorkerPool()
	select {
	case workers <- struct{}{}:
	default:
		s.workerObservation.capacityDeferrals.Add(1)
		return true
	}
	select {
	case <-s.stopCh:
		<-workers
		return true
	default:
	}
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		defer func() { <-workers }()
		defer func() {
			if r := recover(); r != nil {
				log.Printf("Scheduler: panic executing durably accepted flow '%s': %v", developerName, r)
			}
		}()
		successes, failures := 0, acceptanceFailures
		for _, run := range accepted {
			refreshed, refreshErr := store.RefreshLease(ctx, run.lease)
			if refreshErr != nil {
				failures++
				if ctx.Err() == nil {
					log.Printf(
						"Scheduler: failed to refresh accepted flow '%s' before execution: %v",
						def.DeveloperName,
						refreshErr,
					)
				}
				continue
			}
			result, execErr := s.executeFencedScheduledVersion(
				ctx, store, refreshed, plan, run.inputs, systemUser,
			)
			if execErr != nil || (result != nil && !result.IsSuccess) {
				failures++
				if execErr != nil {
					log.Printf("Scheduler: failed to execute flow '%s': %v", def.DeveloperName, execErr)
				}
				continue
			}
			successes++
			log.Printf("Scheduler: flow '%s' executed successfully", def.DeveloperName)
		}
		phase, outcome := classifyScheduledExecutionPhase(
			len(inputSets), successes, failures, nil,
		)
		s.recordTickOutcome(def, version, phase)
		s.emitScheduledExecutionSummary(
			def, version, occurrence, outcome,
			len(inputSets), successes, failures,
		)
	}()
	return true
}

func (s *SchedulerService) scheduledFlowWorkerPool() chan struct{} {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.scheduledFlowWorkers == nil {
		s.scheduledFlowWorkers = make(chan struct{}, scheduledFlowWorkerConcurrency)
	}
	return s.scheduledFlowWorkers
}

func (s *SchedulerService) executeScheduledDefinition(ctx context.Context, def *models.FlowDefinition, version *models.FlowVersion, systemUser *models.UserSession) {
	s.executeScheduledDefinitionAt(ctx, def, version, systemUser, s.now())
}

func (s *SchedulerService) executeScheduledDefinitionAt(ctx context.Context, def *models.FlowDefinition, version *models.FlowVersion, systemUser *models.UserSession, now time.Time) {
	ctx = constants.WithSecurityContext(ctx, constants.SystemContext, "scheduled flow execution")
	now = now.UTC()
	applyFlowDefinitionRuntimeFields(version, def)

	if s.shouldSkipForBackoff(def, version) {
		s.emitScheduledExecutionSummary(def, version, now, scheduledExecutionOutcomeBackoffSkipped, 0, 0, 0)
		return
	}

	inputSets, err := s.loadScheduledInputs(ctx, version, systemUser)
	if err != nil {
		log.Printf("Scheduler: failed to load scheduled inputs for flow '%s': %v", def.DeveloperName, err)
		phase, _ := classifyScheduledExecutionPhase(0, 0, 0, err)
		outcome := classifyScheduledInputLoadOutcome(err)
		s.recordTickOutcome(def, version, phase)
		s.emitScheduledExecutionSummary(def, version, now, outcome, 0, 0, 0)
		return
	}

	if len(inputSets) == 0 {
		log.Printf("Scheduler: no matching records for scheduled flow '%s'", def.DeveloperName)
		phase, outcome := classifyScheduledExecutionPhase(0, 0, 0, nil)
		s.recordTickOutcome(def, version, phase)
		s.emitScheduledExecutionSummary(def, version, now, outcome, 0, 0, 0)
		return
	}

	store := s.scheduledReservationStore()
	if store == nil {
		log.Printf("Scheduler: scheduled run reservation store is unavailable for flow '%s'", def.DeveloperName)
		s.recordTickOutcome(def, version, phaseExecAllFailed)
		s.emitScheduledExecutionSummary(def, version, now, scheduledExecutionOutcomeAllFailure, len(inputSets), 0, len(inputSets))
		return
	}

	windowKey, err := BuildScheduleWindowKey(version, now)
	if err != nil {
		log.Printf("Scheduler: failed to resolve schedule window for flow '%s': %v", def.DeveloperName, err)
		s.recordTickOutcome(def, version, phaseExecAllFailed)
		s.emitScheduledExecutionSummary(def, version, now, scheduledExecutionOutcomeAllFailure, len(inputSets), 0, len(inputSets))
		return
	}

	successes, failures := 0, 0
	for _, inputs := range inputSets {
		reservationKey, err := NewScheduledRunReservationKey(def.Id, version.Id, windowKey, scheduledInputRecordID(inputs))
		if err != nil {
			failures++
			log.Printf("Scheduler: failed to build reservation key for flow '%s': %v", def.DeveloperName, err)
			continue
		}

		acceptanceStarted := time.Now()
		s.workerObservation.acceptanceAttempts.Add(1)
		reservation, reserveErr := store.Reserve(ctx, reservationKey)
		s.workerObservation.acceptanceDurationNano.Add(time.Since(acceptanceStarted).Nanoseconds())
		if reserveErr != nil {
			failures++
			log.Printf("Scheduler: failed to reserve scheduled run for flow '%s': %v", def.DeveloperName, reserveErr)
			continue
		}
		if reservation.Status == ScheduledRunReserveStatusDuplicate {
			s.workerObservation.acceptanceEmpty.Add(1)
			log.Printf("Scheduler: skipping duplicate scheduled run for flow '%s'", def.DeveloperName)
			continue
		}
		s.workerObservation.occurrencesAccepted.Add(1)
		result, execErr := s.executeReservedScheduledVersion(
			ctx, store, reservation.ReservationID, reservationKey,
			version, def.DeveloperName, inputs, systemUser,
		)
		// Both an explicit error AND a result that signals !IsSuccess count
		// as a failure for backoff accounting; without the latter, soft
		// failures (validation errors carried inside the result) would
		// silently re-create the storm shape this fix targets.
		if execErr != nil || (result != nil && !result.IsSuccess) {
			failures++
			if execErr != nil {
				log.Printf("Scheduler: failed to execute flow '%s': %v", def.DeveloperName, execErr)
			} else if len(result.Errors) > 0 {
				log.Printf("Scheduler: flow '%s' completed with errors: %s", def.DeveloperName, result.Errors[0].Message)
			}
			continue
		}
		successes++
		log.Printf("Scheduler: flow '%s' executed successfully", def.DeveloperName)
	}

	phase, outcome := classifyScheduledExecutionPhase(len(inputSets), successes, failures, nil)
	s.recordTickOutcome(def, version, phase)
	s.emitScheduledExecutionSummary(def, version, now, outcome, len(inputSets), successes, failures)
}

func encodeScheduledRunInputs(inputs map[string]interface{}) ([]byte, error) {
	payload, err := json.Marshal(inputs)
	if err != nil {
		return nil, err
	}
	if len(payload) == 0 || len(payload) > maxScheduledRunInputPayloadBytes {
		return nil, fmt.Errorf("canonical scheduled inputs size %d is outside the supported bound", len(payload))
	}
	return payload, nil
}

func decodeScheduledRunInputs(payload []byte) (map[string]interface{}, error) {
	if len(payload) == 0 || len(payload) > maxScheduledRunInputPayloadBytes {
		return nil, errors.New("canonical scheduled inputs are missing or exceed the supported bound")
	}
	var inputs map[string]interface{}
	if err := json.Unmarshal(payload, &inputs); err != nil {
		return nil, err
	}
	// JSON preserves the canonical record fields but not Go's named map type.
	// The existing Flow interpreter deliberately distinguishes an SObject
	// carrier from arbitrary map variables when resolving RecordUpdate and
	// relationship semantics. Rehydrate only the engine-owned root record
	// inputs; all other application variables remain ordinary JSON values.
	for _, name := range []string{"$Record", "$Record__Prior", "$Record__Old"} {
		if record, ok := inputs[name].(map[string]interface{}); ok {
			inputs[name] = models.SObject(record)
		}
	}
	return inputs, nil
}

func applyFlowDefinitionRuntimeFields(version *models.FlowVersion, def *models.FlowDefinition) {
	if version == nil || def == nil {
		return
	}
	version.DefinitionIsCustomized = def.IsCustomized
	version.DefinitionOwnerPack = strings.TrimSpace(def.OwnerPack)
	version.DefinitionRuntimeLoaded = true
	if strings.TrimSpace(version.ProcessType) == "" {
		version.ProcessType = strings.TrimSpace(def.ProcessType)
	}
}

// SetLoadScheduledInputsForTest exposes the existing loadScheduledInputsFn seam
// to black-box integration tests without changing production behavior.
func (s *SchedulerService) SetLoadScheduledInputsForTest(fn func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error)) {
	s.loadScheduledInputsFn = fn
}

// SetNowForTest exposes the scheduler clock seam to black-box integration
// tests. Production callers leave nowFn nil and use time.Now().
func (s *SchedulerService) SetNowForTest(fn func() time.Time) {
	s.nowFn = fn
}

// ExecuteScheduledDefinitionForTest exposes the single-definition execution
// seam so integration tests can drive the real scheduler path directly.
func (s *SchedulerService) ExecuteScheduledDefinitionForTest(ctx context.Context, def *models.FlowDefinition, version *models.FlowVersion, systemUser *models.UserSession) {
	s.executeScheduledDefinition(ctx, def, version, systemUser)
}

func (s *SchedulerService) loadScheduledInputs(ctx context.Context, version *models.FlowVersion, user *models.UserSession) ([]map[string]interface{}, error) {
	if s.loadScheduledInputsFn != nil {
		return s.loadScheduledInputsFn(ctx, version, user)
	}
	if s.flowExecutor == nil {
		return nil, fmt.Errorf("flow executor is not configured")
	}
	return s.flowExecutor.buildScheduledRunInputs(ctx, version, user)
}

func (s *SchedulerService) executeScheduledVersion(ctx context.Context, version *models.FlowVersion, flowAPIName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	if s.executeFlowVersionFn != nil {
		return s.executeFlowVersionFn(ctx, version, flowAPIName, inputs, user)
	}
	if s.flowExecutor == nil {
		return nil, fmt.Errorf("flow executor is not configured")
	}
	return s.flowExecutor.executeFlowVersion(ctx, version, flowAPIName, inputs, user)
}

func (s *SchedulerService) executeReservedScheduledVersion(ctx context.Context, store scheduledRunReservationStore, reservationID string, key ScheduledRunReservationKey, version *models.FlowVersion, flowAPIName string, inputs map[string]interface{}, user *models.UserSession) (result *models.FlowExecutionResult, execErr error) {
	started := time.Now()
	s.workerObservation.executionStarted.Add(1)
	defer func() {
		s.workerObservation.executionDurationNano.Add(time.Since(started).Nanoseconds())
	}()
	defer func() {
		if r := recover(); r != nil {
			execErr = fmt.Errorf("scheduled flow panic: %v", r)
		}

		status := ScheduledRunExecutionStatusCompleted
		lastError := ""
		if execErr != nil {
			status = ScheduledRunExecutionStatusFailed
			lastError = execErr.Error()
		} else if result != nil && !result.IsSuccess {
			status = ScheduledRunExecutionStatusFailed
			if len(result.Errors) > 0 {
				lastError = result.Errors[0].Message
			}
		}

		completeCtx := scheduledRunReservationCompleteContext(ctx)
		completeErr := store.Complete(completeCtx, reservationID, key, status, lastError)
		if completeErr != nil && execErr == nil {
			execErr = fmt.Errorf("scheduled run reservation complete: %w", completeErr)
		} else if completeErr != nil {
			log.Printf("Scheduler: failed to complete scheduled reservation %s: %v", reservationID, completeErr)
		}
		if completeErr == nil {
			s.workerObservation.executionCompleted.Add(1)
			if status == ScheduledRunExecutionStatusFailed {
				s.workerObservation.executionFailed.Add(1)
			}
		}
	}()

	return s.executeScheduledVersion(ctx, version, flowAPIName, inputs, user)
}

func (s *SchedulerService) executeFencedScheduledVersion(
	ctx context.Context,
	store scheduledRunFencedStore,
	lease ScheduledRunLease,
	prepared PreparedScheduledFlowExecution,
	inputs map[string]interface{},
	user *models.UserSession,
) (result *models.FlowExecutionResult, execErr error) {
	if err := prepared.Validate(); err != nil {
		return nil, err
	}
	// Scheduled work has the same system authority whether it is freshly
	// accepted or reclaimed after restart. Elevate at the execution boundary
	// so neither caller path can accidentally inherit worker/user authority.
	executionCtx := constants.WithSecurityContext(
		ctx,
		constants.SystemContext,
		"fenced scheduled flow execution",
	)
	executionCtx, err := prepared.InstallExecution(executionCtx)
	if err != nil {
		return nil, err
	}
	started := time.Now()
	s.workerObservation.executionStarted.Add(1)
	defer func() {
		s.workerObservation.executionDurationNano.Add(time.Since(started).Nanoseconds())
	}()
	execErr = store.Execute(executionCtx, lease, func(txCtx context.Context) error {
		result, err = s.executePreparedScheduledFlow(
			txCtx,
			prepared.ScheduledFlowExecutionPlan,
			inputs,
			user,
		)
		if err != nil {
			return err
		}
		if result == nil {
			return errors.New("scheduler worker role: Flow execution returned no result")
		}
		if !result.IsSuccess {
			return errScheduledFlowResultRollback
		}
		return nil
	})
	if errors.Is(execErr, errScheduledFlowResultRollback) {
		execErr = nil
	}
	// Owner/controller cancellation and lease loss are not terminal Flow
	// outcomes. The effect transaction rolled back; after expiry the next
	// owner can reclaim the unchanged durable attempt.
	if ctx.Err() != nil || errors.Is(execErr, ErrScheduledRunLeaseLost) ||
		errors.Is(execErr, ErrScheduledRunOutcomeUncertain) ||
		errors.Is(execErr, persistence.ErrRuntimeWorkOwnershipLost) {
		return result, execErr
	}

	// A successful effect transaction already wrote COMPLETED. Failed
	// executions rolled back all effects and can be disposed in a separate
	// terminal transaction.
	if execErr == nil && result != nil && result.IsSuccess {
		s.workerObservation.executionCompleted.Add(1)
		return result, nil
	}
	status := ScheduledRunExecutionStatusFailed
	lastError := ""
	if execErr != nil {
		lastError = execErr.Error()
	} else if result != nil && !result.IsSuccess && len(result.Errors) > 0 {
		lastError = result.Errors[0].Message
	}
	completeCtx, completeCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer completeCancel()
	if completeErr := store.Complete(completeCtx, lease, status, lastError); completeErr != nil {
		if execErr == nil {
			execErr = fmt.Errorf("scheduled run reservation complete: %w", completeErr)
		} else {
			log.Printf("Scheduler: failed to complete fenced scheduled reservation %s: %v", lease.ReservationID, completeErr)
		}
		return result, execErr
	}
	s.workerObservation.executionCompleted.Add(1)
	s.workerObservation.executionFailed.Add(1)
	return result, execErr
}

func (s *SchedulerService) executePreparedScheduledFlow(
	ctx context.Context,
	plan ScheduledFlowExecutionPlan,
	inputs map[string]interface{},
	user *models.UserSession,
) (*models.FlowExecutionResult, error) {
	if err := validateScheduledFlowCanonicalInputs(inputs); err != nil {
		return nil, err
	}
	if s.executePreparedScheduledFlowFn != nil {
		return s.executePreparedScheduledFlowFn(ctx, plan, inputs, user)
	}
	if s.flowExecutor == nil {
		return nil, errors.New("scheduler worker role: Flow executor is unavailable")
	}
	ctx, err := withPreparedFlowDirectRecordOperationLimit(
		ctx,
		MaxAtomicFilteredRecordOperationRows,
	)
	if err != nil {
		return nil, err
	}
	// Only a strict AsyncAfterCommit handoff must pass the protected-root gate.
	// Ordinary scheduled plans already carry Catalog-selected immutable closure
	// authority and historically execute pack-owned standard flows directly.
	if plan.AsyncAfterCommit {
		ctx = withPreparedFlowRootAuthorizationRequired(ctx)
	}
	outcome, err := s.flowExecutor.ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		plan.Definition.DeveloperName,
		plan.Versions,
		inputs,
		user,
		plan.MaxElementExecutions,
	)
	if err != nil {
		return nil, err
	}
	if outcome == nil || outcome.Result == nil {
		return nil, errors.New("scheduler worker role: Flow execution returned no outcome")
	}
	return outcome.Result, err
}

func validateScheduledFlowCanonicalInputs(inputs map[string]interface{}) error {
	for name, value := range inputs {
		if value == nil {
			continue
		}
		kind := reflect.TypeOf(value).Kind()
		if kind == reflect.Slice || kind == reflect.Array {
			return fmt.Errorf(
				"%w: canonical input %q is a caller-supplied collection",
				ErrScheduledFlowNotEligible,
				name,
			)
		}
	}
	return nil
}

func scheduledRunReservationCompleteContext(sourceCtx context.Context) context.Context {
	ctx := schedulerDetachedContext()
	if sourceCtx == nil {
		return ctx
	}
	if requestID, _ := sourceCtx.Value(constants.ContextKeyRequestID).(string); requestID != "" {
		ctx = context.WithValue(ctx, constants.ContextKeyRequestID, requestID)
	}
	if traceparent, _ := sourceCtx.Value(constants.ContextKeyTraceparent).(string); traceparent != "" {
		ctx = context.WithValue(ctx, constants.ContextKeyTraceparent, traceparent)
	}
	return ctx
}

func (s *SchedulerService) schedulerControllerContext() context.Context {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.controllerCtx != nil {
		return s.controllerCtx
	}
	return schedulerDetachedContext()
}

func schedulerDetachedContext() context.Context {
	return context.Background()
}

func (s *SchedulerService) scheduledReservationStore() scheduledRunReservationStore {
	if s.reservationStoreFn != nil {
		return s.reservationStoreFn()
	}
	provider, ok := s.persistence.(rawDBProvider)
	if !ok || provider.RawDB() == nil {
		return nil
	}
	return NewScheduledRunReservationService(provider.RawDB())
}

func (s *SchedulerService) fencedScheduledReservationStore() scheduledRunFencedStore {
	if s.fencedReservationStoreFn != nil {
		return s.fencedReservationStoreFn()
	}
	return nil
}

// SetFencedReservationStore installs the independent Runtime Worker authority.
// Legacy StartScheduler composition does not call this and keeps its existing
// reservation path until the maintenance-window deployment switch.
func (s *SchedulerService) SetFencedReservationStore(store *FencedScheduledRunReservationService) error {
	if store == nil {
		return errors.New("scheduler worker role: fenced reservation store is required")
	}
	s.fencedReservationStoreFn = func() scheduledRunFencedStore { return store }
	return nil
}

// SetFencedExecutionAuthority installs the immutable Catalog-view binder used
// by the independent Runtime Worker role. Legacy StartScheduler composition
// does not install one and retains its existing process-local behavior until
// the maintenance-window deployment switch.
func (s *SchedulerService) SetFencedExecutionAuthority(
	authority ScheduledFlowExecutionAuthority,
) error {
	if authority == nil {
		return errors.New("scheduler worker role: immutable scheduled Flow authority is required")
	}
	s.fencedExecutionAuthority = authority
	return nil
}

func (s *SchedulerService) schedulerInterval() time.Duration {
	if s.tickInterval > 0 {
		return s.tickInterval
	}
	return schedulerTickInterval
}

// SetSchedulerIntervalForTest shortens controller-driven loop tests without
// changing the production one-minute discovery interval.
func (s *SchedulerService) SetSchedulerIntervalForTest(interval time.Duration) {
	s.tickInterval = interval
}

// SetWorkerDiscoveryInterval binds scheduled discovery to the Runtime Worker
// role's existing bounded poll interval. Startup recovery runs first; normal
// ticks then discover new Catalog-owned occurrences without a concurrent
// startup scan or an unrelated hard-coded one-minute delay.
func (s *SchedulerService) SetWorkerDiscoveryInterval(interval time.Duration) error {
	if interval <= 0 || interval > time.Minute {
		return errors.New("scheduler worker role: discovery interval is outside the supported bound")
	}
	s.tickInterval = interval
	return nil
}

func (s *SchedulerService) getScheduledFlowDefinitions(
	ctx context.Context,
) ([]*models.FlowDefinition, error) {
	if s.listScheduledFlowDefinitionsFn != nil {
		return s.listScheduledFlowDefinitionsFn(ctx), nil
	}
	if s.metadata == nil {
		return nil, errors.New("scheduler: metadata service is unavailable")
	}
	if s.metadata.UsesRequestBoundRuntimeView() {
		return s.metadata.GetRuntimeScheduledFlowDefinitions(ctx)
	}
	return s.metadata.GetScheduledFlowDefinitions(ctx), nil
}

func (s *SchedulerService) getFlowDefinitionByID(ctx context.Context, id string) *models.FlowDefinition {
	if s.getFlowDefinitionByIDFn != nil {
		return s.getFlowDefinitionByIDFn(ctx, id)
	}
	if s.metadata == nil {
		return nil
	}
	return s.metadata.GetFlowDefinitionById(ctx, id)
}

func (s *SchedulerService) getFlowVersionByID(ctx context.Context, id string) *models.FlowVersion {
	if s.getFlowVersionByIDFn != nil {
		return s.getFlowVersionByIDFn(ctx, id)
	}
	if s.metadata == nil {
		return nil
	}
	return s.metadata.GetFlowVersionById(ctx, id)
}

func classifyScheduledExecutionPhase(matchedCount, successCount, failureCount int, loadErr error) (tickPhase, scheduledExecutionOutcome) {
	if loadErr != nil {
		return phaseLoadFailed, scheduledExecutionOutcomeLoadFailed
	}
	if matchedCount == 0 {
		return phaseInputsEmpty, scheduledExecutionOutcomeNoMatchingRecords
	}
	if failureCount == 0 {
		return phaseExecAllSucceeded, scheduledExecutionOutcomeAllSucceeded
	}
	if successCount == 0 {
		return phaseExecAllFailed, scheduledExecutionOutcomeAllFailure
	}
	return phaseExecMixed, scheduledExecutionOutcomePartialFailure
}

func classifyScheduledInputLoadOutcome(err error) scheduledExecutionOutcome {
	var scanErr *scheduledFlowCandidateScanBudgetError
	if errors.As(err, &scanErr) {
		return scheduledExecutionOutcomeScanBudgetExceeded
	}
	return scheduledExecutionOutcomeLoadFailed
}

func (s *SchedulerService) emitScheduledExecutionSummary(def *models.FlowDefinition, version *models.FlowVersion, now time.Time, outcome scheduledExecutionOutcome, matchedCount, successCount, failureCount int) {
	log.Printf(
		"Scheduler: scheduled_flow_summary flow_developer_name=%s active_version_id=%s window_id=%s outcome=%s matched_count=%d success_count=%d failure_count=%d",
		scheduledFlowDeveloperName(def),
		scheduledFlowActiveVersionID(def, version),
		scheduledFlowWindowID(version, now),
		outcome,
		matchedCount,
		successCount,
		failureCount,
	)
}

func scheduledFlowDeveloperName(def *models.FlowDefinition) string {
	if def == nil {
		return ""
	}
	return def.DeveloperName
}

func scheduledFlowActiveVersionID(def *models.FlowDefinition, version *models.FlowVersion) string {
	if version != nil && version.Id != "" {
		return version.Id
	}
	if def != nil {
		return def.ActiveVersionId
	}
	return ""
}

func scheduledFlowWindowID(version *models.FlowVersion, now time.Time) string {
	key, err := BuildScheduleWindowKey(version, now)
	if err != nil {
		return ""
	}
	return key
}

// shouldRunFlow checks if a scheduled flow should run at the given time.
//
// Timezone contract (issue #227): the caller passes a UTC-located `now` (this
// is the runtime invariant established by s.now() — see the helper above).
// Schedule metadata `StartDate` (date-only "YYYY-MM-DD") and `StartTime`
// ("HH:MM[:SS]") are interpreted as UTC throughout per the
// ScheduledFlow / WaitElement design assumption stated in issue #227. The
// occurrence helpers normalize every boundary to UTC before selecting the
// latest Once/Daily/Weekly target. A scheduled flow configured
// `StartTime: "09:00"` therefore fires at 09:00 UTC for every host regardless
// of host timezone. Org-local-time scheduling requires an explicit metadata
// field declaring the authoring timezone and remains out of scope.
func (s *SchedulerService) shouldRunFlow(version *models.FlowVersion, now time.Time) bool {
	_, ok := scheduledOccurrenceFor(version, now)
	return ok
}

// BuildScheduleWindowKey returns the canonical occurrence key for the current
// due scheduled-flow occurrence. The key is deterministic for replay / restart
// of the same logical occurrence and diverges across different occurrences.
func BuildScheduleWindowKey(version *models.FlowVersion, now time.Time) (string, error) {
	occurrence, ok := scheduledOccurrenceFor(version, now)
	if !ok {
		return "", fmt.Errorf("scheduled flow occurrence is not due at %s", now.UTC().Format(time.RFC3339))
	}
	return occurrence.frequency + ":" + occurrence.start.UTC().Format(time.RFC3339), nil
}

type scheduledOccurrence struct {
	frequency string
	start     time.Time
}

func scheduledOccurrenceFor(version *models.FlowVersion, now time.Time) (scheduledOccurrence, bool) {
	occurrence, frequency, ok := latestScheduledOccurrence(version, now.UTC())
	if !ok {
		return scheduledOccurrence{}, false
	}

	window := time.Minute
	if frequency == constants.FlowScheduleFrequencyOnce {
		window = 2 * time.Minute
	}
	if !now.Before(occurrence.start) && now.Before(occurrence.start.Add(window)) {
		return occurrence, true
	}
	return scheduledOccurrence{}, false
}

// scheduledOccurrenceBetween returns the latest logical occurrence in the
// exclusive/inclusive interval (after, through]. The scheduler advances one
// serial discovery cursor, so adjacent scans cannot both own the same boundary
// occurrence even when a scan itself is delayed past the scheduled minute.
func scheduledOccurrenceBetween(version *models.FlowVersion, after, through time.Time) (scheduledOccurrence, bool) {
	after = after.UTC()
	through = through.UTC()
	if !after.Before(through) {
		return scheduledOccurrence{}, false
	}

	occurrence, frequency, ok := latestScheduledOccurrence(version, through)
	if !ok || !occurrence.start.After(after) {
		return scheduledOccurrence{}, false
	}
	if frequency == constants.FlowScheduleFrequencyOnce && !through.Before(occurrence.start.Add(2*time.Minute)) {
		return scheduledOccurrence{}, false
	}
	return occurrence, true
}

func latestScheduledOccurrence(version *models.FlowVersion, through time.Time) (scheduledOccurrence, string, bool) {
	if version == nil || version.Metadata.Start == nil || version.Metadata.Start.Schedule == nil {
		return scheduledOccurrence{}, "", false
	}

	schedule := version.Metadata.Start.Schedule
	if strings.TrimSpace(schedule.StartTime) == "" {
		return scheduledOccurrence{}, "", false
	}

	var startDate time.Time
	if strings.TrimSpace(schedule.StartDate) != "" {
		parsedDate, err := time.Parse("2006-01-02", schedule.StartDate)
		if err != nil {
			return scheduledOccurrence{}, "", false
		}
		startDate = parsedDate
	}

	targetHour, targetMinute, targetSecond, ok := parseScheduledStartTime(schedule.StartTime)
	if !ok {
		return scheduledOccurrence{}, "", false
	}

	frequency, ok := constants.NormalizeFlowScheduleFrequency(schedule.Frequency)
	if !ok {
		return scheduledOccurrence{}, "", false
	}
	occurrenceKey, _ := constants.FlowScheduleOccurrenceKey(frequency)
	buildOccurrence := func(target time.Time) (scheduledOccurrence, string, bool) {
		return scheduledOccurrence{frequency: occurrenceKey, start: target}, frequency, true
	}

	switch frequency {
	case constants.FlowScheduleFrequencyOnce:
		if startDate.IsZero() {
			return scheduledOccurrence{}, "", false
		}
		targetTime := time.Date(startDate.Year(), startDate.Month(), startDate.Day(),
			targetHour, targetMinute, targetSecond, 0, time.UTC)
		if targetTime.After(through) {
			return scheduledOccurrence{}, "", false
		}
		return buildOccurrence(targetTime)

	case constants.FlowScheduleFrequencyDaily:
		targetTime := time.Date(through.Year(), through.Month(), through.Day(), targetHour, targetMinute, targetSecond, 0, time.UTC)
		if targetTime.After(through) {
			targetTime = targetTime.AddDate(0, 0, -1)
		}
		if !startDate.IsZero() {
			firstTarget := time.Date(startDate.Year(), startDate.Month(), startDate.Day(), targetHour, targetMinute, targetSecond, 0, time.UTC)
			if targetTime.Before(firstTarget) {
				return scheduledOccurrence{}, "", false
			}
		}
		return buildOccurrence(targetTime)

	case constants.FlowScheduleFrequencyWeekly:
		if startDate.IsZero() {
			return scheduledOccurrence{}, "", false
		}
		daysSinceTarget := (int(through.Weekday()) - int(startDate.Weekday()) + 7) % 7
		targetDay := through.AddDate(0, 0, -daysSinceTarget)
		targetTime := time.Date(targetDay.Year(), targetDay.Month(), targetDay.Day(), targetHour, targetMinute, targetSecond, 0, time.UTC)
		if targetTime.After(through) {
			targetTime = targetTime.AddDate(0, 0, -7)
		}
		firstTarget := time.Date(startDate.Year(), startDate.Month(), startDate.Day(), targetHour, targetMinute, targetSecond, 0, time.UTC)
		if targetTime.Before(firstTarget) {
			return scheduledOccurrence{}, "", false
		}
		return buildOccurrence(targetTime)

	default:
		return scheduledOccurrence{}, "", false
	}
}

func parseScheduledStartTime(raw string) (hour, minute, second int, ok bool) {
	normalized := strings.TrimSpace(strings.TrimSuffix(raw, "Z"))
	if normalized == "" {
		return 0, 0, 0, false
	}

	for _, layout := range []string{
		"15:04",
		"15:04:05",
		"15:04:05.999",
		"15:04:05.999999999",
	} {
		parsed, err := time.Parse(layout, normalized)
		if err == nil {
			return parsed.Hour(), parsed.Minute(), parsed.Second(), true
		}
	}

	parts := []int{}
	for _, p := range strings.Split(normalized, ":") {
		if n, err := strconv.Atoi(p); err == nil {
			parts = append(parts, n)
		}
	}
	if len(parts) >= 2 {
		second = 0
		if len(parts) >= 3 {
			second = parts[2]
		}
		return parts[0], parts[1], second, true
	}
	return 0, 0, 0, false
}

func scheduledInputRecordID(inputs map[string]interface{}) string {
	if inputs == nil {
		return ""
	}
	if id := scheduledRecordIDValue(inputs[constants.FieldID]); id != "" {
		return id
	}
	record, ok := inputs["$Record"]
	if !ok || record == nil {
		return ""
	}
	switch typed := record.(type) {
	case models.SObject:
		return scheduledRecordIDValue(typed[constants.FieldID])
	case map[string]interface{}:
		return scheduledRecordIDValue(typed[constants.FieldID])
	default:
		return ""
	}
}

func scheduledRecordIDValue(value interface{}) string {
	stringValue, _ := value.(string)
	return strings.TrimSpace(stringValue)
}