package connector_workqueue

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
)

func claimCandidateQueryPattern(queueClassPattern string, staleOnly bool) string {
	statePattern := `w\.State IN \(\?, \?\).*w\.RunAfter <= \?`
	if staleOnly {
		statePattern = `w\.State = \?.*w\.LeaseUntil IS NOT NULL AND w\.LeaseUntil <= \?`
	}
	return `(?s)SELECT w\.WorkId.*FROM _System_ConnectorWorkItem w.*JOIN _System_ConnectorSchedulerLease l.*l\.OwnerId = \?.*l\.FencingToken = \?.*l\.LeaseUntil > \?.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?` + queueClassPattern + `.*` + statePattern + `.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT 1024`
}

func distributedClaimCandidateQueryPattern(queueClassPattern string, staleOnly bool) string {
	statePattern := `w\.State IN \(\?, \?\).*w\.RunAfter <= \?`
	if staleOnly {
		statePattern = `w\.State = \?.*w\.LeaseUntil IS NOT NULL AND w\.LeaseUntil <= \?`
	}
	return `(?s)SELECT w\.WorkId.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?` + queueClassPattern + `.*` + statePattern + `.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT 1024`
}

func laneHeadClaimCandidateQueryPattern(queueClassPattern string, staleOnly bool, limit int) string {
	statePattern := `w\.State IN \(\?, \?\).*w\.RunAfter <= \?`
	if staleOnly {
		statePattern = `w\.State = \?.*w\.LeaseUntil IS NOT NULL AND w\.LeaseUntil <= \?`
	}
	return fmt.Sprintf(`(?s)SELECT w\.WorkId.*FROM _System_ConnectorWorkItem w.*JOIN _System_ConnectorSchedulerLease l.*l\.OwnerId = \?.*l\.FencingToken = \?.*l\.LeaseUntil > \?.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?%s.*%s.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT %d`, queueClassPattern, statePattern, limit)
}

func preferredClaimCandidateQueryPattern(queueClassPattern string, staleOnly bool) string {
	statePattern := `w\.State IN \(\?, \?\).*w\.RunAfter <= \?`
	if staleOnly {
		statePattern = `w\.State = \?.*w\.LeaseUntil IS NOT NULL AND w\.LeaseUntil <= \?`
	}
	return `(?s)SELECT w\.WorkId.*FROM _System_ConnectorWorkItem w.*JOIN _System_ConnectorSchedulerLease l.*l\.OwnerId = \?.*l\.FencingToken = \?.*l\.LeaseUntil > \?.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?` + queueClassPattern + `.*` + statePattern + `.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY CASE WHEN w\.QueueClass IN \(\?,\?\) THEN 0 ELSE 1 END ASC,.*w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT 1024`
}

func preferredLaneHeadClaimCandidateQueryPattern(queueClassPattern string, limit int) string {
	return fmt.Sprintf(`(?s)SELECT w\.WorkId.*FROM _System_ConnectorWorkItem w.*JOIN _System_ConnectorSchedulerLease l.*l\.OwnerId = \?.*l\.FencingToken = \?.*l\.LeaseUntil > \?.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?%s.*w\.State IN \(\?, \?\).*w\.RunAfter <= \?.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY CASE WHEN w\.QueueClass IN \(\?,\?\) THEN 0 ELSE 1 END ASC,.*w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT %d`, queueClassPattern, limit)
}

func TestWorkItemPreferredQueueClassOrderSupportsSingleClass(t *testing.T) {
	order, args := workItemPreferredQueueClassOrder("w", []string{"bulk"}, false)
	if !strings.Contains(order, "CASE WHEN w.QueueClass = ? THEN 0 ELSE 1 END ASC") {
		t.Fatalf("single-class preferred order = %q", order)
	}
	if len(args) != 1 || args[0] != "bulk" {
		t.Fatalf("single-class preferred args = %v, want [bulk]", args)
	}
}

func normalClaimCandidateArgs(ownerID, fencingToken string, now time.Time, installedPackID string, generation int64, queueClasses ...string) []driver.Value {
	args := []driver.Value{
		ownerID, fencingToken, now,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		installedPackID, generation,
	}
	for _, queueClass := range queueClasses {
		args = append(args, queueClass)
	}
	args = append(args, WorkStatePending, WorkStateRetry, now)
	for _, arg := range laneHeadWorkItemArgs(installedPackID, generation, now) {
		args = append(args, arg)
	}
	for _, arg := range workItemEventCandidateFenceArgs(now) {
		args = append(args, arg)
	}
	return args
}

func normalLaneHeadClaimCandidateArgs(ownerID, fencingToken string, now time.Time, installedPackID string, generation int64, queueClasses ...string) []driver.Value {
	return normalClaimCandidateArgs(ownerID, fencingToken, now, installedPackID, generation, queueClasses...)
}

func scopedNormalClaimCandidateArgs(ownerID, fencingToken string, now time.Time, installedPackID string, generation int64, workTypes []string, queueClasses ...string) []driver.Value {
	args := normalClaimCandidateArgs(ownerID, fencingToken, now, installedPackID, generation, queueClasses...)
	scoped := append([]driver.Value(nil), args[:7]...)
	for _, workType := range workTypes {
		scoped = append(scoped, workType)
	}
	return append(scoped, args[7:]...)
}

func staleClaimCandidateArgs(ownerID, fencingToken string, now time.Time, installedPackID string, generation int64, queueClasses ...string) []driver.Value {
	args := []driver.Value{
		ownerID, fencingToken, now,
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		installedPackID, generation,
	}
	for _, queueClass := range queueClasses {
		args = append(args, queueClass)
	}
	args = append(args, WorkStateProcessing, now)
	for _, arg := range laneHeadWorkItemArgs(installedPackID, generation, now) {
		args = append(args, arg)
	}
	for _, arg := range workItemEventCandidateFenceArgs(now) {
		args = append(args, arg)
	}
	return args
}

func staleLaneHeadClaimCandidateArgs(ownerID, fencingToken string, now time.Time, installedPackID string, generation int64, queueClasses ...string) []driver.Value {
	return staleClaimCandidateArgs(ownerID, fencingToken, now, installedPackID, generation, queueClasses...)
}

func laneReadyHeadQueryPattern() string {
	return `(?s)SELECT w\.WorkId.*USE INDEX \(idx_connector_work_item_target_lane\).*State IN \(\?, \?, \?, \?\).*RunAfter <= \?.*UNION ALL.*USE INDEX \(idx_connector_work_item_source_lane_gen\).*LEFT JOIN _System_ConnectorEvent e.*ORDER BY w\.LaneOrdinal ASC,.*e\.EnqueueOrdinal.*w\.CreatedDate ASC.*w\.WorkId ASC.*LIMIT 1`
}

func laneReadyHeadArgs(installedPackID string, generation int64, laneKey string, now time.Time) []driver.Value {
	args := []driver.Value{
		installedPackID, generation, laneKey,
		WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
		now,
	}
	return append(args, args...)
}

func promoteLaneReadyHeadArgs(now time.Time, workID, installedPackID string, generation int64, laneKey string) []driver.Value {
	args := []driver.Value{
		WorkStateBlocked, WorkStatePending,
		WorkStateBlockedRetry, WorkStateRetry,
		now, workID, WorkStateBlocked, WorkStateBlockedRetry,
	}
	for range 2 {
		args = append(args, installedPackID, generation, laneKey, WorkStateProcessing)
	}
	return args
}

func TestLaneHeadWorkItemPredicateBoundsReadsByActiveStateIndex(t *testing.T) {
	predicate := laneHeadWorkItemPredicate("w")
	for _, want := range []string{
		"USE INDEX (idx_connector_work_item_lease_due)",
		"older.InstalledPackId = ?",
		"older.Generation = ?",
		"COALESCE(NULLIF(older.TargetMutationLaneKey, ''), older.SourceLaneKey, '') = COALESCE(NULLIF(w.TargetMutationLaneKey, ''), w.SourceLaneKey, '')",
		"older.LaneOrdinal < w.LaneOrdinal",
		"older.CreatedDate < w.CreatedDate",
		"older.WorkId < w.WorkId",
	} {
		if !strings.Contains(predicate, want) {
			t.Fatalf("laneHeadWorkItemPredicate missing %q:\n%s", want, predicate)
		}
	}
	if got := strings.Count(predicate, "NOT EXISTS ("); got != 1 {
		t.Fatalf("lane-head blocker lookup count=%d, want 1:\n%s", got, predicate)
	}
	if strings.Contains(predicate, "NO_DECORRELATE") {
		t.Fatalf("lane-head blocker lookup must remain decorrelatable for multi-row candidate reads:\n%s", predicate)
	}
	if strings.Contains(predicate, "_System_ConnectorEvent") {
		t.Fatalf("lane-head blocker lookup must use WorkItem state only:\n%s", predicate)
	}
}

func TestLaneHeadWorkItemArgsGateNonProcessingBlockersByRunAfter(t *testing.T) {
	now := time.Date(2026, 7, 8, 23, 20, 0, 0, time.UTC)
	got := laneHeadWorkItemArgs("pack-1", 7, now)
	want := []any{
		"pack-1", int64(7),
		WorkStateProcessing,
		WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
		now,
	}
	if len(got) != len(want) {
		t.Fatalf("laneHeadWorkItemArgs len=%d, want %d: %#v", len(got), len(want), got)
	}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("laneHeadWorkItemArgs[%d]=%#v, want %#v (all=%#v)", i, got[i], want[i], got)
		}
	}

	predicate := activeWorkItemBlockerPredicate("older")
	if !strings.Contains(predicate, "older.State = ?") ||
		!strings.Contains(predicate, "older.State IN (?, ?, ?, ?)") {
		t.Fatalf("activeWorkItemBlockerPredicate must include blocked states as lane blockers:\n%s", predicate)
	}
	if !strings.Contains(predicate, "older.RunAfter <= ?") {
		t.Fatalf("activeWorkItemBlockerPredicate must let future scheduled/backoff rows yield the lane:\n%s", predicate)
	}
	if strings.Contains(predicate, "_System_ConnectorEvent") || strings.Contains(predicate, "NextAttemptAt") {
		t.Fatalf("activeWorkItemBlockerPredicate must not read Event executor state:\n%s", predicate)
	}
	for _, forbidden := range []string{"DependencyKind", "LeaseOwner", "LeaseToken", "LeaseUntil"} {
		if strings.Contains(predicate, forbidden) {
			t.Fatalf("activeWorkItemBlockerPredicate must keep baseline-schema claims free of %q:\n%s", forbidden, predicate)
		}
	}
}

