package worker

import (
	"context"
	"errors"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func TestControllerDoesNotReportReadyBeforeStartupReconcile(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	source := &testSource{}
	source.onReconcile = func(*testSource) {
		close(started)
		<-release
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("startup reconciliation did not begin")
	}
	if controller.Ready() || controller.Snapshot().ClaimCalls != 0 {
		t.Fatalf("controller became ready or claimed before reconciliation: %+v", controller.Snapshot())
	}
	close(release)
	waitFor(t, controller.Ready, "readiness after startup reconciliation")
	stopController(t, cancel, done)
}

func TestControllerSourceFailureStopsBeforeReady(t *testing.T) {
	failure := errors.New("source unavailable")
	for _, tc := range []struct {
		name         string
		make         func() Source
		want         string
		wantUnusable bool
	}{
		{
			name: "readiness",
			make: func() Source { return &testSource{readyErr: failure} },
			want: "source readiness",
		},
		{
			name:         "reconcile",
			make:         func() Source { return &testSource{reconcileErr: failure} },
			want:         "reconcile",
			wantUnusable: true,
		},
		{
			name:         "claim",
			make:         func() Source { return &testSource{claimErr: failure} },
			want:         "claim",
			wantUnusable: true,
		},
		{
			name: "next work",
			make: func() Source { return &testSource{nextErr: failure} },
			want: "next work",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			controller, err := NewController(tc.make(), nil, testConfig())
			if err != nil {
				t.Fatal(err)
			}
			runErr := controller.Run(context.Background())
			if !errors.Is(runErr, failure) || !strings.Contains(runErr.Error(), tc.want) {
				t.Fatalf("Run error=%v, want wrapped %q failure", runErr, tc.want)
			}
			if controller.Ready() {
				t.Fatal("failed controller reported ready")
			}
			if snapshot := controller.Snapshot(); snapshot.Unusable != tc.wantUnusable {
				t.Fatalf("source failure classification=%+v, want unusable=%t", snapshot, tc.wantUnusable)
			}
		})
	}
}

type reconcileFuncSource struct {
	*testSource
	reconcile func(context.Context) error
}

func (s *reconcileFuncSource) Reconcile(ctx context.Context) error {
	return s.reconcile(ctx)
}

