package runtimeworker

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

	"github.com/nexuscrm/persistence"
)

type blockingParallelMetadataProcessor struct {
	started   chan struct{}
	release   chan struct{}
	completed chan struct{}
	once      sync.Once
	calls     atomic.Int64
	done      atomic.Int64
}

type idleParallelMetadataProcessor struct {
	observes atomic.Int64
	hints    atomic.Int64
	calls    atomic.Int64
	five     chan struct{}
	once     sync.Once
}

func (*idleParallelMetadataProcessor) CheckReady(context.Context) error { return nil }
func (p *idleParallelMetadataProcessor) Observe(context.Context) (metadataWorkState, error) {
	p.observes.Add(1)
	return metadataWorkState{}, nil
}
func (p *idleParallelMetadataProcessor) HasWork(context.Context) (bool, error) {
	if p.hints.Add(1) == 5 {
		p.once.Do(func() { close(p.five) })
	}
	return false, nil
}
func (*idleParallelMetadataProcessor) Maintain(context.Context, int) error { return nil }
func (p *idleParallelMetadataProcessor) ProcessOne(context.Context, int) (int, bool, error) {
	p.calls.Add(1)
	return 0, false, nil
}

func TestMetadataWorkerParallelIdlePollingUsesOnlyNegativeHints(t *testing.T) {
	processor := &idleParallelMetadataProcessor{five: make(chan struct{})}
	worker, err := newMetadataWorker(processor, 10, 5*time.Millisecond, 0, 0, 15)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- worker.Run(ctx) }()
	select {
	case <-processor.five:
	case <-time.After(time.Second):
		cancel()
		t.Fatal("idle work hint did not continue polling")
	}
	cancel()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if calls := processor.calls.Load(); calls != 0 {
		t.Fatalf("idle ProcessOne calls = %d, want zero fenced claims", calls)
	}
	if hints := processor.hints.Load(); hints < 5 || hints >= 15 {
		t.Fatalf("idle HasWork calls = %d, want one coordinator hint stream independent of 15 lanes", hints)
	}
	if observes := processor.observes.Load(); observes != 0 {
		t.Fatalf("idle Observe calls = %d, want no heavyweight sample before the hourly interval", observes)
	}
}

func TestMetadataWorkerObservationUsesFixedHourlyL0Cadence(t *testing.T) {
	if metadataObservationInterval != time.Hour {
		t.Fatalf("metadata observation interval = %s, want fixed L0 interval %s", metadataObservationInterval, time.Hour)
	}
	processor := &idleParallelMetadataProcessor{five: make(chan struct{})}
	worker, err := newMetadataWorker(processor, 10, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	stop := worker.startObservationLoop(t.Context())
	stop()
	if observes := processor.observes.Load(); observes != 0 {
		t.Fatalf("starting observation loop executed %d samples, want zero", observes)
	}
}

type singleItemParallelMetadataProcessor struct {
	remaining atomic.Int64
	calls     atomic.Int64
	empty     chan struct{}
	once      sync.Once
}

func (*singleItemParallelMetadataProcessor) CheckReady(context.Context) error { return nil }
func (p *singleItemParallelMetadataProcessor) Observe(context.Context) (metadataWorkState, error) {
	return metadataWorkState{queueDepth: p.remaining.Load()}, nil
}
func (*singleItemParallelMetadataProcessor) Maintain(context.Context, int) error { return nil }
func (p *singleItemParallelMetadataProcessor) ProcessOne(context.Context, int) (int, bool, error) {
	p.calls.Add(1)
	if p.remaining.CompareAndSwap(1, 0) {
		return 1, true, nil
	}
	p.once.Do(func() { close(p.empty) })
	return 0, false, nil
}

func TestMetadataWorkerSingleItemDoesNotBroadcastToEveryLane(t *testing.T) {
	processor := &singleItemParallelMetadataProcessor{empty: make(chan struct{})}
	processor.remaining.Store(1)
	worker, err := newMetadataWorker(processor, 10, time.Hour, 0, 0, 15)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- worker.Run(ctx) }()
	select {
	case <-processor.empty:
	case <-time.After(time.Second):
		cancel()
		t.Fatal("single-item successor did not finish its bounded drain")
	}
	cancel()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if calls := processor.calls.Load(); calls > 3 {
		t.Fatalf("single-item ProcessOne calls = %d, want at most the useful claim plus two bounded empty drains", calls)
	}
}

