package automation

import (
	"bytes"
	"context"
	"errors"
	"log"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

func scheduledFlowVersion(frequency, startDate, startTime string) *models.FlowVersion {
	return &models.FlowVersion{
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule: &models.FlowSchedule{
					Frequency: frequency,
					StartDate: startDate,
					StartTime: startTime,
				},
			},
		},
	}
}

func scheduledFlowDefinition(id, developerName string) *models.FlowDefinition {
	return &models.FlowDefinition{
		Id:            id,
		DeveloperName: developerName,
	}
}

type scheduledRunReservationStoreStub struct {
	mu              sync.Mutex
	nextID          int
	byID            map[string]ScheduledRunExecutionStatus
	byKey           map[ScheduledRunReservationKey]string
	reserveCalls    []ScheduledRunReservationKey
	completeCalls   []string
	completeCtxErrs []error
	completeCtxs    []context.Context
	completeErr     error
}

func newScheduledRunReservationStoreStub() *scheduledRunReservationStoreStub {
	return &scheduledRunReservationStoreStub{
		byID:  make(map[string]ScheduledRunExecutionStatus),
		byKey: make(map[ScheduledRunReservationKey]string),
	}
}

func (s *scheduledRunReservationStoreStub) Reserve(_ context.Context, key ScheduledRunReservationKey) (ScheduledRunReserveResult, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.reserveCalls = append(s.reserveCalls, key)
	if id, ok := s.byKey[key]; ok {
		return ScheduledRunReserveResult{
			Status:        ScheduledRunReserveStatusDuplicate,
			ReservationID: id,
		}, nil
	}
	s.nextID++
	id := "reservation-" + time.Unix(int64(s.nextID), 0).UTC().Format("150405")
	s.byID[id] = ScheduledRunExecutionStatusReserved
	s.byKey[key] = id
	return ScheduledRunReserveResult{
		Status:        ScheduledRunReserveStatusReserved,
		ReservationID: id,
	}, nil
}

func (s *scheduledRunReservationStoreStub) Complete(ctx context.Context, reservationID string, _ ScheduledRunReservationKey, status ScheduledRunExecutionStatus, _ string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.completeCalls = append(s.completeCalls, reservationID)
	s.completeCtxErrs = append(s.completeCtxErrs, ctx.Err())
	s.completeCtxs = append(s.completeCtxs, ctx)
	s.byID[reservationID] = status
	return s.completeErr
}

func TestReservedScheduledExecutionObservationRequiresDurableTerminalOutcome(t *testing.T) {
	tests := []struct {
		name          string
		result        *models.FlowExecutionResult
		completeErr   error
		wantCompleted uint64
		wantFailed    uint64
	}{
		{name: "success", result: &models.FlowExecutionResult{IsSuccess: true}, wantCompleted: 1},
		{name: "failed", result: &models.FlowExecutionResult{IsSuccess: false}, wantCompleted: 1, wantFailed: 1},
		{name: "terminal write uncertain", result: &models.FlowExecutionResult{IsSuccess: true}, completeErr: errors.New("complete unavailable")},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			store := newScheduledRunReservationStoreStub()
			store.completeErr = tc.completeErr
			s := &SchedulerService{
				executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
					return tc.result, nil
				},
			}
			_, err := s.executeReservedScheduledVersion(
				t.Context(), store, "reservation-1", ScheduledRunReservationKey{},
				&models.FlowVersion{}, "ScheduledFlow", nil, newSystemAutomationUser(),
			)
			if tc.completeErr != nil && err == nil {
				t.Fatal("terminal write failure was not returned")
			}
			snapshot := s.WorkerObservationSnapshot()
			if snapshot.ExecutionStarted != 1 || snapshot.ExecutionCompleted != tc.wantCompleted || snapshot.ExecutionFailed != tc.wantFailed {
				t.Fatalf("worker observation=%+v", snapshot)
			}
		})
	}
}

func TestScheduledOccurrenceBetweenCatchesElapsedMinuteWithoutReowningBoundary(t *testing.T) {
	version := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	target := time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC)

	occurrence, ok := scheduledOccurrenceBetween(
		version,
		time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		time.Date(2026, 7, 14, 0, 20, 0, 0, time.UTC),
	)
	if !ok {
		t.Fatal("scheduledOccurrenceBetween(delayed scan) = not due, want 00:15 occurrence")
	}
	if !occurrence.start.Equal(target) {
		t.Fatalf("occurrence start = %v, want %v", occurrence.start, target)
	}

	if _, ok := scheduledOccurrenceBetween(
		version,
		target,
		time.Date(2026, 7, 14, 0, 16, 0, 0, time.UTC),
	); ok {
		t.Fatal("adjacent interval re-owned occurrence at exclusive lower boundary")
	}

	once := scheduledFlowVersion("Once", "2026-07-14", "00:15:00")
	if _, ok := scheduledOccurrenceBetween(
		once,
		time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		time.Date(2026, 7, 14, 0, 20, 0, 0, time.UTC),
	); ok {
		t.Fatal("delayed interval executed Once occurrence after its replay-safe window")
	}
}