func TestControllerReconcileErrorWithUnknownOutcomePreventsReuse(t *testing.T) {
	failure := errors.New("reconcile response lost after durable acceptance")
	source := &reconcileFuncSource{
		testSource: &testSource{},
		reconcile:  func(context.Context) error { return failure },
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, failure) {
		t.Fatalf("reconcile error=%v", runErr)
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable {
		t.Fatalf("unknown reconcile outcome left controller reusable: %+v", snapshot)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
		t.Fatalf("unsafe controller reuse error=%v", runErr)
	}
}

func TestControllerKnownUncommittedReconcileErrorIsReusable(t *testing.T) {
	failure := errors.New("reconcile transaction rolled back")
	var reconcileCalls atomic.Int64
	source := &reconcileFuncSource{
		testSource: &testSource{},
		reconcile: func(context.Context) error {
			if reconcileCalls.Add(1) == 1 {
				return errors.Join(ErrSourceOperationNotCommitted, failure)
			}
			return nil
		},
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, failure) {
		t.Fatalf("known uncommitted reconcile error=%v", runErr)
	}
	if snapshot := controller.Snapshot(); snapshot.Unusable {
		t.Fatalf("known uncommitted reconcile disabled controller: %+v", snapshot)
	}
	cancel, done := runController(t, controller)
	waitFor(t, controller.Ready, "readiness after safe reconcile restart")
	stopController(t, cancel, done)
}

func TestControllerCancellationDoesNotHideUnknownReconcileOutcome(t *testing.T) {
	started := make(chan struct{})
	source := &reconcileFuncSource{
		testSource: &testSource{},
		reconcile: func(ctx context.Context) error {
			close(started)
			<-ctx.Done()
			return ctx.Err()
		},
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("reconciliation did not start")
	}
	cancel()
	select {
	case runErr := <-done:
		if !errors.Is(runErr, context.Canceled) {
			t.Fatalf("unknown reconcile outcome was hidden: %v", runErr)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("controller did not stop")
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable {
		t.Fatalf("unknown reconcile outcome left controller reusable: %+v", snapshot)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
		t.Fatalf("unsafe controller reuse error=%v", runErr)
	}
}

func TestControllerRuntimeReconcileFailureCancelsAndDrainsAttempts(t *testing.T) {
	failure := errors.New("runtime reconcile response lost")
	peerCanceled := make(chan struct{})
	var reconcileCalls atomic.Int64
	source := &reconcileFuncSource{
		testSource: &testSource{queue: []Attempt{{ID: "peer", Run: func(ctx context.Context) error {
			<-ctx.Done()
			close(peerCanceled)
			return nil
		}}}},
		reconcile: func(context.Context) error {
			if reconcileCalls.Add(1) == 1 {
				return nil
			}
			return failure
		},
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	_, done := runController(t, controller)
	waitFor(t, func() bool { return controller.Snapshot().InFlight == 1 }, "active attempt")
	controller.RequestReconcile()
	select {
	case runErr := <-done:
		if !errors.Is(runErr, failure) {
			t.Fatalf("runtime reconcile error=%v", runErr)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("controller did not stop after runtime reconcile failure")
	}
	select {
	case <-peerCanceled:
	default:
		t.Fatal("runtime reconcile failure did not cancel and drain the active attempt")
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable || snapshot.InFlight != 0 {
		t.Fatalf("runtime reconcile failure accounting=%+v", snapshot)
	}
}

func TestControllerKnownUncommittedRuntimeReconcileCancellationIsReusable(t *testing.T) {
	reconcileStarted := make(chan struct{})
	peerCanceled := make(chan struct{})
	var reconcileCalls atomic.Int64
	source := &reconcileFuncSource{
		testSource: &testSource{queue: []Attempt{{ID: "peer", Run: func(ctx context.Context) error {
			<-ctx.Done()
			close(peerCanceled)
			return nil
		}}}},
		reconcile: func(ctx context.Context) error {
			if reconcileCalls.Add(1) != 2 {
				return nil
			}
			close(reconcileStarted)
			<-ctx.Done()
			return errors.Join(ErrSourceOperationNotCommitted, ctx.Err())
		},
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool { return controller.Snapshot().InFlight == 1 }, "active attempt")
	controller.RequestReconcile()
	select {
	case <-reconcileStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("runtime reconciliation did not start")
	}
	stopController(t, cancel, done)
	select {
	case <-peerCanceled:
	default:
		t.Fatal("safe reconcile cancellation did not drain the active attempt")
	}
	if snapshot := controller.Snapshot(); snapshot.Unusable || snapshot.InFlight != 0 {
		t.Fatalf("known uncommitted runtime reconcile accounting=%+v", snapshot)
	}

	cancel, done = runController(t, controller)
	waitFor(t, controller.Ready, "readiness after safe runtime reconcile restart")
	stopController(t, cancel, done)
}

type blockingStageSource struct {
	stage   string
	started chan<- struct{}
}

func (s *blockingStageSource) wait(ctx context.Context, stage string) error {
	if s.stage != stage {
		return nil
	}
	s.started <- struct{}{}
	<-ctx.Done()
	return ctx.Err()
}

func (s *blockingStageSource) CheckReady(ctx context.Context) error {
	return s.wait(ctx, "readiness")
}

func (s *blockingStageSource) Reconcile(ctx context.Context) error {
	err := s.wait(ctx, "reconcile")
	if err != nil {
		err = errors.Join(ErrSourceOperationNotCommitted, err)
	}
	return err
}

func (s *blockingStageSource) Claim(ctx context.Context, _ int) (ClaimBatch, error) {
	err := s.wait(ctx, "claim")
	if err != nil {
		// This test source never performs a durable transition, so it can
		// certify that cancellation left no claim behind.
		err = errors.Join(ErrSourceOperationNotCommitted, err)
	}
	return ClaimBatch{}, err
}

func (s *blockingStageSource) NextWorkAt(ctx context.Context) (time.Time, bool, error) {
	return time.Time{}, false, s.wait(ctx, "next-work")
}

func TestControllerCancellationStopsEverySourceStage(t *testing.T) {
	for _, stage := range []string{"readiness", "reconcile", "claim", "next-work"} {
		t.Run(stage, func(t *testing.T) {
			started := make(chan struct{}, 1)
			controller, err := NewController(&blockingStageSource{stage: stage, started: started}, nil, testConfig())
			if err != nil {
				t.Fatal(err)
			}
			cancel, done := runController(t, controller)
			select {
			case <-started:
			case <-time.After(2 * time.Second):
				t.Fatalf("%s did not begin", stage)
			}
			stopController(t, cancel, done)
			if snapshot := controller.Snapshot(); snapshot.Unusable {
				t.Fatalf("known read-only or uncommitted %s cancellation disabled controller: %+v", stage, snapshot)
			}
		})
	}
}

type failingGenerationProbe struct{ err error }

func (p failingGenerationProbe) ReadGeneration(context.Context, uint64) (uint64, error) {
	return 0, p.err
}

func TestControllerInitialGenerationFailureStopsBeforeReady(t *testing.T) {
	failure := errors.New("generation unavailable")
	config := testConfig()
	config.GenerationInterval = time.Hour
	controller, err := NewController(&testSource{}, failingGenerationProbe{err: failure}, config)
	if err != nil {
		t.Fatal(err)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, failure) {
		t.Fatalf("Run error=%v", runErr)
	}
	if snapshot := controller.Snapshot(); controller.Ready() || snapshot.Reconciliations != 0 || snapshot.Unusable {
		t.Fatalf("generation failure crossed readiness boundary: %+v", controller.Snapshot())
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, failure) {
		t.Fatalf("read-only generation failure disabled reuse: %v", runErr)
	}
}

func TestControllerAttemptFailureCancelsAndDrainsPeers(t *testing.T) {
	started := make(chan struct{}, 2)
	releaseFailure := make(chan struct{})
	peerCanceled := make(chan struct{})
	source := &testSource{queue: []Attempt{
		{ID: "failure", Run: func(context.Context) error {
			started <- struct{}{}
			<-releaseFailure
			return errors.New("terminal disposition unknown")
		}},
		{ID: "peer", Run: func(ctx context.Context) error {
			started <- struct{}{}
			<-ctx.Done()
			close(peerCanceled)
			return nil
		}},
	}}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	done := make(chan error, 1)
	go func() { done <- controller.Run(context.Background()) }()
	for range 2 {
		select {
		case <-started:
		case <-time.After(2 * time.Second):
			t.Fatal("attempts did not both start")
		}
	}
	close(releaseFailure)
	select {
	case runErr := <-done:
		if runErr == nil {
			t.Fatal("attempt failure did not stop controller")
		}
	case <-time.After(2 * time.Second):
		t.Fatal("controller did not stop after attempt failure")
	}
	select {
	case <-peerCanceled:
	default:
		t.Fatal("peer attempt was not canceled and drained")
	}
	if snapshot := controller.Snapshot(); snapshot.InFlight != 0 || snapshot.AttemptFailures != 1 {
		t.Fatalf("snapshot after fatal attempt=%+v", snapshot)
	}
}

func TestControllerReconcileRequestDuringMorePrecedesNextClaim(t *testing.T) {
	prematureClaim := errors.New("claim ran before requested reconciliation")
	var claimCalls atomic.Int64
	var prematureDetected atomic.Bool
	var controller *Controller
	source := &testSource{}
	source.claim = func(ctx context.Context, _ int) (ClaimBatch, error) {
		switch claimCalls.Add(1) {
		case 1:
			controller.RequestReconcile()
			return ClaimBatch{
				Attempts: []Attempt{{ID: "first", Run: func(ctx context.Context) error {
					<-ctx.Done()
					return nil
				}}},
				More: true,
			}, nil
		case 2:
			reconciles, _ := source.counts()
			if reconciles != 2 {
				prematureDetected.Store(true)
				return ClaimBatch{}, prematureClaim
			}
			return ClaimBatch{}, nil
		default:
			return ClaimBatch{}, nil
		}
	}

	var err error
	controller, err = NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool {
		return controller.Snapshot().ClaimCalls >= 2
	}, "claim pass after in-flight reconciliation request")
	if prematureDetected.Load() {
		runErr := <-done
		if errors.Is(runErr, prematureClaim) {
			t.Fatalf("RequestReconcile did not precede the next More claim: %v", runErr)
		}
		t.Fatalf("controller stopped unexpectedly: %v", runErr)
	}
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.ReconcileRequests != 1 || snapshot.Reconciliations != 2 {
		t.Fatalf("requested reconciliation accounting=%+v", snapshot)
	}
}

func TestControllerQueuedWakeDoesNotBypassReconcileBeforeClaim(t *testing.T) {
	startupReconcile := make(chan struct{})
	releaseStartup := make(chan struct{})
	prematureClaim := errors.New("wake claim ran before queued reconciliation")
	var prematureDetected atomic.Bool
	var claimCalls atomic.Int64
	source := &testSource{}
	source.onReconcile = func(source *testSource) {
		if source.reconciles == 1 {
			close(startupReconcile)
			<-releaseStartup
		}
	}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		claimCalls.Add(1)
		reconciles, _ := source.counts()
		if reconciles != 2 {
			prematureDetected.Store(true)
			return ClaimBatch{}, prematureClaim
		}
		return ClaimBatch{}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-startupReconcile:
	case <-time.After(2 * time.Second):
		t.Fatal("startup reconciliation did not begin")
	}
	controller.Wake()
	controller.RequestReconcile()
	close(releaseStartup)
	waitFor(t, func() bool { return claimCalls.Load() >= 1 }, "first claim turn")
	if prematureDetected.Load() {
		runErr := <-done
		if errors.Is(runErr, prematureClaim) {
			t.Fatalf("queued Wake bypassed RequestReconcile: %v", runErr)
		}
		t.Fatalf("controller stopped unexpectedly: %v", runErr)
	}
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.WakeRequests != 1 || snapshot.ReconcileRequests != 1 ||
		snapshot.Reconciliations != 2 {
		t.Fatalf("queued signal accounting=%+v", snapshot)
	}
}

func TestControllerGenerationChangeDuringMorePrecedesNextClaim(t *testing.T) {
	claimStarted := make(chan struct{})
	releaseClaim := make(chan struct{})
	prematureClaim := errors.New("claim ran before generation reconciliation")
	probe := &testGenerationProbe{}
	probe.generation.Store(1)
	var claimCalls atomic.Int64
	var prematureDetected atomic.Bool
	source := &testSource{}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		switch claimCalls.Add(1) {
		case 1:
			close(claimStarted)
			<-releaseClaim
			return ClaimBatch{
				Attempts: []Attempt{{ID: "first", Run: func(ctx context.Context) error {
					<-ctx.Done()
					return nil
				}}},
				More: true,
			}, nil
		case 2:
			reconciles, _ := source.counts()
			if reconciles != 2 {
				prematureDetected.Store(true)
				return ClaimBatch{}, prematureClaim
			}
			return ClaimBatch{}, nil
		default:
			return ClaimBatch{}, nil
		}
	}
	config := testConfig()
	config.GenerationInterval = 2 * time.Millisecond
	controller, err := NewController(source, probe, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-claimStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("first claim did not begin")
	}
	probe.generation.Store(2)
	time.Sleep(10 * time.Millisecond)
	close(releaseClaim)
	waitFor(t, func() bool { return claimCalls.Load() >= 2 }, "claim after queued generation tick")
	if prematureDetected.Load() {
		runErr := <-done
		if errors.Is(runErr, prematureClaim) {
			t.Fatalf("generation change did not precede the next More claim: %v", runErr)
		}
		t.Fatalf("controller stopped unexpectedly: %v", runErr)
	}
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.GenerationChanges != 1 || snapshot.Reconciliations != 2 {
		t.Fatalf("generation reconciliation accounting=%+v", snapshot)
	}
}

func TestControllerPeriodicReconcileDuringMorePrecedesNextClaim(t *testing.T) {
	claimStarted := make(chan struct{})
	releaseClaim := make(chan struct{})
	prematureClaim := errors.New("claim ran before periodic reconciliation")
	var claimCalls atomic.Int64
	var prematureDetected atomic.Bool
	source := &testSource{}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		switch claimCalls.Add(1) {
		case 1:
			close(claimStarted)
			<-releaseClaim
			return ClaimBatch{
				Attempts: []Attempt{{ID: "first", Run: func(ctx context.Context) error {
					<-ctx.Done()
					return nil
				}}},
				More: true,
			}, nil
		case 2:
			reconciles, _ := source.counts()
			if reconciles != 2 {
				prematureDetected.Store(true)
				return ClaimBatch{}, prematureClaim
			}
			return ClaimBatch{}, nil
		default:
			return ClaimBatch{}, nil
		}
	}
	config := testConfig()
	config.ReconcileInterval = 2 * time.Millisecond
	controller, err := NewController(source, nil, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-claimStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("first claim did not begin")
	}
	time.Sleep(10 * time.Millisecond)
	close(releaseClaim)
	waitFor(t, func() bool { return claimCalls.Load() >= 2 }, "claim after queued periodic reconciliation")
	if prematureDetected.Load() {
		runErr := <-done
		if errors.Is(runErr, prematureClaim) {
			t.Fatalf("periodic reconciliation did not precede the next More claim: %v", runErr)
		}
		t.Fatalf("controller stopped unexpectedly: %v", runErr)
	}
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.Reconciliations < 2 {
		t.Fatalf("periodic reconciliation accounting=%+v", snapshot)
	}
}

func TestControllerShutdownDrainAccountsEveryAttemptResult(t *testing.T) {
	attemptFailure := errors.New("durable disposition failed")
	controller, err := NewController(&testSource{}, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	controller.inFlight.Store(2)
	results := make(chan attemptResult, 2)
	results <- attemptResult{}
	results <- attemptResult{err: attemptFailure}

	runErr := controller.stopAttempts(func() {}, results, 2, nil)
	if !errors.Is(runErr, attemptFailure) {
		t.Fatalf("shutdown drain hid attempt failure: %v", runErr)
	}
	if snapshot := controller.Snapshot(); snapshot.InFlight != 0 || snapshot.Completed != 1 ||
		snapshot.AttemptFailures != 1 || !snapshot.Unusable {
		t.Fatalf("shutdown drain accounting=%+v", snapshot)
	}
}

func TestControllerCanRestartOnlyAfterGracefulShutdown(t *testing.T) {
	controller, err := NewController(&testSource{}, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	for range 2 {
		cancel, done := runController(t, controller)
		waitFor(t, controller.Ready, "controller readiness")
		stopController(t, cancel, done)
	}
	snapshot := controller.Snapshot()
	if snapshot.Unusable || snapshot.ReadinessChecks != 2 || snapshot.Reconciliations != 2 {
		t.Fatalf("graceful restart snapshot=%+v", snapshot)
	}
}

func TestControllerShutdownClearsPendingDueObservation(t *testing.T) {
	source := &testSource{next: time.Now().Add(time.Hour)}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool { return controller.Snapshot().NextWorkScheduled }, "pending due timer")
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.NextWorkScheduled || snapshot.NextWorkUnixNano != 0 {
		t.Fatalf("shutdown retained stopped timer observation: %+v", snapshot)
	}
}

func TestControllerCompletionDoesNotBypassFutureDueTimer(t *testing.T) {
	release := make(chan struct{})
	started := make(chan struct{})
	var claimCalls atomic.Int64
	source := &testSource{next: time.Now().Add(time.Hour)}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		if claimCalls.Add(1) == 1 {
			return ClaimBatch{Attempts: []Attempt{{ID: "active", Run: func(context.Context) error {
				close(started)
				<-release
				return nil
			}}}, More: true}, nil
		}
		return ClaimBatch{}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("attempt did not start")
	}
	waitFor(t, func() bool {
		snapshot := controller.Snapshot()
		return snapshot.InFlight == 1 && snapshot.ClaimCalls == 2 &&
			snapshot.NextWorkLookups == 1 && snapshot.NextWorkScheduled
	}, "future due timer while attempt remains active")
	close(release)
	waitFor(t, func() bool { return controller.Snapshot().Completed == 1 }, "attempt completion")
	time.Sleep(10 * time.Millisecond)
	if snapshot := controller.Snapshot(); snapshot.ClaimCalls != 2 || snapshot.NextWorkLookups != 1 ||
		!snapshot.NextWorkScheduled {
		t.Fatalf("completion bypassed future due timer: %+v", snapshot)
	}
	stopController(t, cancel, done)
}

func TestControllerTerminalNonEmptyBatchSchedulesFutureWork(t *testing.T) {
	started := make(chan struct{})
	source := &testSource{next: time.Now().Add(time.Hour)}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		return ClaimBatch{Attempts: []Attempt{{ID: "active", Run: func(ctx context.Context) error {
			close(started)
			<-ctx.Done()
			return nil
		}}}}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	defer cancel()
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("attempt did not start")
	}
	waitFor(t, func() bool { return controller.Snapshot().NextWorkScheduled }, "future due timer after terminal batch")
	if snapshot := controller.Snapshot(); snapshot.ClaimCalls != 1 || snapshot.NextWorkLookups != 1 ||
		snapshot.InFlight != 1 {
		t.Fatalf("terminal batch scheduling=%+v", snapshot)
	}
	stopController(t, cancel, done)
}

func TestControllerTerminalNonEmptyBatchSuppressesOverdueNextWork(t *testing.T) {
	started := make(chan struct{})
	var claimCalls atomic.Int64
	source := &testSource{next: time.Now().Add(-time.Hour)}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		if claimCalls.Add(1) != 1 {
			return ClaimBatch{}, errors.New("overdue timer bypassed terminal claim authority")
		}
		return ClaimBatch{Attempts: []Attempt{{ID: "active", Run: func(ctx context.Context) error {
			close(started)
			<-ctx.Done()
			return nil
		}}}}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("terminal attempt did not start")
	}
	waitFor(t, func() bool { return controller.Snapshot().OverdueNextWorkSuppressions == 1 }, "overdue terminal-batch suppression")
	time.Sleep(25 * time.Millisecond)
	if snapshot := controller.Snapshot(); snapshot.ClaimCalls != 1 || snapshot.NextWorkLookups != 1 ||
		snapshot.NextWorkScheduled || snapshot.Unusable {
		t.Fatalf("overdue next-work bypassed terminal non-empty claim: %+v", snapshot)
	}
	stopController(t, cancel, done)
}