func (*blockingParallelMetadataProcessor) CheckReady(context.Context) error { return nil }
func (*blockingParallelMetadataProcessor) Observe(context.Context) (metadataWorkState, error) {
	return metadataWorkState{queueDepth: 1}, nil
}
func (*blockingParallelMetadataProcessor) Maintain(context.Context, int) error { return nil }
func (p *blockingParallelMetadataProcessor) ProcessOne(ctx context.Context, _ int) (int, bool, error) {
	if p.calls.Add(1) == 1 {
		p.once.Do(func() { close(p.started) })
		select {
		case <-p.release:
			return 1, true, nil
		case <-ctx.Done():
			return 0, false, ctx.Err()
		}
	}
	if p.done.Add(1) == 3 {
		close(p.completed)
	}
	return 1, true, nil
}

func TestMetadataWorkerParallelLanesHaveNoBlockedBatchBarrier(t *testing.T) {
	processor := &blockingParallelMetadataProcessor{
		started: make(chan struct{}), release: make(chan struct{}), completed: make(chan struct{}),
	}
	worker, err := newMetadataWorker(processor, 1, time.Millisecond, time.Millisecond, 0, 2)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- worker.Run(ctx) }()
	select {
	case <-processor.started:
	case <-time.After(time.Second):
		t.Fatal("blocking lane did not start")
	}
	select {
	case <-processor.completed:
	case <-time.After(time.Second):
		t.Fatal("free lane did not continue across polls while peer remained blocked")
	}
	cancel()
	close(processor.release)
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(time.Second):
		t.Fatal("parallel worker did not join canceled lanes")
	}
}

type coordinatedBatchMetadataProcessor struct {
	mu          sync.Mutex
	pending     []metadataClaimedWork
	claimLimits []int
	hintErr     error
}

func (*coordinatedBatchMetadataProcessor) CheckReady(context.Context) error { return nil }
func (p *coordinatedBatchMetadataProcessor) Observe(context.Context) (metadataWorkState, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return metadataWorkState{queueDepth: int64(len(p.pending))}, nil
}
func (p *coordinatedBatchMetadataProcessor) HasWork(context.Context) (bool, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return len(p.pending) > 0, p.hintErr
}
func (*coordinatedBatchMetadataProcessor) Maintain(context.Context, int) error { return nil }
func (*coordinatedBatchMetadataProcessor) ProcessOne(context.Context, int) (int, bool, error) {
	return 0, false, errors.New("coordinated batch processor must not use ProcessOne")
}
func (p *coordinatedBatchMetadataProcessor) ClaimBatch(_ context.Context, limit int) ([]metadataClaimedWork, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.claimLimits = append(p.claimLimits, limit)
	if limit > len(p.pending) {
		limit = len(p.pending)
	}
	claimed := append([]metadataClaimedWork(nil), p.pending[:limit]...)
	p.pending = p.pending[limit:]
	return claimed, nil
}

func (p *coordinatedBatchMetadataProcessor) limits() []int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return append([]int(nil), p.claimLimits...)
}