func TestDispatchScheduledFlowsBetweenSlowMidnightFlowDoesNotSuppressQuarterPast(t *testing.T) {
	store := newScheduledRunReservationStoreStub()
	midnightDef := &models.FlowDefinition{
		Id:              "3000000000004103A",
		DeveloperName:   "SlowMidnightFlow",
		ActiveVersionId: "3010000000004103A",
	}
	quarterPastDef := &models.FlowDefinition{
		Id:              "3000000000004103B",
		DeveloperName:   "QuarterPastFlow",
		ActiveVersionId: "3010000000004103B",
	}
	midnightVersion := scheduledFlowVersion("Daily", "2026-07-13", "00:00:00")
	midnightVersion.Id = midnightDef.ActiveVersionId
	quarterPastVersion := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	quarterPastVersion.Id = quarterPastDef.ActiveVersionId

	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	secondExecuted := make(chan struct{})
	var signalFirst, signalSecond sync.Once
	var countsMu sync.Mutex
	executeCounts := map[string]int{}

	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{midnightDef, quarterPastDef}
		},
		getFlowDefinitionByIDFn: func(_ context.Context, id string) *models.FlowDefinition {
			switch id {
			case midnightDef.Id:
				return midnightDef
			case quarterPastDef.Id:
				return quarterPastDef
			default:
				t.Fatalf("unexpected flow definition id %q", id)
				return nil
			}
		},
		getFlowVersionByIDFn: func(_ context.Context, id string) *models.FlowVersion {
			switch id {
			case midnightVersion.Id:
				return midnightVersion
			case quarterPastVersion.Id:
				return quarterPastVersion
			default:
				t.Fatalf("unexpected flow version id %q", id)
				return nil
			}
		},
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "0010000000004103A"}}}, nil
		},
		executeFlowVersionFn: func(_ context.Context, version *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			countsMu.Lock()
			executeCounts[version.Id]++
			countsMu.Unlock()
			switch version.Id {
			case midnightVersion.Id:
				signalFirst.Do(func() { close(firstStarted) })
				<-releaseFirst
			case quarterPastVersion.Id:
				signalSecond.Do(func() { close(secondExecuted) })
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn: func() scheduledRunReservationStore { return store },
		// Keep a slot available for the duplicate scan while the two accepted runs overlap.
		// This assertion owns durable reservation deduplication, not capacity deferral.
		scheduledFlowWorkers: make(chan struct{}, 3),
	}

	firstDispatchReturned := make(chan struct{})
	go func() {
		s.dispatchScheduledFlowsBetween(
			time.Date(2026, 7, 13, 23, 59, 0, 0, time.UTC),
			time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		)
		close(firstDispatchReturned)
	}()
	select {
	case <-firstStarted:
	case <-time.After(3 * time.Second):
		t.Fatal("00:00 flow did not start")
	}
	select {
	case <-firstDispatchReturned:
	case <-time.After(3 * time.Second):
		t.Fatal("occurrence discovery waited for the 00:00 execution")
	}

	// The first execution is still blocked. Advancing the discovery interval by
	// 15 minutes models a run that lasts beyond 00:20 without sleeping in real time.
	s.dispatchScheduledFlowsBetween(
		time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC),
	)
	select {
	case <-secondExecuted:
	case <-time.After(3 * time.Second):
		t.Fatal("00:15 flow was suppressed while 00:00 flow was still running")
	}

	// A duplicate scan can race in a multi-worker process, but the persisted
	// reservation identity must still collapse it to one execution.
	s.dispatchScheduledFlowsBetween(
		time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC),
	)
	close(releaseFirst)
	s.wg.Wait()

	countsMu.Lock()
	defer countsMu.Unlock()
	if got := executeCounts[midnightVersion.Id]; got != 1 {
		t.Fatalf("00:00 execution count = %d, want 1", got)
	}
	if got := executeCounts[quarterPastVersion.Id]; got != 1 {
		t.Fatalf("00:15 execution count = %d, want 1", got)
	}
	if snapshot := s.WorkerObservationSnapshot(); snapshot.ExecutionStarted != 2 || snapshot.ExecutionCompleted != 2 || snapshot.ExecutionFailed != 0 || snapshot.OccurrencesAccepted != 2 || snapshot.AcceptanceAttempts != 3 || snapshot.AcceptanceEmpty != 1 {
		t.Fatalf("worker observation=%+v", snapshot)
	}
}

func TestDispatchScheduledFlowsBetweenSaturatedLaneSpawnsNoParkedWorker(t *testing.T) {
	def := &models.FlowDefinition{
		Id:              "3000000000004103C",
		DeveloperName:   "SaturatedLaneFlow",
		ActiveVersionId: "3010000000004103C",
	}
	version := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	version.Id = def.ActiveVersionId
	workers := make(chan struct{}, 1)
	workers <- struct{}{}

	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{def}
		},
		getFlowVersionByIDFn: func(context.Context, string) *models.FlowVersion { return version },
		getFlowDefinitionByIDFn: func(context.Context, string) *models.FlowDefinition {
			t.Fatal("saturated dispatch must not spawn a worker that reads metadata")
			return nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			t.Fatal("saturated dispatch must not execute")
			return nil, nil
		},
		scheduledFlowWorkers: workers,
	}

	complete := s.dispatchScheduledFlowsBetween(
		time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC),
	)
	if complete {
		t.Fatal("saturated discovery interval reported complete; want retryable cursor")
	}
	if got := len(workers); got != 1 {
		t.Fatalf("worker token count = %d, want saturated capacity unchanged", got)
	}
	if snapshot := s.WorkerObservationSnapshot(); snapshot.DiscoveryPasses != 1 || snapshot.DefinitionsConsidered != 1 || snapshot.CapacityDeferrals != 1 || snapshot.DiscoveryFailures != 1 || snapshot.OccurrencesAccepted != 0 {
		t.Fatalf("worker observation=%+v", snapshot)
	}
	<-workers
	s.wg.Wait()
}

func TestScheduledWorkerObservationRecordsConfirmedIdleDiscovery(t *testing.T) {
	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition { return nil },
	}
	if complete := s.dispatchScheduledFlowsBetween(time.Now().Add(-time.Minute), time.Now()); !complete {
		t.Fatal("empty discovery reported incomplete")
	}
	if snapshot := s.WorkerObservationSnapshot(); snapshot.DiscoveryPasses != 1 || snapshot.DiscoveryIdle != 1 || snapshot.DefinitionsConsidered != 0 || snapshot.DiscoveryFailures != 0 || snapshot.DiscoveryDuration < 0 {
		t.Fatalf("worker observation=%+v", snapshot)
	}
}

func TestRetriedDiscoveryIntervalDoesNotStarveLaterDefinition(t *testing.T) {
	firstDef := &models.FlowDefinition{Id: "3000000000004103E", DeveloperName: "FirstDueFlow", ActiveVersionId: "3010000000004103E"}
	secondDef := &models.FlowDefinition{Id: "3000000000004103F", DeveloperName: "SecondDueFlow", ActiveVersionId: "3010000000004103F"}
	firstVersion := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	firstVersion.Id = firstDef.ActiveVersionId
	secondVersion := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	secondVersion.Id = secondDef.ActiveVersionId
	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	var signalFirst sync.Once
	var countsMu sync.Mutex
	executeCounts := map[string]int{}

	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{firstDef, secondDef}
		},
		getFlowDefinitionByIDFn: func(_ context.Context, id string) *models.FlowDefinition {
			if id == firstDef.Id {
				return firstDef
			}
			return secondDef
		},
		getFlowVersionByIDFn: func(_ context.Context, id string) *models.FlowVersion {
			if id == firstVersion.Id {
				return firstVersion
			}
			return secondVersion
		},
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "0010000000004103E"}}}, nil
		},
		executeFlowVersionFn: func(_ context.Context, version *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			countsMu.Lock()
			executeCounts[version.Id]++
			countsMu.Unlock()
			if version.Id == firstVersion.Id {
				signalFirst.Do(func() { close(firstStarted) })
				<-releaseFirst
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn:   func() scheduledRunReservationStore { return newScheduledRunReservationStoreStub() },
		scheduledFlowWorkers: make(chan struct{}, 1),
	}
	accepted := map[string]struct{}{}
	after := time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC)
	through := time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC)

	if complete := s.dispatchScheduledFlowsBetweenWithProgress(after, through, accepted); complete {
		t.Fatal("first saturated scan reported complete")
	}
	select {
	case <-firstStarted:
	case <-time.After(3 * time.Second):
		t.Fatal("first due flow did not start")
	}
	close(releaseFirst)
	s.wg.Wait()

	if complete := s.dispatchScheduledFlowsBetweenWithProgress(after, through, accepted); !complete {
		t.Fatal("retry did not admit later due definition after capacity became available")
	}
	s.wg.Wait()

	countsMu.Lock()
	defer countsMu.Unlock()
	if got := executeCounts[firstVersion.Id]; got != 1 {
		t.Fatalf("first flow execution count = %d, want 1", got)
	}
	if got := executeCounts[secondVersion.Id]; got != 1 {
		t.Fatalf("second flow execution count = %d, want 1", got)
	}
}