func TestControllerFatalAndTimedOutShutdownClearPendingDueObservation(t *testing.T) {
	fatalFailure := errors.New("disposition failed")
	for _, test := range []struct {
		name      string
		attempt   func(context.Context) error
		wantError error
	}{
		{
			name:      "fatal attempt",
			wantError: fatalFailure,
			attempt: func(context.Context) error {
				return fatalFailure
			},
		},
		{
			name:      "shutdown timeout",
			wantError: ErrShutdownTimeout,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			started := make(chan struct{})
			release := make(chan struct{})
			var claimCalls atomic.Int64
			source := &testSource{next: time.Now().Add(time.Hour)}
			source.claim = func(context.Context, int) (ClaimBatch, error) {
				if claimCalls.Add(1) == 1 {
					return ClaimBatch{Attempts: []Attempt{{ID: "active", Run: func(ctx context.Context) error {
						close(started)
						<-release
						if test.attempt != nil {
							return test.attempt(ctx)
						}
						return nil
					}}}, More: true}, nil
				}
				return ClaimBatch{}, nil
			}
			config := testConfig()
			config.ShutdownTimeout = 20 * time.Millisecond
			controller, err := NewController(source, nil, config)
			if err != nil {
				t.Fatal(err)
			}
			cancel, done := runController(t, controller)
			select {
			case <-started:
			case <-time.After(2 * time.Second):
				t.Fatal("attempt did not start")
			}
			waitFor(t, func() bool { return controller.Snapshot().NextWorkScheduled }, "pending due timer")
			if test.attempt != nil {
				close(release)
			} else {
				cancel()
			}
			runErr := <-done
			if !errors.Is(runErr, test.wantError) {
				t.Fatalf("shutdown result=%v, want %v", runErr, test.wantError)
			}
			if test.attempt != nil && controller.Snapshot().AttemptFailures != 1 {
				t.Fatalf("fatal attempt accounting=%+v", controller.Snapshot())
			}
			if snapshot := controller.Snapshot(); snapshot.NextWorkScheduled || snapshot.NextWorkUnixNano != 0 {
				t.Fatalf("stopped controller retained due observation: %+v", snapshot)
			}
			if test.attempt == nil {
				close(release)
			}
		})
	}
}

