package connector_inbound

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

var _ WorkItemRowCASClaimAuthority = (*connector_workqueue.Repository)(nil)

type workItemClaimServiceWithoutRowCASCapability struct {
	WorkItemClaimService
}

type fakeWorkItemClaims struct {
	work  *connector_workqueue.WorkItem
	works []*connector_workqueue.WorkItem

	acquireCalls           int
	renewCalls             int
	claimCalls             int
	completeCalls          []string
	completeWorkIDs        []string
	retryCalls             int
	retryRunAfters         []time.Time
	retryWorkIDs           []string
	retryWorkTokens        []string
	renewWorkCalls         int
	claimByEventCalls      int
	staleClaimByEventCalls int
	claimByEventTypes      []string
	peekCalls              int
	lastClaimInput         connector_workqueue.ClaimWorkItemInput
	lastClaimWorkTypes     []string
	lastClaimByEvent       connector_workqueue.ClaimWorkItemByEventInput
	lastResolveInput       connector_workqueue.ResolveClaimableGenerationInput
	lastResolveWorkTypes   []string
	lastRenewInput         connector_workqueue.SchedulerLeaseInput
	lastDistributed        bool
	completeContexts       []context.Context
	completeCtxErrs        []error
	retryContexts          []context.Context
	retryCtxErrs           []error
	failCtxErrs            []error
	renewWorkIDs           []string
	renewWorkTokens        []string
	renewWorkLeaseUntils   []time.Time
	workReturned           bool
	acquireResult          *bool
	renewResult            *bool
	resolveGeneration      *int64
	resolveCalls           int
	lastPreference         []string
	legacyLeaseAuthority   bool
	executorStateAuthority bool
	failInputs             []connector_workqueue.FailWorkItemInput
}

func (f *fakeWorkItemClaims) UsesWorkItemRowCASClaimAuthority() bool {
	return !f.legacyLeaseAuthority
}

func (f *fakeWorkItemClaims) UsesWorkItemExecutorStateAuthority() bool {
	return f.executorStateAuthority
}

func (f *fakeWorkItemClaims) AcquireSchedulerLease(context.Context, connector_workqueue.SchedulerLeaseInput) (bool, error) {
	f.acquireCalls++
	if f.acquireResult != nil {
		return *f.acquireResult, nil
	}
	return true, nil
}

func (f *fakeWorkItemClaims) RenewSchedulerLease(_ context.Context, in connector_workqueue.SchedulerLeaseInput) (bool, error) {
	f.renewCalls++
	f.lastRenewInput = in
	if f.renewResult != nil {
		return *f.renewResult, nil
	}
	return false, nil
}

func (f *fakeWorkItemClaims) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	f.claimCalls++
	f.lastClaimInput = in
	f.lastClaimWorkTypes = connector_workqueue.ClaimWorkTypes(ctx)
	f.lastPreference = connector_workqueue.ClaimQueueClassPreference(ctx)
	f.lastDistributed = connector_workqueue.DistributedClaim(ctx)
	if len(f.works) > 0 {
		work := f.works[0]
		f.works = f.works[1:]
		if work == nil {
			return nil, false, nil
		}
		return work, true, nil
	}
	if f.work == nil {
		return nil, false, nil
	}
	if f.workReturned {
		return nil, false, nil
	}
	f.workReturned = true
	return f.work, true, nil
}

func (f *fakeWorkItemClaims) PeekClaimableWorkItems(ctx context.Context, in connector_workqueue.ClaimWorkItemInput, _ int) ([]*connector_workqueue.WorkItem, error) {
	f.peekCalls++
	f.lastClaimInput = in
	f.lastClaimWorkTypes = connector_workqueue.ClaimWorkTypes(ctx)
	f.lastPreference = connector_workqueue.ClaimQueueClassPreference(ctx)
	f.lastDistributed = connector_workqueue.DistributedClaim(ctx)
	return append([]*connector_workqueue.WorkItem(nil), f.works...), nil
}

func (f *fakeWorkItemClaims) ResolveClaimableWorkItemGeneration(ctx context.Context, in connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error) {
	f.resolveCalls++
	f.lastResolveInput = in
	f.lastResolveWorkTypes = connector_workqueue.ClaimWorkTypes(ctx)
	if f.resolveGeneration == nil {
		return in.Generation, false, nil
	}
	return *f.resolveGeneration, true, nil
}

func (f *fakeWorkItemClaims) ClaimWorkItemByEvent(_ context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.claimWorkItemByEvent(in, false)
}

func (f *fakeWorkItemClaims) ClaimStaleWorkItemByEvent(_ context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.claimWorkItemByEvent(in, true)
}

func (f *fakeWorkItemClaims) claimWorkItemByEvent(in connector_workqueue.ClaimWorkItemByEventInput, staleOnly bool) (*connector_workqueue.WorkItem, bool, error) {
	f.claimByEventCalls++
	if staleOnly {
		f.staleClaimByEventCalls++
	}
	f.lastClaimByEvent = in
	f.claimByEventTypes = append(f.claimByEventTypes, in.WorkType)
	for i, work := range f.works {
		if work != nil && work.EventID == in.EventID && work.WorkType == in.WorkType && work.Generation == in.Generation {
			claimed := *work
			claimed.LeaseToken = in.LeaseToken
			claimed.LeaseOwner = in.OwnerID
			claimed.LeaseUntil = sql.NullTime{Time: in.LeaseUntil, Valid: true}
			f.works = append(f.works[:i], f.works[i+1:]...)
			return &claimed, true, nil
		}
	}
	if f.work != nil && f.work.EventID == in.EventID && f.work.WorkType == in.WorkType && f.work.Generation == in.Generation {
		claimed := *f.work
		claimed.LeaseToken = in.LeaseToken
		claimed.LeaseOwner = in.OwnerID
		claimed.LeaseUntil = sql.NullTime{Time: in.LeaseUntil, Valid: true}
		return &claimed, true, nil
	}
	return nil, false, nil
}

func (f *fakeWorkItemClaims) CompleteWorkItem(ctx context.Context, workID string, _ string, terminalState string, _ time.Time) (bool, error) {
	f.completeCalls = append(f.completeCalls, terminalState)
	f.completeWorkIDs = append(f.completeWorkIDs, workID)
	f.completeContexts = append(f.completeContexts, ctx)
	f.completeCtxErrs = append(f.completeCtxErrs, ctx.Err())
	return true, nil
}

func (f *fakeWorkItemClaims) RetryWorkItem(ctx context.Context, workID string, leaseToken string, runAfter, _ time.Time) (bool, error) {
	f.retryCalls++
	f.retryContexts = append(f.retryContexts, ctx)
	f.retryCtxErrs = append(f.retryCtxErrs, ctx.Err())
	f.retryRunAfters = append(f.retryRunAfters, runAfter)
	f.retryWorkIDs = append(f.retryWorkIDs, workID)
	f.retryWorkTokens = append(f.retryWorkTokens, leaseToken)
	return true, nil
}

func (f *fakeWorkItemClaims) RenewWorkItemLease(_ context.Context, workID, leaseToken string, leaseUntil, _ time.Time) (bool, error) {
	f.renewWorkCalls++
	f.renewWorkIDs = append(f.renewWorkIDs, workID)
	f.renewWorkTokens = append(f.renewWorkTokens, leaseToken)
	f.renewWorkLeaseUntils = append(f.renewWorkLeaseUntils, leaseUntil)
	return true, nil
}

func (f *fakeWorkItemClaims) FailWorkItem(ctx context.Context, in connector_workqueue.FailWorkItemInput) (*connector_workqueue.FailWorkItemResult, bool, error) {
	f.failInputs = append(f.failInputs, in)
	f.failCtxErrs = append(f.failCtxErrs, ctx.Err())
	return &connector_workqueue.FailWorkItemResult{
		State:        connector_workqueue.WorkStateRetry,
		FailureCount: 1,
		RunAfter:     in.Now.Add(in.RetryDelays[0]),
	}, true, nil
}

type fakeWorkItemEventLoader struct {
	event                     *connector_events.ConnectorEvent
	events                    map[string]*connector_events.ConnectorEvent
	settledProjectionPayloads map[string][]*connector_events.ConnectorEvent
	sequence                  []*connector_events.ConnectorEvent
	calls                     int
}

type fakeBatchWorkItemEventLoader struct {
	*fakeWorkItemEventLoader
	batchCalls int
}

type nilSuccessfulWorkItemEventLoader struct{}

func (nilSuccessfulWorkItemEventLoader) LoadEventForWorkItem(context.Context, string) (*connector_events.ConnectorEvent, error) {
	return nil, nil
}

func (nilSuccessfulWorkItemEventLoader) LoadSourceProjectionPayloads(context.Context, string) ([]*connector_events.ConnectorEvent, error) {
	return nil, nil
}

type fakeWorkItemClaimInputs struct {
	claims *fakeWorkItemClaims
	loader testWorkItemEventLoader
	honors bool
}

// testWorkItemEventLoader is the immutable-input portion of WorkItemService
// needed by the test fakes. Keeping this test-owned interface avoids coupling
// live WorkItem tests to the split compatibility interface retired in v0.2.
type testWorkItemEventLoader interface {
	LoadEventForWorkItem(context.Context, string) (*connector_events.ConnectorEvent, error)
	LoadSourceProjectionPayloads(context.Context, string) ([]*connector_events.ConnectorEvent, error)
}

type workItemClaimInputEvents struct {
	EventsService
	WorkItemClaimInputService
}

func newFakeWorkItemClaimInputs(claims *fakeWorkItemClaims, loader testWorkItemEventLoader) *fakeWorkItemClaimInputs {
	return &fakeWorkItemClaimInputs{claims: claims, loader: loader, honors: true}
}

func (f *fakeWorkItemClaimInputs) HonorsClaimWorkTypes() bool { return f != nil && f.honors }

func (f *fakeWorkItemClaimInputs) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.claims.ClaimNextWorkItem(ctx, in)
}

func (f *fakeWorkItemClaimInputs) PeekClaimableWorkItems(ctx context.Context, in connector_workqueue.ClaimWorkItemInput, limit int) ([]*connector_workqueue.WorkItem, error) {
	return f.claims.PeekClaimableWorkItems(ctx, in, limit)
}

func (f *fakeWorkItemClaimInputs) ClaimWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.claims.ClaimWorkItemByEvent(ctx, in)
}

func (f *fakeWorkItemClaimInputs) ClaimStaleWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return f.claims.ClaimStaleWorkItemByEvent(ctx, in)
}

func (f *fakeWorkItemClaimInputs) ResolveClaimableWorkItemGeneration(ctx context.Context, in connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error) {
	return f.claims.ResolveClaimableWorkItemGeneration(ctx, in)
}

func (f *fakeWorkItemClaimInputs) LoadEventForWorkItem(ctx context.Context, eventID string) (*connector_events.ConnectorEvent, error) {
	return f.loader.LoadEventForWorkItem(ctx, eventID)
}

func (f *fakeWorkItemClaimInputs) LoadEventsForWorkItems(ctx context.Context, eventIDs []string) (map[string]*connector_events.ConnectorEvent, error) {
	if loader, ok := f.loader.(workItemEventBatchLoader); ok {
		return loader.LoadEventsForWorkItems(ctx, eventIDs)
	}
	result := make(map[string]*connector_events.ConnectorEvent, len(eventIDs))
	for _, eventID := range eventIDs {
		event, err := f.loader.LoadEventForWorkItem(ctx, eventID)
		if err != nil {
			return nil, err
		}
		result[eventID] = event
	}
	return result, nil
}

func (f *fakeWorkItemClaimInputs) LoadSourceProjectionPayloads(ctx context.Context, eventID string) ([]*connector_events.ConnectorEvent, error) {
	return f.loader.LoadSourceProjectionPayloads(ctx, eventID)
}

type nilSuccessfulWorkItemClaimInputs struct {
	*fakeWorkItemClaimInputs
}

func (f *nilSuccessfulWorkItemClaimInputs) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	_, _, _ = f.fakeWorkItemClaimInputs.ClaimNextWorkItem(ctx, in)
	return nil, true, nil
}

func (*nilSuccessfulWorkItemClaimInputs) ClaimWorkItemByEvent(context.Context, connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return nil, true, nil
}

func (*nilSuccessfulWorkItemClaimInputs) ClaimStaleWorkItemByEvent(context.Context, connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error) {
	return nil, true, nil
}

func (f *fakeBatchWorkItemEventLoader) LoadEventsForWorkItems(_ context.Context, eventIDs []string) (map[string]*connector_events.ConnectorEvent, error) {
	f.batchCalls++
	result := make(map[string]*connector_events.ConnectorEvent, len(eventIDs))
	for _, eventID := range eventIDs {
		ev := f.events[eventID]
		if ev != nil {
			copy := *ev
			result[eventID] = &copy
		}
	}
	return result, nil
}

func (f *fakeWorkItemEventLoader) LoadEventForWorkItem(_ context.Context, eventID string) (*connector_events.ConnectorEvent, error) {
	f.calls++
	event := f.event
	if len(f.sequence) > 0 {
		event = f.sequence[0]
		f.sequence = f.sequence[1:]
	}
	if f.events != nil {
		event = f.events[eventID]
	}
	if event == nil {
		return nil, connector_events.ErrEventNotFound
	}
	ev := *event
	return &ev, nil
}

func (f *fakeWorkItemEventLoader) LoadSourceProjectionPayloads(ctx context.Context, eventID string) ([]*connector_events.ConnectorEvent, error) {
	if events, ok := f.settledProjectionPayloads[eventID]; ok {
		out := make([]*connector_events.ConnectorEvent, 0, len(events))
		for _, event := range events {
			if event == nil {
				out = append(out, nil)
				continue
			}
			copy := *event
			out = append(out, &copy)
		}
		return out, nil
	}
	event, err := f.LoadEventForWorkItem(ctx, eventID)
	if err != nil {
		return nil, err
	}
	return []*connector_events.ConnectorEvent{event}, nil
}

type fakeWorkItemEventBridge struct {
	processingCalls         []string
	processingErr           error
	coalesceCalls           []string
	coalesceOwners          []string
	coalesceFences          []string
	coalescedClaim          *connector_events.WorkItemCoalescedSiblingClaim
	terminalCalls           []string
	projectionTerminalCalls []string
	projectionTerminalOK    bool
	projectionTerminalErr   error
	projectionRetryCalls    []projectionRetryCall
	projectionRetryOK       bool
	projectionRetryErr      error
	retryCalls              []string
	renewCalls              []string
	renewErr                error
}

type atomicFakeWorkItemEventBridge struct {
	fakeWorkItemEventBridge
	atomicCalls  int
	failureCalls []string
}

func (f *atomicFakeWorkItemEventBridge) CompleteEventForWorkItem(context.Context, string, string, string, string, string, time.Time) (bool, error) {
	f.atomicCalls++
	return true, nil
}

func (f *atomicFakeWorkItemEventBridge) FailEventForWorkItem(_ context.Context, eventID, workID, leaseToken, errorMsg string, now time.Time) (*connector_workqueue.FailWorkItemResult, bool, error) {
	f.failureCalls = append(f.failureCalls, strings.Join([]string{eventID, workID, leaseToken, errorMsg}, ":"))
	return &connector_workqueue.FailWorkItemResult{
		State:        connector_workqueue.WorkStateRetry,
		FailureCount: 1,
		RunAfter:     now.Add(5 * time.Second),
	}, true, nil
}

type ownedAtomicFakeWorkItemEventBridge struct {
	atomicFakeWorkItemEventBridge
	repo *connector_workqueue.Repository
}

func (f *ownedAtomicFakeWorkItemEventBridge) WorkItemRepository() *connector_workqueue.Repository {
	return f.repo
}

type projectionRetryCall struct {
	eventID, workID, leaseToken, errorMsg string
	runAfter                              time.Time
	retryCount                            int
	ctxErr                                error
}