func TestRetainedDiscoveryIntervalDoesNotReadmitOccurrenceAfterActiveVersionChange(t *testing.T) {
	targetDef := &models.FlowDefinition{Id: "3000000000004103G", DeveloperName: "VersionChangedFlow", ActiveVersionId: "3010000000004103G"}
	laterDef := &models.FlowDefinition{Id: "3000000000004103H", DeveloperName: "LaterDueFlow", ActiveVersionId: "3010000000004103I"}
	versionA := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	versionA.Id = targetDef.ActiveVersionId
	versionB := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	versionB.Id = "3010000000004103H"
	laterVersion := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	laterVersion.Id = laterDef.ActiveVersionId
	versionAStarted := make(chan struct{})
	releaseVersionA := make(chan struct{})
	releaseVersionB := make(chan struct{})
	var signalVersionA sync.Once
	var countsMu sync.Mutex
	executeCounts := map[string]int{}

	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{targetDef, laterDef}
		},
		getFlowDefinitionByIDFn: func(_ context.Context, id string) *models.FlowDefinition {
			if id == targetDef.Id {
				return targetDef
			}
			return laterDef
		},
		getFlowVersionByIDFn: func(_ context.Context, id string) *models.FlowVersion {
			switch id {
			case versionA.Id:
				return versionA
			case versionB.Id:
				return versionB
			case laterVersion.Id:
				return laterVersion
			default:
				return nil
			}
		},
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "0010000000004103G"}}}, nil
		},
		executeFlowVersionFn: func(_ context.Context, version *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			countsMu.Lock()
			executeCounts[version.Id]++
			countsMu.Unlock()
			switch version.Id {
			case versionA.Id:
				signalVersionA.Do(func() { close(versionAStarted) })
				<-releaseVersionA
			case versionB.Id:
				<-releaseVersionB
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn:   func() scheduledRunReservationStore { return newScheduledRunReservationStoreStub() },
		scheduledFlowWorkers: make(chan struct{}, 1),
	}
	accepted := map[string]struct{}{}
	after := time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC)
	through := time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC)

	if complete := s.dispatchScheduledFlowsBetweenWithProgress(after, through, accepted); complete {
		t.Fatal("first saturated scan reported complete")
	}
	select {
	case <-versionAStarted:
	case <-time.After(3 * time.Second):
		t.Fatal("original active version did not start")
	}
	close(releaseVersionA)
	s.wg.Wait()

	targetDef.ActiveVersionId = versionB.Id
	complete := s.dispatchScheduledFlowsBetweenWithProgress(after, through, accepted)
	close(releaseVersionB)
	s.wg.Wait()
	if !complete {
		t.Fatal("retained scan did not admit the remaining due definition")
	}

	countsMu.Lock()
	defer countsMu.Unlock()
	if got := executeCounts[versionA.Id]; got != 1 {
		t.Fatalf("original version execution count = %d, want 1", got)
	}
	if got := executeCounts[versionB.Id]; got != 0 {
		t.Fatalf("replacement version execution count = %d, want 0 for the already accepted occurrence", got)
	}
	if got := executeCounts[laterVersion.Id]; got != 1 {
		t.Fatalf("later flow execution count = %d, want 1", got)
	}
}

func TestDispatchedScheduledFlowRechecksActiveVersionBeforeExecution(t *testing.T) {
	discoveredDef := &models.FlowDefinition{
		Id:              "3000000000004103D",
		DeveloperName:   "DeactivatedWhileDispatched",
		ActiveVersionId: "3010000000004103D",
	}
	currentDef := *discoveredDef
	version := scheduledFlowVersion("Daily", "2026-07-13", "00:15:00")
	version.Id = discoveredDef.ActiveVersionId
	refreshStarted := make(chan struct{})
	allowRefresh := make(chan struct{})
	var signalRefresh sync.Once
	var executeCalls int

	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{discoveredDef}
		},
		getFlowDefinitionByIDFn: func(_ context.Context, id string) *models.FlowDefinition {
			if id != discoveredDef.Id {
				t.Fatalf("flow definition id = %q, want %q", id, discoveredDef.Id)
			}
			signalRefresh.Do(func() { close(refreshStarted) })
			<-allowRefresh
			return &currentDef
		},
		getFlowVersionByIDFn: func(_ context.Context, id string) *models.FlowVersion {
			if id != version.Id {
				t.Fatalf("flow version id = %q, want %q", id, version.Id)
			}
			return version
		},
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "0010000000004103D"}}}, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			executeCalls++
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn:   func() scheduledRunReservationStore { return newScheduledRunReservationStoreStub() },
		scheduledFlowWorkers: make(chan struct{}, 1),
	}

	if complete := s.dispatchScheduledFlowsBetween(
		time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
		time.Date(2026, 7, 14, 0, 15, 0, 0, time.UTC),
	); !complete {
		t.Fatal("available worker lane reported incomplete dispatch")
	}
	select {
	case <-refreshStarted:
	case <-time.After(3 * time.Second):
		t.Fatal("worker did not re-read FlowDefinition after acquiring capacity")
	}
	currentDef.ActiveVersionId = ""
	close(allowRefresh)
	s.wg.Wait()

	if executeCalls != 0 {
		t.Fatalf("deactivated scheduled flow executed %d time(s), want 0", executeCalls)
	}
}

// TestSchedulerNowReturnsUTC verifies conjunct (a): the now() helper's
// fallback path (nowFn == nil, line 128) returns a UTC-located time.
// Failure mode caught: line 128 reverted to bare time.Now().
func TestSchedulerNowReturnsUTC(t *testing.T) {
	s := &SchedulerService{} // nowFn is nil — exercises fallback at line 128

	got := s.now()

	if got.Location() != time.UTC {
		t.Errorf("s.now().Location() = %v; want time.UTC", got.Location())
	}
	if got.Location().String() != "UTC" {
		t.Errorf("s.now().Location().String() = %q; want %q", got.Location().String(), "UTC")
	}
}