func TestControllerIdleDomainCallBudget(t *testing.T) {
	controller, err := NewController(&testSource{}, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool {
		snapshot := controller.Snapshot()
		return controller.Ready() && snapshot.ClaimCalls == 1 && snapshot.NextWorkLookups == 1
	}, "initial empty claim and next-work lookup")

	baseline := controller.Snapshot()
	time.Sleep(25 * time.Millisecond)
	if got := controller.Snapshot(); got.Reconciliations != baseline.Reconciliations ||
		got.ClaimCalls != baseline.ClaimCalls || got.NextWorkLookups != baseline.NextWorkLookups {
		t.Fatalf("idle controller polled domain state: before=%+v after=%+v", baseline, got)
	}

	controller.RequestReconcile()
	waitFor(t, func() bool {
		snapshot := controller.Snapshot()
		return snapshot.Reconciliations == baseline.Reconciliations+1 &&
			snapshot.ClaimCalls == baseline.ClaimCalls+1 &&
			snapshot.NextWorkLookups == baseline.NextWorkLookups+1
	}, "one reconciled empty claim pass")
	stopController(t, cancel, done)
}

func TestControllerOverdueNextWorkAfterEmptyClaimDoesNotHotLoop(t *testing.T) {
	source := &testSource{next: time.Now().Add(-time.Hour)}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, controller.Ready, "controller readiness")
	time.Sleep(25 * time.Millisecond)
	if snapshot := controller.Snapshot(); snapshot.ClaimCalls != 1 ||
		snapshot.NextWorkLookups != 1 || snapshot.OverdueNextWorkSuppressions != 1 || snapshot.NextWorkScheduled {
		t.Fatalf("overdue next-work signal caused an empty-claim hot loop: %+v", snapshot)
	}
	stopController(t, cancel, done)
}