func TestWorkItemClaimPredicatesUseOnlyWorkItemExecutorState(t *testing.T) {
	for _, predicate := range []string{
		claimCandidateWorkItemPredicate("w", false),
		claimCandidateWorkItemPredicate("w", true),
	} {
		if !strings.Contains(predicate, "_System_ConnectorEvent immutable_event") ||
			!strings.Contains(predicate, "immutable_event.Id = w.EventId") {
			t.Fatalf("Event-backed WorkItem candidate must require immutable Event input:\n%s", predicate)
		}
		if !strings.Contains(predicate, "w.ProjectionKey") ||
			!strings.Contains(predicate, "w.ProjectionActiveKey") {
			t.Fatalf("WorkItem candidate must exclude inactive source-projection successors:\n%s", predicate)
		}
		for _, forbidden := range []string{".Status", "NextAttemptAt", "LastModifiedDate", "RetryCount", "ProcessedAt"} {
			if strings.Contains(predicate, forbidden) {
				t.Fatalf("WorkItem readiness predicate contains Event executor fact %q:\n%s", forbidden, predicate)
			}
		}
	}
	for _, predicate := range []string{
		activeWorkItemBlockerPredicate("w"),
		laneHeadWorkItemPredicate("w"),
	} {
		if strings.Contains(predicate, "_System_ConnectorEvent") {
			t.Fatalf("WorkItem lane ownership predicate must use WorkItem state only:\n%s", predicate)
		}
	}
}

func expectNoBlockedLaneHeadsForClaim(mock sqlmock.Sqlmock, now time.Time, installedPackID string, generation int64, queueClasses ...string) {
	queueClassPattern := ``
	args := []driver.Value{installedPackID, generation}
	for _, queueClass := range queueClasses {
		args = append(args, queueClass)
	}
	if len(queueClasses) == 1 {
		queueClassPattern = `.*w\.QueueClass = \?`
	} else if len(queueClasses) > 1 {
		queueClassPattern = `.*w\.QueueClass IN`
	}
	args = append(args, blockedLaneHeadPromotionArgs(installedPackID, generation, now, claimWorkItemCandidateWindow)...)
	mock.ExpectQuery(blockedLaneHeadPromotionQueryPattern(queueClassPattern, claimWorkItemCandidateWindow)).
		WithArgs(args...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "EventId", "WorkType", "QueueClass", "RunAfter", "SourceLaneKey", "TargetMutationLaneKey", "LaneOrdinal"}))
}

func blockedLaneHeadPromotionArgs(installedPackID string, generation int64, now time.Time, limit int) []driver.Value {
	args := []driver.Value{WorkStateBlocked, WorkStateBlockedRetry, now}
	for _, arg := range laneHeadWorkItemArgs(installedPackID, generation, now) {
		args = append(args, arg)
	}
	return append(args, limit)
}

func blockedLaneHeadPromotionQueryPattern(queueClassPattern string, limit int) string {
	return fmt.Sprintf(`(?s)SELECT w\.WorkId,.*w\.LaneOrdinal.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?%s.*w\.State IN \(\?, \?\).*w\.RunAfter <= \?.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT \?`, queueClassPattern)
}

func preferredBlockedLaneHeadPromotionQueryPattern(queueClassPattern string) string {
	return fmt.Sprintf(`(?s)SELECT w\.WorkId,.*w\.LaneOrdinal.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.Generation = \?%s.*w\.State IN \(\?, \?\).*w\.RunAfter <= \?.*NOT EXISTS.*USE INDEX \(idx_connector_work_item_lease_due\).*older\.InstalledPackId = \?.*ORDER BY CASE WHEN w\.QueueClass IN \(\?,\?\) THEN 0 ELSE 1 END ASC,.*w\.LaneOrdinal ASC,.*w\.RunAfter ASC,.*w\.CreatedDate ASC,.*w\.WorkId ASC.*LIMIT \?`, queueClassPattern)
}

func claimUpdatePattern(staleOnly bool) string {
	statePattern := `w\.State IN \(\?, \?\).*w\.RunAfter <= \?`
	if staleOnly {
		statePattern = `w\.State = \?.*w\.LeaseUntil IS NOT NULL AND w\.LeaseUntil <= \?`
	}
	return `(?s)UPDATE _System_ConnectorWorkItem AS w.*LEFT JOIN _System_ConnectorEvent AS e.*w\.WorkType IN \(\?, \?\).*w\.OwnershipEpoch = w\.OwnershipEpoch \+ 1.*e\.ExecutionOwner = \?.*WHERE w\.WorkId = \?.*AND w\.InstalledPackId = \?.*AND w\.Generation = \?.*` + statePattern + `.*NOT EXISTS.*older\.InstalledPackId = \?.*e\.ExecutionOwner = \?.*EXISTS.*_System_ConnectorSchedulerLease`
}

func claimUpdateArgs(ownerID, fencingToken, leaseToken string, leaseUntil, now time.Time, workID, installedPackID string, generation int64, staleOnly bool) []driver.Value {
	args := []driver.Value{
		WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
		WorkStateProcessing, EventExecutionOwnerEvent,
		ownerID, leaseToken, leaseUntil, now,
		EventExecutionOwnerEvent, EventWorkItemCompatibilityFenceStatus,
		EventExecutionOwnerEvent, EventExecutionOwnerEvent, EventExecutionOwnerEvent, EventExecutionOwnerEvent,
		EventExecutionOwnerWorkItem,
		workID, installedPackID, generation,
	}
	if staleOnly {
		args = append(args, WorkStateProcessing, now)
	} else {
		args = append(args, WorkStatePending, WorkStateRetry, now)
	}
	for _, arg := range laneHeadWorkItemArgs(installedPackID, generation, now) {
		args = append(args, arg)
	}
	for _, arg := range workItemEventOwnershipFenceArgs(now) {
		args = append(args, arg)
	}
	return append(args, ownerID, fencingToken, now)
}

func distributedClaimCandidateArgs(now time.Time, installedPackID string, generation int64, queueClasses ...string) []driver.Value {
	args := normalClaimCandidateArgs("ignored-owner", "ignored-fence", now, installedPackID, generation, queueClasses...)
	return args[3:]
}

func scopedDistributedClaimCandidateArgs(now time.Time, installedPackID string, generation int64, workTypes []string, queueClasses ...string) []driver.Value {
	args := distributedClaimCandidateArgs(now, installedPackID, generation, queueClasses...)
	scoped := append([]driver.Value(nil), args[:4]...)
	for _, workType := range workTypes {
		scoped = append(scoped, workType)
	}
	return append(scoped, args[4:]...)
}

func distributedClaimUpdateArgs(ownerID, leaseToken string, leaseUntil, now time.Time, workID, installedPackID string, generation int64) []driver.Value {
	args := claimUpdateArgs(ownerID, "ignored-fence", leaseToken, leaseUntil, now, workID, installedPackID, generation, false)
	return args[:len(args)-3]
}

func expectNoLegacyEventTerminalConvergence(mock sqlmock.Sqlmock, workID string) {
	mock.ExpectQuery(`(?s)SELECT w\.EventId.*FROM _System_ConnectorWorkItem w.*JOIN _System_ConnectorEvent e.*WHERE w\.WorkId = \?.*w\.Attempt = 0.*e\.ExecutionOwner = \?.*e\.Status IN \(\?, \?, \?\).*LIMIT 1`).
		WithArgs(
			workID,
			WorkTypeConnectorEvent, WorkTypeWorkflowActionApply,
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry,
			EventExecutionOwnerEvent,
			WorkStateProcessed, WorkStateSkipped, WorkStateDead,
		).
		WillReturnRows(sqlmock.NewRows([]string{"EventId"}))
}