// TestResumePausedInterviewsPassesUTCTime verifies conjunct (b): line 288
// passes UTC-located time to listFn after routing through s.now().
// Failure mode caught: line 288 left as direct time.Now() (raw local time).
func TestResumePausedInterviewsPassesUTCTime(t *testing.T) {
	// A fixed UTC instant used as the injected clock value.
	fixedUTC := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC)

	var capturedTime time.Time
	var nowFnCalled bool

	s := &SchedulerService{
		nowFn: func() time.Time {
			nowFnCalled = true
			return fixedUTC
		},
		listDuePausedInterviewsFn: func(_ context.Context, t time.Time, _ int, _ *models.UserSession) ([]string, error) {
			capturedTime = t
			return nil, nil // empty list — no resume work needed
		},
		resumeInterviewFn: func(_ context.Context, _ string, _ *models.UserSession) error {
			return nil
		},
	}

	s.resumePausedInterviews()

	// Conjunct (b1): the nowFn seam was invoked.
	if !nowFnCalled {
		t.Error("nowFn was not called; expected resumePausedInterviews to route through s.now()")
	}

	// Conjunct (b2): the time passed to listFn has UTC location.
	if capturedTime.Location() != time.UTC {
		t.Errorf("listFn received time with Location %v; want time.UTC", capturedTime.Location())
	}

	// Conjunct (b3): the captured time matches the fixed UTC instant exactly.
	if !capturedTime.Equal(fixedUTC) {
		t.Errorf("listFn received time %v; want %v", capturedTime, fixedUTC)
	}
}

// TestSchedulerNowForcesUTCEvenIfNowFnReturnsLocal verifies that even a
// misbehaving test seam returning a non-UTC time.Time cannot bypass the
// UTC-alignment guarantee. Closes the nowFn contract gap raised by Lane C
// Pass B Probe 1 (PR #245 round 0).
func TestSchedulerNowForcesUTCEvenIfNowFnReturnsLocal(t *testing.T) {
	shanghai, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		t.Skipf("Asia/Shanghai timezone not available on this host: %v", err)
	}
	localInstant := time.Date(2026, 5, 5, 14, 0, 0, 0, shanghai)

	s := &SchedulerService{
		nowFn: func() time.Time {
			return localInstant
		},
	}

	got := s.now()

	if got.Location() != time.UTC {
		t.Errorf("s.now().Location() = %v; want time.UTC even when nowFn returns local time", got.Location())
	}
	if !got.Equal(localInstant) {
		t.Errorf("s.now() = %v; want underlying instant equal to localInstant %v (Location should change but instant should match)", got, localInstant)
	}
}

// TestSchedulerDiscoveryClockUsesUTC verifies that the discovery cursor uses
// the same UTC-normalized clock seam as every other scheduler time boundary.
func TestSchedulerDiscoveryClockUsesUTC(t *testing.T) {
	fixedUTC := time.Date(2026, 5, 5, 18, 30, 0, 0, time.UTC)
	var nowFnCalled bool

	s := &SchedulerService{
		nowFn: func() time.Time {
			nowFnCalled = true
			return fixedUTC
		},
	}

	// Verify the seam returns UTC-located time.
	got := s.now()

	if !nowFnCalled {
		t.Error("nowFn was not called")
	}
	if got.Location() != time.UTC {
		t.Errorf("s.now().Location() = %v; want time.UTC", got.Location())
	}
	if !got.Equal(fixedUTC) {
		t.Errorf("s.now() = %v; want %v", got, fixedUTC)
	}
}