func TestWorkItemExecutorAuthorityIgnoresDivergentEventLifecycle(t *testing.T) {
	now := time.Now().UTC()
	future := now.Add(time.Hour)
	claims := &fakeWorkItemClaims{
		executorStateAuthority: true,
		work: &connector_workqueue.WorkItem{
			WorkID: "work-owned", EventID: "event-owned",
			WorkType: connector_workqueue.WorkTypeConnectorEvent,
			State:    connector_workqueue.WorkStateProcessing, LeaseToken: "work-lease",
			LastModifiedDate: now, Generation: 7,
		},
	}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID: "event-owned", Status: connector_workqueue.WorkStateDead,
		RetryCount: 99, NextAttemptAt: &future, ClaimToken: "stale-event-lease",
		LastError: "stale event error",
	}}
	bridge := &fakeWorkItemEventBridge{}
	svc := &workItemEventsService{
		claims: claims, loader: loader, bridge: bridge,
		cfg: WorkerConfig{
			WorkItemSchedulerOwnerID:      "owner-1",
			WorkItemSchedulerFencingToken: "fence-1",
		},
		claimed: make(map[string]claimedWorkItemRef),
		peeked:  make(map[string]string),
	}

	ev, ok, err := svc.ClaimNext(context.Background(), "pack-1", "worker-1")
	if err != nil || !ok {
		t.Fatalf("ClaimNext = (%+v, %v, %v), want claimed WorkItem", ev, ok, err)
	}
	if ev.Status != connector_workqueue.WorkStateProcessing || ev.ClaimToken != "work-lease" ||
		ev.ClaimedBy != "worker-1" || ev.NextAttemptAt != nil || ev.RetryCount != 0 || ev.LastError != "" {
		t.Fatalf("claimed compatibility Event = %+v, want fields synthesized from WorkItem", ev)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 || len(bridge.retryCalls) != 0 {
		t.Fatalf("Event bridge calls = processing %v terminal %v retry %v, want none", bridge.processingCalls, bridge.terminalCalls, bridge.retryCalls)
	}
	if err := svc.RenewLease(context.Background(), ev.ID, ev.ClaimToken); err != nil {
		t.Fatalf("RenewLease: %v", err)
	}
	if claims.renewWorkCalls != 1 || len(bridge.renewCalls) != 0 {
		t.Fatalf("renew calls = WorkItem %d Event %v, want 1/none", claims.renewWorkCalls, bridge.renewCalls)
	}
	if err := svc.MarkProcessed(context.Background(), ev.ID, ev.ClaimToken); err != nil {
		t.Fatalf("MarkProcessed: %v", err)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("CompleteWorkItem calls = %v, want processed", got)
	}
	if len(bridge.terminalCalls) != 0 {
		t.Fatalf("Event terminal bridge calls = %v, want none", bridge.terminalCalls)
	}
}

func TestWorkItemExecutorAuthorityFailsAndReleasesThroughWorkItemOnly(t *testing.T) {
	newService := func() (*workItemEventsService, *fakeWorkItemClaims, *fakeWorkItemEventBridge) {
		claims := &fakeWorkItemClaims{executorStateAuthority: true}
		bridge := &fakeWorkItemEventBridge{}
		return &workItemEventsService{
			claims: claims, bridge: bridge,
			cfg: WorkerConfig{FinalizeTimeout: time.Second},
			claimed: map[string]claimedWorkItemRef{
				"event-owned": {workID: "work-owned", leaseToken: "work-lease"},
			},
			peeked: make(map[string]string),
		}, claims, bridge
	}

	failedSvc, failedClaims, failedBridge := newService()
	if err := failedSvc.MarkFailed(context.Background(), "event-owned", "work-lease", "raw provider error"); err != nil {
		t.Fatalf("MarkFailed: %v", err)
	}
	if len(failedClaims.failInputs) != 1 {
		t.Fatalf("FailWorkItem calls = %d, want 1", len(failedClaims.failInputs))
	}
	if in := failedClaims.failInputs[0]; in.FailureClass != connector_workqueue.FailureClassInboundExecution ||
		in.FailureSummary != connector_workqueue.FailureSummaryInbound ||
		in.RetryThreshold != connector_events.MaxRetryAttempts {
		t.Fatalf("FailWorkItem input = %+v", in)
	}
	if len(failedBridge.retryCalls) != 0 || len(failedBridge.terminalCalls) != 0 {
		t.Fatalf("Event failure bridge calls = retry %v terminal %v, want none", failedBridge.retryCalls, failedBridge.terminalCalls)
	}

	releasedSvc, releasedClaims, releasedBridge := newService()
	if err := releasedSvc.ReleaseClaimForRetry(context.Background(), "event-owned", "work-lease", "transient"); err != nil {
		t.Fatalf("ReleaseClaimForRetry: %v", err)
	}
	if releasedClaims.retryCalls != 1 || len(releasedBridge.retryCalls) != 0 {
		t.Fatalf("retry calls = WorkItem %d Event %v, want 1/none", releasedClaims.retryCalls, releasedBridge.retryCalls)
	}
}

func (f *fakeWorkItemEventBridge) MarkEventProcessingForWorkItem(_ context.Context, eventID, _, leaseToken string, _, _ time.Time) error {
	f.processingCalls = append(f.processingCalls, eventID+":"+leaseToken)
	return f.processingErr
}

func (f *fakeWorkItemEventBridge) ClaimNextCoalescedWorkItemSibling(_ context.Context, currentEventID, currentWorkID, currentLeaseToken, _, ownerID, fencingToken, nextLeaseToken string, _ int64, _, _ time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error) {
	f.coalesceCalls = append(f.coalesceCalls, currentEventID+":"+currentWorkID+":"+currentLeaseToken)
	f.coalesceOwners = append(f.coalesceOwners, ownerID)
	f.coalesceFences = append(f.coalesceFences, fencingToken)
	if f.coalescedClaim == nil {
		return nil, false, nil
	}
	claim := *f.coalescedClaim
	if claim.Work != nil {
		work := *claim.Work
		work.LeaseToken = nextLeaseToken
		claim.Work = &work
	}
	if claim.Event != nil {
		ev := *claim.Event
		ev.ClaimToken = nextLeaseToken
		claim.Event = &ev
	}
	f.coalescedClaim = nil
	return &claim, true, nil
}

func (f *fakeWorkItemEventBridge) ClaimNextCoalescedExecutorWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error) {
	return f.ClaimNextCoalescedWorkItemSibling(ctx, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now)
}

func (f *fakeWorkItemEventBridge) ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error) {
	return f.ClaimNextCoalescedExecutorWorkItemSibling(ctx, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now)
}

func (f *fakeWorkItemEventBridge) MarkEventTerminalForWorkItem(_ context.Context, eventID, leaseToken, terminalStatus, _ string, _ time.Time) error {
	f.terminalCalls = append(f.terminalCalls, eventID+":"+leaseToken+":"+terminalStatus)
	return nil
}

func (f *fakeWorkItemEventBridge) CompleteSourceProjectionEventForWorkItem(_ context.Context, eventID, workID, leaseToken, terminalStatus, errorMsg string, _ time.Time) (bool, error) {
	f.projectionTerminalCalls = append(f.projectionTerminalCalls, strings.Join([]string{eventID, workID, leaseToken, terminalStatus, errorMsg}, ":"))
	return f.projectionTerminalOK, f.projectionTerminalErr
}

func (f *fakeWorkItemEventBridge) RetrySourceProjectionEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, errorMsg string, runAfter, _ time.Time, retryCount int) (bool, error) {
	f.projectionRetryCalls = append(f.projectionRetryCalls, projectionRetryCall{
		eventID: eventID, workID: workID, leaseToken: leaseToken, errorMsg: errorMsg,
		runAfter: runAfter, retryCount: retryCount, ctxErr: ctx.Err(),
	})
	return f.projectionRetryOK, f.projectionRetryErr
}

func (f *fakeWorkItemEventBridge) ReleaseEventForWorkItemRetry(_ context.Context, eventID, leaseToken, _ string, _ time.Time, _ time.Time, retryCount int) error {
	f.retryCalls = append(f.retryCalls, eventID+":"+leaseToken+":"+strconv.Itoa(retryCount))
	return nil
}

func (f *fakeWorkItemEventBridge) RenewEventLeaseForWorkItem(_ context.Context, eventID, leaseToken string, _ time.Time) error {
	f.renewCalls = append(f.renewCalls, eventID+":"+leaseToken)
	return f.renewErr
}

func TestWorkItemCoalescedBridgeSkipsSourceProjectionWorkItem(t *testing.T) {
	bridge := &fakeWorkItemEventBridge{coalescedClaim: &connector_events.WorkItemCoalescedSiblingClaim{}}
	svc := &workItemEventsService{
		bridge: bridge,
		claimed: map[string]claimedWorkItemRef{
			"event-current": {
				workID:     "csp:source-projection",
				leaseToken: "lease-current",
				generation: 7,
			},
		},
	}

	next, ok, err := svc.ClaimNextCoalescedSiblingSourceScoped(context.Background(), "event-current", "lease-current", "worker-1")
	if err != nil {
		t.Fatalf("ClaimNextCoalescedSiblingSourceScoped: %v", err)
	}
	if ok || next != nil {
		t.Fatalf("source-projection coalesced sibling = (%+v, %v), want no legacy bridge handoff", next, ok)
	}
	if len(bridge.coalesceCalls) != 0 {
		t.Fatalf("legacy coalesced bridge calls = %v, want none for source projection", bridge.coalesceCalls)
	}
}

func TestWorkItemTerminalSourceProjectionUsesAtomicBridge(t *testing.T) {
	tests := []struct {
		name          string
		terminalState string
		errorMsg      string
		retryCount    int
		complete      func(*workItemEventsService) error
	}{
		{
			name:          "processed",
			terminalState: connector_workqueue.WorkStateProcessed,
			complete: func(s *workItemEventsService) error {
				return s.MarkProcessed(context.Background(), "event-current", "lease-current")
			},
		},
		{
			name:          "dead",
			terminalState: connector_workqueue.WorkStateDead,
			errorMsg:      "terminal failure",
			retryCount:    connector_events.MaxRetryAttempts - 1,
			complete: func(s *workItemEventsService) error {
				return s.MarkFailed(context.Background(), "event-current", "lease-current", "terminal failure")
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			claims := &fakeWorkItemClaims{}
			bridge := &fakeWorkItemEventBridge{projectionTerminalOK: true}
			svc := &workItemEventsService{
				claims: claims,
				bridge: bridge,
				claimed: map[string]claimedWorkItemRef{
					"event-current": {
						workID:     "csp:source-projection",
						leaseToken: "lease-current",
						retryCount: tt.retryCount,
					},
				},
			}

			if err := tt.complete(svc); err != nil {
				t.Fatalf("complete source projection: %v", err)
			}
			want := strings.Join([]string{"event-current", "csp:source-projection", "lease-current", tt.terminalState, tt.errorMsg}, ":")
			if len(bridge.projectionTerminalCalls) != 1 || bridge.projectionTerminalCalls[0] != want {
				t.Fatalf("atomic projection terminal calls = %v, want [%s]", bridge.projectionTerminalCalls, want)
			}
			if len(bridge.terminalCalls) != 0 || len(claims.completeCalls) != 0 {
				t.Fatalf("two-phase terminal/complete calls = %v/%v, want none", bridge.terminalCalls, claims.completeCalls)
			}
			if _, err := svc.claimedRef("event-current", "lease-current"); !errors.Is(err, connector_events.ErrLeaseLost) {
				t.Fatalf("completed projection remained claimed: %v", err)
			}
		})
	}
}

func TestWorkItemMarkFailedUsesRepositoryWorkItemAuthority(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectBegin()
	tx, err := db.Begin()
	if err != nil {
		t.Fatalf("Begin: %v", err)
	}
	repo := connector_workqueue.NewRepository(db, nil, nil).WithTx(tx)
	bridge := &ownedAtomicFakeWorkItemEventBridge{repo: repo}
	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*w.WorkId = \?.*w.EventId = \?.*w.State = \?.*w.LeaseToken = \?.*FOR UPDATE`).
		WithArgs("work-current", "event-current", connector_workqueue.WorkStateProcessing, "lease-current").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).
			AddRow("event-current", connector_events.MaxRetryAttempts-1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*FailureCount = \?.*WHERE WorkId = \?.*FailureCount = \?`).
		WithArgs(
			connector_workqueue.WorkStateDead, sqlmock.AnyArg(), connector_events.MaxRetryAttempts,
			connector_workqueue.FailureClassInboundExecution, connector_workqueue.FailureSummaryInbound,
			sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),
			"work-current", "event-current", connector_workqueue.WorkStateProcessing, "lease-current",
			connector_events.MaxRetryAttempts-1,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`(?s)SELECT InstalledPackId,.*Generation,.*SourceLaneKey.*TargetMutationLaneKey.*FROM _System_ConnectorWorkItem.*WorkId = \?`).
		WithArgs("work-current").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-current", int64(0), "", ""))
	svc := &workItemEventsService{
		claims: repo,
		bridge: bridge,
		cfg:    WorkerConfig{FinalizeTimeout: time.Second},
		claimed: map[string]claimedWorkItemRef{
			"event-current": {
				workID:     "work-current",
				leaseToken: "lease-current",
				retryCount: connector_events.MaxRetryAttempts - 1,
			},
		},
	}

	if err := svc.MarkFailed(context.Background(), "event-current", "lease-current", "boom"); err != nil {
		t.Fatalf("MarkFailed: %v", err)
	}
	mock.ExpectCommit()
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	if len(bridge.failureCalls) != 0 {
		t.Fatalf("Event failure bridge calls = %v, want none", bridge.failureCalls)
	}
	if len(bridge.terminalCalls) != 0 || len(bridge.retryCalls) != 0 {
		t.Fatalf("legacy Event terminal/retry calls = %v/%v, want none", bridge.terminalCalls, bridge.retryCalls)
	}
	if _, err := svc.claimedRef("event-current", "lease-current"); !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("failed WorkItem remained claimed: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestWorkItemRetrySourceProjectionUsesAtomicBridge(t *testing.T) {
	tests := []struct {
		name           string
		retryCount     int
		wantRetryCount int
		wantErrorMsg   string
		retry          func(context.Context, *workItemEventsService) error
		checkRunAfter  func(*testing.T, time.Time, time.Time)
	}{
		{
			name:           "mark failed increments budget",
			retryCount:     1,
			wantRetryCount: 2,
			wantErrorMsg:   "retry failure",
			retry: func(ctx context.Context, s *workItemEventsService) error {
				return s.MarkFailed(ctx, "event-current", "lease-current", "retry failure")
			},
			checkRunAfter: func(t *testing.T, before, got time.Time) {
				t.Helper()
				if got.Before(before.Add(29*time.Second)) || got.After(time.Now().UTC().Add(31*time.Second)) {
					t.Fatalf("MarkFailed runAfter = %v, want about 30 seconds", got)
				}
			},
		},
		{
			name:           "release claim keeps budget",
			retryCount:     2,
			wantRetryCount: 2,
			wantErrorMsg:   "work item retry",
			retry: func(ctx context.Context, s *workItemEventsService) error {
				return s.ReleaseClaim(ctx, "event-current", "lease-current")
			},
			checkRunAfter: func(t *testing.T, before, got time.Time) {
				t.Helper()
				if got.Before(before) || got.After(time.Now().UTC().Add(time.Second)) {
					t.Fatalf("ReleaseClaim runAfter = %v, want immediately due", got)
				}
			},
		},
		{
			name:           "release claim for retry keeps budget",
			retryCount:     3,
			wantRetryCount: 3,
			wantErrorMsg:   "work item retry",
			retry: func(ctx context.Context, s *workItemEventsService) error {
				return s.ReleaseClaimForRetry(ctx, "event-current", "lease-current", "transient")
			},
			checkRunAfter: func(t *testing.T, before, got time.Time) {
				t.Helper()
				if got.Before(before.Add(29*time.Second)) || got.After(time.Now().UTC().Add(31*time.Second)) {
					t.Fatalf("ReleaseClaimForRetry runAfter = %v, want about 30 seconds", got)
				}
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			claims := &fakeWorkItemClaims{}
			bridge := &fakeWorkItemEventBridge{projectionRetryOK: true}
			svc := &workItemEventsService{
				claims: claims,
				bridge: bridge,
				cfg:    WorkerConfig{FinalizeTimeout: time.Second},
				claimed: map[string]claimedWorkItemRef{"event-current": {
					workID: "csp:source-projection", leaseToken: "lease-current", retryCount: tt.retryCount,
				}},
			}
			ctx, cancel := context.WithCancel(context.Background())
			cancel()
			before := time.Now().UTC()
			if err := tt.retry(ctx, svc); err != nil {
				t.Fatalf("retry source projection: %v", err)
			}
			if len(bridge.projectionRetryCalls) != 1 {
				t.Fatalf("atomic projection retry calls = %d, want 1", len(bridge.projectionRetryCalls))
			}
			call := bridge.projectionRetryCalls[0]
			if call.eventID != "event-current" || call.workID != "csp:source-projection" || call.leaseToken != "lease-current" || call.errorMsg != tt.wantErrorMsg || call.retryCount != tt.wantRetryCount {
				t.Fatalf("atomic projection retry call = %+v", call)
			}
			if call.ctxErr != nil {
				t.Fatalf("atomic projection retry context err = %v, want fresh live context", call.ctxErr)
			}
			tt.checkRunAfter(t, before, call.runAfter)
			if len(bridge.retryCalls) != 0 || claims.retryCalls != 0 {
				t.Fatalf("two-phase Event/WorkItem retry calls = %v/%d, want none", bridge.retryCalls, claims.retryCalls)
			}
		})
	}
}