func TestControllerOverdueNextWorkRecoversOnPeriodicReconciliation(t *testing.T) {
	var completed atomic.Bool
	source := &testSource{next: time.Now().Add(-time.Hour)}
	source.onReconcile = func(source *testSource) {
		if source.reconciles == 3 {
			source.next = time.Time{}
			source.queue = append(source.queue, Attempt{ID: "recovered", Run: func(context.Context) error {
				completed.Store(true)
				return nil
			}})
		}
	}
	config := testConfig()
	config.ReconcileInterval = 10 * time.Millisecond
	controller, err := NewController(source, nil, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, completed.Load, "periodic recovery of overdue next-work")
	stopController(t, cancel, done)
	snapshot := controller.Snapshot()
	if snapshot.ClaimCalls > snapshot.Reconciliations || snapshot.NextWorkLookups > snapshot.Reconciliations ||
		snapshot.OverdueNextWorkSuppressions == 0 || snapshot.OverdueNextWorkSuppressions >= snapshot.NextWorkLookups ||
		snapshot.Completed != 1 || snapshot.NextWorkScheduled {
		t.Fatalf("overdue next-work bypassed bounded reconciliation cadence: %+v", snapshot)
	}
}

func TestControllerUnchangedGenerationDoesNotReadDomainQueue(t *testing.T) {
	probe := &testGenerationProbe{}
	probe.generation.Store(7)
	config := testConfig()
	config.GenerationInterval = 2 * time.Millisecond
	controller, err := NewController(&testSource{}, probe, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool { return controller.Snapshot().GenerationProbes >= 11 }, "ten unchanged generation probes")
	snapshot := controller.Snapshot()
	if snapshot.GenerationChanges != 0 || snapshot.Reconciliations != 1 ||
		snapshot.ClaimCalls != 1 || snapshot.NextWorkLookups != 1 {
		t.Fatalf("unchanged generation caused domain queue reads: %+v", snapshot)
	}
	stopController(t, cancel, done)
}