func TestExecuteScheduledDefinition_EmitsStructuredSummaryAndAlignedFailureAccounting(t *testing.T) {
	fixedNow := time.Date(2026, 5, 13, 10, 15, 0, 0, time.UTC)
	def := &models.FlowDefinition{
		Id:              "300648000000001AAA",
		DeveloperName:   "Issue648ScheduledSummary",
		ActiveVersionId: "301648000000001AAA",
	}
	version := &models.FlowVersion{
		Id: "301648000000001AAA",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule: &models.FlowSchedule{
					Frequency: "Daily",
					StartTime: "10:15",
				},
			},
		},
	}
	systemUser := &models.UserSession{ID: "system"}

	type wantState struct {
		consecutiveFailures int
		nextRetryAtZero     bool
		loggedStage         int
	}

	cases := []struct {
		name        string
		loadInputs  []map[string]interface{}
		loadErr     error
		execResults []struct {
			result *models.FlowExecutionResult
			err    error
		}
		initialState      *scheduledFlowRetryState
		wantOutcome       string
		wantMatchedCount  string
		wantSuccessCount  string
		wantFailureCount  string
		wantNoSecretValue string
		wantState         wantState
	}{
		{
			name:    "load_failed",
			loadErr: errors.New("load boom"),
			initialState: &scheduledFlowRetryState{
				consecutiveFailures: 2,
				nextRetryAt:         fixedNow.Add(-5 * time.Minute),
				lastLoggedStage:     1,
			},
			wantOutcome:       "load_failed",
			wantMatchedCount:  "0",
			wantSuccessCount:  "0",
			wantFailureCount:  "0",
			wantNoSecretValue: "do-not-log",
			wantState: wantState{
				consecutiveFailures: 3,
				nextRetryAtZero:     false,
				loggedStage:         2,
			},
		},
		{
			name: "scan_budget_exceeded",
			loadErr: &scheduledFlowCandidateScanBudgetError{
				ObjectName:          "GenericObject__c",
				ScannedCount:        scheduledFlowMaxScanned,
				MatchedCount:        0,
				ScanCap:             scheduledFlowMaxScanned,
				MatchedCap:          scheduledFlowMaxMatched,
				SQLPrefilterApplied: false,
			},
			initialState: &scheduledFlowRetryState{
				consecutiveFailures: 2,
				nextRetryAt:         fixedNow.Add(-5 * time.Minute),
				lastLoggedStage:     1,
			},
			wantOutcome:       "scan_budget_exceeded",
			wantMatchedCount:  "0",
			wantSuccessCount:  "0",
			wantFailureCount:  "0",
			wantNoSecretValue: "do-not-log",
			wantState: wantState{
				consecutiveFailures: 3,
				nextRetryAtZero:     false,
				loggedStage:         2,
			},
		},
		{
			name:       "no_matching_records",
			loadInputs: []map[string]interface{}{},
			initialState: &scheduledFlowRetryState{
				consecutiveFailures: 2,
				nextRetryAt:         fixedNow.Add(-5 * time.Minute),
				lastLoggedStage:     1,
			},
			wantOutcome:       "no_matching_records",
			wantMatchedCount:  "0",
			wantSuccessCount:  "0",
			wantFailureCount:  "0",
			wantNoSecretValue: "do-not-log",
			wantState: wantState{
				consecutiveFailures: 2,
				nextRetryAtZero:     false,
				loggedStage:         1,
			},
		},
		{
			name: "all_succeeded",
			loadInputs: []map[string]interface{}{
				{"Id": "001A", "Secret__c": "do-not-log"},
				{"Id": "001B", "Secret__c": "do-not-log"},
			},
			execResults: []struct {
				result *models.FlowExecutionResult
				err    error
			}{
				{result: &models.FlowExecutionResult{IsSuccess: true}},
				{result: &models.FlowExecutionResult{IsSuccess: true}},
			},
			initialState: &scheduledFlowRetryState{
				consecutiveFailures: 2,
				nextRetryAt:         fixedNow.Add(-5 * time.Minute),
				lastLoggedStage:     1,
			},
			wantOutcome:       "all_succeeded",
			wantMatchedCount:  "2",
			wantSuccessCount:  "2",
			wantFailureCount:  "0",
			wantNoSecretValue: "do-not-log",
			wantState: wantState{
				consecutiveFailures: 0,
				nextRetryAtZero:     true,
				loggedStage:         -1,
			},
		},
		{
			name: "partial_failure",
			loadInputs: []map[string]interface{}{
				{"Id": "001A", "Secret__c": "do-not-log"},
				{"Id": "001B", "Secret__c": "do-not-log"},
			},
			execResults: []struct {
				result *models.FlowExecutionResult
				err    error
			}{
				{result: &models.FlowExecutionResult{IsSuccess: true}},
				{err: errors.New("boom")},
			},
			initialState: &scheduledFlowRetryState{
				consecutiveFailures: 2,
				nextRetryAt:         fixedNow.Add(-5 * time.Minute),
				lastLoggedStage:     1,
			},
			wantOutcome:       "partial_failure",
			wantMatchedCount:  "2",
			wantSuccessCount:  "1",
			wantFailureCount:  "1",
			wantNoSecretValue: "do-not-log",
			wantState: wantState{
				consecutiveFailures: 0,
				nextRetryAtZero:     true,
				loggedStage:         -1,
			},
		},
		{
			name: "all_failure",
			loadInputs: []map[string]interface{}{
				{"Id": "001A", "Secret__c": "do-not-log"},
				{"Id": "001B", "Secret__c": "do-not-log"},
			},
			execResults: []struct {
				result *models.FlowExecutionResult
				err    error
			}{
				{err: errors.New("first failure")},
				{result: &models.FlowExecutionResult{IsSuccess: false, Errors: []models.FlowExecutionError{{Message: "second failure"}}}},
			},
			initialState: &scheduledFlowRetryState{
				consecutiveFailures: 2,
				lastLoggedStage:     1,
			},
			wantOutcome:       "all_failure",
			wantMatchedCount:  "2",
			wantSuccessCount:  "0",
			wantFailureCount:  "2",
			wantNoSecretValue: "do-not-log",
			wantState: wantState{
				consecutiveFailures: 3,
				nextRetryAtZero:     false,
				loggedStage:         2,
			},
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			var execIndex int
			buf, restoreLogs := captureSchedulerLogs(t)
			defer restoreLogs()

			s := &SchedulerService{
				nowFn: func() time.Time { return fixedNow },
				reservationStoreFn: func() scheduledRunReservationStore {
					return newScheduledRunReservationStoreStub()
				},
				loadScheduledInputsFn: func(_ context.Context, gotVersion *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
					if gotVersion != version {
						t.Fatalf("loadScheduledInputs version = %p, want %p", gotVersion, version)
					}
					return tc.loadInputs, tc.loadErr
				},
				executeFlowVersionFn: func(_ context.Context, gotVersion *models.FlowVersion, flowAPIName string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
					if gotVersion != version {
						t.Fatalf("executeScheduledVersion version = %p, want %p", gotVersion, version)
					}
					if flowAPIName != def.DeveloperName {
						t.Fatalf("flowAPIName = %q, want %q", flowAPIName, def.DeveloperName)
					}
					if execIndex >= len(tc.execResults) {
						t.Fatalf("unexpected executeFlowVersion call #%d", execIndex+1)
					}
					result := tc.execResults[execIndex]
					execIndex++
					return result.result, result.err
				},
				retryState: map[string]*scheduledFlowRetryState{
					effectiveStateKey(def, version): {
						consecutiveFailures: tc.initialState.consecutiveFailures,
						nextRetryAt:         tc.initialState.nextRetryAt,
						lastLoggedStage:     tc.initialState.lastLoggedStage,
					},
				},
			}

			s.executeScheduledDefinition(context.Background(), def, version, systemUser)

			summaryLine := findScheduledSummaryLine(t, buf.String())
			for _, want := range []string{
				"outcome=" + tc.wantOutcome,
				"matched_count=" + tc.wantMatchedCount,
				"success_count=" + tc.wantSuccessCount,
				"failure_count=" + tc.wantFailureCount,
				"flow_developer_name=" + def.DeveloperName,
				"active_version_id=" + version.Id,
			} {
				if !strings.Contains(summaryLine, want) {
					t.Fatalf("summary line %q missing %q", summaryLine, want)
				}
			}
			if !strings.Contains(summaryLine, "window_id=") {
				t.Fatalf("summary line %q missing window_id field", summaryLine)
			}
			if strings.Contains(summaryLine, tc.wantNoSecretValue) {
				t.Fatalf("summary line leaked input value %q: %q", tc.wantNoSecretValue, summaryLine)
			}

			state := s.retryState[effectiveStateKey(def, version)]
			if state == nil {
				t.Fatal("retry state missing after executeScheduledDefinition")
			}
			if state.consecutiveFailures != tc.wantState.consecutiveFailures {
				t.Fatalf("retry consecutiveFailures = %d, want %d", state.consecutiveFailures, tc.wantState.consecutiveFailures)
			}
			if gotZero := state.nextRetryAt.IsZero(); gotZero != tc.wantState.nextRetryAtZero {
				t.Fatalf("retry nextRetryAt zero = %t, want %t (value %v)", gotZero, tc.wantState.nextRetryAtZero, state.nextRetryAt)
			}
			if state.lastLoggedStage != tc.wantState.loggedStage {
				t.Fatalf("retry lastLoggedStage = %d, want %d", state.lastLoggedStage, tc.wantState.loggedStage)
			}
		})
	}
}