func TestMetadataWorkerCoordinatedAdmissionRefillsOnlyFreeSlotsWithoutJoinBarrier(t *testing.T) {
	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	thirdCompleted := make(chan struct{})
	processor := &coordinatedBatchMetadataProcessor{pending: []metadataClaimedWork{
		{run: func(ctx context.Context) error {
			close(firstStarted)
			select {
			case <-releaseFirst:
				return nil
			case <-ctx.Done():
				return ctx.Err()
			}
		}},
		{run: func(context.Context) error { return nil }},
		{run: func(context.Context) error { close(thirdCompleted); return nil }},
	}}
	worker, err := newMetadataWorker(processor, 10, time.Hour, 0, 0, 2)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- worker.Run(ctx) }()
	select {
	case <-firstStarted:
	case <-time.After(time.Second):
		cancel()
		t.Fatal("first coordinated effect did not start")
	}
	select {
	case <-thirdCompleted:
	case <-time.After(time.Second):
		cancel()
		t.Fatal("completion did not refill the free slot while its peer remained blocked")
	}
	close(releaseFirst)
	deadline := time.Now().Add(time.Second)
	for worker.snapshot().Processed != 3 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	cancel()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	limits := processor.limits()
	if len(limits) != 2 || limits[0] != 2 || limits[1] != 1 {
		t.Fatalf("coordinated claim limits = %v, want exactly current free capacity [2 1]", limits)
	}
	snapshot := worker.snapshot()
	if snapshot.Attempts != 3 || snapshot.Processed != 3 || snapshot.InFlight != 0 ||
		snapshot.ContinuationAttempts != 0 || snapshot.ContinuationEmptyAttempts != 0 ||
		snapshot.SuccessorFirstEmptyAttempts != 0 {
		t.Fatalf("coordinated worker snapshot = %+v", snapshot)
	}
}

func TestMetadataWorkerCoordinatedShutdownWaitsForInFlightEffect(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	processor := &coordinatedBatchMetadataProcessor{pending: []metadataClaimedWork{{run: func(ctx context.Context) error {
		close(started)
		<-ctx.Done()
		<-release
		return ctx.Err()
	}}}}
	worker, err := newMetadataWorker(processor, 10, time.Hour, 0, 0, 2)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- worker.Run(ctx) }()
	<-started
	cancel()
	select {
	case err := <-done:
		t.Fatalf("worker returned before its in-flight durable effect: %v", err)
	case <-time.After(20 * time.Millisecond):
	}
	close(release)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if snapshot := worker.snapshot(); snapshot.InFlight != 0 || snapshot.Failures != 0 {
		t.Fatalf("coordinated shutdown snapshot = %+v, want zero in-flight and no cancellation failure", snapshot)
	}
}

func TestMetadataWorkerCoordinatedOwnershipLossStopsAdmission(t *testing.T) {
	processor := &coordinatedBatchMetadataProcessor{pending: []metadataClaimedWork{{run: func(context.Context) error {
		return persistence.ErrRuntimeWorkOwnershipLost
	}}}}
	worker, err := newMetadataWorker(processor, 10, time.Hour, 0, 0, 2)
	if err != nil {
		t.Fatal(err)
	}
	err = worker.Run(t.Context())
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("coordinated ownership error = %v", err)
	}
	if limits := processor.limits(); len(limits) != 1 || limits[0] != 2 {
		t.Fatalf("claim limits after ownership loss = %v, want one admission", limits)
	}
	if snapshot := worker.snapshot(); snapshot.Attempts != 1 || snapshot.Failures != 1 || snapshot.InFlight != 0 {
		t.Fatalf("snapshot after ownership loss = %+v", snapshot)
	}
}

type metadataProcessorStep struct {
	processed int
	found     bool
	err       error
}

type fakeMetadataProcessor struct {
	readyErr      error
	maintainErr   error
	observeErr    error
	state         metadataWorkState
	steps         []metadataProcessorStep
	calls         int
	maintainCalls int
}

type failingHintMetadataProcessor struct {
	fakeMetadataProcessor
	hintErr error
}

func (p *failingHintMetadataProcessor) HasWork(context.Context) (bool, error) {
	return false, p.hintErr
}

func (p *fakeMetadataProcessor) CheckReady(context.Context) error { return p.readyErr }
func (p *fakeMetadataProcessor) Observe(context.Context) (metadataWorkState, error) {
	return p.state, p.observeErr
}
func (p *fakeMetadataProcessor) Maintain(context.Context, int) error {
	p.maintainCalls++
	return p.maintainErr
}
func (p *fakeMetadataProcessor) ProcessOne(context.Context, int) (int, bool, error) {
	if p.calls >= len(p.steps) {
		return 0, false, nil
	}
	step := p.steps[p.calls]
	p.calls++
	return step.processed, step.found, step.err
}