func TestWorkItemSourceProjectionLeaseLossConvergesByEventState(t *testing.T) {
	now := time.Now().UTC()
	future := now.Add(2 * time.Minute)
	processingAt := now.Add(-time.Second)
	tests := []struct {
		name              string
		event             *connector_events.ConnectorEvent
		wantCompleteState string
		wantRunAfter      time.Time
	}{
		{
			name:              "terminal advances projection",
			event:             &connector_events.ConnectorEvent{ID: "event-current", Status: connector_workqueue.WorkStateProcessed},
			wantCompleteState: connector_workqueue.WorkStateProcessed,
		},
		{
			name:         "failed keeps event backoff",
			event:        &connector_events.ConnectorEvent{ID: "event-current", Status: "failed", NextAttemptAt: &future},
			wantRunAfter: future,
		},
		{
			name:         "fresh processing waits for stale boundary",
			event:        &connector_events.ConnectorEvent{ID: "event-current", Status: connector_workqueue.WorkStateProcessing, LastModifiedDate: processingAt},
			wantRunAfter: processingAt.Add((connector_events.StaleProcessingSeconds + 1) * time.Second),
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			claims := &fakeWorkItemClaims{}
			svc := &workItemEventsService{
				claims: claims,
				loader: &fakeWorkItemEventLoader{event: tt.event},
			}
			err := svc.retryWorkItemAfterEventLeaseLost(context.Background(), &connector_workqueue.WorkItem{
				WorkID: "csp:source-projection", LeaseToken: "lease-current",
			}, tt.event, now)
			if err != nil {
				t.Fatalf("converge source projection after Event lease loss: %v", err)
			}
			if tt.wantCompleteState != "" {
				if len(claims.completeCalls) != 1 || claims.completeCalls[0] != tt.wantCompleteState || claims.retryCalls != 0 {
					t.Fatalf("complete/retry calls = %v/%d, want [%s]/0", claims.completeCalls, claims.retryCalls, tt.wantCompleteState)
				}
				return
			}
			if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || !claims.retryRunAfters[0].Equal(tt.wantRunAfter) || len(claims.completeCalls) != 0 {
				t.Fatalf("retry runAfter/complete = %v/%v, want [%v]/none", claims.retryRunAfters, claims.completeCalls, tt.wantRunAfter)
			}
		})
	}
}

func TestWorkItemClaimSourceProjectionTerminalCollisionConvergesActualState(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID: "csp:source-projection", EventID: "event-terminal-race",
		WorkType: connector_workqueue.WorkTypeConnectorEvent, State: connector_workqueue.WorkStateProcessing,
		LeaseToken: "lease-terminal-race", LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{sequence: []*connector_events.ConnectorEvent{
		{ID: "event-terminal-race", Status: connector_workqueue.WorkStatePending},
		{ID: "event-terminal-race", Status: connector_workqueue.WorkStateProcessed},
	}}
	bridge := &fakeWorkItemEventBridge{processingErr: connector_events.ErrEventAlreadyTerminal}
	svc := &workItemEventsService{
		claims: claims, loader: loader, bridge: bridge,
		cfg:     WorkerConfig{WorkItemLeaseDuration: time.Minute},
		claimed: map[string]claimedWorkItemRef{},
	}

	ev, ok, err := svc.claimNext(context.Background(), "pack", "worker", nil, false)
	if err != nil || ok || ev != nil {
		t.Fatalf("claimNext terminal collision = (%v, %v, %v), want nil, false, nil", ev, ok, err)
	}
	if len(claims.completeCalls) != 1 || claims.completeCalls[0] != connector_workqueue.WorkStateProcessed || claims.retryCalls != 0 {
		t.Fatalf("complete/retry calls = %v/%d, want [processed]/0", claims.completeCalls, claims.retryCalls)
	}
}

func TestWorkItemRenewLeaseRenewsEventMirrorBeforeWorkItem(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{}
	svc := &workItemEventsService{
		claims:  claims,
		bridge:  bridge,
		cfg:     WorkerConfig{WorkItemLeaseDuration: time.Minute},
		claimed: map[string]claimedWorkItemRef{"event-renew": {workID: "work-renew", leaseToken: "lease-renew"}},
	}

	if err := svc.RenewLease(context.Background(), "event-renew", "lease-renew"); err != nil {
		t.Fatalf("RenewLease: %v", err)
	}
	if got := bridge.renewCalls; len(got) != 1 || got[0] != "event-renew:lease-renew" {
		t.Fatalf("event mirror renew calls = %v, want event mirror renewed", got)
	}
	if claims.renewWorkCalls != 1 {
		t.Fatalf("RenewWorkItemLease calls = %d, want 1 after event mirror renewal", claims.renewWorkCalls)
	}
	if claims.retryCalls != 0 {
		t.Fatalf("RetryWorkItem calls = %d, want no release on successful heartbeat", claims.retryCalls)
	}
}

func TestWorkItemRenewLeaseReleasesWorkItemWhenEventMirrorRenewalFails(t *testing.T) {
	renewErr := errors.New("event mirror renewal failed")
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{renewErr: renewErr}
	svc := &workItemEventsService{
		claims: claims,
		bridge: bridge,
		cfg:    WorkerConfig{WorkItemLeaseDuration: time.Minute},
		claimed: map[string]claimedWorkItemRef{"event-renew-fail": {
			workID:     "work-renew-fail",
			leaseToken: "lease-renew-fail",
			retryCount: 3,
		}},
	}

	err := svc.RenewLease(context.Background(), "event-renew-fail", "lease-renew-fail")
	if !errors.Is(err, renewErr) {
		t.Fatalf("RenewLease error = %v, want event mirror renewal error", err)
	}
	if got := bridge.renewCalls; len(got) != 1 || got[0] != "event-renew-fail:lease-renew-fail" {
		t.Fatalf("event mirror renew calls = %v, want attempted mirror renewal", got)
	}
	if claims.renewWorkCalls != 0 {
		t.Fatalf("RenewWorkItemLease calls = %d, want no WorkItem heartbeat after mirror renewal failed", claims.renewWorkCalls)
	}
	if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || claims.retryRunAfters[0].After(time.Now().UTC()) {
		t.Fatalf("RetryWorkItem calls/runAfter = %d/%v, want immediate due release after mirror renewal failed", claims.retryCalls, claims.retryRunAfters)
	}
	if len(claims.completeCalls) != 0 {
		t.Fatalf("complete calls = %v, want no terminal WorkItem completion", claims.completeCalls)
	}
	if _, err := svc.claimedRef("event-renew-fail", "lease-renew-fail"); !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("claimedRef after failed renewal = %v, want forgotten claim", err)
	}
}

func TestWorkItemRenewLeaseConvergesSourceProjectionWhenEventIsTerminal(t *testing.T) {
	renewErr := errors.New("event mirror lease lost")
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{renewErr: renewErr}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID: "event-renew-terminal", Status: connector_workqueue.WorkStateProcessed,
	}}
	svc := &workItemEventsService{
		claims: claims,
		bridge: bridge,
		loader: loader,
		cfg:    WorkerConfig{WorkItemLeaseDuration: time.Minute, FinalizeTimeout: time.Second},
		claimed: map[string]claimedWorkItemRef{"event-renew-terminal": {
			workID: "csp:source-projection", leaseToken: "lease-renew-terminal",
		}},
	}

	err := svc.RenewLease(context.Background(), "event-renew-terminal", "lease-renew-terminal")
	if !errors.Is(err, renewErr) {
		t.Fatalf("RenewLease error = %v, want original Event renewal error", err)
	}
	if len(claims.completeCalls) != 1 || claims.completeCalls[0] != connector_workqueue.WorkStateProcessed || claims.retryCalls != 0 {
		t.Fatalf("complete/retry calls = %v/%d, want [processed]/0", claims.completeCalls, claims.retryCalls)
	}
	if _, err := svc.claimedRef("event-renew-terminal", "lease-renew-terminal"); !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("terminal source projection remained claimed: %v", err)
	}
}

func TestWorkerDefaultClaimPathRejectsWorkItemConfig(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	_, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:      "ns",
		PackInstallID:  "pack",
		WorkerID:       "worker",
		WorkItemClaims: claims,
	}, newFakeEvents(), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err == nil {
		t.Fatalf("NewWorker with WorkItem config and default claim path succeeded; want fail-closed error")
	}
}

func TestWorkerWorkItemClaimPathUsesWorkItemPrimitive(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-1",
		EventID:          "event-1",
		InstalledPackID:  "pack",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-token",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-1",
		Namespace:       "ns",
		InstalledPackID: "pack",
		EventKind:       "object.changed",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		CreatedDate:     now.Add(-time.Second),
	}}
	bridge := &fakeWorkItemEventBridge{}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	ev, ok, err := w.claimNext(context.Background(), false)
	if err != nil || !ok {
		t.Fatalf("claimNext = (%v, %v, %v), want claimed WorkItem event", ev, ok, err)
	}
	if ev.ID != "event-1" || ev.ClaimToken != "lease-token" || ev.Status != "processing" {
		t.Fatalf("loaded event = %#v, want WorkItem lease stamped onto event", ev)
	}
	if claims.claimCalls != 1 || loader.calls != 1 {
		t.Fatalf("WorkItem path calls claim/load = %d/%d, want 1/1", claims.claimCalls, loader.calls)
	}
	if claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease calls acquire/renew = %d/%d, want none", claims.acquireCalls, claims.renewCalls)
	}
	if got := claims.lastClaimInput.OwnerID; got != "scheduler-owner" {
		t.Fatalf("WorkItem LeaseOwner = %q, want stable scheduler owner", got)
	}
	if got := claims.lastClaimInput.FencingToken; got != "fence" {
		t.Fatalf("WorkItem compatibility fencing token = %q, want fence", got)
	}
	if len(bridge.processingCalls) != 0 {
		t.Fatalf("processing bridge calls = %v, want none", bridge.processingCalls)
	}
	if err := w.markProcessed(ev); err != nil {
		t.Fatalf("markProcessed via WorkItem finalizer: %v", err)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("complete states = %v, want [processed]", got)
	}
	if len(bridge.terminalCalls) != 0 {
		t.Fatalf("terminal bridge calls = %v, want none", bridge.terminalCalls)
	}
}

func TestWorkItemCompleteUsesFreshContextAfterEventBridgeSuccess(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{}
	svc := &workItemEventsService{
		claims: claims,
		bridge: bridge,
		cfg: WorkerConfig{
			FinalizeTimeout: time.Second,
		},
		claimed: map[string]claimedWorkItemRef{"event-finalize": {
			workID:     "work-finalize",
			leaseToken: "lease-finalize",
		}},
	}

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	if err := svc.complete(ctx, "event-finalize", "lease-finalize", connector_workqueue.WorkStateProcessed); err != nil {
		t.Fatalf("complete with canceled caller context: %v", err)
	}
	if got := bridge.terminalCalls; len(got) != 1 || got[0] != "event-finalize:lease-finalize:processed" {
		t.Fatalf("terminal bridge calls = %v, want Event terminal bridge success", got)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("CompleteWorkItem calls = %v, want processed follow-up", got)
	}
	if len(claims.completeContexts) != 1 {
		t.Fatalf("CompleteWorkItem contexts = %d, want 1", len(claims.completeContexts))
	}
	if err := claims.completeCtxErrs[0]; err != nil {
		t.Fatalf("CompleteWorkItem context err at call time = %v, want fresh live context", err)
	}
}

func TestWorkItemCompleteKeepsCustomClaimsAuthoritativeOverAtomicBridge(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	bridge := &atomicFakeWorkItemEventBridge{}
	svc := &workItemEventsService{
		claims: claims,
		bridge: bridge,
		cfg:    WorkerConfig{FinalizeTimeout: time.Second},
		claimed: map[string]claimedWorkItemRef{"event-custom-claims": {
			workID:     "work-custom-claims",
			leaseToken: "lease-custom-claims",
		}},
	}

	if err := svc.complete(context.Background(), "event-custom-claims", "lease-custom-claims", connector_workqueue.WorkStateProcessed); err != nil {
		t.Fatalf("complete with custom claims: %v", err)
	}
	if bridge.atomicCalls != 0 {
		t.Fatalf("atomic bridge calls = %d, want 0 when custom WorkItemClaims is injected", bridge.atomicCalls)
	}
	if got := bridge.terminalCalls; len(got) != 1 || got[0] != "event-custom-claims:lease-custom-claims:processed" {
		t.Fatalf("terminal bridge calls = %v, want legacy Event terminal follow-up", got)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("custom CompleteWorkItem calls = %v, want processed", got)
	}
}