func TestControllerWakeStormCoalescesDomainReads(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	var calls atomic.Uint64
	source := &testSource{}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		if calls.Add(1) == 1 {
			close(started)
			<-release
		}
		return ClaimBatch{}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("initial claim did not block")
	}

	const wakes = 100_000
	for range wakes {
		controller.Wake()
	}
	close(release)
	waitFor(t, func() bool { return controller.Snapshot().ClaimCalls == 2 }, "one coalesced wake claim")
	time.Sleep(25 * time.Millisecond)
	snapshot := controller.Snapshot()
	if snapshot.WakeRequests != wakes || snapshot.WakeCoalesced != wakes-1 ||
		snapshot.ClaimCalls != 2 || snapshot.NextWorkLookups != 2 {
		t.Fatalf("wake storm was not bounded: %+v", snapshot)
	}
	stopController(t, cancel, done)
}

func TestControllerSignalsAreConcurrentSafeAndBounded(t *testing.T) {
	controller, err := NewController(&testSource{}, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	const goroutines = 32
	const requestsPerGoroutine = 1_000
	var group sync.WaitGroup
	group.Add(goroutines)
	for range goroutines {
		go func() {
			defer group.Done()
			for range requestsPerGoroutine {
				controller.Wake()
				controller.RequestReconcile()
			}
		}()
	}
	group.Wait()
	want := uint64(goroutines * requestsPerGoroutine)
	snapshot := controller.Snapshot()
	if snapshot.WakeRequests != want || snapshot.WakeCoalesced != want-1 ||
		snapshot.ReconcileRequests != want || snapshot.ReconcileCoalesced != want-1 {
		t.Fatalf("concurrent signals were not bounded: %+v", snapshot)
	}
}

func TestControllerRejectsClaimBeyondFreeCapacity(t *testing.T) {
	source := &testSource{}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		return ClaimBatch{Attempts: []Attempt{
			{ID: "1", Run: func(context.Context) error { return nil }},
			{ID: "2", Run: func(context.Context) error { return nil }},
			{ID: "3", Run: func(context.Context) error { return nil }},
		}}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	if runErr := controller.Run(context.Background()); runErr == nil || !strings.Contains(runErr.Error(), "capacity") {
		t.Fatalf("oversized claim error=%v", runErr)
	}
	if snapshot := controller.Snapshot(); snapshot.Claimed != 0 || snapshot.InFlight != 0 {
		t.Fatalf("oversized batch was partially accepted: %+v", snapshot)
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable {
		t.Fatalf("oversized durable claim result left controller reusable: %+v", snapshot)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
		t.Fatalf("oversized durable claim result allowed reuse: %v", runErr)
	}
}