type blockingObservationMetadataProcessor struct{}

func (*blockingObservationMetadataProcessor) CheckReady(context.Context) error { return nil }
func (*blockingObservationMetadataProcessor) Observe(ctx context.Context) (metadataWorkState, error) {
	<-ctx.Done()
	return metadataWorkState{}, ctx.Err()
}
func (*blockingObservationMetadataProcessor) HasWork(context.Context) (bool, error) {
	return false, nil
}
func (*blockingObservationMetadataProcessor) Maintain(context.Context, int) error { return nil }
func (*blockingObservationMetadataProcessor) ProcessOne(context.Context, int) (int, bool, error) {
	return 0, false, nil
}

type isolatedObservationMetadataProcessor struct {
	observationStarted chan struct{}
	observationClosed  chan struct{}
}

func (*isolatedObservationMetadataProcessor) CheckReady(context.Context) error { return nil }
func (p *isolatedObservationMetadataProcessor) Observe(ctx context.Context) (metadataWorkState, error) {
	close(p.observationStarted)
	<-ctx.Done()
	return metadataWorkState{}, ctx.Err()
}
func (*isolatedObservationMetadataProcessor) HasWork(context.Context) (bool, error) {
	return true, nil
}
func (*isolatedObservationMetadataProcessor) Maintain(context.Context, int) error { return nil }
func (*isolatedObservationMetadataProcessor) ProcessOne(context.Context, int) (int, bool, error) {
	return 1, false, nil
}
func (p *isolatedObservationMetadataProcessor) closeObservation() error {
	close(p.observationClosed)
	return nil
}