func TestWorkItemCompleteUsesRepositoryWorkItemAuthority(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	repo := connector_workqueue.NewRepository(db, nil, nil)
	bridge := &ownedAtomicFakeWorkItemEventBridge{repo: repo}
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*TerminalAt = \?.*WHERE WorkId = \?.*State = \?.*LeaseToken = \?`).
		WithArgs(
			connector_workqueue.WorkStateProcessed, sqlmock.AnyArg(), sqlmock.AnyArg(),
			"work-owned-claims", connector_workqueue.WorkStateProcessing, "lease-owned-claims",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`(?s)SELECT InstalledPackId,.*Generation,.*SourceLaneKey.*TargetMutationLaneKey.*FROM _System_ConnectorWorkItem.*WorkId = \?`).
		WithArgs("work-owned-claims").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-owned", int64(0), "", ""))
	svc := &workItemEventsService{
		claims: repo,
		bridge: bridge,
		cfg:    WorkerConfig{FinalizeTimeout: time.Second},
		claimed: map[string]claimedWorkItemRef{"event-owned-claims": {
			workID:     "work-owned-claims",
			leaseToken: "lease-owned-claims",
		}},
	}

	if err := svc.complete(context.Background(), "event-owned-claims", "lease-owned-claims", connector_workqueue.WorkStateProcessed); err != nil {
		t.Fatalf("complete with bridge-owned claims: %v", err)
	}
	if bridge.atomicCalls != 0 {
		t.Fatalf("atomic Event bridge calls = %d, want 0 under WorkItem authority", bridge.atomicCalls)
	}
	if len(bridge.terminalCalls) != 0 {
		t.Fatalf("legacy terminal bridge calls = %v, want none", bridge.terminalCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestWorkItemFailedRetryUsesFreshContextAfterEventBridgeSuccess(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{}
	svc := &workItemEventsService{
		claims: claims,
		bridge: bridge,
		cfg:    WorkerConfig{FinalizeTimeout: time.Second},
		claimed: map[string]claimedWorkItemRef{"event-retry": {
			workID:     "work-retry",
			leaseToken: "lease-retry",
			retryCount: 1,
		}},
	}

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	if err := svc.MarkFailed(ctx, "event-retry", "lease-retry", "retry me"); err != nil {
		t.Fatalf("MarkFailed retry with canceled caller context: %v", err)
	}
	if got := bridge.retryCalls; len(got) != 1 || !strings.HasPrefix(got[0], "event-retry:lease-retry:2") {
		t.Fatalf("retry bridge calls = %v, want Event retry bridge success", got)
	}
	if claims.retryCalls != 1 || len(claims.retryCtxErrs) != 1 {
		t.Fatalf("RetryWorkItem calls/context errs = %d/%v, want one follow-up", claims.retryCalls, claims.retryCtxErrs)
	}
	if err := claims.retryCtxErrs[0]; err != nil {
		t.Fatalf("RetryWorkItem context err at call time = %v, want fresh live context", err)
	}
}

func TestWorkItemFailedDeadUsesFreshContextAfterEventBridgeSuccess(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	bridge := &fakeWorkItemEventBridge{}
	svc := &workItemEventsService{
		claims: claims,
		bridge: bridge,
		cfg:    WorkerConfig{FinalizeTimeout: time.Second},
		claimed: map[string]claimedWorkItemRef{"event-dead": {
			workID:     "work-dead",
			leaseToken: "lease-dead",
			retryCount: connector_events.MaxRetryAttempts - 1,
		}},
	}

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	if err := svc.MarkFailed(ctx, "event-dead", "lease-dead", "dead me"); err != nil {
		t.Fatalf("MarkFailed dead with canceled caller context: %v", err)
	}
	if got := bridge.terminalCalls; len(got) != 1 || got[0] != "event-dead:lease-dead:dead" {
		t.Fatalf("terminal bridge calls = %v, want Event dead bridge success", got)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateDead {
		t.Fatalf("CompleteWorkItem calls = %v, want dead follow-up", got)
	}
	if err := claims.completeCtxErrs[0]; err != nil {
		t.Fatalf("CompleteWorkItem context err at call time = %v, want fresh live context", err)
	}
}

func TestWorkerWorkItemClaimPathResolvesActiveGeneration(t *testing.T) {
	now := time.Now().UTC()
	activeGeneration := int64(7)
	claims := &fakeWorkItemClaims{
		resolveGeneration: &activeGeneration,
		work: &connector_workqueue.WorkItem{
			WorkID:           "work-1",
			EventID:          "event-1",
			InstalledPackID:  "pack",
			WorkType:         connector_workqueue.WorkTypeConnectorEvent,
			QueueClass:       connector_workqueue.DefaultQueueClass,
			State:            connector_workqueue.WorkStateProcessing,
			LeaseToken:       "lease-token",
			Attempt:          1,
			Generation:       activeGeneration,
			LastModifiedDate: now,
		},
	}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-1",
		Namespace:       "ns",
		InstalledPackID: "pack",
		EventKind:       "object.changed",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		CreatedDate:     now.Add(-time.Second),
	}}
	service := newCompleteWorkItemServiceFake(claims, loader, &fakeWorkItemEventBridge{})
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		WorkItemGeneration:            0,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if _, ok, err := w.claimNext(context.Background(), false); err != nil || !ok {
		t.Fatalf("claimNext = %t, %v; want claim on resolved generation", ok, err)
	}
	if claims.resolveCalls == 0 {
		t.Fatal("ResolveClaimableWorkItemGeneration was not called")
	}
	if claims.lastClaimInput.Generation != activeGeneration {
		t.Fatalf("claim generation = %d, want %d", claims.lastClaimInput.Generation, activeGeneration)
	}
	wantWorkTypes := []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
	if got := strings.Join(claims.lastResolveWorkTypes, ","); got != strings.Join(wantWorkTypes, ",") {
		t.Fatalf("resolve work types = %q, want %q", got, strings.Join(wantWorkTypes, ","))
	}
	if got := strings.Join(claims.lastClaimWorkTypes, ","); got != strings.Join(wantWorkTypes, ",") {
		t.Fatalf("claim work types = %q, want %q", got, strings.Join(wantWorkTypes, ","))
	}
}

func TestWorkerWorkItemClaimPathRejectsOutOfScopeClaimResult(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:          "work-outbound",
		EventID:         "event-outbound",
		InstalledPackID: "pack",
		WorkType:        "external_effect_attempt",
		State:           connector_workqueue.WorkStateProcessing,
		LeaseToken:      "lease-outbound",
		Generation:      0,
	}}
	loader := &fakeWorkItemEventLoader{}
	svc := newMandatoryWorkItemEventsService(
		newCompleteWorkItemServiceFake(claims, loader, nil),
		WorkerConfig{
			WorkItemSchedulerOwnerID:      "scheduler-owner",
			WorkItemSchedulerFencingToken: "fence",
			WorkItemLeaseDuration:         time.Minute,
		},
	)

	ev, ok, err := svc.ClaimNext(context.Background(), "pack", "worker")
	if err == nil || !strings.Contains(err.Error(), "unsupported WorkItem type") || ok || ev != nil {
		t.Fatalf("ClaimNext = (%v, %v, %v), want rejected out-of-scope WorkItem", ev, ok, err)
	}
	if claims.retryCalls != 1 {
		t.Fatalf("RetryWorkItem calls = %d, want one defensive release", claims.retryCalls)
	}
	if loader.calls != 0 {
		t.Fatalf("event loader calls = %d, want none for out-of-scope WorkItem", loader.calls)
	}
	wantWorkTypes := []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
	if got := strings.Join(claims.lastClaimWorkTypes, ","); got != strings.Join(wantWorkTypes, ",") {
		t.Fatalf("claim work types = %q, want %q", got, strings.Join(wantWorkTypes, ","))
	}
	if got := claims.retryRunAfters[0]; got.Before(now.Add(59*time.Second)) || got.After(now.Add(61*time.Second)) {
		t.Fatalf("retry run after = %s, want approximately one minute", got)
	}
}

func TestWorkItemClaimInputServiceRejectsNilSuccessfulClaims(t *testing.T) {
	newService := func() *workItemEventsService {
		claims := &fakeWorkItemClaims{}
		inputs := &nilSuccessfulWorkItemClaimInputs{
			fakeWorkItemClaimInputs: newFakeWorkItemClaimInputs(claims, &fakeWorkItemEventLoader{}),
		}
		service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
		service.inputs = inputs
		return newMandatoryWorkItemEventsService(service, WorkerConfig{
			WorkItemSchedulerOwnerID:      "scheduler-owner",
			WorkItemSchedulerFencingToken: "fence",
			WorkItemLeaseDuration:         time.Minute,
		})
	}

	t.Run("claim next", func(t *testing.T) {
		event, ok, err := newService().ClaimNext(context.Background(), "pack", "worker")
		if err == nil || !strings.Contains(err.Error(), "nil WorkItem") || ok || event != nil {
			t.Fatalf("ClaimNext = (%v, %v, %v), want fail-closed nil-success rejection", event, ok, err)
		}
	})

	t.Run("claim by event", func(t *testing.T) {
		token, _, ok, err := newService().ClaimByID(context.Background(), "event-1", "worker", "pack", sql.NullString{}, 0)
		if err == nil || !strings.Contains(err.Error(), "nil WorkItem") || ok || token != "" {
			t.Fatalf("ClaimByID = (%q, %v, %v), want fail-closed nil-success rejection", token, ok, err)
		}
	})
}

func TestWorkItemClaimInputServiceRejectsMissingImmutableEvent(t *testing.T) {
	newService := func() (*workItemEventsService, *fakeWorkItemClaims) {
		claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
			WorkID: "work-1", EventID: "event-1",
			WorkType: connector_workqueue.WorkTypeConnectorEvent,
			State:    connector_workqueue.WorkStateProcessing,
		}}
		service := newCompleteWorkItemServiceFake(claims, nilSuccessfulWorkItemEventLoader{}, nil)
		return newMandatoryWorkItemEventsService(service, WorkerConfig{
			WorkItemSchedulerOwnerID:      "scheduler-owner",
			WorkItemSchedulerFencingToken: "fence",
			WorkItemLeaseDuration:         time.Minute,
		}), claims
	}

	t.Run("claim next", func(t *testing.T) {
		service, claims := newService()
		event, ok, err := service.ClaimNext(context.Background(), "pack", "worker")
		if !errors.Is(err, connector_events.ErrEventNotFound) || ok || event != nil {
			t.Fatalf("ClaimNext = (%v, %v, %v), want fail-closed immutable Event error", event, ok, err)
		}
		if claims.retryCalls != 1 {
			t.Fatalf("RetryWorkItem calls = %d, want claimed WorkItem released", claims.retryCalls)
		}
	})

	t.Run("claim by event", func(t *testing.T) {
		service, claims := newService()
		token, _, ok, err := service.ClaimByID(context.Background(), "event-1", "worker", "pack", sql.NullString{}, 0)
		if !errors.Is(err, connector_events.ErrEventNotFound) || ok || token != "" {
			t.Fatalf("ClaimByID = (%q, %v, %v), want fail-closed immutable Event error", token, ok, err)
		}
		if claims.retryCalls != 1 {
			t.Fatalf("RetryWorkItem calls = %d, want claimed WorkItem released", claims.retryCalls)
		}
	})
}

func TestNewWorkerRejectsUnscopedWorkItemClaimInputService(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
	service.inputs.(*fakeWorkItemClaimInputs).honors = false
	_, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if !errors.Is(err, ErrInvalidWorkerConfig) || !strings.Contains(err.Error(), "ClaimWorkTypes") {
		t.Fatalf("NewWorker error = %v, want ClaimWorkTypes admission failure", err)
	}
}

func TestNewWorkerPreservesConfiguredLegacyClaimInputs(t *testing.T) {
	legacyClaims := &fakeWorkItemClaims{
		executorStateAuthority: true,
		work: &connector_workqueue.WorkItem{
			WorkID:     "legacy-work",
			EventID:    "legacy-event",
			WorkType:   connector_workqueue.WorkTypeConnectorEvent,
			State:      connector_workqueue.WorkStateProcessing,
			LeaseToken: "legacy-lease",
		},
	}
	legacyLoader := &fakeWorkItemEventLoader{event: noopEvent("legacy-event", "noop.created")}
	consolidatedClaims := &fakeWorkItemClaims{
		executorStateAuthority: true,
		work: &connector_workqueue.WorkItem{
			WorkID:     "consolidated-work",
			EventID:    "consolidated-event",
			WorkType:   connector_workqueue.WorkTypeConnectorEvent,
			State:      connector_workqueue.WorkStateProcessing,
			LeaseToken: "consolidated-lease",
		},
	}
	consolidatedLoader := &fakeWorkItemEventLoader{event: noopEvent("consolidated-event", "noop.created")}
	events := workItemClaimInputEvents{
		EventsService:             newFakeEvents(),
		WorkItemClaimInputService: newFakeWorkItemClaimInputs(consolidatedClaims, consolidatedLoader),
	}
	worker, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                legacyClaims,
		WorkItemEventLoader:           legacyLoader,
		WorkItemEventBridge:           &fakeWorkItemEventBridge{},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, events, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	event, claimed, err := worker.events.ClaimNext(context.Background(), "pack-test", "worker")
	if err != nil {
		t.Fatalf("ClaimNext: %v", err)
	}
	if !claimed || event == nil || event.ID != "legacy-event" {
		t.Fatalf("ClaimNext = (%+v, %v), want configured legacy event", event, claimed)
	}
	if legacyClaims.claimCalls != 1 || legacyLoader.calls != 1 {
		t.Fatalf("configured legacy claim/loader calls = %d/%d, want 1/1", legacyClaims.claimCalls, legacyLoader.calls)
	}
	if consolidatedClaims.claimCalls != 0 || consolidatedLoader.calls != 0 {
		t.Fatalf("implicit consolidated claim/loader calls = %d/%d, want 0/0", consolidatedClaims.claimCalls, consolidatedLoader.calls)
	}

	t.Run("published compatibility lookup and batch load", func(t *testing.T) {
		now := time.Now().UTC()
		claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{
			{WorkID: "legacy-peek-work", EventID: "legacy-peek-event", WorkType: connector_workqueue.WorkTypeConnectorEvent, Generation: 3},
			{WorkID: "legacy-stale-work", EventID: "legacy-stale-event", WorkType: connector_workqueue.WorkTypeConnectorEvent, Generation: 3},
		}}
		baseLoader := &fakeWorkItemEventLoader{
			events: map[string]*connector_events.ConnectorEvent{
				"legacy-peek-event":  noopEvent("legacy-peek-event", "noop.created"),
				"legacy-stale-event": noopEvent("legacy-stale-event", "noop.created"),
			},
			settledProjectionPayloads: map[string][]*connector_events.ConnectorEvent{
				"legacy-stale-event": {noopEvent("legacy-stale-event", "noop.created")},
			},
		}
		loader := &fakeBatchWorkItemEventLoader{fakeWorkItemEventLoader: baseLoader}
		shim := legacyWorkItemClaimInputs{claims: claims, loader: loader}
		input := connector_workqueue.ClaimWorkItemInput{Generation: 3, Now: now}
		peeked, err := shim.PeekClaimableWorkItems(context.Background(), input, 2)
		if err != nil || len(peeked) != 2 {
			t.Fatalf("PeekClaimableWorkItems = (%v, %v), want two published-shim candidates", peeked, err)
		}
		claimed, ok, err := shim.ClaimStaleWorkItemByEvent(context.Background(), connector_workqueue.ClaimWorkItemByEventInput{
			EventID: "legacy-stale-event", WorkType: connector_workqueue.WorkTypeConnectorEvent,
			Generation: 3, LeaseToken: "legacy-stale-lease", LeaseUntil: now.Add(time.Minute), Now: now,
		})
		if err != nil || !ok || claimed == nil || claimed.EventID != "legacy-stale-event" {
			t.Fatalf("ClaimStaleWorkItemByEvent = (%+v, %v, %v), want stale compatibility claim", claimed, ok, err)
		}
		loaded, err := shim.LoadEventsForWorkItems(context.Background(), []string{"legacy-peek-event", "legacy-stale-event"})
		if err != nil || len(loaded) != 2 || loader.batchCalls != 1 {
			t.Fatalf("LoadEventsForWorkItems = (%v, %v), batch calls=%d; want two events in one batch", loaded, err, loader.batchCalls)
		}
		payloads, err := shim.LoadSourceProjectionPayloads(context.Background(), "legacy-stale-event")
		if err != nil || len(payloads) != 1 {
			t.Fatalf("LoadSourceProjectionPayloads = (%v, %v), want one payload", payloads, err)
		}
	})

	t.Run("published compatibility single-load fallback", func(t *testing.T) {
		claims := &fakeWorkItemClaims{}
		loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{
			"legacy-fallback-a": noopEvent("legacy-fallback-a", "noop.created"),
			"legacy-fallback-b": noopEvent("legacy-fallback-b", "noop.created"),
		}}
		shim := legacyWorkItemClaimInputs{claims: claims, loader: loader}
		loaded, err := shim.LoadEventsForWorkItems(context.Background(), []string{"legacy-fallback-a", "legacy-fallback-b"})
		if err != nil || len(loaded) != 2 || loader.calls != 2 {
			t.Fatalf("fallback LoadEventsForWorkItems = (%v, %v), single calls=%d; want two", loaded, err, loader.calls)
		}
	})
}

func TestNewWorkerRequiresOneRepositoryForConsolidatedClaimInputs(t *testing.T) {
	events := connector_events.NewService(nil, nil, nil)
	baseConfig := WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                events.WorkItemRepository(),
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}

	if _, err := newWorkerForTestWithNoopBatchRead(baseConfig, events, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{}); err != nil {
		t.Fatalf("NewWorker with one consolidated repository: %v", err)
	}

	baseConfig.WorkItemClaims = connector_workqueue.NewRepository(nil, nil, nil)
	_, err := newWorkerForTestWithNoopBatchRead(baseConfig, events, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if !errors.Is(err, ErrInvalidWorkerConfig) || !strings.Contains(err.Error(), "share the configured WorkItemClaims repository") {
		t.Fatalf("NewWorker error = %v, want split-authority rejection", err)
	}
}

func TestWorkerWorkflowActionApplyWorkItemExecutesAndFinalizes(t *testing.T) {
	w, claims, bridge, flow := newEventBackedWorkItemWorkerForTest(t, connector_workqueue.WorkTypeWorkflowActionApply, "workflow.action.execute", "event-workflow-action")

	result, err := w.executeClaim(context.Background(), nil, claimExecutionRequest{MaxClaims: 1})
	if err != nil {
		t.Fatalf("executeClaim workflow_action_apply WorkItem: %v", err)
	}
	if !result.Claimed || result.ClaimedRows != 1 {
		t.Fatalf("executeClaim result = %+v, want one claimed row", result)
	}
	if claims.claimCalls != 1 || claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("claim/acquire/renew calls = %d/%d/%d, want 1/0/0", claims.claimCalls, claims.acquireCalls, claims.renewCalls)
	}
	if claims.retryCalls != 0 {
		t.Fatalf("RetryWorkItem calls = %d, want no unsupported-kind retry", claims.retryCalls)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 {
		t.Fatalf("mutable Event bridge calls = processing:%v terminal:%v, want none", bridge.processingCalls, bridge.terminalCalls)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("complete states = %v, want [processed]", got)
	}
	if flow.calls != 1 {
		t.Fatalf("flow calls = %d, want workflow-action apply executed once", flow.calls)
	}
}

func TestWorkerRouteBackedHubSpotInboundWorkItemExecutesAndFinalizes(t *testing.T) {
	w, claims, bridge, flow := newEventBackedWorkItemWorkerForTest(t, connector_workqueue.WorkTypeConnectorEvent, "contact.propertychange", "event-contact-propertychange")

	result, err := w.executeClaim(context.Background(), nil, claimExecutionRequest{MaxClaims: 1})
	if err != nil {
		t.Fatalf("executeClaim route-backed connector_event WorkItem: %v", err)
	}
	if !result.Claimed || result.ClaimedRows != 1 {
		t.Fatalf("executeClaim result = %+v, want one claimed row", result)
	}
	if claims.claimCalls != 1 || claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("claim/acquire/renew calls = %d/%d/%d, want 1/0/0", claims.claimCalls, claims.acquireCalls, claims.renewCalls)
	}
	if claims.retryCalls != 0 {
		t.Fatalf("RetryWorkItem calls = %d, want no retry on ordinary route-backed inbound event", claims.retryCalls)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 {
		t.Fatalf("mutable Event bridge calls = processing:%v terminal:%v, want none", bridge.processingCalls, bridge.terminalCalls)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("complete states = %v, want [processed]", got)
	}
	if flow.calls != 1 {
		t.Fatalf("flow calls = %d, want route-backed inbound event executed once", flow.calls)
	}
}

func newEventBackedWorkItemWorkerForTest(t *testing.T, workType, eventKind, eventID string) (*Worker, *fakeWorkItemClaims, *fakeWorkItemEventBridge, *fakeFlowInvoker) {
	t.Helper()
	now := time.Now().UTC()
	leaseToken := "lease-" + eventID
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-" + eventID,
		EventID:          eventID,
		InstalledPackID:  "pack-test",
		WorkType:         workType,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       leaseToken,
		Attempt:          1,
		LastModifiedDate: now,
	}}
	event := noopEvent(eventID, eventKind)
	event.Status = "pending"
	event.ClaimToken = ""
	event.ClaimedAt = nil
	loader := &fakeWorkItemEventLoader{event: event}
	bridge := &fakeWorkItemEventBridge{}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, service, &fakeSyncState{}, fakeBatchRead{}, flow, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/" + eventKind: "NoopFlow__c",
	}})
	return w, claims, bridge, flow
}

func TestWorkerWorkItemClaimPathIgnoresRetiredSchedulerLeaseAPI(t *testing.T) {
	acquire := false
	claims := &fakeWorkItemClaims{acquireResult: &acquire}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	ev, ok, err := w.claimNext(context.Background(), false)
	if err != nil {
		t.Fatalf("claimNext: %v", err)
	}
	if ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v), want empty WorkItem queue", ev, ok)
	}
	if claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease calls acquire/renew = %d/%d, want none", claims.acquireCalls, claims.renewCalls)
	}
	if claims.claimCalls != 1 {
		t.Fatalf("ClaimNextWorkItem calls = %d, want 1", claims.claimCalls)
	}
}

func TestWorkerLegacyCustomClaimServiceRetainsSchedulerLeaseHooks(t *testing.T) {
	renew := true
	rawClaims := &fakeWorkItemClaims{
		legacyLeaseAuthority: true,
		renewResult:          &renew,
	}
	claims := &workItemClaimServiceWithoutRowCASCapability{WorkItemClaimService: rawClaims}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           &fakeWorkItemEventLoader{},
		WorkItemEventBridge:           &fakeWorkItemEventBridge{},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "stable-fence",
		WorkItemLeaseDuration:         time.Minute,
	}, newFakeEvents(), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	ev, ok, err := w.claimNext(context.Background(), false)
	if err != nil || ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v, %v), want empty custom queue after lease renewal", ev, ok, err)
	}
	if rawClaims.renewCalls != 1 || rawClaims.acquireCalls != 0 {
		t.Fatalf("legacy custom scheduler renew/acquire calls = %d/%d, want 1/0", rawClaims.renewCalls, rawClaims.acquireCalls)
	}
	if got := rawClaims.lastRenewInput.OwnerID; got != "scheduler-owner" {
		t.Fatalf("renew owner = %q, want scheduler-owner", got)
	}
	if got := rawClaims.lastRenewInput.FencingToken; got != "stable-fence" {
		t.Fatalf("renew fencing token = %q, want stable-fence", got)
	}
	if rawClaims.claimCalls != 1 {
		t.Fatalf("ClaimNextWorkItem calls = %d, want 1", rawClaims.claimCalls)
	}
	wantWorkTypes := []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
	if got := strings.Join(rawClaims.lastClaimWorkTypes, ","); got != strings.Join(wantWorkTypes, ",") {
		t.Fatalf("legacy custom claim work types = %q, want %q", got, strings.Join(wantWorkTypes, ","))
	}
	if rawClaims.lastDistributed {
		t.Fatal("legacy custom claim context carried distributed marker without opt-in")
	}
}

func TestWorkerLegacyDistributedOptInSkipsLeaseAndMarksClaims(t *testing.T) {
	acquire := false
	rawClaims := &fakeWorkItemClaims{
		legacyLeaseAuthority: true,
		acquireResult:        &acquire,
	}
	claims := &workItemClaimServiceWithoutRowCASCapability{WorkItemClaimService: rawClaims}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           &fakeWorkItemEventLoader{},
		WorkItemEventBridge:           &fakeWorkItemEventBridge{},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "stable-fence",
		WorkItemLeaseDuration:         time.Minute,
	}, newFakeEvents(), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	if w.DistributedWorkItemClaimsEnabled() {
		t.Fatal("legacy custom claim service reported distributed claims before opt-in")
	}

	w.UseDistributedWorkItemClaims()
	if !w.DistributedWorkItemClaimsEnabled() {
		t.Fatal("legacy custom claim service did not retain distributed opt-in")
	}

	ev, ok, err := w.claimNext(context.Background(), false)
	if err != nil || ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v, %v), want empty distributed custom queue", ev, ok, err)
	}
	if rawClaims.renewCalls != 0 || rawClaims.acquireCalls != 0 {
		t.Fatalf("scheduler renew/acquire calls = %d/%d, want none after distributed opt-in", rawClaims.renewCalls, rawClaims.acquireCalls)
	}
	if rawClaims.claimCalls != 1 || !rawClaims.lastDistributed {
		t.Fatalf("claim calls/distributed marker = %d/%v, want 1/true", rawClaims.claimCalls, rawClaims.lastDistributed)
	}
}

func TestWorkerWorkItemClaimDoesNotRequireDistributedOptIn(t *testing.T) {
	acquire := false
	claims := &fakeWorkItemClaims{acquireResult: &acquire}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		WorkItemSchedulerOwnerID:      "replica-b",
		WorkItemSchedulerFencingToken: "not-the-current-scheduler-fence",
		WorkItemLeaseDuration:         time.Minute,
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	ev, ok, err := w.claimNext(context.Background(), false)
	if err != nil || ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v, %v), want empty queue without scheduler conflict", ev, ok, err)
	}
	if claims.renewCalls != 0 || claims.acquireCalls != 0 {
		t.Fatalf("scheduler renew/acquire calls = %d/%d, want none", claims.renewCalls, claims.acquireCalls)
	}
	if claims.claimCalls != 1 || !claims.lastDistributed {
		t.Fatalf("WorkItem claim calls/compatibility distributed marker = %d/%v, want 1/true", claims.claimCalls, claims.lastDistributed)
	}
}

func TestWorkerWorkItemClaimForwardsStableCompatibilityIdentity(t *testing.T) {
	renew := true
	claims := &fakeWorkItemClaims{renewResult: &renew}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "stable-fence",
		WorkItemLeaseDuration:         time.Minute,
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	ev, ok, err := w.claimNext(context.Background(), false)
	if err != nil {
		t.Fatalf("claimNext: %v", err)
	}
	if ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v), want empty queue", ev, ok)
	}
	if claims.renewCalls != 0 || claims.acquireCalls != 0 {
		t.Fatalf("scheduler renew/acquire calls = %d/%d, want none", claims.renewCalls, claims.acquireCalls)
	}
	if got := claims.lastClaimInput.OwnerID; got != "scheduler-owner" {
		t.Fatalf("claim owner = %q, want stable scheduler owner", got)
	}
	if got := claims.lastClaimInput.FencingToken; got != "stable-fence" {
		t.Fatalf("claim compatibility fencing token = %q, want stable-fence", got)
	}
	if claims.claimCalls != 1 {
		t.Fatalf("ClaimNextWorkItem calls = %d, want 1 after scheduler renew", claims.claimCalls)
	}
}

func TestWorkerWorkItemClaimPathReleasesDueRetryWhenLegacyBridgeWins(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-bridge-lost",
		EventID:          "event-bridge-lost",
		InstalledPackID:  "pack",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-bridge-lost",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:               "event-bridge-lost",
		Namespace:        "ns",
		InstalledPackID:  "pack",
		EventKind:        "object.changed",
		QueueClass:       connector_workqueue.DefaultQueueClass,
		Status:           "pending",
		LastModifiedDate: now,
	}}
	bridge := &fakeWorkItemEventBridge{processingErr: connector_events.ErrLeaseLost}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, newFakeEvents(), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	ev, ok, err := w.claimNext(context.Background(), false)
	if !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("claimNext error = %v, want ErrLeaseLost", err)
	}
	if ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v), want no executable event after legacy bridge wins", ev, ok)
	}
	if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || claims.retryRunAfters[0].After(time.Now().UTC()) {
		t.Fatalf("RetryWorkItem calls/runAfter = %d/%v, want one immediately due retry", claims.retryCalls, claims.retryRunAfters)
	}
	if len(claims.completeCalls) != 0 {
		t.Fatalf("CompleteWorkItem calls = %v, want no terminal completion on legacy bridge collision", claims.completeCalls)
	}
	if got := bridge.processingCalls; len(got) != 1 || got[0] != "event-bridge-lost:lease-bridge-lost" {
		t.Fatalf("processing bridge calls = %v, want one attempted bridge with WorkItem lease", got)
	}
}

func TestWorkerWorkItemClaimPathReleasesDueRetryWhenLegacyEventAlreadyProcessing(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-legacy-processing",
		EventID:          "event-legacy-processing",
		InstalledPackID:  "pack",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-legacy-processing",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:               "event-legacy-processing",
		Namespace:        "ns",
		InstalledPackID:  "pack",
		EventKind:        "object.changed",
		QueueClass:       connector_workqueue.DefaultQueueClass,
		Status:           connector_workqueue.WorkStateProcessing,
		LastModifiedDate: now,
	}}
	bridge := &fakeWorkItemEventBridge{}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     "ns",
		PackInstallID:                 "pack",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	}, newFakeEvents(), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}

	ev, ok, err := w.claimNext(context.Background(), false)
	if !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("claimNext error = %v, want ErrLeaseLost", err)
	}
	if ok || ev != nil {
		t.Fatalf("claimNext = (%v, %v), want no executable event while legacy event is already processing", ev, ok)
	}
	if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || claims.retryRunAfters[0].After(time.Now().UTC()) {
		t.Fatalf("RetryWorkItem calls/runAfter = %d/%v, want one immediately due retry", claims.retryCalls, claims.retryRunAfters)
	}
	if len(claims.completeCalls) != 0 {
		t.Fatalf("CompleteWorkItem calls = %v, want no terminal completion on legacy processing collision", claims.completeCalls)
	}
	if len(bridge.processingCalls) != 0 {
		t.Fatalf("processing bridge calls = %v, want none after observing fresh legacy processing state", bridge.processingCalls)
	}
}

func TestWorkerWorkItemClaimPathPreservesMultiClassQueueFilter(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-classes",
		EventID:          "event-classes",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_inbound_routing.QueueClassInteractive,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-classes",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-classes",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_inbound_routing.QueueClassInteractive,
		Payload:         noopPayload("noop_object", "src-event-classes"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	service := newCompleteWorkItemServiceFake(claims, loader, nil)
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-classes",
		QueueClasses:                  []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	if _, ok, err := w.claimNext(context.Background(), false); err != nil || !ok {
		t.Fatalf("claimNext = ok:%v err:%v, want WorkItem claim", ok, err)
	}
	if got := strings.Join(claims.lastClaimInput.QueueClasses, ","); got != "interactive,default" {
		t.Fatalf("WorkItem QueueClasses = %q, want interactive,default", got)
	}
}

func TestWorkItemClaimEntryPointsCarryPreferredQueueTier(t *testing.T) {
	allowed := []string{
		connector_inbound_routing.QueueClassInteractive,
		connector_inbound_routing.QueueClassDefault,
		connector_inbound_routing.QueueClassBulk,
	}
	preferred := []string{
		connector_inbound_routing.QueueClassInteractive,
		connector_inbound_routing.QueueClassDefault,
	}
	tests := []struct {
		name string
		call func(*workItemEventsService) error
	}{
		{
			name: "direct claim",
			call: func(svc *workItemEventsService) error {
				_, _, err := svc.ClaimNextForClasses(context.Background(), "pack-test", "worker-test", allowed)
				return err
			},
		},
		{
			name: "ordinary peek",
			call: func(svc *workItemEventsService) error {
				_, err := svc.PeekBatchForClasses(context.Background(), "pack-test", allowed, 1)
				return err
			},
		},
		{
			name: "scheduler peek",
			call: func(svc *workItemEventsService) error {
				_, err := svc.peekClaimCandidatesForScheduler(context.Background(), "pack-test", allowed, preferred, 1, false)
				return err
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			claims := &fakeWorkItemClaims{}
			service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
			svc := newMandatoryWorkItemEventsService(service, WorkerConfig{
				Namespace:                     workerTestNS,
				PackInstallID:                 "pack-test",
				WorkerID:                      "worker-test",
				QueueClasses:                  allowed,
				WorkItemSchedulerOwnerID:      "scheduler-owner",
				WorkItemSchedulerFencingToken: "fence",
			})
			svc.setPreferredQueueClasses(preferred)
			if err := tc.call(svc); err != nil {
				t.Fatalf("entry point: %v", err)
			}
			if got := strings.Join(claims.lastPreference, ","); got != "interactive,default" {
				t.Fatalf("claim queue-class preference = %q, want interactive,default", got)
			}
			wantWorkTypes := []string{
				connector_workqueue.WorkTypeConnectorEvent,
				connector_workqueue.WorkTypeWorkflowActionApply,
			}
			if got := strings.Join(claims.lastClaimWorkTypes, ","); got != strings.Join(wantWorkTypes, ",") {
				t.Fatalf("claim work types = %q, want %q", got, strings.Join(wantWorkTypes, ","))
			}
		})
	}
}

func TestWorkItemClaimFiltersPreferenceToAllowedClasses(t *testing.T) {
	claims := &fakeWorkItemClaims{}
	service := newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{}, nil)
	svc := newMandatoryWorkItemEventsService(service, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-test",
		QueueClasses:                  []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault, connector_inbound_routing.QueueClassBulk},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	})
	svc.setPreferredQueueClasses([]string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault})

	_, _, err := svc.ClaimNextForClasses(context.Background(), "pack-test", "worker-test", []string{connector_inbound_routing.QueueClassBulk})
	if err != nil {
		t.Fatalf("ClaimNextForClasses: %v", err)
	}
	if len(claims.lastPreference) != 0 {
		t.Fatalf("claim queue-class preference = %v, want empty intersection for bulk-only claim", claims.lastPreference)
	}
}

func TestWorkerWorkflowActionWorkItemBatchCollectionCapsClaimLimitAndFinalizes(t *testing.T) {
	now := time.Now().UTC()
	works := make([]*connector_workqueue.WorkItem, 0, 3)
	events := make(map[string]*connector_events.ConnectorEvent, 3)
	eventResults := make([]any, 0, 3)
	for i := 0; i < 3; i++ {
		eventID := "event-batch-" + strconv.Itoa(i)
		works = append(works, &connector_workqueue.WorkItem{
			WorkID:           "work-batch-" + strconv.Itoa(i),
			EventID:          eventID,
			InstalledPackID:  "pack-test",
			WorkType:         connector_workqueue.WorkTypeWorkflowActionApply,
			QueueClass:       connector_workqueue.DefaultQueueClass,
			State:            connector_workqueue.WorkStateProcessing,
			LeaseToken:       "lease-batch-" + strconv.Itoa(i),
			Attempt:          1,
			LastModifiedDate: now,
		})
		events[eventID] = &connector_events.ConnectorEvent{
			ID:              eventID,
			Namespace:       workerTestNS,
			InstalledPackID: "pack-test",
			EventKind:       "noop.created",
			QueueClass:      connector_workqueue.DefaultQueueClass,
			Payload:         noopPayload("noop_object", "src-event-batch-"+strconv.Itoa(i)),
			RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
			CreatedDate:     now.Add(time.Duration(i) * time.Microsecond),
		}
		eventResults = append(eventResults, map[string]any{"eventID": eventID, "success": true})
	}
	claims := &fakeWorkItemClaims{works: works}
	loader := &fakeWorkItemEventLoader{events: events}
	flow := &fakeFlowInvoker{successResult: &FlowResult{
		IsSuccess: true,
		OutputValues: map[string]any{
			"eventResults": eventResults,
		},
	}}
	service := newCompleteWorkItemServiceFake(claims, loader, &fakeWorkItemEventBridge{})
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-batch",
		BatchSize:                     32,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, fakeBatchRead{}, flow, batchCollectionResolver(32))
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	claimed, rows, err := w.processOneWithDrain(context.Background(), nil, w.cfg.BatchSize, false)
	if err != nil {
		t.Fatalf("processOneWithDrain: %v", err)
	}
	if !claimed || rows != 3 {
		t.Fatalf("processOneWithDrain claimed=%v rows=%d, want three batched WorkItems", claimed, rows)
	}
	if flow.calls != 1 {
		t.Fatalf("flow calls = %d, want one batch flow", flow.calls)
	}
	if got := claims.completeCalls; len(got) != 3 {
		t.Fatalf("complete states = %v, want three processed completions", got)
	}
	if got := claims.claimByEventTypes; strings.Join(got, ",") != "workflow_action_apply,workflow_action_apply" {
		t.Fatalf("ClaimByID WorkItem types = %v, want only peeked workflow_action_apply candidates", got)
	}
	workItemEvents, ok := w.events.(*workItemEventsService)
	if !ok {
		t.Fatalf("worker events = %T, want WorkItem-backed events service", w.events)
	}
	workItemEvents.mu.Lock()
	peekedLen := len(workItemEvents.peeked)
	workItemEvents.mu.Unlock()
	if peekedLen != 0 {
		t.Fatalf("peeked work type cache len = %d, want 0 after batch assembly", peekedLen)
	}
}

func TestWorkItemClaimTypesWithoutPeekCoverBothEventBackedIdentities(t *testing.T) {
	svc := &workItemEventsService{
		peeked: make(map[string]string),
	}
	got := svc.claimWorkTypesForEvent("event-without-peek")
	want := []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
	if strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("claim WorkTypes without peek = %v, want %v", got, want)
	}
}

func TestWorkerWorkItemSchedulerPermitCapsLimiterMaxClaims(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-cap",
		EventID:          "event-cap",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-cap",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: noopEvent("event-cap", "noop.created")}
	limiter := &recordingClaimLimiter{decision: ClaimLimiterDecision{Allow: true, MaxClaims: 32}}
	service := newCompleteWorkItemServiceFake(claims, loader, nil)
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-cap",
		BatchSize:                     32,
		ClaimLimiter:                  limiter,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	decision, allowed, _, err := w.requestClaimPermitForScheduler(context.Background(), w.cfg.BatchSize)
	if err != nil || !allowed {
		t.Fatalf("requestClaimPermitForScheduler = (%+v, %v, %v), want allowed", decision, allowed, err)
	}
	if decision.MaxClaims != 32 {
		t.Fatalf("WorkItem scheduler MaxClaims = %d, want batch-compatible permit width 32", decision.MaxClaims)
	}
	if got := limiter.requests[0].BatchSize; got != 32 {
		t.Fatalf("limiter request BatchSize = %d, want configured batch size 32", got)
	}
}

func TestWorkItemClaimedRefOwnership(t *testing.T) {
	now := time.Now().UTC()
	eventID := "event-owned-ref"
	workID := "work-owned-ref"
	event := &connector_events.ConnectorEvent{
		ID:              eventID,
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStatePending,
		Payload:         noopPayload("noop_object", "src-owned-ref"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}
	claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{{
		WorkID:          workID,
		EventID:         eventID,
		InstalledPackID: "pack-test",
		WorkType:        connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:      connector_workqueue.DefaultQueueClass,
		State:           connector_workqueue.WorkStatePending,
		RunAfter:        now.Add(-time.Second),
		Attempt:         1,
		CreatedDate:     now.Add(-time.Second),
	}}}
	loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{eventID: event}}
	bridge := &fakeWorkItemEventBridge{}
	cfgA := WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-owner-a",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}
	cfgB := cfgA
	cfgB.WorkerID = "worker-owner-b"
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	ownerA := newMandatoryWorkItemEventsService(service, cfgA)
	ownerB := newMandatoryWorkItemEventsService(service, cfgB)

	token, _, ok, err := ownerA.ClaimByID(context.Background(), eventID, cfgA.WorkerID, cfgA.PackInstallID, sql.NullString{}, 0)
	if err != nil || !ok || token == "" {
		t.Fatalf("ownerA ClaimByID token=%q ok=%t err=%v, want claimed", token, ok, err)
	}
	if err := ownerB.MarkProcessed(context.Background(), eventID, token); !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("ownerB MarkProcessed error = %v, want ErrLeaseLost", err)
	}
	if err := ownerA.MarkProcessed(context.Background(), eventID, token); err != nil {
		t.Fatalf("ownerA MarkProcessed: %v", err)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("complete calls = %v, want one processed completion", got)
	}
}

func TestWorkItemClaimByIDDefersFailedFutureBackoffMirror(t *testing.T) {
	now := time.Now().UTC()
	future := now.Add(10 * time.Minute)
	eventID := "event-failed-future"
	claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{{
		WorkID:           "work-failed-future",
		EventID:          eventID,
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStatePending,
		RunAfter:         now.Add(-time.Minute),
		Attempt:          2,
		CreatedDate:      now.Add(-time.Minute),
		LastModifiedDate: now.Add(-time.Minute),
	}}}
	loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{eventID: {
		ID:              eventID,
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          "failed",
		NextAttemptAt:   &future,
		Payload:         noopPayload("noop_object", "src-failed-future"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now.Add(-time.Minute),
	}}}
	bridge := &fakeWorkItemEventBridge{}
	svc := newWorkItemEventsService(newFakeEvents(), WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	})

	token, _, ok, err := svc.ClaimByID(context.Background(), eventID, "worker", "pack-test", sql.NullString{}, 0)
	if err != nil {
		t.Fatalf("ClaimByID: %v", err)
	}
	if ok || token != "" {
		t.Fatalf("ClaimByID token=%q ok=%t, want no executable claim while event mirror backs off", token, ok)
	}
	if len(bridge.processingCalls) != 0 {
		t.Fatalf("processing bridge calls = %v, want none while event mirror backs off", bridge.processingCalls)
	}
	if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || !claims.retryRunAfters[0].Equal(future) {
		t.Fatalf("RetryWorkItem calls/runAfter = %d/%v, want one retry at event NextAttemptAt %v", claims.retryCalls, claims.retryRunAfters, future)
	}
}

type hostileWorkItemClaimFixture struct {
	worker    *Worker
	adapter   *workItemEventsService
	claims    *fakeWorkItemClaims
	bridge    *fakeWorkItemEventBridge
	rawEvent  *connector_events.ConnectorEvent
	workID    string
	claimedAt time.Time
}

func newHostileWorkItemClaimFixture(t *testing.T, executorStateCompatibility bool, workState string) hostileWorkItemClaimFixture {
	t.Helper()
	now := time.Now().UTC().Truncate(time.Microsecond)
	claimedAt := now.Add(-time.Minute)
	legacyClaimedAt := now.Add(-time.Hour)
	legacyErrorAt := now.Add(-30 * time.Minute)
	legacyProcessedAt := now.Add(-2 * time.Hour)
	future := now.Add(time.Hour)
	eventID := "event-hostile-workitem-projection"
	workID := "work-hostile-workitem-projection"
	rawEvent := &connector_events.ConnectorEvent{
		ID:               eventID,
		Namespace:        workerTestNS,
		InstalledPackID:  "pack-test",
		EventKind:        "noop.created",
		QueueClass:       connector_inbound_routing.QueueClassBulk,
		Status:           connector_workqueue.WorkStateDead,
		Payload:          noopPayload("noop_object", "src-hostile-projection"),
		RouteVersion:     sql.NullInt64{Valid: true, Int64: 1},
		RetryCount:       99,
		LastError:        "legacy event lifecycle must not execute",
		LastErrorAt:      &legacyErrorAt,
		NextAttemptAt:    &future,
		ClaimedBy:        "legacy-worker",
		ClaimedAt:        &legacyClaimedAt,
		ClaimToken:       "legacy-claim-token",
		ProcessedAt:      &legacyProcessedAt,
		CreatedDate:      now.Add(-3 * time.Hour),
		LastModifiedDate: legacyProcessedAt,
	}
	claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{{
		WorkID:           workID,
		EventID:          eventID,
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_inbound_routing.QueueClassBulk,
		State:            workState,
		Generation:       7,
		RunAfter:         now.Add(-time.Minute),
		CreatedDate:      now.Add(-2 * time.Minute),
		LastModifiedDate: claimedAt,
	}}}
	loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{eventID: rawEvent}}
	bridge := &fakeWorkItemEventBridge{}
	cfg := mandatoryWorkerTestConfig()
	cfg.QueueClasses = []string{connector_inbound_routing.QueueClassBulk}
	cfg.WorkItemGeneration = 7

	var worker *Worker
	if executorStateCompatibility {
		claims.executorStateAuthority = true
		cfg.ClaimPath = ClaimPathWorkItem
		cfg.WorkItemClaims = claims
		cfg.WorkItemEventLoader = loader
		cfg.WorkItemEventBridge = bridge
		var err error
		worker, err = newWorkerForTestWithNoopBatchRead(cfg, newFakeEvents(), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
		if err != nil {
			t.Fatalf("NewWorkerForTest executor-state compatibility lane: %v", err)
		}
	} else {
		worker = newMandatoryWorkerForTest(t, cfg, newCompleteWorkItemServiceFake(claims, loader, bridge))
	}
	adapter, ok := worker.events.(*workItemEventsService)
	if !ok {
		t.Fatalf("worker events = %T, want WorkItem adapter", worker.events)
	}
	return hostileWorkItemClaimFixture{
		worker: worker, adapter: adapter, claims: claims, bridge: bridge,
		rawEvent: rawEvent, workID: workID, claimedAt: claimedAt,
	}
}

func assertNormalizedWorkItemClaimProjection(t *testing.T, ev *connector_events.ConnectorEvent, workerID string, claimedAt time.Time, leaseToken string) {
	t.Helper()
	if ev == nil {
		t.Fatal("claimed Event is nil")
	}
	if ev.Status != connector_workqueue.WorkStateProcessing || ev.ClaimedBy != workerID || ev.ClaimToken != leaseToken {
		t.Fatalf("claimed Event lifecycle = status:%q worker:%q token:%q, want processing/%q/%q", ev.Status, ev.ClaimedBy, ev.ClaimToken, workerID, leaseToken)
	}
	if ev.ClaimedAt == nil || !ev.ClaimedAt.Equal(claimedAt) || !ev.LastModifiedDate.Equal(claimedAt) {
		t.Fatalf("claimed Event timestamps = claimed_at:%v last_modified:%v, want both %v", ev.ClaimedAt, ev.LastModifiedDate, claimedAt)
	}
	if ev.RetryCount != 0 || ev.LastError != "" || ev.LastErrorAt != nil || ev.NextAttemptAt != nil || ev.ProcessedAt != nil {
		t.Fatalf("claimed Event retained stale lifecycle state: %+v", ev)
	}
}

func TestWorkItemSchedulerClaimUsesNormalizedEventProjection(t *testing.T) {
	for _, tc := range []struct {
		name                       string
		executorStateCompatibility bool
	}{
		{name: "mandatory work item service"},
		{name: "executor-state compatibility lane", executorStateCompatibility: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			fixture := newHostileWorkItemClaimFixture(t, tc.executorStateCompatibility, connector_workqueue.WorkStateProcessing)
			if tc.executorStateCompatibility && (fixture.adapter.workItems != nil || !fixture.adapter.workItemExecutorStateEnabled()) {
				t.Fatalf("compatibility adapter authority = work_items:%v executor_state:%v, want nil/true", fixture.adapter.workItems, fixture.adapter.workItemExecutorStateEnabled())
			}
			candidates, err := fixture.worker.peekClaimCandidatesForScheduler(context.Background(), 1, true)
			if err != nil || len(candidates) != 1 {
				t.Fatalf("peekClaimCandidatesForScheduler = (%v, %v), want one candidate", candidates, err)
			}
			claimed, ok, err := fixture.worker.claimByIDForScheduler(context.Background(), candidates[0])
			if err != nil || !ok {
				t.Fatalf("claimByIDForScheduler = (%+v, %v, %v), want claimed projection", claimed, ok, err)
			}
			assertNormalizedWorkItemClaimProjection(t, claimed, fixture.worker.cfg.WorkerID, fixture.claimedAt, fixture.claims.lastClaimByEvent.LeaseToken)
			if fixture.claims.claimByEventCalls != 1 || fixture.claims.staleClaimByEventCalls != 1 || fixture.claims.lastClaimByEvent.Generation != 7 {
				t.Fatalf("WorkItem scheduler claim calls/stale/generation = %d/%d/%d, want 1/1/7", fixture.claims.claimByEventCalls, fixture.claims.staleClaimByEventCalls, fixture.claims.lastClaimByEvent.Generation)
			}
			if fixture.claims.acquireCalls != 0 || fixture.claims.renewCalls != 0 {
				t.Fatalf("SchedulerLease acquire/renew calls = %d/%d, want 0/0 for row-CAS authority", fixture.claims.acquireCalls, fixture.claims.renewCalls)
			}
			if len(fixture.bridge.processingCalls) != 0 || len(fixture.bridge.terminalCalls) != 0 || len(fixture.bridge.retryCalls) != 0 {
				t.Fatalf("Event bridge calls = processing:%v terminal:%v retry:%v, want none", fixture.bridge.processingCalls, fixture.bridge.terminalCalls, fixture.bridge.retryCalls)
			}
			if fixture.rawEvent.Status != connector_workqueue.WorkStateDead || fixture.rawEvent.ClaimToken != "legacy-claim-token" {
				t.Fatalf("peeked immutable Event was mutated: %+v", fixture.rawEvent)
			}
			ref, refErr := fixture.adapter.claimedRef(claimed.ID, claimed.ClaimToken)
			if refErr != nil || ref.workID != fixture.workID || ref.generation != 7 {
				t.Fatalf("claimed WorkItem ref = %+v, %v; want work %q generation 7", ref, refErr, fixture.workID)
			}
		})
	}
}

func TestMandatoryWorkItemBatchCandidateUsesNormalizedEventProjection(t *testing.T) {
	fixture := newHostileWorkItemClaimFixture(t, false, connector_workqueue.WorkStatePending)
	peeked, err := fixture.worker.peekBatch(context.Background(), 1, false)
	if err != nil || len(peeked) != 1 {
		t.Fatalf("peekBatch = (%v, %v), want one candidate", peeked, err)
	}
	claimed, ok, err := fixture.worker.claimBatchCandidate(context.Background(), peeked[0])
	if err != nil || !ok {
		t.Fatalf("claimBatchCandidate = (%+v, %v, %v), want claimed projection", claimed, ok, err)
	}
	assertNormalizedWorkItemClaimProjection(t, claimed, fixture.worker.cfg.WorkerID, fixture.claimedAt, fixture.claims.lastClaimByEvent.LeaseToken)
	if fixture.claims.claimByEventCalls != 1 || fixture.claims.staleClaimByEventCalls != 0 || fixture.claims.lastClaimByEvent.Generation != 7 {
		t.Fatalf("WorkItem batch claim calls/stale/generation = %d/%d/%d, want 1/0/7", fixture.claims.claimByEventCalls, fixture.claims.staleClaimByEventCalls, fixture.claims.lastClaimByEvent.Generation)
	}
	if fixture.claims.acquireCalls != 0 || fixture.claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease acquire/renew calls = %d/%d, want 0/0 for row-CAS authority", fixture.claims.acquireCalls, fixture.claims.renewCalls)
	}
	if fixture.rawEvent.Status != connector_workqueue.WorkStateDead || fixture.rawEvent.ClaimToken != "legacy-claim-token" {
		t.Fatalf("peeked immutable Event was mutated: %+v", fixture.rawEvent)
	}
}

func TestWorkItemClaimByIDForClassesUsesPeekedGenerationAndQueueClasses(t *testing.T) {
	now := time.Now().UTC()
	eventID := "event-peeked-generation"
	claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{{
		WorkID:           "work-peeked-generation",
		EventID:          eventID,
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_inbound_routing.QueueClassBulk,
		State:            connector_workqueue.WorkStatePending,
		RunAfter:         now.Add(-time.Minute),
		Attempt:          1,
		Generation:       7,
		CreatedDate:      now.Add(-time.Minute),
		LastModifiedDate: now.Add(-time.Minute),
	}}}
	loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{eventID: {
		ID:              eventID,
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_inbound_routing.QueueClassBulk,
		Status:          "pending",
		Payload:         noopPayload("noop_object", "src-peeked-generation"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now.Add(-time.Minute),
	}}}
	svc := newMandatoryWorkItemEventsService(newCompleteWorkItemServiceFake(claims, loader, nil), WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-bulk",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
		WorkItemLeaseDuration:         time.Minute,
	})

	token, _, ok, err := svc.ClaimByIDForClasses(context.Background(), eventID, "worker-bulk", "pack-test", sql.NullString{}, 0, []string{connector_inbound_routing.QueueClassBulk}, 7, true)
	if err != nil || !ok || token == "" {
		t.Fatalf("ClaimByIDForClasses token=%q ok=%t err=%v, want claimed", token, ok, err)
	}
	if claims.resolveCalls != 0 {
		t.Fatalf("ResolveClaimableWorkItemGeneration calls = %d, want 0 for already-peeked generation", claims.resolveCalls)
	}
	if got := claims.lastClaimByEvent.Generation; got != 7 {
		t.Fatalf("ClaimWorkItemByEvent generation = %d, want peeked generation 7", got)
	}
}

func TestWorkItemClaimByIDForClassesWithModePreservesStalePeek(t *testing.T) {
	now := time.Now().UTC()
	eventID := "event-stale-peek"
	claims := &fakeWorkItemClaims{works: []*connector_workqueue.WorkItem{{
		WorkID: "work-stale-peek", EventID: eventID, InstalledPackID: "pack-test",
		WorkType: connector_workqueue.WorkTypeConnectorEvent, QueueClass: connector_inbound_routing.QueueClassBulk,
		State: connector_workqueue.WorkStateProcessing, Generation: 7, LeaseUntil: sql.NullTime{Time: now.Add(-time.Minute), Valid: true},
	}}}
	loader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{eventID: {
		ID: eventID, Namespace: workerTestNS, InstalledPackID: "pack-test", EventKind: "noop.created",
		QueueClass: connector_inbound_routing.QueueClassBulk, Status: "pending", Payload: noopPayload("noop_object", "src-stale-peek"),
	}}}
	svc := newMandatoryWorkItemEventsService(newCompleteWorkItemServiceFake(claims, loader, nil), WorkerConfig{
		Namespace: workerTestNS, PackInstallID: "pack-test", WorkerID: "worker-bulk",
		WorkItemSchedulerOwnerID: "scheduler-owner", WorkItemSchedulerFencingToken: "fence", WorkItemLeaseDuration: time.Minute,
	})
	token, _, ok, err := svc.ClaimByIDForClassesWithMode(context.Background(), eventID, "worker-bulk", "pack-test", sql.NullString{}, 0, []string{connector_inbound_routing.QueueClassBulk}, 7, true, true)
	if err != nil || !ok || token == "" {
		t.Fatalf("ClaimByIDForClassesWithMode token=%q ok=%t err=%v, want stale claim", token, ok, err)
	}
	if claims.staleClaimByEventCalls != 1 {
		t.Fatalf("ClaimStaleWorkItemByEvent calls = %d, want 1", claims.staleClaimByEventCalls)
	}
}

func TestWorkItemSchedulerPeekBatchHydratesEventsOnce(t *testing.T) {
	now := time.Now().UTC()
	works := []*connector_workqueue.WorkItem{
		{WorkID: "work-1", EventID: "event-1", WorkType: connector_workqueue.WorkTypeConnectorEvent, Generation: 7, RunAfter: now},
		{WorkID: "work-2", EventID: "event-2", WorkType: connector_workqueue.WorkTypeConnectorEvent, Generation: 7, RunAfter: now},
	}
	claims := &fakeWorkItemClaims{works: works}
	baseLoader := &fakeWorkItemEventLoader{events: map[string]*connector_events.ConnectorEvent{
		"event-1": {ID: "event-1", InstalledPackID: "pack-test", EventKind: "noop.skip", Status: "pending"},
		"event-2": {ID: "event-2", InstalledPackID: "pack-test", EventKind: "noop.skip", Status: "pending"},
	}}
	loader := &fakeBatchWorkItemEventLoader{fakeWorkItemEventLoader: baseLoader}
	svc := newMandatoryWorkItemEventsService(newCompleteWorkItemServiceFake(claims, loader, nil), WorkerConfig{
		PackInstallID: "pack-test", WorkItemGeneration: 7,
		WorkItemSchedulerOwnerID: "scheduler-owner", WorkItemSchedulerFencingToken: "fence", WorkItemLeaseDuration: time.Minute,
	})
	candidates, err := svc.peekClaimCandidatesForScheduler(context.Background(), "pack-test", []string{connector_inbound_routing.QueueClassBulk}, nil, 2, false)
	if err != nil {
		t.Fatalf("peekClaimCandidatesForScheduler: %v", err)
	}
	if len(candidates) != 2 || loader.batchCalls != 1 || baseLoader.calls != 0 {
		t.Fatalf("candidates=%d batch_calls=%d single_calls=%d, want 2/1/0", len(candidates), loader.batchCalls, baseLoader.calls)
	}
}

func TestWorkerWorkItemSchedulerSoftDBThrottlePreserved(t *testing.T) {
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-progress",
		BatchSize:                     32,
		ClaimLimiter:                  &recordingClaimLimiter{decision: ClaimLimiterDecision{Allow: false, Reason: "db_pool_waiting"}},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, newCompleteWorkItemServiceFake(&fakeWorkItemClaims{}, &fakeWorkItemEventLoader{}, nil), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	decision, allowed, _, err := w.requestClaimPermitForScheduler(context.Background(), w.cfg.BatchSize)
	if err != nil {
		t.Fatalf("requestClaimPermitForScheduler err = %v", err)
	}
	if allowed || decision.Allow {
		t.Fatalf("decision = %+v allowed=%v, want soft DB throttle preserved", decision, allowed)
	}
	if decision.Reason != "db_pool_waiting" {
		t.Fatalf("decision = %+v, want db_pool_waiting reason preserved", decision)
	}
}

func TestWorkerWorkItemSchedulerDBPoolExhaustedPreserved(t *testing.T) {
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-db-exhausted-progress",
		BatchSize:                     32,
		ClaimLimiter:                  &recordingClaimLimiter{decision: ClaimLimiterDecision{Allow: false, Reason: "db_pool_exhausted"}},
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, newCompleteWorkItemServiceFake(&fakeWorkItemClaims{}, &fakeWorkItemEventLoader{}, nil), &fakeSyncState{}, nil, &fakeFlowInvoker{}, &stubRouteResolver{})
	decision, allowed, _, err := w.requestClaimPermitForScheduler(context.Background(), w.cfg.BatchSize)
	if err != nil {
		t.Fatalf("requestClaimPermitForScheduler err = %v", err)
	}
	if allowed || decision.Allow {
		t.Fatalf("decision = %+v allowed=%v, want DB-exhausted WorkItem throttle preserved", decision, allowed)
	}
	if decision.Reason != "db_pool_exhausted" {
		t.Fatalf("decision = %+v, want db_pool_exhausted reason preserved", decision)
	}
}

func TestWorkerWorkItemExecutorAuthorityLatestBySourceCoalescesSibling(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-latest",
		EventID:          "event-latest",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-latest",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-latest",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Payload:         noopPayload("noop_object", "src-event-latest"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	bridge := &fakeWorkItemEventBridge{coalescedClaim: &connector_events.WorkItemCoalescedSiblingClaim{
		Work: &connector_workqueue.WorkItem{
			WorkID:           "work-latest-next",
			EventID:          "event-latest-next",
			InstalledPackID:  "pack-test",
			WorkType:         connector_workqueue.WorkTypeConnectorEvent,
			QueueClass:       connector_workqueue.DefaultQueueClass,
			State:            connector_workqueue.WorkStateProcessing,
			Attempt:          2,
			LastModifiedDate: now,
		},
		Event: &connector_events.ConnectorEvent{
			ID:              "event-latest-next",
			Namespace:       workerTestNS,
			InstalledPackID: "pack-test",
			EventKind:       "noop.created",
			QueueClass:      connector_workqueue.DefaultQueueClass,
			Payload:         noopPayload("noop_object", "src-event-latest"),
			RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
			CreatedDate:     now.Add(time.Second),
		},
	}}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	resolver := &stubRouteResolver{specs: map[string]connector_inbound_routing.RouteExecutionSpec{
		workerTestNS + "/noop.created": {
			FlowDeveloperName: "NoopLatestFlow",
			InvocationMode:    connector_inbound_routing.InvocationModeLatestBySource,
		},
	}}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	w, err := NewWorkItemWorkerForTest(context.Background(), WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-latest",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, fakeBatchRead{}, flow, resolver)
	if err != nil {
		t.Fatalf("NewWorkItemWorkerForTest: %v", err)
	}
	claimed, rows, err := w.processOneWithDrain(context.Background(), nil, w.cfg.BatchSize, false)
	if err != nil {
		t.Fatalf("processOneWithDrain: %v", err)
	}
	if !claimed || rows != 2 {
		t.Fatalf("processOneWithDrain claimed=%v rows=%d, want current coalesced plus sibling executed", claimed, rows)
	}
	if flow.calls != 1 {
		t.Fatalf("flow calls = %d, want 1 coalesced execution", flow.calls)
	}
	if !w.DistributedWorkItemClaimsEnabled() || claims.claimCalls != 1 || !claims.lastDistributed {
		t.Fatalf("mandatory WorkItem claim path = enabled:%v calls:%d distributed:%v, want one row-CAS claim", w.DistributedWorkItemClaimsEnabled(), claims.claimCalls, claims.lastDistributed)
	}
	if claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease calls acquire/renew = %d/%d, want native WorkItem row-CAS only", claims.acquireCalls, claims.renewCalls)
	}
	if got := claims.lastClaimInput.OwnerID; got != "scheduler-owner" {
		t.Fatalf("WorkItem claim owner = %q, want scheduler-owner", got)
	}
	if got := claims.lastClaimInput.FencingToken; got != "fence" {
		t.Fatalf("WorkItem claim fence = %q, want fence", got)
	}
	if claims.retryCalls != 0 || len(claims.failInputs) != 0 || len(claims.completeCalls) != 1 ||
		claims.completeCalls[0] != connector_workqueue.WorkStateProcessed || len(claims.completeWorkIDs) != 1 || claims.completeWorkIDs[0] != "work-latest-next" {
		t.Fatalf("WorkItem retry/failure/complete calls = %d/%d/%v (%v), want coalesced sibling processed only", claims.retryCalls, len(claims.failInputs), claims.completeCalls, claims.completeWorkIDs)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 || len(bridge.retryCalls) != 0 {
		t.Fatalf("legacy Event bridge mutations = processing:%v terminal:%v retry:%v, want none", bridge.processingCalls, bridge.terminalCalls, bridge.retryCalls)
	}
	if got := bridge.coalesceCalls; len(got) != 2 || got[0] != "event-latest:work-latest:lease-latest" || !strings.HasPrefix(got[1], "event-latest-next:work-latest-next:") {
		t.Fatalf("coalesce bridge calls = %v, want current bridge claim then empty sibling probe", got)
	}
	if got := bridge.coalesceOwners; len(got) != 2 || got[0] != "scheduler-owner" || got[1] != "scheduler-owner" {
		t.Fatalf("coalesce owners = %v, want stable scheduler owner", got)
	}
	if got := bridge.coalesceFences; len(got) != 2 || got[0] != "fence" || got[1] != "fence" {
		t.Fatalf("coalesce fencing tokens = %v, want stable fence", got)
	}
}

func TestWorkerWorkItemSettledProjectionRestoresCrossDeliveryGenerationPayloads(t *testing.T) {
	now := time.Now().UTC()
	first := propertyChangeEvent("delivery-first", "firstname")
	latest := propertyChangeEvent("delivery-latest", "country_picklist__sfdc_")
	for _, event := range []*connector_events.ConnectorEvent{first, latest} {
		event.SourceLaneKey = "source-lane-contact-123"
	}
	first.DeliveryId = "delivery-a"
	first.DeliverySequence = 0
	latest.DeliveryId = "delivery-b"
	latest.DeliverySequence = 11
	latest.Status = connector_workqueue.WorkStatePending
	latest.ClaimToken = ""
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "csp:contact-123",
		EventID:          latest.ID,
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-settled",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: latest, settledProjectionPayloads: map[string][]*connector_events.ConnectorEvent{
		claims.work.WorkID: {first, latest},
	}}
	flow := &capturingFlowInvoker{}
	service := newCompleteWorkItemServiceFake(claims, loader, nil)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-settled",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, &returningBatchRead{result: map[string]json.RawMessage{
		"contact-123": json.RawMessage(`{"id":"contact-123"}`),
	}}, flow, &stubRouteResolver{specs: map[string]connector_inbound_routing.RouteExecutionSpec{
		workerTestNS + "/contact.propertychange": {
			FlowDeveloperName: "DeltaFullRecordFlow",
			InvocationMode:    connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
		},
	}})

	claimed, err := w.processOne(context.Background())
	if err != nil {
		t.Fatalf("processOne: %v", err)
	}
	if !claimed || len(flow.capturedArgs) != 1 {
		t.Fatalf("claimed/flow calls = %v/%d, want true/1", claimed, len(flow.capturedArgs))
	}
	coalesced, ok := flow.capturedArgs[0]["coalescedPayloads"].([]any)
	if !ok || len(coalesced) != 2 {
		t.Fatalf("coalescedPayloads = %#v, want two delivery payloads", flow.capturedArgs[0]["coalescedPayloads"])
	}
	for i, want := range []string{"firstname", "country_picklist__sfdc_"} {
		payload, ok := coalesced[i].(map[string]any)["payload"].(map[string]any)
		if !ok || payload["propertyName"] != want {
			t.Fatalf("coalescedPayloads[%d] = %#v, want propertyName %q", i, coalesced[i], want)
		}
	}
}

func TestWorkerWorkItemSkipsLegacyTerminalMirror(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-already-processed",
		EventID:          "event-already-processed",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-already-processed",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-already-processed",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStateProcessed,
		Payload:         noopPayload("noop_object", "src-event-already-processed"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	bridge := &fakeWorkItemEventBridge{}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-terminal-mirror",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, newFakeEvents(), &fakeSyncState{}, fakeBatchRead{}, flow, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	claimed, rows, err := w.processOneWithDrain(context.Background(), nil, w.cfg.BatchSize, false)
	if err != nil {
		t.Fatalf("processOneWithDrain: %v", err)
	}
	if claimed || rows != 0 {
		t.Fatalf("processOneWithDrain claimed=%v rows=%d, want terminal mirror skipped before execution", claimed, rows)
	}
	if flow.calls != 0 {
		t.Fatalf("flow calls = %d, want 0 for already-terminal legacy event", flow.calls)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("complete calls = %v, want WorkItem completed as processed", got)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 {
		t.Fatalf("bridge calls = processing:%v terminal:%v, want no bridge for already-terminal mirror", bridge.processingCalls, bridge.terminalCalls)
	}
}

func TestWorkerWorkItemSelectorSkipsTerminalMirrorsBeforeLiveEvent(t *testing.T) {
	now := time.Now().UTC()
	events := make(map[string]*connector_events.ConnectorEvent, 21)
	for i := 0; i < 20; i++ {
		suffix := strconv.Itoa(i)
		eventID := "event-terminal-" + suffix
		status := connector_workqueue.WorkStateProcessed
		if i%2 == 1 {
			status = connector_workqueue.WorkStateDead
		}
		events[eventID] = &connector_events.ConnectorEvent{
			ID:              eventID,
			Namespace:       workerTestNS,
			InstalledPackID: "pack-test",
			EventKind:       "noop.created",
			QueueClass:      connector_workqueue.DefaultQueueClass,
			Status:          status,
			Payload:         noopPayload("noop_object", "src-"+eventID),
			RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
			CreatedDate:     now,
		}
	}
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-live",
		EventID:          "event-live",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStatePending,
		LeaseToken:       "lease-live",
		LastModifiedDate: now,
	}}
	events["event-live"] = &connector_events.ConnectorEvent{
		ID:              "event-live",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStatePending,
		Payload:         noopPayload("noop_object", "src-event-live"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}
	loader := &fakeWorkItemEventLoader{events: events}
	bridge := &fakeWorkItemEventBridge{}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-terminal-cleanup",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, newFakeEvents(), &fakeSyncState{}, fakeBatchRead{}, flow, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	claimed, rows, err := w.processOneWithDrain(context.Background(), nil, w.cfg.BatchSize, false)
	if err != nil {
		t.Fatalf("processOneWithDrain: %v", err)
	}
	if !claimed || rows != 1 {
		t.Fatalf("processOneWithDrain claimed=%v rows=%d, want live event claimed without terminal cleanup", claimed, rows)
	}
	if claims.claimCalls != 1 {
		t.Fatalf("WorkItem claim calls = %d, want selector to return live event without terminal backlog claims", claims.claimCalls)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("WorkItem complete calls = %v, want only live event finalization", got)
	}
	if flow.calls != 1 {
		t.Fatalf("flow calls = %d, want live event executed once", flow.calls)
	}
	if got := bridge.processingCalls; len(got) != 1 || got[0] != "event-live:lease-live" {
		t.Fatalf("processing bridge calls = %v, want live event only", got)
	}
}

func TestWorkerRunWorkItemUsesClaimFirstPath(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-run",
		EventID:          "event-run",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-run",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-run",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStatePending,
		Payload:         noopPayload("noop_object", "src-event-run"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	service := newCompleteWorkItemServiceFake(claims, loader, nil)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-run",
		MaxClaims:                     1,
		BatchSize:                     32,
		PollInterval:                  time.Millisecond,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, fakeBatchRead{}, flow, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if w.cfg.BatchSize != 32 {
		t.Fatalf("WorkItem BatchSize = %d, want configured batch size 32", w.cfg.BatchSize)
	}
	if err := w.Run(context.Background()); err != nil {
		t.Fatalf("Run: %v", err)
	}
	if claims.claimCalls != 1 || flow.calls != 1 {
		t.Fatalf("WorkItem claim/flow calls = %d/%d, want Run to drain through claim-first path", claims.claimCalls, flow.calls)
	}
	adapter, ok := w.events.(*workItemEventsService)
	if !ok || adapter.workItems != service || adapter.claims != nil || adapter.bridge != nil {
		t.Fatalf("Run adapter = %+v, want mandatory WorkItem service only", adapter)
	}
	if got := claims.completeCalls; len(got) != 1 || got[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("WorkItem complete calls = %v, want processed", got)
	}
}

func TestWorkerWorkItemSkipsLegacyProcessingMirror(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-legacy-processing",
		EventID:          "event-legacy-processing",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-legacy-processing",
		Attempt:          1,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-legacy-processing",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStateProcessing,
		ClaimToken:      "legacy-token",
		Payload:         noopPayload("noop_object", "src-event-legacy-processing"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	bridge := &fakeWorkItemEventBridge{}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-legacy-processing",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, newFakeEvents(), &fakeSyncState{}, fakeBatchRead{}, flow, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	claimed, rows, err := w.processOneWithDrain(context.Background(), nil, w.cfg.BatchSize, false)
	if !errors.Is(err, connector_events.ErrLeaseLost) {
		t.Fatalf("processOneWithDrain error = %v, want ErrLeaseLost", err)
	}
	if claimed || rows != 0 {
		t.Fatalf("processOneWithDrain claimed=%v rows=%d, want no execution while preserving lane-blocking WorkItem", claimed, rows)
	}
	if flow.calls != 0 {
		t.Fatalf("flow calls = %d, want 0 while legacy event is already processing", flow.calls)
	}
	if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || claims.retryRunAfters[0].After(time.Now().UTC()) || len(claims.completeCalls) != 0 {
		t.Fatalf("WorkItem retry/runAfter/complete calls = %d/%v/%v, want immediately due retry without terminal completion", claims.retryCalls, claims.retryRunAfters, claims.completeCalls)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 {
		t.Fatalf("bridge calls = processing:%v terminal:%v, want no event mutation for legacy processing row", bridge.processingCalls, bridge.terminalCalls)
	}
}

func TestWorkerWorkItemReclaimsStaleLegacyProcessingMirror(t *testing.T) {
	now := time.Now().UTC()
	stale := now.Add(-(connector_events.StaleProcessingSeconds + 1) * time.Second)
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-stale-processing",
		EventID:          "event-stale-processing",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-stale-processing",
		Attempt:          4,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:               "event-stale-processing",
		Namespace:        workerTestNS,
		InstalledPackID:  "pack-test",
		EventKind:        "noop.created",
		QueueClass:       connector_workqueue.DefaultQueueClass,
		Status:           connector_workqueue.WorkStateProcessing,
		ClaimToken:       "legacy-stale-token",
		Payload:          noopPayload("noop_object", "src-event-stale-processing"),
		RouteVersion:     sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:      stale,
		LastModifiedDate: stale,
	}}
	bridge := &fakeWorkItemEventBridge{}
	flow := &fakeFlowInvoker{successResult: &FlowResult{IsSuccess: true}}
	w, err := newWorkerForTestWithNoopBatchRead(WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-stale-processing",
		ClaimPath:                     ClaimPathWorkItem,
		WorkItemClaims:                claims,
		WorkItemEventLoader:           loader,
		WorkItemEventBridge:           bridge,
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, newFakeEvents(), &fakeSyncState{}, fakeBatchRead{}, flow, &stubRouteResolver{routes: map[string]string{
		workerTestNS + "/noop.created": "NoopFlow__c",
	}})
	if err != nil {
		t.Fatalf("NewWorker: %v", err)
	}
	claimed, rows, err := w.processOneWithDrain(context.Background(), nil, w.cfg.BatchSize, false)
	if err != nil {
		t.Fatalf("processOneWithDrain: %v", err)
	}
	if !claimed || rows != 1 {
		t.Fatalf("processOneWithDrain claimed=%v rows=%d, want stale processing takeover", claimed, rows)
	}
	if flow.calls != 1 {
		t.Fatalf("flow calls = %d, want stale processing event executed", flow.calls)
	}
	if got := bridge.processingCalls; len(got) != 1 || got[0] != "event-stale-processing:lease-stale-processing" {
		t.Fatalf("processing bridge calls = %v, want stale processing takeover", got)
	}
	if claims.retryCalls != 0 || len(claims.completeCalls) != 1 || claims.completeCalls[0] != connector_workqueue.WorkStateProcessed {
		t.Fatalf("WorkItem retry/complete calls = %d/%v, want processed completion", claims.retryCalls, claims.completeCalls)
	}
}

func TestWorkerWorkItemStaleOnlyPermitUsesStaleWorkItemClaim(t *testing.T) {
	now := time.Now().UTC()
	stale := now.Add(-(connector_events.StaleProcessingSeconds + 1) * time.Second)
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-stale-only",
		EventID:          "event-stale-only",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-stale-only",
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:               "event-stale-only",
		Namespace:        workerTestNS,
		InstalledPackID:  "pack-test",
		EventKind:        "noop.created",
		QueueClass:       connector_workqueue.DefaultQueueClass,
		Status:           connector_workqueue.WorkStateProcessing,
		ClaimToken:       "legacy-stale-token",
		Payload:          noopPayload("noop_object", "src-event-stale-only"),
		RouteVersion:     sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:      stale,
		LastModifiedDate: stale,
	}}
	bridge := &fakeWorkItemEventBridge{}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-stale-only",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})

	ev, ok, err := w.claimNext(context.Background(), true)
	if err != nil {
		t.Fatalf("claimNext staleOnly: %v", err)
	}
	if !ok || ev == nil || ev.ID != "event-stale-only" {
		t.Fatalf("claimNext staleOnly = (%+v, %v), want stale WorkItem event", ev, ok)
	}
	if !claims.lastClaimInput.StaleOnly {
		t.Fatalf("ClaimNextWorkItem StaleOnly = false, want true for limp permit")
	}
	if len(bridge.processingCalls) != 0 {
		t.Fatalf("processing bridge calls = %v, want WorkItem-only stale takeover", bridge.processingCalls)
	}
}

func TestMandatoryWorkItemStaleReclaimNormalizesEventLifecycle(t *testing.T) {
	workClaimedAt := time.Now().UTC()
	staleEventAt := workClaimedAt.Add(-time.Hour)
	futureRetryAt := workClaimedAt.Add(time.Hour)
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-stale-normalized",
		EventID:          "event-stale-normalized",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "workitem-stale-lease",
		Generation:       7,
		LastModifiedDate: workClaimedAt,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:               "event-stale-normalized",
		Namespace:        workerTestNS,
		InstalledPackID:  "pack-test",
		EventKind:        "noop.created",
		QueueClass:       connector_workqueue.DefaultQueueClass,
		Status:           connector_workqueue.WorkStateDead,
		RetryCount:       connector_events.MaxRetryAttempts,
		LastError:        "stale legacy failure",
		LastErrorAt:      &staleEventAt,
		NextAttemptAt:    &futureRetryAt,
		ClaimedBy:        "legacy-worker",
		ClaimedAt:        &staleEventAt,
		ClaimToken:       "legacy-lease",
		ProcessedAt:      &staleEventAt,
		Payload:          noopPayload("noop_object", "src-stale-normalized"),
		RouteVersion:     sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:      staleEventAt,
		LastModifiedDate: staleEventAt,
	}}
	bridge := &fakeWorkItemEventBridge{}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	cfg := mandatoryWorkerTestConfig()
	cfg.WorkerID = "worker-stale-normalized"
	w := newMandatoryWorkerForTest(t, cfg, service)

	ev, ok, err := w.claimNext(context.Background(), true)
	if err != nil || !ok {
		t.Fatalf("claimNext staleOnly = (%+v, %v, %v), want WorkItem reclaim", ev, ok, err)
	}
	if ev.ID != "event-stale-normalized" || ev.Status != connector_workqueue.WorkStateProcessing ||
		ev.ClaimedBy != "worker-stale-normalized" || ev.ClaimToken != "workitem-stale-lease" {
		t.Fatalf("reclaimed Event = %+v, want WorkItem execution projection", ev)
	}
	if ev.RetryCount != 0 || ev.LastError != "" || ev.LastErrorAt != nil || ev.NextAttemptAt != nil || ev.ProcessedAt != nil {
		t.Fatalf("reclaimed Event retained stale lifecycle fields: %+v", ev)
	}
	if ev.ClaimedAt == nil || !ev.ClaimedAt.Equal(workClaimedAt) || !ev.LastModifiedDate.Equal(workClaimedAt) {
		t.Fatalf("reclaimed Event timestamps = %+v, want WorkItem claim time %v", ev, workClaimedAt)
	}
	if !claims.lastClaimInput.StaleOnly || !claims.lastDistributed || claims.claimCalls != 1 {
		t.Fatalf("WorkItem stale claim = %+v distributed=%v calls=%d, want one distributed stale claim", claims.lastClaimInput, claims.lastDistributed, claims.claimCalls)
	}
	if claims.acquireCalls != 0 || claims.renewCalls != 0 {
		t.Fatalf("SchedulerLease calls acquire/renew = %d/%d, want none", claims.acquireCalls, claims.renewCalls)
	}
	if len(bridge.processingCalls) != 0 || len(bridge.terminalCalls) != 0 || len(bridge.retryCalls) != 0 || len(bridge.renewCalls) != 0 || len(bridge.coalesceCalls) != 0 {
		t.Fatalf("Event bridge calls = processing:%v terminal:%v retry:%v renew:%v coalesce:%v, want none", bridge.processingCalls, bridge.terminalCalls, bridge.retryCalls, bridge.renewCalls, bridge.coalesceCalls)
	}
	adapter, ok := w.events.(*workItemEventsService)
	if !ok || adapter.workItems != service || adapter.claims != nil || adapter.loader != nil || adapter.bridge != nil {
		t.Fatalf("mandatory adapter = %+v, want only complete WorkItem service", adapter)
	}
	ref, err := adapter.claimedRef(ev.ID, ev.ClaimToken)
	if err != nil || ref.workID != "work-stale-normalized" || ref.leaseToken != "workitem-stale-lease" || ref.generation != 7 {
		t.Fatalf("reclaimed WorkItem ref = %+v, %v, want work-stale-normalized/workitem-stale-lease/generation 7", ref, err)
	}
}

func TestWorkerWorkItemReleaseForRetryDoesNotBurnRetryBudget(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-release-budget",
		EventID:          "event-release-budget",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-release-budget",
		Attempt:          connector_events.MaxRetryAttempts,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-release-budget",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStatePending,
		RetryCount:      2,
		Payload:         noopPayload("noop_object", "src-event-release-budget"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	bridge := &fakeWorkItemEventBridge{}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-release-budget",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
	ev, ok, err := w.events.ClaimNext(context.Background(), "pack-test", "worker-release-budget")
	if err != nil || !ok {
		t.Fatalf("ClaimNext = (%v, %v, %v), want WorkItem claim", ev, ok, err)
	}
	releaser := w.events.(claimReleasingEventsService)
	if err := releaser.ReleaseClaimForRetry(context.Background(), ev.ID, ev.ClaimToken, "transient"); err != nil {
		t.Fatalf("ReleaseClaimForRetry: %v", err)
	}
	if claims.retryCalls != 1 || len(claims.retryRunAfters) != 1 || claims.retryRunAfters[0].Before(now.Add(29*time.Second)) {
		t.Fatalf("WorkItem retry calls/runAfter = %d/%v, want one non-budget-burning retry", claims.retryCalls, claims.retryRunAfters)
	}
	if len(bridge.retryCalls) != 0 || len(claims.failInputs) != 0 {
		t.Fatalf("Event retry/WorkItem failure calls = %v/%v, want no Event mutation or failure-budget increment", bridge.retryCalls, claims.failInputs)
	}
}

func TestWorkerWorkItemMarkFailedUsesEventRetryBudget(t *testing.T) {
	now := time.Now().UTC()
	claims := &fakeWorkItemClaims{work: &connector_workqueue.WorkItem{
		WorkID:           "work-fail-budget",
		EventID:          "event-fail-budget",
		InstalledPackID:  "pack-test",
		WorkType:         connector_workqueue.WorkTypeConnectorEvent,
		QueueClass:       connector_workqueue.DefaultQueueClass,
		State:            connector_workqueue.WorkStateProcessing,
		LeaseToken:       "lease-fail-budget",
		Attempt:          connector_events.MaxRetryAttempts,
		LastModifiedDate: now,
	}}
	loader := &fakeWorkItemEventLoader{event: &connector_events.ConnectorEvent{
		ID:              "event-fail-budget",
		Namespace:       workerTestNS,
		InstalledPackID: "pack-test",
		EventKind:       "noop.created",
		QueueClass:      connector_workqueue.DefaultQueueClass,
		Status:          connector_workqueue.WorkStatePending,
		RetryCount:      2,
		Payload:         noopPayload("noop_object", "src-event-fail-budget"),
		RouteVersion:    sql.NullInt64{Valid: true, Int64: 1},
		CreatedDate:     now,
	}}
	bridge := &fakeWorkItemEventBridge{}
	service := newCompleteWorkItemServiceFake(claims, loader, bridge)
	w := newMandatoryWorkerWithDependenciesForTest(t, WorkerConfig{
		Namespace:                     workerTestNS,
		PackInstallID:                 "pack-test",
		WorkerID:                      "worker-fail-budget",
		WorkItemSchedulerOwnerID:      "scheduler-owner",
		WorkItemSchedulerFencingToken: "fence",
	}, service, &fakeSyncState{}, fakeBatchRead{}, &fakeFlowInvoker{}, &stubRouteResolver{})
	ev, ok, err := w.events.ClaimNext(context.Background(), "pack-test", "worker-fail-budget")
	if err != nil || !ok {
		t.Fatalf("ClaimNext = (%v, %v, %v), want WorkItem claim", ev, ok, err)
	}
	if err := w.events.MarkFailed(context.Background(), ev.ID, ev.ClaimToken, "boom"); err != nil {
		t.Fatalf("MarkFailed: %v", err)
	}
	if len(claims.failInputs) != 1 {
		t.Fatalf("FailWorkItem calls = %d, want one WorkItem-owned failure transition", len(claims.failInputs))
	}
	if got := claims.failInputs[0]; got.WorkID != "work-fail-budget" || got.EventID != "event-fail-budget" || got.LeaseToken != "lease-fail-budget" || got.RetryThreshold != connector_events.MaxRetryAttempts || len(got.RetryDelays) != connector_events.MaxRetryAttempts {
		t.Fatalf("FailWorkItem input = %+v, want complete WorkItem retry policy", got)
	}
	if len(bridge.retryCalls) != 0 || len(bridge.terminalCalls) != 0 {
		t.Fatalf("Event bridge retry/terminal calls = %v/%v, want none", bridge.retryCalls, bridge.terminalCalls)
	}
}