func TestExecuteScheduledDefinition_BackoffSkipped_EmitsSummaryWithoutLoadingInputs(t *testing.T) {
	fixedNow := time.Date(2026, 5, 13, 10, 15, 0, 0, time.UTC)
	def := &models.FlowDefinition{
		Id:              "300648000000002AAA",
		DeveloperName:   "Issue648BackoffSkipped",
		ActiveVersionId: "301648000000002AAA",
	}
	version := &models.FlowVersion{
		Id: "301648000000002AAA",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule: &models.FlowSchedule{
					Frequency: "Daily",
					StartTime: "10:15",
				},
			},
		},
	}
	key := effectiveStateKey(def, version)
	nextRetryAt := fixedNow.Add(1 * time.Hour)

	buf, restoreLogs := captureSchedulerLogs(t)
	defer restoreLogs()

	s := &SchedulerService{
		nowFn: func() time.Time { return fixedNow },
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			t.Fatal("loadScheduledInputs should not run for backoff-skipped window")
			return nil, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			t.Fatal("executeFlowVersion should not run for backoff-skipped window")
			return nil, nil
		},
		retryState: map[string]*scheduledFlowRetryState{
			key: {
				consecutiveFailures: 3,
				nextRetryAt:         nextRetryAt,
				lastLoggedStage:     2,
			},
		},
	}

	s.executeScheduledDefinition(context.Background(), def, version, &models.UserSession{ID: "system"})

	summaryLine := findScheduledSummaryLine(t, buf.String())
	for _, want := range []string{
		"outcome=backoff_skipped",
		"matched_count=0",
		"success_count=0",
		"failure_count=0",
		"flow_developer_name=" + def.DeveloperName,
		"active_version_id=" + version.Id,
	} {
		if !strings.Contains(summaryLine, want) {
			t.Fatalf("summary line %q missing %q", summaryLine, want)
		}
	}
	if !strings.Contains(summaryLine, "window_id=") {
		t.Fatalf("summary line %q missing window_id field", summaryLine)
	}

	state := s.retryState[key]
	if state == nil {
		t.Fatal("retry state missing after executeScheduledDefinition")
	}
	if state.consecutiveFailures != 3 {
		t.Fatalf("retry consecutiveFailures = %d, want 3", state.consecutiveFailures)
	}
	if !state.nextRetryAt.Equal(nextRetryAt) {
		t.Fatalf("retry nextRetryAt = %v, want %v", state.nextRetryAt, nextRetryAt)
	}
	if state.lastLoggedStage != 2 {
		t.Fatalf("retry lastLoggedStage = %d, want 2", state.lastLoggedStage)
	}
}

func TestExecuteScheduledFlows_EmitsStructuredSummaryForSkippedAndDueOutcomes(t *testing.T) {
	fixedNow := time.Date(2026, 5, 13, 10, 15, 0, 0, time.UTC)
	defNoActive := &models.FlowDefinition{Id: "300-no-active", DeveloperName: "NoActive"}
	defNotDue := &models.FlowDefinition{Id: "300-not-due", DeveloperName: "NotDue", ActiveVersionId: "301-not-due"}
	defNoMatches := &models.FlowDefinition{Id: "300-no-match", DeveloperName: "NoMatches", ActiveVersionId: "301-no-match"}

	notDueVersion := &models.FlowVersion{
		Id: "301-not-due",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule: &models.FlowSchedule{
					Frequency: "Daily",
					StartTime: "10:16",
				},
			},
		},
	}
	dueVersion := &models.FlowVersion{
		Id: "301-no-match",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule: &models.FlowSchedule{
					Frequency: "Daily",
					StartTime: "10:15",
				},
			},
		},
	}

	buf, restoreLogs := captureSchedulerLogs(t)
	defer restoreLogs()

	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{defNoActive, defNotDue, defNoMatches}
		},
		getFlowDefinitionByIDFn: func(_ context.Context, id string) *models.FlowDefinition {
			switch id {
			case defNoActive.Id:
				return defNoActive
			case defNotDue.Id:
				return defNotDue
			case defNoMatches.Id:
				return defNoMatches
			default:
				t.Fatalf("unexpected GetFlowDefinitionById(%q)", id)
				return nil
			}
		},
		getFlowVersionByIDFn: func(_ context.Context, id string) *models.FlowVersion {
			switch id {
			case notDueVersion.Id:
				return notDueVersion
			case dueVersion.Id:
				return dueVersion
			default:
				t.Fatalf("unexpected GetFlowVersionById(%q)", id)
				return nil
			}
		},
		loadScheduledInputsFn: func(_ context.Context, version *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
			if version.Id != dueVersion.Id {
				t.Fatalf("loadScheduledInputs called for version %q, want due version", version.Id)
			}
			return nil, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			t.Fatal("executeFlowVersion should not run for no-match scenario")
			return nil, nil
		},
	}

	s.dispatchScheduledFlowsBetween(fixedNow.Add(-time.Minute), fixedNow)
	s.wg.Wait()

	logs := buf.String()
	for _, want := range []string{
		"flow_developer_name=NoActive",
		"outcome=no_active_version",
		"flow_developer_name=NotDue",
		"outcome=not_due",
		"flow_developer_name=NoMatches",
		"outcome=no_matching_records",
	} {
		if !strings.Contains(logs, want) {
			t.Fatalf("logs missing %q: %s", want, logs)
		}
	}
	if strings.Contains(logs, "Secret__c") {
		t.Fatalf("structured summary leaked field name in logs: %s", logs)
	}
}

func TestShouldRunFlowFailClosedOnNilFlowVersion(t *testing.T) {
	s := &SchedulerService{}
	if s.shouldRunFlow(nil, time.Date(2026, 5, 11, 9, 0, 0, 0, time.UTC)) {
		t.Fatal("shouldRunFlow(nil, now) = true, want false")
	}
}

func TestShouldRunFlowDailyWeeklySkippedMinute(t *testing.T) {
	s := &SchedulerService{}

	tests := []struct {
		name    string
		version *models.FlowVersion
		now     time.Time
	}{
		{
			name:    "daily skipped minute",
			version: scheduledFlowVersion("Daily", "2026-05-11", "09:00"),
			now:     time.Date(2026, 5, 12, 9, 1, 0, 0, time.UTC),
		},
		{
			name:    "weekly skipped minute",
			version: scheduledFlowVersion("Weekly", "2026-05-11", "09:00"),
			now:     time.Date(2026, 5, 18, 9, 1, 0, 0, time.UTC),
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			if s.shouldRunFlow(tc.version, tc.now) {
				t.Fatalf("shouldRunFlow(%s) = true, want false", tc.name)
			}
		})
	}
}

func TestShouldRunFlowAcceptsSalesforceTimeValues(t *testing.T) {
	s := &SchedulerService{}

	tests := []struct {
		name      string
		frequency string
		startDate string
		startTime string
		now       time.Time
		want      bool
	}{
		{
			name:      "once accepts milliseconds with z suffix",
			frequency: "Once",
			startDate: "2026-05-11",
			startTime: "09:00:00.000Z",
			now:       time.Date(2026, 5, 11, 9, 0, 30, 0, time.UTC),
			want:      true,
		},
		{
			name:      "daily accepts utc suffix without fractional seconds",
			frequency: "Daily",
			startDate: "2026-05-11",
			startTime: "09:00:00Z",
			now:       time.Date(2026, 5, 12, 9, 0, 59, 0, time.UTC),
			want:      true,
		},
		{
			name:      "weekly accepts fractional seconds with z suffix",
			frequency: "Weekly",
			startDate: "2026-05-11",
			startTime: "09:00:00.123Z",
			now:       time.Date(2026, 5, 18, 9, 0, 30, 0, time.UTC),
			want:      true,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			version := scheduledFlowVersion(tc.frequency, tc.startDate, tc.startTime)
			if got := s.shouldRunFlow(version, tc.now); got != tc.want {
				t.Fatalf(
					"shouldRunFlow(%s, %q, %s) = %v, want %v",
					tc.frequency,
					tc.startTime,
					tc.now.Format(time.RFC3339),
					got,
					tc.want,
				)
			}
		})
	}
}