func TestMetadataWorkerObservationCachesSuccessAndMarksFailureStale(t *testing.T) {
	processor := &failingHintMetadataProcessor{fakeMetadataProcessor: fakeMetadataProcessor{
		state: metadataWorkState{queueDepth: 9, poison: 2, oldestAge: 3 * time.Second},
	}}
	worker, err := newMetadataWorker(processor, 10, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	initial := worker.snapshot()
	if !initial.ObservationEnabled || !initial.ObservationStale || !initial.ObservationLastSuccess.IsZero() {
		t.Fatalf("initial observation snapshot = %+v, want enabled stale cache without a sample", initial)
	}

	worker.refreshObservationWithTimeout(t.Context(), time.Second)
	success := worker.snapshot()
	if success.QueueDepth != 9 || success.PoisonDepth != 2 || success.OldestAge != 3*time.Second ||
		success.ObservationStale || success.ObservationLastSuccess.IsZero() || success.ObservationFailures != 0 ||
		success.OldestObservedAt.IsZero() {
		t.Fatalf("successful observation snapshot = %+v", success)
	}
	if age := time.Since(success.OldestObservedAt); age < 2*time.Second || age > 4*time.Second {
		t.Fatalf("oldest observed timestamp age = %s, want approximately 3s", age)
	}

	processor.observeErr = errors.New("observation unavailable")
	processor.state = metadataWorkState{queueDepth: 99, poison: 22, oldestAge: 30 * time.Second}
	worker.refreshObservationWithTimeout(t.Context(), time.Second)
	failed := worker.snapshot()
	if failed.QueueDepth != success.QueueDepth || failed.PoisonDepth != success.PoisonDepth ||
		failed.OldestAge != success.OldestAge || !failed.ObservationStale ||
		failed.ObservationFailures != 1 || !failed.ObservationLastSuccess.Equal(success.ObservationLastSuccess) ||
		!failed.OldestObservedAt.Equal(success.OldestObservedAt) {
		t.Fatalf("failed observation snapshot = %+v, want retained success %+v", failed, success)
	}
}

func TestMetadataWorkerObservationTimeoutFailsOpen(t *testing.T) {
	worker, err := newMetadataWorker(&blockingObservationMetadataProcessor{}, 10, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	worker.refreshObservationWithTimeout(t.Context(), 5*time.Millisecond)
	snapshot := worker.snapshot()
	if !snapshot.ObservationStale || snapshot.ObservationFailures != 1 ||
		snapshot.ObservationDuration < 5*time.Millisecond || !snapshot.ObservationLastSuccess.IsZero() ||
		snapshot.Attempts != 0 || snapshot.Processed != 0 {
		t.Fatalf("timed-out observation snapshot = %+v", snapshot)
	}
}

func TestMetadataWorkerObservationDoesNotBlockProcessing(t *testing.T) {
	processor := &isolatedObservationMetadataProcessor{
		observationStarted: make(chan struct{}), observationClosed: make(chan struct{}),
	}
	worker, err := newMetadataWorker(processor, 10, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	stop := worker.startObservationLoopWithTiming(t.Context(), time.Millisecond, time.Second)
	t.Cleanup(stop)
	select {
	case <-processor.observationStarted:
	case <-time.After(time.Second):
		t.Fatal("observation did not start")
	}

	processed := make(chan error, 1)
	go func() { processed <- worker.runPass(t.Context()) }()
	select {
	case err := <-processed:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(100 * time.Millisecond):
		t.Fatal("blocked observation also blocked processing")
	}
	stop()
	select {
	case <-processor.observationClosed:
	default:
		t.Fatal("observation lifecycle was not closed before the loop stopped")
	}
}

func TestMetadataWorkerAttributesEmptyAttemptsToWakeAndContinuation(t *testing.T) {
	pollProcessor := &fakeMetadataProcessor{steps: []metadataProcessorStep{
		{processed: 1, found: true},
		{processed: 0, found: false},
	}}
	pollWorker, err := newMetadataWorker(pollProcessor, 10, time.Second, 0, 0, 2)
	if err != nil {
		t.Fatal(err)
	}
	if err := pollWorker.runDrainPassFrom(t.Context(), nil, metadataWakeSourcePoll); err != nil {
		t.Fatal(err)
	}
	pollSnapshot := pollWorker.snapshot()
	if pollSnapshot.ContinuationAttempts != 1 || pollSnapshot.ContinuationEmptyAttempts != 1 ||
		pollSnapshot.PollFirstEmptyAttempts != 0 || pollSnapshot.SuccessorFirstEmptyAttempts != 0 {
		t.Fatalf("poll drain attribution = %+v", pollSnapshot)
	}

	successorProcessor := &fakeMetadataProcessor{steps: []metadataProcessorStep{{processed: 0, found: false}}}
	successorWorker, err := newMetadataWorker(successorProcessor, 10, time.Second, 0, 0, 2)
	if err != nil {
		t.Fatal(err)
	}
	if err := successorWorker.runDrainPassFrom(t.Context(), nil, metadataWakeSourceSuccessor); err != nil {
		t.Fatal(err)
	}
	successorSnapshot := successorWorker.snapshot()
	if successorSnapshot.SuccessorFirstEmptyAttempts != 1 || successorSnapshot.PollFirstEmptyAttempts != 0 ||
		successorSnapshot.ContinuationAttempts != 0 || successorSnapshot.ContinuationEmptyAttempts != 0 {
		t.Fatalf("successor drain attribution = %+v", successorSnapshot)
	}
}

func TestMetadataWorkerBoundsPassAndReportsDurableState(t *testing.T) {
	processor := &fakeMetadataProcessor{
		state: metadataWorkState{queueDepth: 9, poison: 2, oldestAge: 3 * time.Second},
		steps: []metadataProcessorStep{{processed: 1, found: true}, {processed: 2, found: true}},
	}
	worker, err := newMetadataWorker(processor, 3, time.Second, 0, time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.Prepare(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := worker.runMaintenance(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(context.Background()); err != nil {
		t.Fatal(err)
	}
	snapshot := worker.snapshot()
	if processor.maintainCalls != 1 || processor.calls != 2 || snapshot.Processed != 3 || snapshot.Attempts != 2 || snapshot.QueueDepth != 9 ||
		snapshot.PoisonDepth != 2 || snapshot.OldestAge != 3*time.Second || snapshot.InFlight != 0 {
		t.Fatalf("processor calls=%d snapshot=%+v", processor.calls, snapshot)
	}
}

func TestMetadataWorkerStopsPassWhenMaintenanceLosesOwnership(t *testing.T) {
	processor := &fakeMetadataProcessor{
		maintainErr: persistence.ErrRuntimeWorkOwnershipLost,
		steps:       []metadataProcessorStep{{processed: 1, found: true}},
	}
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runMaintenance(context.Background()); !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("runMaintenance error=%v", err)
	}
	if processor.calls != 0 || processor.maintainCalls != 1 || worker.snapshot().Failures != 1 {
		t.Fatalf("processor calls=%d maintenance=%d snapshot=%+v", processor.calls, processor.maintainCalls, worker.snapshot())
	}
}

func TestMetadataWorkerPropagatesOwnershipLoss(t *testing.T) {
	processor := &fakeMetadataProcessor{
		state: metadataWorkState{queueDepth: 1},
		steps: []metadataProcessorStep{{found: true, err: persistence.ErrRuntimeWorkOwnershipLost}},
	}
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(context.Background()); !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("runPass error=%v", err)
	}
	if snapshot := worker.snapshot(); snapshot.Failures != 1 || snapshot.Attempts != 1 {
		t.Fatalf("snapshot=%+v", snapshot)
	}
}

func TestMetadataWorkerSuccessfulZeroHintSkipsFencedProcessor(t *testing.T) {
	processor := &fakeMetadataProcessor{steps: []metadataProcessorStep{{processed: 1, found: true}}}
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	if processor.calls != 0 || worker.snapshot().Attempts != 0 {
		t.Fatalf("processor calls=%d snapshot=%+v, want no fenced attempt", processor.calls, worker.snapshot())
	}
	if loop := worker.snapshot().WorkLoop; loop.PollChecks != 1 || loop.ConfirmedIdle != 1 || loop.ClaimAttempts != 0 {
		t.Fatalf("work loop after idle hint=%+v", loop)
	}
	// Work arriving immediately after the negative hint is deliberately not
	// lost: the next bounded poll sees a positive hint and enters the unchanged
	// authoritative processor.
	processor.state = metadataWorkState{queueDepth: 1}
	if err := worker.runPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	if processor.calls != 1 || worker.snapshot().Attempts != 1 || worker.snapshot().Processed != 1 {
		t.Fatalf("processor calls=%d snapshot=%+v, want next-poll fenced processing", processor.calls, worker.snapshot())
	}
	if loop := worker.snapshot().WorkLoop; loop.PollChecks != 2 || loop.ConfirmedIdle != 1 || loop.ClaimAttempts != 1 || loop.Claimed != 1 {
		t.Fatalf("work loop after next-poll processing=%+v", loop)
	}
}

func TestMetadataWorkerObservationFailureFallsThroughToFencedProcessor(t *testing.T) {
	processor := &fakeMetadataProcessor{
		observeErr: errors.New("observation unavailable"),
		steps:      []metadataProcessorStep{{processed: 1, found: true}},
	}
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	if processor.calls != 1 || worker.snapshot().Attempts != 1 || worker.snapshot().Processed != 1 {
		t.Fatalf("processor calls=%d snapshot=%+v, want authoritative fallback", processor.calls, worker.snapshot())
	}
	if loop := worker.snapshot().WorkLoop; loop.PollChecks != 1 || loop.HintFailures != 1 || loop.ClaimAttempts != 1 || loop.Claimed != 1 {
		t.Fatalf("work loop after hint failure=%+v", loop)
	}
}

func TestMetadataWorkerHintFailureFallsThroughToFencedProcessor(t *testing.T) {
	processor := &failingHintMetadataProcessor{
		fakeMetadataProcessor: fakeMetadataProcessor{
			steps: []metadataProcessorStep{{processed: 1, found: true}},
		},
		hintErr: errors.New("hint unavailable"),
	}
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	if processor.calls != 1 || worker.snapshot().Attempts != 1 || worker.snapshot().Processed != 1 {
		t.Fatalf("processor calls=%d snapshot=%+v, want authoritative fallback", processor.calls, worker.snapshot())
	}
}