func connectorEventCoverageArgs(installedPackID string) []driver.Value {
	return []driver.Value{
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
		installedPackID,
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
	}
}

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

	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 7, 1, 57, 0, 0, time.UTC)

	mock.ExpectQuery(`(?s)SELECT w\.Generation.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.WorkType IN \(\?,\?\).*w\.State IN \(\?, \?\).*w\.RunAfter <= \?.*ORDER BY w\.Generation DESC.*LIMIT 1`).
		WithArgs(
			"pack-1",
			WorkTypeConnectorEvent,
			WorkTypeWorkflowActionApply,
			WorkStatePending,
			WorkStateRetry,
			now,
		).
		WillReturnRows(sqlmock.NewRows([]string{"Generation"}).AddRow(int64(7)))

	generation, found, err := repo.ResolveClaimableWorkItemGeneration(WithClaimWorkTypes(context.Background(), []string{
		WorkTypeConnectorEvent,
		WorkTypeWorkflowActionApply,
	}), ResolveClaimableGenerationInput{
		InstalledPackID: "pack-1",
		Generation:      0,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ResolveClaimableWorkItemGeneration: %v", err)
	}
	if !found {
		t.Fatal("ResolveClaimableWorkItemGeneration found=false, want true")
	}
	if generation != 7 {
		t.Fatalf("ResolveClaimableWorkItemGeneration generation=%d, want 7", generation)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 7, 1, 57, 0, 0, time.UTC)

	mock.ExpectQuery(`(?s)SELECT w\.Generation.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.State IN \(\?, \?\).*w\.RunAfter <= \?.*ORDER BY w\.Generation DESC.*LIMIT 1`).
		WithArgs("pack-1", WorkStatePending, WorkStateRetry, now).
		WillReturnRows(sqlmock.NewRows([]string{"Generation"}))
	mock.ExpectQuery(`(?s)SELECT w\.Generation.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.State IN \(\?, \?, \?, \?, \?\).*ORDER BY w\.Generation DESC.*LIMIT 1`).
		WithArgs("pack-1", WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateProcessing).
		WillReturnRows(sqlmock.NewRows([]string{"Generation"}).AddRow(int64(6)))

	generation, found, err := repo.ResolveClaimableWorkItemGeneration(context.Background(), ResolveClaimableGenerationInput{
		InstalledPackID: "pack-1",
		Generation:      0,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ResolveClaimableWorkItemGeneration: %v", err)
	}
	if !found {
		t.Fatal("ResolveClaimableWorkItemGeneration found=false, want true")
	}
	if generation != 6 {
		t.Fatalf("ResolveClaimableWorkItemGeneration generation=%d, want 6", generation)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 7, 1, 57, 0, 0, time.UTC)

	mock.ExpectQuery(`(?s)SELECT w\.Generation.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.State IN \(\?, \?\).*w\.RunAfter <= \?.*ORDER BY w\.Generation DESC.*LIMIT 1`).
		WithArgs("pack-1", WorkStatePending, WorkStateRetry, now).
		WillReturnRows(sqlmock.NewRows([]string{"Generation"}).AddRow(int64(0)))

	generation, found, err := repo.ResolveClaimableWorkItemGeneration(context.Background(), ResolveClaimableGenerationInput{
		InstalledPackID: "pack-1",
		Generation:      0,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ResolveClaimableWorkItemGeneration: %v", err)
	}
	if !found {
		t.Fatal("ResolveClaimableWorkItemGeneration found=false, want true")
	}
	if generation != 0 {
		t.Fatalf("ResolveClaimableWorkItemGeneration generation=%d, want older claimable generation 0", generation)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 7, 1, 57, 0, 0, time.UTC)

	mock.ExpectQuery(`(?s)SELECT w\.Generation.*FROM _System_ConnectorWorkItem w.*WHERE w\.InstalledPackId = \?.*w\.State = \?.*w\.LeaseUntil IS NOT NULL AND w\.LeaseUntil <= \?.*ORDER BY w\.Generation DESC.*LIMIT 1`).
		WithArgs("pack-1", WorkStateProcessing, now).
		WillReturnRows(sqlmock.NewRows([]string{"Generation"}).AddRow(int64(0)))

	generation, found, err := repo.ResolveClaimableWorkItemGeneration(context.Background(), ResolveClaimableGenerationInput{
		InstalledPackID: "pack-1",
		Generation:      0,
		Now:             now,
		StaleOnly:       true,
	})
	if err != nil {
		t.Fatalf("ResolveClaimableWorkItemGeneration staleOnly: %v", err)
	}
	if !found {
		t.Fatal("ResolveClaimableWorkItemGeneration staleOnly found=false, want true")
	}
	if generation != 0 {
		t.Fatalf("ResolveClaimableWorkItemGeneration staleOnly generation=%d, want stale generation 0", generation)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(`(?s)SELECT e.Id.*FROM _System_ConnectorEvent e.*LEFT JOIN _System_ConnectorWorkItem w.*e.Status NOT IN \('processed', 'skipped', 'dead'\).*w.WorkId IS NULL.*LOWER\(TRIM\(selected.EventKind\)\) = LOWER\(TRIM\(e.EventKind\)\).*LIMIT 1`).
		WithArgs(connectorEventCoverageArgs("pack-1")...).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("event-failed-future"))

	missing, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(context.Background(), LegacyEventMirrorReadinessInput{
		InstalledPackID: " pack-1 ",
		Generation:      0,
	})
	if err != nil {
		t.Fatalf("HasNonTerminalLegacyEventWithoutWorkItem: %v", err)
	}
	if !missing {
		t.Fatal("HasNonTerminalLegacyEventWithoutWorkItem = false, want true for mirrorless non-terminal failed event")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(`(?s)SELECT e.Id.*FROM _System_ConnectorEvent e.*LEFT JOIN _System_ConnectorWorkItem w.*e.Status NOT IN \('processed', 'skipped', 'dead'\).*w.WorkId IS NULL.*LIMIT 1`).
		WithArgs(connectorEventCoverageArgs("pack-1")...).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("event-processing-fresh"))

	missing, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(context.Background(), LegacyEventMirrorReadinessInput{
		InstalledPackID: "pack-1",
		Generation:      0,
	})
	if err != nil {
		t.Fatalf("HasNonTerminalLegacyEventWithoutWorkItem: %v", err)
	}
	if !missing {
		t.Fatal("HasNonTerminalLegacyEventWithoutWorkItem = false, want true for mirrorless non-terminal processing event")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(`(?s)SELECT e.Id.*FROM _System_ConnectorEvent e.*LEFT JOIN _System_ConnectorWorkItem w.*e.Status NOT IN \('processed', 'skipped', 'dead'\).*w.WorkId IS NULL.*LIMIT 1`).
		WithArgs(connectorEventCoverageArgs("pack-1")...).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	missing, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(context.Background(), LegacyEventMirrorReadinessInput{
		InstalledPackID: "pack-1",
		Generation:      0,
	})
	if err != nil {
		t.Fatalf("HasNonTerminalLegacyEventWithoutWorkItem: %v", err)
	}
	if missing {
		t.Fatal("HasNonTerminalLegacyEventWithoutWorkItem = true, want false when every non-terminal legacy event has a WorkItem mirror")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryHasNonTerminalLegacyEventWithoutWorkItemAllowsSourceProjectionSuccessor(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
		for _, want := range []string{
			"AND NOT EXISTS",
			"w.WorkType IN (?, ?)",
			"projection.WorkType IN (?, ?)",
			"projection.WorkId LIKE 'csp:%'",
			"selected.RouteVersion",
			"selected.CompanionPackInstallId",
			"selected.CompanionFlowVersionId",
			"e.EnqueueOrdinal BETWEEN projection.InputStartOrdinal AND projection.LaneOrdinal",
		} {
			if !strings.Contains(actualSQL, want) {
				return fmt.Errorf("readiness mirror check missing %q: %s", want, actualSQL)
			}
		}
		for _, forbidden := range []string{"w.InstalledPackId = e.InstalledPackId"} {
			if strings.Contains(actualSQL, forbidden) {
				return fmt.Errorf("readiness mirror check must fail closed across compatibility ownership; found %q: %s", forbidden, actualSQL)
			}
		}
		return nil
	})))
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	repo := NewRepository(db, nil, nil)
	mock.ExpectQuery("source projection readiness check").
		WithArgs(connectorEventCoverageArgs("pack-1")...).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	missing, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(context.Background(), LegacyEventMirrorReadinessInput{
		InstalledPackID: "pack-1",
		Generation:      0,
	})
	if err != nil {
		t.Fatalf("HasNonTerminalLegacyEventWithoutWorkItem: %v", err)
	}
	if missing {
		t.Fatal("HasNonTerminalLegacyEventWithoutWorkItem = true, want false when a source-projection successor is covered by its durable WorkItem")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryHasNonTerminalLegacyEventWithoutWorkItemAcceptsAnyGenerationMirror(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
		if strings.Contains(actualSQL, "w.Generation = ?") {
			return fmt.Errorf("readiness mirror check must not be constrained to one WorkItem generation: %s", actualSQL)
		}
		if !strings.Contains(actualSQL, "w.EventId = e.Id") {
			return fmt.Errorf("readiness mirror check must still join WorkItems by EventId: %s", actualSQL)
		}
		if !strings.Contains(actualSQL, "w.WorkType IN (?, ?)") ||
			!strings.Contains(actualSQL, "projection.WorkType IN (?, ?)") {
			return fmt.Errorf("readiness mirror check must only accept inbound executor WorkTypes: %s", actualSQL)
		}
		if strings.Contains(actualSQL, "w.InstalledPackId = e.InstalledPackId") {
			return fmt.Errorf("readiness mirror check must treat any Event-owned WorkItem row as an authority fence: %s", actualSQL)
		}
		return nil
	})))
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery("readiness mirror check").
		WithArgs(connectorEventCoverageArgs("pack-1")...).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	missing, err := repo.HasNonTerminalLegacyEventWithoutWorkItem(context.Background(), LegacyEventMirrorReadinessInput{
		InstalledPackID: "pack-1",
		Generation:      7,
	})
	if err != nil {
		t.Fatalf("HasNonTerminalLegacyEventWithoutWorkItem: %v", err)
	}
	if missing {
		t.Fatal("HasNonTerminalLegacyEventWithoutWorkItem = true, want false when a non-current-generation WorkItem mirror exists")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	runAfter := now.Add(time.Minute)
	repo := NewRepository(db, func() string { return "work-generated" }, func() time.Time { return now })

	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			"work-generated", "event-1", "pack-1", "materialize", DefaultQueueClass, WorkStatePending, runAfter,
			"source:1", "target:1", int64(0), int64(7), now, now,
		).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "Duplicate entry 'event-1-materialize' for key 'uniq_connector_work_event_type'"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(allWorkItemIdentityColumns())).
		WithArgs("work-generated", "event-1", "materialize").
		WillReturnRows(ordinaryWorkItemIdentityRows(allWorkItemIdentityColumns()).AddRow(
			"work-existing", "event-1", "pack-1", "materialize", DefaultQueueClass,
			"source:1", "target:1", int64(0), int64(7), "", "", "",
		))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ? FOR UPDATE`)).
		WithArgs("work-existing").
		WillReturnRows(workItemRows().AddRow(
			"work-existing", "event-1", "pack-1", "materialize", DefaultQueueClass, WorkStatePending, runAfter,
			0, "", "", nil, "source:1", "target:1", int64(0), int64(7), now, now,
		))

	work, created, err := repo.CreateWorkItem(context.Background(), CreateWorkItemInput{
		EventID:               "event-1",
		InstalledPackID:       "pack-1",
		WorkType:              "materialize",
		RunAfter:              runAfter,
		SourceLaneKey:         "source:1",
		TargetMutationLaneKey: "target:1",
		Generation:            7,
	})
	if err != nil {
		t.Fatalf("CreateWorkItem: %v", err)
	}
	if created {
		t.Fatal("CreateWorkItem created=true on duplicate event/work type")
	}
	if work == nil || work.WorkID != "work-existing" {
		t.Fatalf("CreateWorkItem loaded %+v, want existing work item", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, func() string { return "work-collides" }, func() time.Time { return now })

	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			"work-collides", "event-new", "pack-1", "materialize", DefaultQueueClass, WorkStatePending, now,
			"", "", int64(0), int64(0), now, now,
		).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "Duplicate entry 'work-collides' for key 'PRIMARY'"})
	mock.ExpectQuery(ordinaryWorkItemIdentityQueryPattern(allWorkItemIdentityColumns())).
		WithArgs("work-collides", "event-new", "materialize").
		WillReturnRows(ordinaryWorkItemIdentityRows(allWorkItemIdentityColumns()).AddRow(
			"work-collides", "event-other", "pack-1", "materialize", DefaultQueueClass,
			"", "", int64(0), int64(0), "", "", "",
		))

	work, created, err := repo.CreateWorkItem(context.Background(), CreateWorkItemInput{
		EventID:         "event-new",
		InstalledPackID: "pack-1",
		WorkType:        "materialize",
	})
	if !errors.Is(err, ErrWorkItemIdentityConflict) {
		t.Fatalf("CreateWorkItem error = %v, want typed identity conflict", err)
	}
	if created {
		t.Fatal("CreateWorkItem created=true on duplicate WorkId")
	}
	if work != nil {
		t.Fatalf("CreateWorkItem work = %+v, want nil on duplicate WorkId without business-key row", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, func() string { return "work-1" }, func() time.Time { return now })

	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			"work-1", "event-1", "pack-1", "materialize", DefaultQueueClass, WorkStatePending, now,
			"", "", int64(0), int64(0), now, now,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-1").
		WillReturnRows(workItemRows().AddRow(
			"work-1", "event-1", "pack-1", "materialize", DefaultQueueClass, WorkStatePending, now,
			0, "", "", nil, "", "", int64(0), int64(0), now, now,
		))

	work, created, err := repo.CreateWorkItem(context.Background(), CreateWorkItemInput{
		EventID:         " event-1 ",
		InstalledPackID: " pack-1 ",
		WorkType:        " materialize ",
		State:           " pending ",
	})
	if err != nil {
		t.Fatalf("CreateWorkItem: %v", err)
	}
	if !created {
		t.Fatal("CreateWorkItem created=false, want true")
	}
	if work == nil || work.State != WorkStatePending || work.EventID != "event-1" {
		t.Fatalf("CreateWorkItem returned %+v, want normalized pending work", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, func() string { return "work-1" }, func() time.Time { return now })

	work, created, err := repo.CreateWorkItem(context.Background(), CreateWorkItemInput{
		EventID:         "event-1",
		InstalledPackID: "pack-1",
		WorkType:        "materialize",
		State:           WorkStateProcessing,
	})
	if !errors.Is(err, ErrInvalidWorkItem) {
		t.Fatalf("CreateWorkItem error = %v, want ErrInvalidWorkItem", err)
	}
	if created {
		t.Fatal("CreateWorkItem created=true for processing initial state")
	}
	if work != nil {
		t.Fatalf("CreateWorkItem work = %+v, want nil for invalid initial state", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 4, 18, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`INSERT INTO _System_ConnectorWorkItem`).
		WithArgs(
			"work-earlier", "event-earlier", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateBlocked, now,
			"source:lane", "", int64(10), int64(7), now, now,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-earlier"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-earlier",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-earlier",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-earlier", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	created, err := repo.EnsureWorkItem(context.Background(), CreateWorkItemInput{
		WorkID:          "work-earlier",
		EventID:         "event-earlier",
		InstalledPackID: "pack-1",
		WorkType:        WorkTypeConnectorEvent,
		State:           WorkStatePending,
		RunAfter:        now,
		LaneOrdinal:     10,
		SourceLaneKey:   "source:lane",
		Generation:      7,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("EnsureWorkItem: %v", err)
	}
	if !created {
		t.Fatal("EnsureWorkItem created=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 4, 18, 30, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-earliest"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-earliest",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-earliest",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-earliest", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := repo.ReconcileLaneReadyState(context.Background(), CreateWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		SourceLaneKey:   "source:lane",
		Now:             now,
	}); err != nil {
		t.Fatalf("ReconcileLaneReadyState: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 22, 13, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-dependency"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-dependency",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-dependency",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	// Promotion only targets ordinary blocked states, so the lease-free
	// dependency suspension stays in processing while its pending tail is
	// demoted.
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-dependency", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 0))

	if err := repo.ReconcileLaneReadyState(context.Background(), CreateWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		SourceLaneKey:   "source:lane",
		Now:             now,
	}); err != nil {
		t.Fatalf("ReconcileLaneReadyState: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*AND State IN \(\?, \?\).*AND RunAfter <= \?`).
		WithArgs(WorkStateProcessing, "scheduler-1", "work-token-1", leaseUntil, now, "work-1", WorkStatePending, WorkStateRetry, now).
		WillReturnResult(sqlmock.NewResult(0, 1))

	claimed, err := NewRepository(db, nil, nil).ClaimWorkItem(
		context.Background(), "work-1", "scheduler-1", "work-token-1", leaseUntil, now,
	)
	if err != nil || !claimed {
		t.Fatalf("ClaimWorkItem = (%v, %v), want (true, nil)", claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	claimed, err := repo.ClaimWorkItem(context.Background(), "work-1", "scheduler-1", "work-token-1", now, now)
	if !errors.Is(err, ErrInvalidWorkItem) {
		t.Fatalf("ClaimWorkItem error = %v, want ErrInvalidWorkItem", err)
	}
	if claimed {
		t.Fatal("ClaimWorkItem claimed=true for an already-expired lease")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestWorkItemWorkTypeFilterNormalizesExecutorScope(t *testing.T) {
	ctx := WithClaimWorkTypes(context.Background(), []string{
		"  " + WorkTypeConnectorEvent + "  ",
		WorkTypeWorkflowActionApply,
		WorkTypeConnectorEvent,
		"",
	})
	workTypes := ClaimWorkTypes(ctx)
	filter, args := workItemWorkTypeFilter("candidate", workTypes)
	if filter != " AND candidate.WorkType IN (?,?)" {
		t.Fatalf("work type filter = %q", filter)
	}
	if got, want := fmt.Sprint(args), fmt.Sprint([]any{WorkTypeConnectorEvent, WorkTypeWorkflowActionApply}); got != want {
		t.Fatalf("work type args = %s, want %s", got, want)
	}
	workTypes[0] = "mutated"
	if got := ClaimWorkTypes(ctx)[0]; got != WorkTypeConnectorEvent {
		t.Fatalf("ClaimWorkTypes returned shared state: %q", got)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(`.*w\.QueueClass IN \(\?,\?\)`, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, "interactive", DefaultQueueClass)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-default"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-default", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-default").
		WillReturnRows(workItemRows().AddRow(
			"work-default", "event-default", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now,
			1, "owner-1", "work-token-1", leaseUntil, "source:2", "target:campaign-1", int64(2), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		QueueClasses:    []string{"interactive", DefaultQueueClass},
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-default" || work.QueueClass != DefaultQueueClass {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want in-profile default item", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	leaseToken := claimTestLeaseTokenForOffset(t, 2, 1)
	repo := NewRepository(db, nil, nil)
	allowed := []string{"interactive", DefaultQueueClass, "bulk"}
	preferred := []string{"interactive", DefaultQueueClass}
	queryArgs := normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, allowed...)
	queryArgs = append(queryArgs, preferred[0], preferred[1])

	mock.ExpectQuery(preferredClaimCandidateQueryPattern(`.*w\.QueueClass IN \(\?,\?,\?\)`, false)).
		WithArgs(queryArgs...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "QueueClass"}).
			AddRow("work-interactive", "interactive").
			AddRow("work-default", DefaultQueueClass).
			AddRow("work-bulk", "bulk"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", leaseToken, leaseUntil, now, "work-default", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-default").
		WillReturnRows(workItemRows().AddRow(
			"work-default", "event-default", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now,
			1, "owner-1", leaseToken, leaseUntil, "source:default", "", int64(1), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(WithClaimQueueClassPreference(context.Background(), preferred), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      leaseToken,
		LeaseUntil:      leaseUntil,
		QueueClasses:    allowed,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-default" {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want rotated preferred default item before bulk fallback", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestClaimQueueClassPreferenceFiltersClassesOutsideEligibility(t *testing.T) {
	ctx := WithClaimQueueClassPreference(context.Background(), []string{"interactive"})
	if got := claimQueueClassPreferenceForEligibility(ctx, []string{"bulk"}); len(got) != 0 {
		t.Fatalf("claimQueueClassPreferenceForEligibility = %v, want empty", got)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)
	workTypes := []string{WorkTypeConnectorEvent, WorkTypeWorkflowActionApply}

	mock.ExpectQuery(laneHeadClaimCandidateQueryPattern(`.*w\.WorkType IN \(\?,\?\)`, false, 2)).
		WithArgs(scopedNormalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, workTypes)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-head-a").AddRow("work-head-b"))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL+` WHERE WorkId IN (?,?)`)).
		WithArgs("work-head-a", "work-head-b").
		WillReturnRows(workItemRows().AddRow(
			"work-head-a", "event-a", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStatePending, now,
			0, "", "", nil, "source:a", "", int64(1), int64(7), now, now,
		).AddRow(
			"work-head-b", "event-b", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStatePending, now,
			0, "", "", nil, "source:b", "", int64(1), int64(7), now, now,
		))

	works, err := repo.PeekClaimableWorkItems(WithClaimWorkTypes(context.Background(), workTypes), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      now.Add(30 * time.Second),
		Now:             now,
	}, 2)
	if err != nil {
		t.Fatalf("PeekClaimableWorkItems: %v", err)
	}
	if len(works) != 2 || works[0].WorkID != "work-head-a" || works[1].WorkID != "work-head-b" {
		t.Fatalf("PeekClaimableWorkItems = %+v, want two lane-head candidates", works)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)
	allowed := []string{"interactive", DefaultQueueClass, "bulk"}
	preferred := []string{"interactive", DefaultQueueClass}
	queryArgs := normalLaneHeadClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, allowed...)
	queryArgs = append(queryArgs, preferred[0], preferred[1])

	mock.ExpectQuery(preferredLaneHeadClaimCandidateQueryPattern(`.*w\.QueueClass IN \(\?,\?,\?\)`, 2)).
		WithArgs(queryArgs...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-interactive").AddRow("work-bulk"))
	items := []struct {
		workID     string
		eventID    string
		queueClass string
		sourceLane string
	}{
		{workID: "work-interactive", eventID: "event-interactive", queueClass: "interactive", sourceLane: "source:interactive"},
		{workID: "work-bulk", eventID: "event-bulk", queueClass: "bulk", sourceLane: "source:bulk"},
	}
	workRows := workItemRows()
	for _, item := range items {
		workRows.AddRow(
			item.workID, item.eventID, "pack-1", WorkTypeConnectorEvent, item.queueClass, WorkStatePending, now,
			0, "", "", nil, item.sourceLane, "", int64(1), int64(7), now, now,
		)
	}
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL+` WHERE WorkId IN (?,?)`)).
		WithArgs("work-interactive", "work-bulk").
		WillReturnRows(workRows)

	works, err := repo.PeekClaimableWorkItems(WithClaimQueueClassPreference(context.Background(), preferred), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      now.Add(30 * time.Second),
		QueueClasses:    allowed,
		Now:             now,
	}, 2)
	if err != nil {
		t.Fatalf("PeekClaimableWorkItems: %v", err)
	}
	if len(works) != 2 || works[0].QueueClass != "interactive" || works[1].QueueClass != "bulk" {
		t.Fatalf("PeekClaimableWorkItems = %+v, want preferred item followed by borrowed bulk", works)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)
	mock.ExpectQuery(laneHeadClaimCandidateQueryPattern(`.*w\.QueueClass IN \(\?,\?,\?\)`, true, 2)).
		WithArgs(staleLaneHeadClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, "interactive", DefaultQueueClass, "bulk")...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))

	works, err := repo.PeekClaimableWorkItems(WithClaimQueueClassPreference(context.Background(), []string{"interactive", DefaultQueueClass}), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      now.Add(30 * time.Second),
		QueueClasses:    []string{"interactive", DefaultQueueClass, "bulk"},
		StaleOnly:       true,
		Now:             now,
	}, 2)
	if err != nil || len(works) != 0 {
		t.Fatalf("PeekClaimableWorkItems stale = (%+v, %v), want empty without preference ordering", works, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	allowed := []string{"interactive", DefaultQueueClass, "bulk"}
	preferred := []string{"interactive", DefaultQueueClass}
	promotionArgs := blockedLaneHeadPromotionArgs("pack-1", 7, now, 2)
	args := []driver.Value{"pack-1", int64(7), allowed[0], allowed[1], allowed[2]}
	args = append(args, promotionArgs[:len(promotionArgs)-1]...)
	args = append(args, preferred[0], preferred[1], 2)
	mock.ExpectQuery(preferredBlockedLaneHeadPromotionQueryPattern(`.*w\.QueueClass IN \(\?,\?,\?\)`)).
		WithArgs(args...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "EventId", "WorkType", "QueueClass", "RunAfter", "SourceLaneKey", "TargetMutationLaneKey", "LaneOrdinal"}))

	promoted, err := NewRepository(db, nil, nil).promoteBlockedLaneHeadsForClaim(WithClaimQueueClassPreference(context.Background(), preferred), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		QueueClasses:    allowed,
		Now:             now,
	}, 2)
	if err != nil || promoted != 0 {
		t.Fatalf("promoteBlockedLaneHeadsForClaim = (%d, %v), want 0,nil", promoted, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(laneHeadClaimCandidateQueryPattern(``, false, 1)).
		WithArgs(normalLaneHeadClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-due"))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId IN (?)`)).
		WithArgs("work-due").
		WillReturnRows(workItemRows().AddRow(
			"work-due", "event-future-mirror", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStatePending, now,
			0, "", "", nil, "lane-a", "", int64(1), int64(7), now, now,
		))

	works, err := repo.PeekClaimableWorkItems(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      now.Add(30 * time.Second),
		Now:             now,
	}, 1)
	if err != nil {
		t.Fatalf("PeekClaimableWorkItems: %v", err)
	}
	if len(works) != 1 || works[0].WorkID != "work-due" {
		t.Fatalf("PeekClaimableWorkItems = %+v, want due WorkItem independent of Event mirror", works)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 6, 14, 59, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(laneHeadClaimCandidateQueryPattern(``, false, 4)).
		WithArgs(normalLaneHeadClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-ready"))
	mock.ExpectQuery(blockedLaneHeadPromotionQueryPattern(``, 3)).
		WithArgs(append([]driver.Value{"pack-1", int64(7)}, blockedLaneHeadPromotionArgs("pack-1", 7, now, 3)...)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "EventId", "WorkType", "QueueClass", "RunAfter", "SourceLaneKey", "TargetMutationLaneKey", "LaneOrdinal"}).
			AddRow("work-blocked", "event-blocked", WorkTypeConnectorEvent, DefaultQueueClass, now.Add(-time.Second), "lane-b", "", int64(2)))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "lane-b", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-blocked"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = CASE.*WorkId <> \?`).
		WithArgs(WorkStatePending, WorkStateBlocked, WorkStateRetry, WorkStateBlockedRetry, now, "pack-1", int64(7), "lane-b", WorkStatePending, WorkStateRetry, now, "work-blocked").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = CASE.*WorkId <> \?`).
		WithArgs(WorkStatePending, WorkStateBlocked, WorkStateRetry, WorkStateBlockedRetry, now, "pack-1", int64(7), "lane-b", WorkStatePending, WorkStateRetry, now, "work-blocked").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = CASE.*NOT EXISTS.*State = \?`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-blocked", "pack-1", 7, "lane-b")...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(laneHeadClaimCandidateQueryPattern(``, false, 4)).
		WithArgs(normalLaneHeadClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-ready").AddRow("work-blocked"))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL+` WHERE WorkId IN (?,?)`)).
		WithArgs("work-ready", "work-blocked").
		WillReturnRows(workItemRows().AddRow(
			"work-ready", "event-ready", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStatePending, now,
			0, "", "", nil, "lane-a", "", int64(1), int64(7), now, now,
		).AddRow(
			"work-blocked", "event-blocked", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStatePending, now,
			0, "", "", nil, "lane-b", "", int64(2), int64(7), now, now,
		))

	works, err := repo.PeekClaimableWorkItems(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      now.Add(30 * time.Second),
		Now:             now,
	}, 4)
	if err != nil {
		t.Fatalf("PeekClaimableWorkItems: %v", err)
	}
	if len(works) != 2 || works[0].WorkID != "work-ready" || works[1].WorkID != "work-blocked" {
		t.Fatalf("PeekClaimableWorkItems = %+v, want ready plus promoted blocked lane head", works)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryDistributedClaimCandidateSelectFiltersBeforeLimit(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
		if strings.Contains(actualSQL, "JSON_EXTRACT") {
			return fmt.Errorf("candidate SELECT must not evaluate JSON fallback: %s", actualSQL)
		}
		if strings.Contains(actualSQL, "_System_ConnectorSchedulerLease") {
			return fmt.Errorf("WorkItem claim SQL must not depend on SchedulerLease: %s", actualSQL)
		}
		if strings.Contains(actualSQL, "_System_ConnectorEvent current_event") {
			return fmt.Errorf("WorkItem claim SQL must not read mutable Event executor state: %s", actualSQL)
		}
		if strings.Contains(actualSQL, "SELECT w.WorkId") {
			limitIndex := strings.Index(actualSQL, "LIMIT 1024")
			for _, want := range []string{
				"NOT EXISTS",
				"older.InstalledPackId = ?",
			} {
				idx := strings.Index(actualSQL, want)
				if idx < 0 {
					return fmt.Errorf("candidate SELECT missing %q before LIMIT: %s", want, actualSQL)
				}
				if limitIndex >= 0 && idx > limitIndex {
					return fmt.Errorf("candidate SELECT applies %q after LIMIT: %s", want, actualSQL)
				}
			}
			if strings.Contains(actualSQL, "older_event.DeliveryId = current_event.DeliveryId") {
				return fmt.Errorf("candidate SELECT must stay on materialized WorkItem ordering and not block on delivery siblings: %s", actualSQL)
			}
		}
		matched, err := regexp.MatchString(expectedSQL, actualSQL)
		if err != nil {
			return err
		}
		if !matched {
			return fmt.Errorf("actual SQL does not match expected pattern\nexpected: %s\nactual: %s", expectedSQL, actualSQL)
		}
		return nil
	})))
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(distributedClaimCandidateQueryPattern(``, false)).
		WithArgs(distributedClaimCandidateArgs(now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))
	expectNoBlockedLaneHeadsForClaim(mock, now, "pack-1", 7)
	mock.ExpectQuery(distributedClaimCandidateQueryPattern(``, false)).
		WithArgs(distributedClaimCandidateArgs(now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))

	work, claimed, err := repo.ClaimNextWorkItem(WithDistributedClaim(context.Background()), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      now.Add(30 * time.Second),
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed || work != nil {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want no candidates", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 8, 7, 30, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-live"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-live",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-live",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem AS w.*WHERE WorkId = \?.*NOT EXISTS`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-live", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := repo.ReconcileLaneReadyState(context.Background(), CreateWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		SourceLaneKey:   "source:lane",
		Now:             now,
	}); err != nil {
		t.Fatalf("ReconcileLaneReadyState: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

// TestRepositoryClaimNextWorkItemRechecksCandidateInFinalCAS verifies that when
// the final claim UPDATE's recheck conditions no longer match — because the work
// item state changed or the lane head moved — the CAS affects 0 rows and the
// claim returns no work instead of claiming a stale candidate.
func TestRepositoryClaimNextWorkItemRechecksCandidateInFinalCAS(t *testing.T) {
	tests := []struct {
		name     string
		workID   string
		wantNote string
	}{
		{name: "WorkItemStateInFinalCAS", workID: "work-raced", wantNote: "WorkItem CAS miss"},
		{name: "LaneHeadOnUpdate", workID: "work-later", wantNote: "lane-head CAS recheck failed"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock.New: %v", err)
			}
			defer db.Close()

			now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
			leaseUntil := now.Add(30 * time.Second)
			repo := NewRepository(db, nil, nil)

			mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
				WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
				WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow(tt.workID))
			mock.ExpectExec(claimUpdatePattern(false)).
				WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, tt.workID, "pack-1", int64(7), false)...).
				WillReturnResult(sqlmock.NewResult(0, 0))
			expectNoLegacyEventTerminalConvergence(mock, tt.workID)

			work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
				InstalledPackID: "pack-1",
				Generation:      7,
				OwnerID:         "owner-1",
				FencingToken:    "fence-1",
				LeaseToken:      "work-token-1",
				LeaseUntil:      leaseUntil,
				Now:             now,
			})
			if err != nil {
				t.Fatalf("ClaimNextWorkItem: %v", err)
			}
			if claimed {
				t.Fatalf("ClaimNextWorkItem claimed=true, want no claim after %s", tt.wantNote)
			}
			if work != nil {
				t.Fatalf("ClaimNextWorkItem work = %+v, want nil after %s", work, tt.wantNote)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sqlmock expectations: %v", err)
			}
		})
	}
}

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

	now := time.Date(2026, 7, 5, 9, 5, 0, 0, time.UTC)
	leaseUntil := now.Add(5 * time.Minute)
	repo := NewRepository(db, nil, nil)
	queueClasses := []string{"bulk"}
	workTypes := []string{WorkTypeConnectorEvent, WorkTypeWorkflowActionApply}
	scopeAndQueuePattern := `.*w\.WorkType IN \(\?,\?\).*w\.QueueClass = \?`

	mock.ExpectQuery(claimCandidateQueryPattern(scopeAndQueuePattern, false)).
		WithArgs(scopedNormalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, workTypes, queueClasses...)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))
	mock.ExpectQuery(blockedLaneHeadPromotionQueryPattern(scopeAndQueuePattern, claimWorkItemCandidateWindow)).
		WithArgs(append([]driver.Value{"pack-1", int64(7), WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, "bulk"}, blockedLaneHeadPromotionArgs("pack-1", 7, now, claimWorkItemCandidateWindow)...)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId", "EventId", "WorkType", "QueueClass", "RunAfter", "SourceLaneKey", "TargetMutationLaneKey", "LaneOrdinal"}).
			AddRow("work-blocked", "event-blocked", WorkTypeConnectorEvent, "bulk", now.Add(-time.Minute), "lane-a", "", int64(12)))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "lane-a", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-blocked"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = CASE.*WorkId <> \?`).
		WithArgs(WorkStatePending, WorkStateBlocked, WorkStateRetry, WorkStateBlockedRetry, now, "pack-1", int64(7), "lane-a", WorkStatePending, WorkStateRetry, now, "work-blocked").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = CASE.*WorkId <> \?`).
		WithArgs(WorkStatePending, WorkStateBlocked, WorkStateRetry, WorkStateBlockedRetry, now, "pack-1", int64(7), "lane-a", WorkStatePending, WorkStateRetry, now, "work-blocked").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = CASE.*NOT EXISTS.*State = \?`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-blocked", "pack-1", 7, "lane-a")...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(claimCandidateQueryPattern(scopeAndQueuePattern, false)).
		WithArgs(scopedNormalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7, workTypes, queueClasses...)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-blocked"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "lease-blocked", leaseUntil, now, "work-blocked", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-blocked").
		WillReturnRows(workItemRows().AddRow(
			"work-blocked", "event-blocked", "pack-1", WorkTypeConnectorEvent, "bulk", WorkStateProcessing, now,
			1, "owner-1", "lease-blocked", leaseUntil, "lane-a", "", int64(12), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(WithClaimWorkTypes(context.Background(), workTypes), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-blocked",
		LeaseUntil:      leaseUntil,
		QueueClasses:    queueClasses,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-blocked" || work.State != WorkStateProcessing {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want promoted blocked head claimed", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryLegacyClaimByEventRequiresSchedulerLeaseAndLaneHeadCAS(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 2, 6, 45, 0, 0, time.UTC)
	leaseUntil := now.Add(5 * time.Minute)
	rows := sqlmock.NewRows([]string{"WorkId"}).AddRow("work-event-2")
	args := []driver.Value{"event-2", WorkTypeConnectorEvent, "pack-1", int64(7), WorkStatePending, WorkStateRetry, now}
	for _, arg := range laneHeadWorkItemArgs("pack-1", 7, now) {
		args = append(args, arg)
	}
	for _, arg := range workItemEventCandidateFenceArgs(now) {
		args = append(args, arg)
	}
	args = append(args, "scheduler-1", "fence-1", now)
	mock.ExpectQuery(`(?s)SELECT w.WorkId\s+FROM _System_ConnectorWorkItem w.*LEFT JOIN _System_ConnectorEvent e.*WHERE w.EventId = \?.*w.WorkType = \?.*e\.ExecutionOwner = \?.*EXISTS.*_System_ConnectorSchedulerLease`).
		WithArgs(args...).
		WillReturnRows(rows)
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("scheduler-1", "fence-1", "lease-event-2", leaseUntil, now, "work-event-2", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-event-2").
		WillReturnRows(workItemRows().AddRow("work-event-2", "event-2", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now, 2, "scheduler-1", "lease-event-2", leaseUntil, "source-lane", "", int64(2), int64(7), now, now))

	work, claimed, err := repo.ClaimWorkItemByEvent(context.Background(), ClaimWorkItemByEventInput{
		EventID:         "event-2",
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "scheduler-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-event-2",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimWorkItemByEvent: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-event-2" || work.LeaseToken != "lease-event-2" {
		t.Fatalf("ClaimWorkItemByEvent = (%+v, %v), want claimed event work item", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryClaimWorkItemByEventRejectsTypeOutsideExecutorScope(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	now := time.Date(2026, 7, 17, 9, 0, 0, 0, time.UTC)

	work, claimed, err := NewRepository(db, nil, nil).ClaimWorkItemByEvent(
		WithClaimWorkTypes(context.Background(), []string{WorkTypeConnectorEvent}),
		ClaimWorkItemByEventInput{
			EventID:         "event-action",
			WorkType:        WorkTypeWorkflowActionApply,
			InstalledPackID: "pack-1",
			Generation:      7,
			OwnerID:         "scheduler-1",
			FencingToken:    "fence-1",
			LeaseToken:      "lease-action",
			LeaseUntil:      now.Add(time.Minute),
			Now:             now,
		},
	)
	if err != nil || claimed || work != nil {
		t.Fatalf("ClaimWorkItemByEvent = (%+v, %v, %v), want out-of-scope no-op", work, claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryClaimStaleWorkItemByEventReclaimsStalePeekedCandidate(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 11, 14, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(5 * time.Minute)
	args := []driver.Value{"event-stale", WorkTypeConnectorEvent, "pack-1", int64(7), WorkStateProcessing, now}
	for _, arg := range laneHeadWorkItemArgs("pack-1", 7, now) {
		args = append(args, arg)
	}
	for _, arg := range workItemEventCandidateFenceArgs(now) {
		args = append(args, arg)
	}
	args = append(args, "scheduler-1", "fence-1", now)
	mock.ExpectQuery(`(?s)SELECT w.WorkId\s+FROM _System_ConnectorWorkItem w.*LEFT JOIN _System_ConnectorEvent e.*WHERE w.EventId = \?.*w.State = \?.*w.LeaseUntil IS NOT NULL.*e\.ExecutionOwner = \?.*EXISTS.*_System_ConnectorSchedulerLease`).
		WithArgs(args...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-stale"))
	mock.ExpectExec(claimUpdatePattern(true)).
		WithArgs(claimUpdateArgs("scheduler-1", "fence-1", "lease-stale", leaseUntil, now, "work-stale", "pack-1", int64(7), true)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-stale").
		WillReturnRows(workItemRows().AddRow("work-stale", "event-stale", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now, 2, "scheduler-1", "lease-stale", leaseUntil, "source-lane", "", int64(2), int64(7), now.Add(-time.Hour), now))

	work, claimed, err := repo.ClaimStaleWorkItemByEvent(context.Background(), ClaimWorkItemByEventInput{
		EventID:         "event-stale",
		WorkType:        WorkTypeConnectorEvent,
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "scheduler-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-stale",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil || !claimed || work == nil || work.WorkID != "work-stale" {
		t.Fatalf("ClaimStaleWorkItemByEvent = (%+v, %v, %v), want reclaimed", work, claimed, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryClaimWorkItemByEventUsesWorkItemRunAfterDespiteEventMirror(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	repo := NewRepository(db, nil, nil)
	now := time.Date(2026, 7, 2, 6, 45, 0, 0, time.UTC)
	leaseUntil := now.Add(5 * time.Minute)
	args := []driver.Value{"event-future", WorkTypeConnectorEvent, "pack-1", int64(7), WorkStatePending, WorkStateRetry, now}
	for _, arg := range laneHeadWorkItemArgs("pack-1", 7, now) {
		args = append(args, arg)
	}
	for _, arg := range workItemEventCandidateFenceArgs(now) {
		args = append(args, arg)
	}
	args = append(args, "scheduler-1", "fence-1", now)
	mock.ExpectQuery(`(?s)SELECT w.WorkId\s+FROM _System_ConnectorWorkItem w.*LEFT JOIN _System_ConnectorEvent e.*WHERE w.EventId = \?.*w.State IN \(\?, \?\).*w.RunAfter <= \?.*e\.ExecutionOwner = \?.*EXISTS.*_System_ConnectorSchedulerLease`).
		WithArgs(args...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-event-future"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("scheduler-1", "fence-1", "lease-event-future", leaseUntil, now, "work-event-future", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-event-future").
		WillReturnRows(workItemRows().AddRow(
			"work-event-future", "event-future", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now,
			1, "scheduler-1", "lease-event-future", leaseUntil, "source-lane", "", int64(1), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimWorkItemByEvent(context.Background(), ClaimWorkItemByEventInput{
		EventID:         "event-future",
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "scheduler-1",
		FencingToken:    "fence-1",
		LeaseToken:      "lease-event-future",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimWorkItemByEvent: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-event-future" {
		t.Fatalf("ClaimWorkItemByEvent = (%+v, %v), want due WorkItem independent of Event mirror", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "stale-fence", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))
	expectNoBlockedLaneHeadsForClaim(mock, now, "pack-1", 7)
	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "stale-fence", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "stale-fence",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed {
		t.Fatal("ClaimNextWorkItem claimed=true without current SchedulerLease")
	}
	if work != nil {
		t.Fatalf("ClaimNextWorkItem work = %+v, want nil without current SchedulerLease", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	leaseToken := claimTestLeaseTokenForOffset(t, 2, 0)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-lost-race").AddRow("work-independent-lane"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", leaseToken, leaseUntil, now, "work-lost-race", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-lost-race")
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", leaseToken, leaseUntil, now, "work-independent-lane", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-independent-lane").
		WillReturnRows(workItemRows().AddRow(
			"work-independent-lane", "event-independent", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now,
			1, "owner-1", leaseToken, leaseUntil, "source:independent", "", int64(2), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      leaseToken,
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-independent-lane" {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want retry to claim independent lane", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-tail-a").AddRow("work-tail-b"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-tail-a", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-tail-a")
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-tail-b", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-tail-b")
	expectNoBlockedLaneHeadsForClaim(mock, now, "pack-1", 7)
	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed || work != nil {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want empty after refill finds no heads", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRotateCandidateSliceSpreadsWorkerStart(t *testing.T) {
	workIDs := []string{"work-1", "work-2", "work-3", "work-4"}
	leaseToken := claimTestLeaseTokenForOffset(t, len(workIDs), 2)

	rotated := rotateCandidateSlice(workIDs, leaseToken)
	want := []string{"work-3", "work-4", "work-1", "work-2"}
	if strings.Join(rotated, ",") != strings.Join(want, ",") {
		t.Fatalf("rotateCandidateSlice = %v, want %v", rotated, want)
	}
	if strings.Join(workIDs, ",") != "work-1,work-2,work-3,work-4" {
		t.Fatalf("rotateCandidateSlice mutated input: %v", workIDs)
	}
}

func claimTestLeaseTokenForOffset(t *testing.T, n, offset int) string {
	t.Helper()
	for i := 0; i < 1000; i++ {
		token := fmt.Sprintf("lease-token-%d", i)
		if stableClaimOffset(token, n) == offset {
			return token
		}
	}
	t.Fatalf("no lease token found for offset %d mod %d", offset, n)
	return ""
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-source-head"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-source-head", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-source-head").
		WillReturnRows(workItemRows().AddRow(
			"work-source-head", "event-source-head", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now,
			1, "owner-1", "work-token-1", leaseUntil, "source:shared", "", int64(2), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-source-head" {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want source-ordered head", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-source-later"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-source-later", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-source-later")

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed {
		t.Fatal("ClaimNextWorkItem claimed=true after event source-order CAS recheck failed")
	}
	if work != nil {
		t.Fatalf("ClaimNextWorkItem work = %+v, want nil after event source-order CAS recheck failed", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-target-later"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-target-later", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-target-later")

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed {
		t.Fatal("ClaimNextWorkItem claimed=true after source-lane-before-target-lane CAS recheck failed")
	}
	if work != nil {
		t.Fatalf("ClaimNextWorkItem work = %+v, want nil after source-lane-before-target-lane CAS recheck failed", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-independent-lane"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-independent-lane", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-independent-lane").
		WillReturnRows(workItemRows().AddRow(
			"work-independent-lane", "event-independent-lane", "pack-1", WorkTypeConnectorEvent, DefaultQueueClass, WorkStateProcessing, now,
			1, "owner-1", "work-token-1", leaseUntil, "source:independent", "", int64(2), int64(7), now, now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.WorkID != "work-independent-lane" {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want independent lane candidate claimed", work, claimed)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-json-later"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-json-later", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-json-later")

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed {
		t.Fatal("ClaimNextWorkItem claimed=true after JSON source fallback CAS recheck failed")
	}
	if work != nil {
		t.Fatalf("ClaimNextWorkItem work = %+v, want nil after JSON source fallback CAS recheck failed", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-after-stale"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-after-stale", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-after-stale")

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed {
		t.Fatal("ClaimNextWorkItem claimed=true after stale-processing lane-head CAS recheck failed")
	}
	if work != nil {
		t.Fatalf("ClaimNextWorkItem work = %+v, want nil after stale-processing lane-head CAS recheck failed", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, false)).
		WithArgs(normalClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-delivery-later"))
	mock.ExpectExec(claimUpdatePattern(false)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-1", leaseUntil, now, "work-delivery-later", "pack-1", int64(7), false)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	expectNoLegacyEventTerminalConvergence(mock, "work-delivery-later")

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-1",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if claimed {
		t.Fatal("ClaimNextWorkItem claimed=true after delivery-head CAS recheck failed")
	}
	if work != nil {
		t.Fatalf("ClaimNextWorkItem work = %+v, want nil after delivery-head CAS recheck failed", work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(claimCandidateQueryPattern(``, true)).
		WithArgs(staleClaimCandidateArgs("owner-1", "fence-1", now, "pack-1", 7)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-stale"))
	mock.ExpectExec(claimUpdatePattern(true)).
		WithArgs(claimUpdateArgs("owner-1", "fence-1", "work-token-new", leaseUntil, now, "work-stale", "pack-1", int64(7), true)...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta(selectWorkItemSQL + ` WHERE WorkId = ?`)).
		WithArgs("work-stale").
		WillReturnRows(workItemRows().AddRow(
			"work-stale", "event-stale", "pack-1", "connector_event", DefaultQueueClass, WorkStateProcessing, now,
			2, "owner-1", "work-token-new", leaseUntil, "source:stale", "", int64(2), int64(7), now.Add(-time.Hour), now,
		))

	work, claimed, err := repo.ClaimNextWorkItem(context.Background(), ClaimWorkItemInput{
		InstalledPackID: "pack-1",
		Generation:      7,
		OwnerID:         "owner-1",
		FencingToken:    "fence-1",
		LeaseToken:      "work-token-new",
		LeaseUntil:      leaseUntil,
		StaleOnly:       true,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ClaimNextWorkItem: %v", err)
	}
	if !claimed || work == nil || work.LeaseUntil.Time != leaseUntil {
		t.Fatalf("ClaimNextWorkItem = (%+v, %v), want reclaimed stale processing row", work, claimed)
	}
	if work.WorkID != "work-stale" {
		t.Fatalf("ClaimNextWorkItem stale-only claimed=%v work=%+v, want stale legacy processing mirror", claimed, work)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	renewed, err := repo.RenewWorkItemLease(context.Background(), "work-1", "work-token-1", now.Add(-time.Nanosecond), now)
	if !errors.Is(err, ErrInvalidWorkItem) {
		t.Fatalf("RenewWorkItemLease error = %v, want ErrInvalidWorkItem", err)
	}
	if renewed {
		t.Fatal("RenewWorkItemLease renewed=true for an already-expired lease")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	runAfter := now.Add(5 * time.Minute)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(regexp.QuoteMeta(`
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`)).
		WithArgs(WorkStateRetry, runAfter, now, "work-1", WorkStateProcessing, "work-token-1").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-1").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "", ""))

	retried, err := repo.RetryWorkItem(context.Background(), "work-1", "work-token-1", runAfter, now)
	if err != nil {
		t.Fatalf("RetryWorkItem: %v", err)
	}
	if !retried {
		t.Fatal("RetryWorkItem retried=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 5, 11, 0, 0, 0, time.UTC)
	runAfter := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(regexp.QuoteMeta(`
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`)).
		WithArgs(WorkStateRetry, runAfter, now, "work-high", WorkStateProcessing, "work-token-high").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-high").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "source:lane", ""))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-low"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-low",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-low",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-low", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	retried, err := repo.RetryWorkItem(context.Background(), "work-high", "work-token-high", runAfter, now)
	if err != nil {
		t.Fatalf("RetryWorkItem: %v", err)
	}
	if !retried {
		t.Fatal("RetryWorkItem retried=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 5, 11, 30, 0, 0, time.UTC)
	runAfter := now.Add(5 * time.Minute)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(regexp.QuoteMeta(`
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?`)).
		WithArgs(WorkStateRetry, runAfter, now, "work-future", WorkStateProcessing, "work-token-future").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-future").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "source:lane", ""))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-due"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*RunAfter <= \?.*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-due",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*RunAfter <= \?.*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-due",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-due", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	retried, err := repo.RetryWorkItem(context.Background(), "work-future", "work-token-future", runAfter, now)
	if err != nil {
		t.Fatalf("RetryWorkItem: %v", err)
	}
	if !retried {
		t.Fatal("RetryWorkItem retried=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 17, 9, 0, 0, 0, time.UTC)
	runAfter := now.Add(2 * time.Minute)
	mock.ExpectBegin()
	tx, err := db.Begin()
	if err != nil {
		t.Fatalf("Begin: %v", err)
	}
	repo := NewRepository(db, nil, nil).WithTx(tx)

	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*FOR UPDATE`).
		WithArgs("work-failure", "event-failure", WorkStateProcessing, "lease-failure").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).AddRow("event-failure", 2))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*FailureCount = \?.*FailureClass = \?.*FailureSummary = \?.*AND FailureCount = \?`).
		WithArgs(
			WorkStateRetry, runAfter, 3,
			"inbound_execution_failed", "safe operator summary", now, nil, now,
			"work-failure", "event-failure", WorkStateProcessing, "lease-failure", 2,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-failure").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(1), "", ""))

	result, failed, err := repo.FailWorkItem(context.Background(), FailWorkItemInput{
		WorkID:         "work-failure",
		EventID:        "event-failure",
		LeaseToken:     "lease-failure",
		FailureClass:   "inbound_execution_failed",
		FailureSummary: "safe operator summary",
		RetryThreshold: 5,
		RetryDelays:    []time.Duration{5 * time.Second, 30 * time.Second, 2 * time.Minute},
		Now:            now,
	})
	if err != nil {
		t.Fatalf("FailWorkItem: %v", err)
	}
	if !failed || result == nil || result.State != WorkStateRetry || result.FailureCount != 3 || !result.RunAfter.Equal(runAfter) {
		t.Fatalf("FailWorkItem = (%+v, %v), want retry count 3 at %v", result, failed, runAfter)
	}
	mock.ExpectCommit()
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 17, 9, 3, 0, 0, time.UTC)
	mock.ExpectBegin()
	tx, err := db.Begin()
	if err != nil {
		t.Fatalf("Begin: %v", err)
	}
	repo := NewRepository(db, nil, nil).WithTx(tx)

	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*FOR UPDATE`).
		WithArgs("work-migrated", "event-migrated", WorkStateProcessing, "lease-current").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).
			AddRow("event-migrated", 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*FailureCount = \?.*AND FailureCount = \?`).
		WithArgs(
			WorkStateRetry, now.Add(5*time.Second), 1,
			"inbound_execution_failed", "safe operator summary", now, nil, now,
			"work-migrated", "event-migrated", WorkStateProcessing, "lease-current", 0,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	result, failed, err := repo.FailWorkItemDeferredLaneReconcile(context.Background(), FailWorkItemInput{
		WorkID:         "work-migrated",
		EventID:        "event-migrated",
		LeaseToken:     "lease-current",
		FailureClass:   "inbound_execution_failed",
		FailureSummary: "safe operator summary",
		RetryThreshold: 5,
		RetryDelays:    []time.Duration{5 * time.Second},
		Now:            now,
	})
	if err != nil {
		t.Fatalf("FailWorkItem migrated: %v", err)
	}
	if !failed || result == nil || result.State != WorkStateRetry || result.FailureCount != 1 {
		t.Fatalf("FailWorkItem migrated = (%+v, %v), want WorkItem-owned retry count 1", result, failed)
	}

	mock.ExpectCommit()
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 17, 9, 5, 0, 0, time.UTC)
	mock.ExpectBegin()
	tx, err := db.Begin()
	if err != nil {
		t.Fatalf("Begin: %v", err)
	}
	repo := NewRepository(db, nil, nil).WithTx(tx)

	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*FOR UPDATE`).
		WithArgs("work-dead", "event-dead", WorkStateProcessing, "lease-current").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).AddRow("event-dead", 4))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = \?.*AND FailureCount = \?`).
		WithArgs(
			WorkStateDead, now, 5,
			"inbound_execution_failed", "safe operator summary", now, now, now,
			"work-dead", "event-dead", WorkStateProcessing, "lease-current", 4,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	result, failed, err := repo.FailWorkItemDeferredLaneReconcile(context.Background(), FailWorkItemInput{
		WorkID:         "work-dead",
		EventID:        "event-dead",
		LeaseToken:     "lease-current",
		FailureClass:   "inbound_execution_failed",
		FailureSummary: "safe operator summary",
		RetryThreshold: 5,
		RetryDelays:    []time.Duration{5 * time.Second},
		Now:            now,
	})
	if err != nil {
		t.Fatalf("FailWorkItem dead: %v", err)
	}
	if !failed || result == nil || result.State != WorkStateDead || result.FailureCount != 5 {
		t.Fatalf("FailWorkItem dead = (%+v, %v), want dead count 5", result, failed)
	}

	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FROM _System_ConnectorWorkItem w.*FOR UPDATE`).
		WithArgs("work-dead", "event-dead", WorkStateProcessing, "lease-stale").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}))
	result, failed, err = repo.FailWorkItemDeferredLaneReconcile(context.Background(), FailWorkItemInput{
		WorkID:         "work-dead",
		EventID:        "event-dead",
		LeaseToken:     "lease-stale",
		FailureClass:   "inbound_execution_failed",
		FailureSummary: "safe operator summary",
		RetryThreshold: 5,
		RetryDelays:    []time.Duration{5 * time.Second},
		Now:            now,
	})
	if err != nil || failed || result != nil {
		t.Fatalf("stale FailWorkItem = (%+v, %v, %v), want nil false nil", result, failed, err)
	}

	mock.ExpectCommit()
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 17, 9, 10, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = \?.*WHERE WorkId = \?`).
		WithArgs(WorkStateProcessed, now, now, "work-legacy-schema", WorkStateProcessing, "work-token").
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'TerminalAt' in 'field list'"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*LeaseToken = NULL.*LastModifiedDate = \?.*WHERE WorkId = \?`).
		WithArgs(WorkStateProcessed, now, "work-legacy-schema", WorkStateProcessing, "work-token").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-legacy-schema").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(1), "", ""))

	completed, err := repo.CompleteWorkItem(context.Background(), "work-legacy-schema", "work-token", WorkStateProcessed, now)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want old-schema compatibility update")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 5, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*LeaseToken = NULL.*WHERE WorkId = \?.*AND State = \?.*AND LeaseToken = \?`).
		WithArgs(WorkStateProcessed, now, now, "work-high", WorkStateProcessing, "work-token-high").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-high").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "source:lane", ""))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-low"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-low",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-low",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-low", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	completed, err := repo.CompleteWorkItem(context.Background(), "work-high", "work-token-high", WorkStateProcessed, now)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 17, 12, 0, 0, 0, time.UTC)
	mock.ExpectQuery(`(?s)SELECT WorkId.*FROM _System_ConnectorWorkItem.*WHERE EventId = \?.*FOR UPDATE`).
		WithArgs("event-owned").
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-owned"))

	workID, prepared, err := NewRepository(db, nil, nil).PrepareEventMirrorForTerminalization(
		context.Background(),
		"event-owned",
		"legacy-event-lease",
		now,
	)
	if err != nil {
		t.Fatalf("PrepareEventMirrorForTerminalization: %v", err)
	}
	if workID != "work-owned" || prepared {
		t.Fatalf("PrepareEventMirrorForTerminalization = (%q, %v), want durable fence without preparation", workID, prepared)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 11, 16, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT w\.EventId.*COALESCE\(w\.TargetMutationLaneKey, ''\), w\.Generation.*FOR UPDATE`).
		WithArgs("csp:source-projection", WorkStateProcessing, "work-token").
		WillReturnRows(sqlmock.NewRows([]string{
			"EventId", "InstalledPackId", "WorkType", "ProjectionKey", "ProjectionActiveKey", "TerminalAt",
			"SourceLaneKey", "TargetMutationLaneKey", "Generation",
		}).AddRow("event-current", "pack-1", WorkTypeConnectorEvent, "csp:source-projection", "csp:source-projection", nil, "source:lane", "target:lane", int64(7)))
	mock.ExpectQuery(`(?s)SELECT WorkId, EventId, InstalledPackId, WorkType.*FROM _System_ConnectorWorkItem.*ProjectionKey = \?.*ProjectionActiveKey IS NULL.*LIMIT 2.*FOR UPDATE`).
		WithArgs("csp:source-projection", "csp:source-projection", WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry).
		WillReturnRows(sqlmock.NewRows([]string{
			"WorkId", "EventId", "InstalledPackId", "WorkType",
			"SourceLaneKey", "State", "ProjectionKey", "ProjectionActiveKey",
		}))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*LeaseToken = NULL.*WHERE WorkId = \? AND EventId = \? AND State = \? AND LeaseToken = \?`).
		WithArgs(WorkStateProcessed, "csp:source-projection", now, now, "csp:source-projection", "event-current", WorkStateProcessing, "work-token").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "target:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("target-successor"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "target:lane",
			WorkStatePending, WorkStateRetry, now, "target-successor",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "target:lane",
			WorkStatePending, WorkStateRetry, now, "target-successor",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "target-successor", "pack-1", 7, "target:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	completed, err := repo.CompleteWorkItem(context.Background(), "csp:source-projection", "work-token", WorkStateProcessed, now)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	mock.ExpectQuery(`(?s)SELECT WorkId, EventId, InstalledPackId, WorkType.*FROM _System_ConnectorWorkItem.*ProjectionKey = \?.*ProjectionActiveKey IS NULL.*LIMIT 2.*FOR UPDATE`).
		WithArgs("csp:ambiguous", "work-active", WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry).
		WillReturnRows(sqlmock.NewRows([]string{
			"WorkId", "EventId", "InstalledPackId", "WorkType",
			"SourceLaneKey", "State", "ProjectionKey", "ProjectionActiveKey", "InputStartOrdinal", "LaneOrdinal", "Attempt",
		}).
			AddRow("work-successor-a", "event-a", "pack-1", WorkTypeConnectorEvent, "source:1", WorkStateBlocked, "csp:ambiguous", "", 1, 2, 0).
			AddRow("work-successor-b", "event-b", "pack-1", WorkTypeConnectorEvent, "source:1", WorkStateBlocked, "csp:ambiguous", "", 3, 4, 0))

	successor, err := NewRepository(db, nil, nil).loadQueuedSourceProjectionSuccessorForUpdate(
		context.Background(),
		"csp:ambiguous",
		"work-active",
	)
	if !errors.Is(err, ErrInvalidWorkItem) {
		t.Fatalf("loadQueuedSourceProjectionSuccessorForUpdate error = %v, want ErrInvalidWorkItem", err)
	}
	if successor != nil {
		t.Fatalf("ambiguous queued successor = %+v, want nil", successor)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 16, 22, 30, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT w\.EventId.*w\.ProjectionKey.*w\.ProjectionActiveKey.*w\.TerminalAt.*FOR UPDATE`).
		WithArgs("csp:legacy-schema", WorkStateProcessing, "legacy-token").
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'w.ProjectionKey' in 'field list'"))
	mock.ExpectQuery(`(?s)SELECT w\.EventId, e\.InstalledPackId, w\.WorkType,.*e\.EventKind.*FOR UPDATE`).
		WithArgs("csp:legacy-schema", WorkStateProcessing, "legacy-token").
		WillReturnRows(sqlmock.NewRows([]string{
			"EventId", "InstalledPackId", "WorkType", "EventKind", "RouteVersion",
			"CompanionPackInstallId", "CompanionFlowVersionId", "SourceLaneKey", "TargetMutationLaneKey", "Generation",
		}).AddRow("legacy-event", "pack-legacy", WorkTypeConnectorEvent, "contact.propertychange", int64(1), "", "", "", "", int64(0)))
	mock.ExpectQuery(`(?s)SELECT Id, QueueClass, Status, NextAttemptAt, EnqueueOrdinal.*FROM _System_ConnectorEvent.*Status IN \('pending', 'failed'\)`).
		WithArgs("pack-legacy", "contact.propertychange", int64(1), "", "", "").
		WillReturnRows(sqlmock.NewRows([]string{"Id", "QueueClass", "Status", "NextAttemptAt", "EnqueueOrdinal"}))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*LeaseToken = NULL.*WHERE WorkId = \? AND EventId = \? AND State = \? AND LeaseToken = \?`).
		WithArgs(WorkStateProcessed, now, "csp:legacy-schema", "legacy-event", WorkStateProcessing, "legacy-token").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	completed, err := repo.CompleteWorkItem(context.Background(), "csp:legacy-schema", "legacy-token", WorkStateProcessed, now)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want legacy-schema terminal fallback")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 9, 0, 10, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*LeaseToken = NULL.*WHERE WorkId = \?.*AND State = \?.*AND LeaseToken = \?`).
		WithArgs(WorkStateSkipped, now, now, "work-head", WorkStateProcessing, "work-token-head").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-head").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "source:lane", ""))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnError(errors.New("invalid connection"))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-head").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "source:lane", ""))
	mock.ExpectQuery(laneReadyHeadQueryPattern()).
		WithArgs(laneReadyHeadArgs("pack-1", 7, "source:lane", now)...).
		WillReturnRows(sqlmock.NewRows([]string{"WorkId"}).AddRow("work-tail"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-tail",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*State = CASE.*State IN \(\?, \?\).*WorkId <> \?`).
		WithArgs(
			WorkStatePending, WorkStateBlocked,
			WorkStateRetry, WorkStateBlockedRetry,
			now, "pack-1", int64(7), "source:lane",
			WorkStatePending, WorkStateRetry, now, "work-tail",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*WHERE WorkId = \?.*State IN \(\?, \?\)`).
		WithArgs(promoteLaneReadyHeadArgs(now, "work-tail", "pack-1", 7, "source:lane")...).
		WillReturnResult(sqlmock.NewResult(0, 1))

	completed, err := repo.CompleteWorkItem(context.Background(), "work-head", "work-token-head", WorkStateSkipped, now)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*SET State = \?.*WHERE WorkId = \?.*AND State = \?.*AND LeaseToken = \?`).
		WithArgs(WorkStateDead, now, now, "work-1", WorkStateProcessing, "work-token-1").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-1").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(7), "", ""))

	completed, err := repo.CompleteWorkItem(context.Background(), "work-1", "work-token-1", WorkStateDead, now)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want true for dead terminal state")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	acquired, err := repo.AcquireSchedulerLease(context.Background(), SchedulerLeaseInput{
		InstalledPackID: "pack-1",
		Generation:      4,
		OwnerID:         "owner-a",
		FencingToken:    "fence-a",
		LeaseUntil:      now,
		Now:             now,
	})
	if !errors.Is(err, ErrInvalidSchedulerLease) {
		t.Fatalf("AcquireSchedulerLease error = %v, want ErrInvalidSchedulerLease", err)
	}
	if acquired {
		t.Fatal("AcquireSchedulerLease acquired=true for an already-expired lease")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorSchedulerLease.*WHERE InstalledPackId = \?.*AND Generation = \?.*AND LeaseUntil <= \?`).
		WithArgs("owner-b", "fence-b", leaseUntil, now, now, "pack-1", int64(4), now).
		WillReturnResult(sqlmock.NewResult(0, 1))

	acquired, err := repo.AcquireSchedulerLease(context.Background(), SchedulerLeaseInput{
		InstalledPackID: "pack-1",
		Generation:      4,
		OwnerID:         "owner-b",
		FencingToken:    "fence-b",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("AcquireSchedulerLease: %v", err)
	}
	if !acquired {
		t.Fatal("AcquireSchedulerLease acquired=false, want true for expired row update")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)UPDATE _System_ConnectorSchedulerLease.*AND LeaseUntil <= \?`).
		WithArgs("owner-b", "fence-b", leaseUntil, now, now, "pack-1", int64(4), now).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorSchedulerLease l.*NOT EXISTS.*_System_ConnectorWorkItem`).
		WithArgs("owner-b", "fence-b", leaseUntil, now, now, "pack-1", int64(4), now, WorkStateProcessing, now).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(`INSERT INTO _System_ConnectorSchedulerLease`).
		WithArgs("pack-1", int64(4), "owner-b", "fence-b", leaseUntil, now, now, now).
		WillReturnError(&mysql.MySQLError{Number: 1062, Message: "Duplicate entry 'pack-1-4' for key 'PRIMARY'"})

	acquired, err := repo.AcquireSchedulerLease(context.Background(), SchedulerLeaseInput{
		InstalledPackID: "pack-1",
		Generation:      4,
		OwnerID:         "owner-b",
		FencingToken:    "fence-b",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("AcquireSchedulerLease: %v", err)
	}
	if acquired {
		t.Fatal("AcquireSchedulerLease acquired=true, want false for live holder")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	repo := NewRepository(db, nil, nil)

	mock.ExpectExec(`(?s)DELETE FROM _System_ConnectorSchedulerLease.*WHERE InstalledPackId = \?.*AND Generation = \?.*AND OwnerId = \?.*AND FencingToken = \?`).
		WithArgs("pack-1", int64(4), "owner-b", "fence-b").
		WillReturnResult(sqlmock.NewResult(0, 1))

	released, err := repo.ReleaseSchedulerLease(context.Background(), SchedulerLeaseReleaseInput{
		InstalledPackID: "pack-1",
		Generation:      4,
		OwnerID:         "owner-b",
		FencingToken:    "fence-b",
		Now:             now,
	})
	if err != nil {
		t.Fatalf("ReleaseSchedulerLease: %v", err)
	}
	if !released {
		t.Fatal("ReleaseSchedulerLease released=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

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

	now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC)
	leaseUntil := now.Add(30 * time.Second)
	repo := NewRepository(db, nil, nil)

	mock.ExpectQuery(`(?s)SELECT 1\s+FROM _System_ConnectorSchedulerLease.*WHERE InstalledPackId = \?.*AND Generation = \?.*AND OwnerId = \?.*AND FencingToken = \?.*AND LeaseUntil > \?`).
		WithArgs("pack-1", int64(4), "owner-a", "fence-a", now.Add(15*time.Second)).
		WillReturnError(sql.ErrNoRows)
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorSchedulerLease.*WHERE InstalledPackId = \?.*AND Generation = \?.*AND OwnerId = \?.*AND FencingToken = \?`).
		WithArgs(leaseUntil, now, now, "pack-1", int64(4), "owner-a", "fence-a").
		WillReturnResult(sqlmock.NewResult(0, 1))

	renewed, err := repo.RenewSchedulerLease(context.Background(), SchedulerLeaseInput{
		InstalledPackID: "pack-1",
		Generation:      4,
		OwnerID:         "owner-a",
		FencingToken:    "fence-a",
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		t.Fatalf("RenewSchedulerLease: %v", err)
	}
	if !renewed {
		t.Fatal("RenewSchedulerLease renewed=false, want true")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func workItemRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"WorkId",
		"EventId",
		"InstalledPackId",
		"WorkType",
		"QueueClass",
		"State",
		"RunAfter",
		"Attempt",
		"LeaseOwner",
		"LeaseToken",
		"LeaseUntil",
		"SourceLaneKey",
		"TargetMutationLaneKey",
		"LaneOrdinal",
		"Generation",
		"CreatedDate",
		"LastModifiedDate",
	})
}