func TestShouldRunFlowOnceNonZeroTargetSecond(t *testing.T) {
	s := &SchedulerService{}
	version := scheduledFlowVersion("Once", "2026-05-11", "09:00:30")

	tests := []struct {
		name string
		now  time.Time
		want bool
	}{
		{
			name: "before target second",
			now:  time.Date(2026, 5, 11, 9, 0, 29, 0, time.UTC),
			want: false,
		},
		{
			name: "at target second",
			now:  time.Date(2026, 5, 11, 9, 0, 30, 0, time.UTC),
			want: true,
		},
		{
			name: "inside replay-safe once window",
			now:  time.Date(2026, 5, 11, 9, 1, 29, 0, time.UTC),
			want: true,
		},
		{
			name: "outside replay-safe once window",
			now:  time.Date(2026, 5, 11, 9, 2, 30, 0, time.UTC),
			want: false,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			if got := s.shouldRunFlow(version, tc.now); got != tc.want {
				t.Fatalf("shouldRunFlow(Once@09:00:30, %s) = %v, want %v", tc.now.Format(time.RFC3339), got, tc.want)
			}
		})
	}
}

func TestShouldRunFlowDailyWeeklyRespectTargetSecond(t *testing.T) {
	s := &SchedulerService{}

	tests := []struct {
		name    string
		version *models.FlowVersion
		now     time.Time
		want    bool
	}{
		{
			name:    "daily before target second",
			version: scheduledFlowVersion("Daily", "2026-05-11", "09:00:30"),
			now:     time.Date(2026, 5, 12, 9, 0, 29, 0, time.UTC),
			want:    false,
		},
		{
			name:    "daily at target second",
			version: scheduledFlowVersion("Daily", "2026-05-11", "09:00:30"),
			now:     time.Date(2026, 5, 12, 9, 0, 30, 0, time.UTC),
			want:    true,
		},
		{
			name:    "weekly before target second",
			version: scheduledFlowVersion("Weekly", "2026-05-11", "09:00:30"),
			now:     time.Date(2026, 5, 18, 9, 0, 29, 0, time.UTC),
			want:    false,
		},
		{
			name:    "weekly at target second",
			version: scheduledFlowVersion("Weekly", "2026-05-11", "09:00:30"),
			now:     time.Date(2026, 5, 18, 9, 0, 30, 0, time.UTC),
			want:    true,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			if got := s.shouldRunFlow(tc.version, tc.now); got != tc.want {
				t.Fatalf("shouldRunFlow(%s, %s) = %v, want %v", tc.version.Metadata.Start.Schedule.Frequency, tc.now.Format(time.RFC3339), got, tc.want)
			}
		})
	}
}

func TestBuildScheduleWindowKeyDeterministicPerOccurrence(t *testing.T) {
	version := scheduledFlowVersion("Once", "2026-05-11", "09:00:30")

	first, err := BuildScheduleWindowKey(version, time.Date(2026, 5, 11, 9, 0, 30, 0, time.UTC))
	if err != nil {
		t.Fatalf("BuildScheduleWindowKey(first): %v", err)
	}
	replay, err := BuildScheduleWindowKey(version, time.Date(2026, 5, 11, 9, 1, 29, 0, time.UTC))
	if err != nil {
		t.Fatalf("BuildScheduleWindowKey(replay): %v", err)
	}
	if first != replay {
		t.Fatalf("same logical occurrence keys differ: first=%q replay=%q", first, replay)
	}
	if _, err := BuildScheduleWindowKey(version, time.Date(2026, 5, 12, 9, 0, 30, 0, time.UTC)); err == nil {
		t.Fatal("BuildScheduleWindowKey(second day) = nil error, want fail-closed for a Once flow outside its occurrence")
	}
}

func TestDispatchScheduledFlowsUsesOccurrenceForWindowKey(t *testing.T) {
	store := newScheduledRunReservationStoreStub()
	def := &models.FlowDefinition{
		Id:              "300000000000001AAA",
		DeveloperName:   "Issue718ClockSample",
		ActiveVersionId: "301000000000001AAA",
	}
	version := scheduledFlowVersion("Daily", "2026-05-11", "09:00:30")
	version.Id = def.ActiveVersionId

	executeCalls := 0
	s := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{def}
		},
		getFlowDefinitionByIDFn: func(context.Context, string) *models.FlowDefinition {
			return def
		},
		getFlowVersionByIDFn: func(context.Context, string) *models.FlowVersion {
			return version
		},
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "001000000000001AAA"}}}, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			executeCalls++
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn: func() scheduledRunReservationStore {
			return store
		},
	}

	s.dispatchScheduledFlowsBetween(
		time.Date(2026, 5, 12, 8, 59, 59, 0, time.UTC),
		time.Date(2026, 5, 12, 9, 1, 0, 0, time.UTC),
	)
	s.wg.Wait()

	if executeCalls != 1 {
		t.Fatalf("executeScheduledVersion call count = %d, want 1", executeCalls)
	}
	if len(store.reserveCalls) != 1 {
		t.Fatalf("Reserve call count = %d, want 1", len(store.reserveCalls))
	}
	if got, want := store.reserveCalls[0].WindowToken, "daily:2026-05-12T09:00:30Z"; got != want {
		t.Fatalf("ScheduleWindowKey = %q, want due-check clock key %q", got, want)
	}
}

func TestExecuteScheduledDefinitionDuplicateSkipOnReplay(t *testing.T) {
	store := newScheduledRunReservationStoreStub()
	version := scheduledFlowVersion("Once", "2026-05-11", "09:00:30")
	version.Id = "301000000000001AAA"
	def := scheduledFlowDefinition("300000000000001AAA", "Issue718ReplaySafe")
	now := time.Date(2026, 5, 11, 9, 1, 29, 0, time.UTC)

	var executeCalls int
	s := &SchedulerService{
		nowFn: func() time.Time { return now },
		loadScheduledInputsFn: func(_ context.Context, _ *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{
				{
					"$Record": map[string]interface{}{"Id": "001000000000001AAA"},
				},
			}, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			executeCalls++
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn: func() scheduledRunReservationStore {
			return store
		},
	}

	s.executeScheduledDefinition(context.Background(), def, version, newSystemAutomationUser())
	s.executeScheduledDefinition(context.Background(), def, version, newSystemAutomationUser())

	if executeCalls != 1 {
		t.Fatalf("executeScheduledVersion call count = %d, want 1 after duplicate replay", executeCalls)
	}
	if len(store.completeCalls) != 1 {
		t.Fatalf("Complete call count = %d, want 1", len(store.completeCalls))
	}
}

func TestExecuteScheduledDefinitionCompletesReservationOnDispatchPanic(t *testing.T) {
	store := newScheduledRunReservationStoreStub()
	version := scheduledFlowVersion("Once", "2026-05-11", "09:00:30")
	version.Id = "301000000000001AAA"
	def := scheduledFlowDefinition("300000000000001AAA", "Issue718PanicSafe")
	now := time.Date(2026, 5, 11, 9, 1, 29, 0, time.UTC)

	s := &SchedulerService{
		nowFn: func() time.Time { return now },
		loadScheduledInputsFn: func(_ context.Context, _ *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "001000000000001AAA"}}}, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			panic("boom")
		},
		reservationStoreFn: func() scheduledRunReservationStore {
			return store
		},
	}

	defer func() {
		if r := recover(); r != nil {
			t.Fatalf("executeScheduledDefinition panicked; want panic-safe reservation completion: %v", r)
		}
		if len(store.completeCalls) != 1 {
			t.Fatalf("Complete call count = %d, want 1", len(store.completeCalls))
		}
		if got := store.byID[store.completeCalls[0]]; got != ScheduledRunExecutionStatusFailed {
			t.Fatalf("reservation state = %q, want %q", got, ScheduledRunExecutionStatusFailed)
		}
	}()

	s.executeScheduledDefinition(context.Background(), def, version, newSystemAutomationUser())
}

func TestExecuteScheduledDefinitionCompletesUnderContextCancellation(t *testing.T) {
	store := newScheduledRunReservationStoreStub()
	version := scheduledFlowVersion("Once", "2026-05-11", "09:00:30")
	version.Id = "301000000000001AAA"
	def := scheduledFlowDefinition("300000000000001AAA", "Issue718CancelSafe")
	now := time.Date(2026, 5, 11, 9, 1, 29, 0, time.UTC)

	ctx, cancel := context.WithCancel(context.Background())
	s := &SchedulerService{
		nowFn: func() time.Time { return now },
		loadScheduledInputsFn: func(_ context.Context, _ *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "001000000000001AAA"}}}, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			cancel()
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn: func() scheduledRunReservationStore {
			return store
		},
	}

	s.executeScheduledDefinition(ctx, def, version, newSystemAutomationUser())

	if len(store.completeCtxErrs) != 1 {
		t.Fatalf("Complete context captures = %d, want 1", len(store.completeCtxErrs))
	}
	if store.completeCtxErrs[0] != nil {
		t.Fatalf("Complete observed cancelled context: %v", store.completeCtxErrs[0])
	}
}

func TestExecuteScheduledDefinitionCompleteContextDropsUnsafeValuesAfterCancel(t *testing.T) {
	type unsafeKey struct{}

	store := newScheduledRunReservationStoreStub()
	version := scheduledFlowVersion("Once", "2026-05-11", "09:00:30")
	version.Id = "301000000000001AAA"
	def := scheduledFlowDefinition("300000000000001AAA", "Issue2861CompleteContext")
	now := time.Date(2026, 5, 11, 9, 1, 29, 0, time.UTC)

	ctx := context.WithValue(context.Background(), constants.ContextKeyRequestID, "req-123")
	ctx = context.WithValue(ctx, constants.ContextKeyTraceparent, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
	ctx = context.WithValue(ctx, unsafeKey{}, "request-scoped")
	ctx, cancel := context.WithCancel(ctx)

	s := &SchedulerService{
		nowFn: func() time.Time { return now },
		loadScheduledInputsFn: func(_ context.Context, _ *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{"$Record": map[string]interface{}{"Id": "001000000000001AAA"}}}, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			cancel()
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		reservationStoreFn: func() scheduledRunReservationStore {
			return store
		},
	}

	s.executeScheduledDefinition(ctx, def, version, newSystemAutomationUser())

	if len(store.completeCtxs) != 1 {
		t.Fatalf("Complete context captures = %d, want 1", len(store.completeCtxs))
	}
	completeCtx := store.completeCtxs[0]
	if err := completeCtx.Err(); err != nil {
		t.Fatalf("complete context inherited source cancellation: %v", err)
	}
	if got, _ := completeCtx.Value(constants.ContextKeyRequestID).(string); got != "req-123" {
		t.Fatalf("request id = %q, want req-123", got)
	}
	if got, _ := completeCtx.Value(constants.ContextKeyTraceparent).(string); got == "" {
		t.Fatal("traceparent was not preserved")
	}
	if got := completeCtx.Value(unsafeKey{}); got != nil {
		t.Fatalf("unsafe request-scoped value leaked into complete context: %#v", got)
	}
}

var scheduledExecutionOutcomeAllValues = []scheduledExecutionOutcome{
	scheduledExecutionOutcomeNoActiveVersion,
	scheduledExecutionOutcomeNotDue,
	scheduledExecutionOutcomeLoadFailed,
	scheduledExecutionOutcomeScanBudgetExceeded,
	scheduledExecutionOutcomeNoMatchingRecords,
	scheduledExecutionOutcomePartialFailure,
	scheduledExecutionOutcomeAllFailure,
	scheduledExecutionOutcomeAllSucceeded,
	scheduledExecutionOutcomeBackoffSkipped,
}

func TestAllScheduledOutcomesCovered(t *testing.T) {
	// Most explicit viable guard for this codebase: it is intentionally
	// hand-maintained and therefore not structurally bulletproof.
	assertedOutcomes := map[string]struct{}{
		"no_active_version":    {},
		"not_due":              {},
		"load_failed":          {},
		"scan_budget_exceeded": {},
		"no_matching_records":  {},
		"partial_failure":      {},
		"all_failure":          {},
		"all_succeeded":        {},
		"backoff_skipped":      {},
	}

	for _, outcome := range scheduledExecutionOutcomeAllValues {
		if _, ok := assertedOutcomes[string(outcome)]; !ok {
			t.Fatalf("scheduledExecutionOutcome %q is not asserted by any test outcome map", outcome)
		}
	}
}

func captureSchedulerLogs(t *testing.T) (*bytes.Buffer, func()) {
	t.Helper()

	var buf bytes.Buffer
	oldWriter := log.Writer()
	oldFlags := log.Flags()
	log.SetOutput(&buf)
	log.SetFlags(0)

	return &buf, func() {
		log.SetOutput(oldWriter)
		log.SetFlags(oldFlags)
	}
}

func findScheduledSummaryLine(t *testing.T, logs string) string {
	t.Helper()

	for _, line := range strings.Split(logs, "\n") {
		if strings.Contains(line, "scheduled_flow_summary") {
			return line
		}
	}
	t.Fatalf("no scheduled_flow_summary line found in logs: %s", logs)
	return ""
}