package automation

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

type scheduledRunIntegrationTransactionOwner struct {
	*persistence.TransactionManager
}

type scheduledRunRecoveryAuthority struct {
	goodBinding string
	plan        ScheduledFlowExecutionPlan
}

func (a scheduledRunRecoveryAuthority) PrepareScheduledFlow(
	ctx context.Context,
	binding string,
	_ string,
	_ string,
	_ string,
) (PreparedScheduledFlowExecution, error) {
	if binding != a.goodBinding {
		return PreparedScheduledFlowExecution{}, errors.New("immutable artifact is temporarily unavailable")
	}
	return NewPreparedScheduledFlowExecution(
		a.plan,
		func(ctx context.Context) (context.Context, error) { return ctx, nil },
	)
}

func (scheduledRunRecoveryAuthority) IsPermanentScheduledFlowAuthorityFailure(error) bool {
	return false
}

func (o scheduledRunIntegrationTransactionOwner) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	var completion ports.TransactionCompletionOutcome
	err := o.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
			completion = outcome
		})
		return fn(txCtx)
	}, 3)
	return completion, err
}

func ageScheduledRunOwnershipCutover(t *testing.T, db *sql.DB, workFamily string) {
	t.Helper()
	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET LastModifiedDate = UTC_TIMESTAMP(6) - INTERVAL 2 MINUTE
WHERE WorkFamily = ?`, workFamily); err != nil {
		t.Fatal(err)
	}
}

func TestStoppedUpgradeBridgeKeepsLegacyReservationWriterAndFencesOwnershipSwitch(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	legacy := NewScheduledRunReservationService(db)
	firstKey, err := NewScheduledRunReservationKey(
		"legacy-definition", "legacy-version", "legacy-window-one", "__global__",
	)
	if err != nil {
		t.Fatal(err)
	}
	first, err := legacy.Reserve(t.Context(), firstKey)
	if err != nil || first.Status != ScheduledRunReserveStatusReserved {
		t.Fatalf("first legacy reserve = %#v, %v", first, err)
	}
	duplicate, err := legacy.Reserve(t.Context(), firstKey)
	if err != nil || duplicate.Status != ScheduledRunReserveStatusDuplicate {
		t.Fatalf("duplicate legacy reserve = %#v, %v", duplicate, err)
	}
	secondKey, err := NewScheduledRunReservationKey(
		"legacy-definition", "legacy-version", "legacy-window-two", "__global__",
	)
	if err != nil {
		t.Fatal(err)
	}
	second, err := legacy.Reserve(t.Context(), secondKey)
	if err != nil || second.Status != ScheduledRunReserveStatusReserved {
		t.Fatalf("second legacy window reserve = %#v, %v", second, err)
	}

	owner := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeScheduledFlowWorkFamily, DeploymentID: "scheduled-stopped-upgrade",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("a", 64), OwnershipEpoch: 1,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-stopped-upgrade')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(),
	); err != nil {
		t.Fatal(err)
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	binding := catalogsnapshot.RuntimeWorkBinding{
		OrgID: "org-stopped-upgrade", AppInstallationID: "installation-stopped-upgrade",
		RuntimeCapabilityID: "runtime.automation.execute-flow.v1",
		CatalogRevision:     1, SchemaGeneration: 1, EngineGeneration: 1,
		CatalogSnapshotDigest:     catalogsnapshot.Digest("sha256:" + strings.Repeat("b", 64)),
		RuntimeMetadataViewDigest: catalogsnapshot.Digest("sha256:" + strings.Repeat("c", 64)),
	}
	fenced, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := fenced.CheckReady(t.Context()); err == nil ||
		!strings.Contains(err.Error(), "ownership cutover still overlaps") {
		t.Fatalf("fresh ownership cutover did not fence startup: %v", err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET CreatedDate = UTC_TIMESTAMP(6) - INTERVAL 3 MINUTE,
    LastModifiedDate = UTC_TIMESTAMP(6) - INTERVAL 3 MINUTE,
    ReservedAt = UTC_TIMESTAMP(6) - INTERVAL 3 MINUTE
WHERE Id IN (?, ?)`, first.ReservationID, second.ReservationID); err != nil {
		t.Fatal(err)
	}
	ageScheduledRunOwnershipCutover(t, db, owner.WorkFamily)
	if err := fenced.CheckReady(t.Context()); err != nil {
		t.Fatalf("pre-cutover legacy rows blocked exact Runtime owner: %v", err)
	}
	postCutoverKey, err := NewScheduledRunReservationKey(
		"legacy-definition", "legacy-version", "legacy-window-after-cutover", "__global__",
	)
	if err != nil {
		t.Fatal(err)
	}
	postCutover, err := legacy.Reserve(t.Context(), postCutoverKey)
	if err != nil || postCutover.Status != ScheduledRunReserveStatusReserved {
		t.Fatalf("post-cutover legacy reserve = %#v, %v", postCutover, err)
	}
	if err := fenced.CheckReady(t.Context()); err == nil ||
		!strings.Contains(err.Error(), "active rows contain missing") {
		t.Fatalf("post-cutover unscoped row did not fail closed: %v", err)
	}
	if err := legacy.Complete(
		t.Context(), postCutover.ReservationID, postCutoverKey,
		ScheduledRunExecutionStatusCompleted, "",
	); err != nil {
		t.Fatal(err)
	}
	if err := fenced.CheckReady(t.Context()); err != nil {
		t.Fatalf("terminal post-cutover legacy row blocked stopped ownership switch: %v", err)
	}
}

func TestSchedulerWorkerRoleRecoversAcceptedOccurrenceAfterProcessRestart(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	owner := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeScheduledFlowWorkFamily, DeploymentID: "scheduled-restart-worker",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("d", 64), OwnershipEpoch: 1,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-scheduled-restart')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(),
	); err != nil {
		t.Fatal(err)
	}
	ageScheduledRunOwnershipCutover(t, db, owner.WorkFamily)
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	binding := catalogsnapshot.RuntimeWorkBinding{
		OrgID: "org-restart", AppInstallationID: "installation-restart",
		RuntimeCapabilityID: "runtime.records.mutate.v1",
		CatalogRevision:     11, SchemaGeneration: 4, EngineGeneration: 6,
		CatalogSnapshotDigest:     catalogsnapshot.Digest("sha256:" + strings.Repeat("e", 64)),
		RuntimeMetadataViewDigest: catalogsnapshot.Digest("sha256:" + strings.Repeat("f", 64)),
	}
	store, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}

	def := &models.FlowDefinition{
		Id: "flow-def-restart", DeveloperName: "RestartRecoveredFlow",
		ActiveVersionId: "flow-version-restart",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = def.ActiveVersionId
	version.DefinitionId = def.Id
	plan := testScheduledFlowExecutionPlan(def, version)
	occurrence := time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC)
	firstStarted := make(chan struct{})
	var signalFirstStarted sync.Once
	firstCtx, stopFirst := context.WithCancel(t.Context())
	firstScheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{def}
		},
		getFlowDefinitionByIDFn: func(context.Context, string) *models.FlowDefinition { return def },
		getFlowVersionByIDFn:    func(context.Context, string) *models.FlowVersion { return version },
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{
				{"$Record": map[string]interface{}{"Id": "record-restart-a", "Stage": "accepted-value-a"}},
				{"$Record": map[string]interface{}{"Id": "record-restart-b", "Stage": "accepted-value-b"}},
			}, nil
		},
		fencedExecutionAuthority: scheduledFlowAuthorityStub{plan: plan},
		executePreparedScheduledFlowFn: func(ctx context.Context, _ ScheduledFlowExecutionPlan, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			signalFirstStarted.Do(func() { close(firstStarted) })
			<-ctx.Done()
			return nil, ctx.Err()
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore { return store },
		scheduledFlowWorkers:     make(chan struct{}, 1),
		stopCh:                   make(chan struct{}),
		controllerCtx:            firstCtx,
	}
	firstAccepted := map[string]struct{}{}
	firstDispatchDone := make(chan bool, 1)
	go func() {
		firstDispatchDone <- firstScheduler.dispatchScheduledFlowsBetweenWithProgress(
			occurrence.Add(-time.Minute), occurrence, firstAccepted,
		)
	}()
	select {
	case <-firstStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("first process did not reach scheduled execution")
	}
	select {
	case complete := <-firstDispatchDone:
		if !complete || len(firstAccepted) != 1 {
			t.Fatalf("first durable dispatch complete/accepted = %v/%d, want true/1", complete, len(firstAccepted))
		}
	case <-time.After(5 * time.Second):
		t.Fatal("first process did not finish durable acceptance")
	}
	stopFirst()
	firstDone := make(chan struct{})
	go func() {
		firstScheduler.wg.Wait()
		close(firstDone)
	}()
	select {
	case <-firstDone:
	case <-time.After(5 * time.Second):
		t.Fatal("first process did not drain after controller cancellation")
	}

	rows, err := db.Query(`SELECT Id FROM _System_ScheduledRunReservation
WHERE FlowDefinitionId = ? AND FlowVersionId = ? ORDER BY InputKey`,
		def.Id, version.Id,
	)
	if err != nil {
		t.Fatal(err)
	}
	var reservationIDs []string
	for rows.Next() {
		var reservationID string
		if err := rows.Scan(&reservationID); err != nil {
			rows.Close()
			t.Fatal(err)
		}
		reservationIDs = append(reservationIDs, reservationID)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		t.Fatal(err)
	}
	if err := rows.Close(); err != nil {
		t.Fatal(err)
	}
	if len(reservationIDs) != 2 {
		t.Fatalf("durably accepted reservation rows = %d, want two before first effect", len(reservationIDs))
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND
WHERE FlowDefinitionId = ? AND FlowVersionId = ?`, def.Id, version.Id); err != nil {
		t.Fatal(err)
	}

	restartedStore, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	var discoveryCalls atomic.Int64
	var reloadInputCalls atomic.Int64
	recovered := make(chan map[string]interface{}, 2)
	restartedScheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			discoveryCalls.Add(1)
			return nil
		},
		getFlowDefinitionByIDFn: func(context.Context, string) *models.FlowDefinition { return def },
		getFlowVersionByIDFn:    func(context.Context, string) *models.FlowVersion { return version },
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			reloadInputCalls.Add(1)
			return nil, errors.New("restart recovery unexpectedly reloaded mutable inputs")
		},
		fencedExecutionAuthority: scheduledFlowAuthorityStub{plan: plan},
		executePreparedScheduledFlowFn: func(ctx context.Context, _ ScheduledFlowExecutionPlan, inputs map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			if err := transactions.WithTransaction(ctx, func(context.Context) error { return nil }); err != nil {
				return nil, err
			}
			recovered <- inputs
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore { return restartedStore },
	}
	restartedScheduler.SetSchedulerIntervalForTest(5 * time.Millisecond)
	controllerCtx, stopRestarted := context.WithCancel(t.Context())
	roleDone := make(chan error, 1)
	go func() { roleDone <- restartedScheduler.RunWorkerRole(controllerCtx) }()
	recoveredRecords := make(map[string]string, 2)
	for len(recoveredRecords) < 2 {
		select {
		case inputs := <-recovered:
			record, _ := inputs["$Record"].(models.SObject)
			id, _ := record["Id"].(string)
			stage, _ := record["Stage"].(string)
			recoveredRecords[id] = stage
		case <-time.After(5 * time.Second):
			t.Fatal("restarted Worker did not rediscover and execute every accepted occurrence")
		}
	}
	if recoveredRecords["record-restart-a"] != "accepted-value-a" ||
		recoveredRecords["record-restart-b"] != "accepted-value-b" {
		t.Fatalf("recovered records = %#v, want both exact accepted payloads", recoveredRecords)
	}
	completedDeadline := time.Now().Add(5 * time.Second)
	for {
		var completed int
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE FlowDefinitionId = ? AND FlowVersionId = ? AND Status = 'completed'`,
			def.Id, version.Id,
		).Scan(&completed); err != nil {
			t.Fatal(err)
		}
		if completed == 2 {
			break
		}
		if time.Now().After(completedDeadline) {
			t.Fatalf("recovered reservations did not reach terminal completion; completed = %d", completed)
		}
		time.Sleep(10 * time.Millisecond)
	}
	stopRestarted()
	select {
	case err := <-roleDone:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("restarted Worker did not drain")
	}
	if discoveryCalls.Load() == 0 {
		t.Fatal("normal schedule discovery did not run; test did not prove recovery is independent of it")
	}
	if reloadInputCalls.Load() != 0 {
		t.Fatalf("restart input reload calls = %d, want zero", reloadInputCalls.Load())
	}
	var completedAttempts int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE FlowDefinitionId = ? AND FlowVersionId = ? AND Status = 'completed' AND Attempt = 2`,
		def.Id, version.Id,
	).Scan(&completedAttempts); err != nil {
		t.Fatal(err)
	}
	if completedAttempts != 2 {
		t.Fatalf("completed attempt-2 reservations = %d, want two", completedAttempts)
	}
}

func TestFencedScheduledRunTiDBCommitsEffectsAndCompletionTogetherAndFencesStaleOwner(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	owner := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeScheduledFlowWorkFamily, DeploymentID: "scheduled-worker-one",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("a", 64), OwnershipEpoch: 1,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-scheduled-one')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(),
	); err != nil {
		t.Fatal(err)
	}
	ageScheduledRunOwnershipCutover(t, db, owner.WorkFamily)
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	binding := catalogsnapshot.RuntimeWorkBinding{
		OrgID: "org-test", AppInstallationID: "installation-test",
		RuntimeCapabilityID: "runtime.records.mutate.v1",
		CatalogRevision:     7, SchemaGeneration: 2, EngineGeneration: 3,
		CatalogSnapshotDigest:     catalogsnapshot.Digest("sha256:" + strings.Repeat("b", 64)),
		RuntimeMetadataViewDigest: catalogsnapshot.Digest("sha256:" + strings.Repeat("c", 64)),
	}
	service, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := service.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	key, err := NewScheduledRunReservationKey("flow-def-1", "flow-version-1", "window-1", "record-1")
	if err != nil {
		t.Fatal(err)
	}
	first, err := service.Claim(t.Context(), key, testScheduledRunCanonicalInputs())
	if err != nil || first.Status != ScheduledRunLeaseClaimed || first.Attempt != 1 {
		t.Fatalf("first Claim() = %#v, %v", first, err)
	}
	activeRecoverable, err := service.ListRecoverable(t.Context(), 10)
	if err != nil || len(activeRecoverable) != 0 {
		t.Fatalf("active ListRecoverable() = %#v, %v, want none", activeRecoverable, err)
	}
	otherBinding := binding
	otherBinding.AppInstallationID = "installation-other"
	otherBinding.RuntimeMetadataViewDigest = catalogsnapshot.Digest(
		"sha256:" + strings.Repeat("d", 64),
	)
	otherScope, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, otherBinding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	otherLease, err := otherScope.Claim(
		t.Context(),
		key,
		[]byte(`{"$Record":{"Id":"record-other-installation"}}`),
	)
	if err != nil || otherLease.Status != ScheduledRunLeaseClaimed ||
		otherLease.ReservationID == first.ReservationID {
		t.Fatalf("other installation Claim() = %#v, %v", otherLease, err)
	}
	if err := service.CheckReady(t.Context()); err != nil {
		t.Fatalf("CheckReady() rejected another valid installation scope: %v", err)
	}
	var scopedIdentities int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE FlowDefinitionId = ? AND FlowVersionId = ? AND ScheduleWindowKey = ? AND InputKey = ?`,
		key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.InputKey,
	).Scan(&scopedIdentities); err != nil {
		t.Fatal(err)
	}
	if scopedIdentities != 2 {
		t.Fatalf("scoped occurrence identities = %d, want two", scopedIdentities)
	}
	driftKey, err := NewScheduledRunReservationKey("flow-def-1", "flow-version-1", "window-1", "record-added-after-acceptance")
	if err != nil {
		t.Fatal(err)
	}
	alreadyAccepted, err := service.ClaimBatch(t.Context(), []ScheduledRunAcceptance{
		{Key: key, CanonicalInputs: testScheduledRunCanonicalInputs()},
		{Key: driftKey, CanonicalInputs: []byte(`{"$Record":{"Id":"record-added-after-acceptance"}}`)},
	})
	if err != nil || !alreadyAccepted.AlreadyAccepted {
		t.Fatalf("changed input-set ClaimBatch() = %#v, %v, want existing batch authority", alreadyAccepted, err)
	}
	var acceptedRows int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND FlowDefinitionId = 'flow-def-1' AND FlowVersionId = 'flow-version-1'
  AND ScheduleWindowKey = 'window-1'`,
		binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID,
	).Scan(&acceptedRows); err != nil {
		t.Fatal(err)
	}
	if acceptedRows != 1 {
		t.Fatalf("rows after changed input-set retry = %d, want original atomic batch of one", acceptedRows)
	}
	duplicate, err := service.Claim(t.Context(), key, testScheduledRunCanonicalInputs())
	if err != nil || duplicate.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("active duplicate Claim() = %#v, %v", duplicate, err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`, first.ReservationID); err != nil {
		t.Fatal(err)
	}
	expiredRecoverable, err := service.ListRecoverable(t.Context(), 10)
	if err != nil || len(expiredRecoverable) != 1 || expiredRecoverable[0].Key != key {
		t.Fatalf("expired ListRecoverable() = %#v, %v, want exact key %#v", expiredRecoverable, err, key)
	}
	reclaimed, err := service.Reclaim(t.Context(), key)
	if err != nil || reclaimed.Status != ScheduledRunLeaseClaimed || reclaimed.Attempt != 2 ||
		reclaimed.LeaseToken == first.LeaseToken {
		t.Fatalf("safe reclaim Claim() = %#v, %v", reclaimed, err)
	}

	if err := service.Execute(t.Context(), reclaimed, func(ctx context.Context) error {
		tx, ok := transactions.ExtractTx(ctx).(*sql.Tx)
		if !ok {
			return errors.New("effect transaction is absent")
		}
		_, err := tx.ExecContext(ctx, `INSERT INTO ScheduledEffect (Id) VALUES ('effect-1')`)
		return err
	}); err != nil {
		t.Fatal(err)
	}
	var status string
	if err := db.QueryRow(`SELECT Status FROM _System_ScheduledRunReservation WHERE Id = ?`,
		reclaimed.ReservationID,
	).Scan(&status); err != nil || status != string(ScheduledRunExecutionStatusCompleted) {
		t.Fatalf("effect transaction status = %q, %v, want completed", status, err)
	}
	var committedEffects int
	if err := db.QueryRow(`SELECT COUNT(*) FROM ScheduledEffect WHERE Id = 'effect-1'`).Scan(&committedEffects); err != nil {
		t.Fatal(err)
	}
	if committedEffects != 1 {
		t.Fatalf("committed effects = %d, want one", committedEffects)
	}
	tamperedLease := reclaimed
	tamperedLease.MetadataBinding = `{"different":"binding"}`
	if err := service.Execute(t.Context(), tamperedLease, func(context.Context) error {
		return errors.New("tampered lease effect must not run")
	}); err == nil ||
		!strings.Contains(err.Error(), "lease immutable authority") {
		t.Fatalf("tampered lease Execute() error = %v", err)
	}

	completed, err := service.Reclaim(t.Context(), key)
	if err != nil || completed.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("post-effect Reclaim() = %#v, %v, want completed duplicate", completed, err)
	}
	restarted, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	afterRestart, err := restarted.Reclaim(t.Context(), key)
	if err != nil || afterRestart.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("restart Reclaim() = %#v, %v, want durable completed duplicate", afterRestart, err)
	}
	mixedBinding := binding
	mixedBinding.CatalogRevision++
	mixed, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, mixedBinding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := mixed.CheckReady(t.Context()); err != nil {
		t.Fatalf("new revision CheckReady() rejected a valid completed old-revision row: %v", err)
	}
	assertDuplicateAcrossRevision := func(
		name string,
		key ScheduledRunReservationKey,
		canonicalInputs []byte,
	) {
		t.Helper()
		duplicate, claimErr := mixed.Claim(t.Context(), key, canonicalInputs)
		if claimErr != nil || duplicate.Status != ScheduledRunLeaseDuplicate ||
			duplicate.MetadataBinding != service.MetadataBinding() {
			t.Fatalf(
				"new revision Claim(%s old row) = %#v, %v, want duplicate with first accepted binding",
				name,
				duplicate,
				claimErr,
			)
		}
	}
	assertDuplicateAcrossRevision("completed", key, testScheduledRunCanonicalInputs())
	failedKey, err := NewScheduledRunReservationKey(
		"flow-def-failed-revision",
		"flow-version-failed-revision",
		"window-failed-revision",
		"record-failed-revision",
	)
	if err != nil {
		t.Fatal(err)
	}
	failedLease, err := service.Claim(t.Context(), failedKey, testScheduledRunCanonicalInputs())
	if err != nil {
		t.Fatal(err)
	}
	if err := service.Complete(
		t.Context(),
		failedLease,
		ScheduledRunExecutionStatusFailed,
		"bounded failure",
	); err != nil {
		t.Fatal(err)
	}
	assertDuplicateAcrossRevision("failed", failedKey, testScheduledRunCanonicalInputs())
	revisionKey, err := NewScheduledRunReservationKey(
		"flow-def-old-revision",
		"flow-version-old-revision",
		"window-old-revision",
		"record-old-revision",
	)
	if err != nil {
		t.Fatal(err)
	}
	oldRevisionLease, err := service.Claim(
		t.Context(),
		revisionKey,
		[]byte(`{"$Record":{"Id":"record-old-revision"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	assertDuplicateAcrossRevision(
		"active",
		revisionKey,
		[]byte(`{"$Record":{"Id":"record-old-revision"}}`),
	)
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`,
		oldRevisionLease.ReservationID,
	); err != nil {
		t.Fatal(err)
	}
	if err := mixed.CheckReady(t.Context()); err != nil {
		t.Fatalf("new revision CheckReady() rejected valid active old-revision authority: %v", err)
	}
	oldCandidates, err := mixed.ListRecoverable(t.Context(), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(oldCandidates) != 1 ||
		oldCandidates[0].Key != revisionKey ||
		oldCandidates[0].MetadataBinding != service.MetadataBinding() {
		t.Fatalf("new revision recovery candidates = %#v, want exact old-revision row authority", oldCandidates)
	}
	oldRevisionReclaimed, err := mixed.Reclaim(t.Context(), revisionKey)
	if err != nil || oldRevisionReclaimed.Status != ScheduledRunLeaseClaimed ||
		oldRevisionReclaimed.MetadataBinding != service.MetadataBinding() {
		t.Fatalf("new revision Reclaim(old row) = %#v, %v, want claimed old binding", oldRevisionReclaimed, err)
	}

	poisonKey, err := NewScheduledRunReservationKey("flow-def-poison", "flow-version-poison", "window-poison", "record-poison")
	if err != nil {
		t.Fatal(err)
	}
	bounded, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 2,
	)
	if err != nil {
		t.Fatal(err)
	}
	poisonLease, err := bounded.Claim(t.Context(), poisonKey, testScheduledRunCanonicalInputs())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`, poisonLease.ReservationID); err != nil {
		t.Fatal(err)
	}
	poisonLease, err = bounded.Reclaim(t.Context(), poisonKey)
	if err != nil || poisonLease.Status != ScheduledRunLeaseClaimed || poisonLease.Attempt != 2 {
		t.Fatalf("last safe reclaim Claim() = %#v, %v", poisonLease, err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`, poisonLease.ReservationID); err != nil {
		t.Fatal(err)
	}
	exhausted, err := bounded.Reclaim(t.Context(), poisonKey)
	if err != nil || exhausted.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("exhausted Claim() = %#v, %v", exhausted, err)
	}
	if err := db.QueryRow(`SELECT Status FROM _System_ScheduledRunReservation WHERE Id = ?`,
		poisonLease.ReservationID,
	).Scan(&status); err != nil || status != string(ScheduledRunExecutionStatusPoisoned) {
		t.Fatalf("exhausted durable status = %q, %v, want poisoned", status, err)
	}
	assertDuplicateAcrossRevision("poisoned", poisonKey, testScheduledRunCanonicalInputs())
	laterRevisionKey, err := NewScheduledRunReservationKey(
		"flow-def-later-revision",
		"flow-version-later-revision",
		"window-later-revision",
		"record-later-revision",
	)
	if err != nil {
		t.Fatal(err)
	}
	laterRevisionLease, err := mixed.Claim(
		t.Context(),
		laterRevisionKey,
		[]byte(`{"$Record":{"Id":"record-later-revision"}}`),
	)
	if err != nil || laterRevisionLease.Status != ScheduledRunLeaseClaimed ||
		laterRevisionLease.MetadataBinding != mixed.MetadataBinding() {
		t.Fatalf(
			"new revision Claim(later occurrence) = %#v, %v, want new claim with current binding",
			laterRevisionLease,
			err,
		)
	}

	legacyKey, err := NewScheduledRunReservationKey("flow-def-legacy", "flow-version-legacy", "window-legacy", "record-legacy")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId,
 OccurrenceIdentityDigest,
 FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey,
 State, Status, Attempt, ReservedAt)
VALUES ('legacy-row', '', '', '', '', ?, ?, ?, ?, ?, 'RESERVED', 'reserved', 1, UTC_TIMESTAMP(6))`,
		legacyKey.FlowDefinitionID, legacyKey.FlowVersionID, legacyKey.WindowToken,
		legacyKey.WindowToken, legacyKey.InputKey,
	); err != nil {
		t.Fatal(err)
	}
	if err := service.CheckReady(t.Context()); err == nil ||
		!strings.Contains(err.Error(), "active rows contain missing") {
		t.Fatalf("CheckReady() with legacy active row error = %v", err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET State = 'FAILED', Status = 'failed',
    CompletedAt = UTC_TIMESTAMP(6) - INTERVAL 2 MINUTE
WHERE Id = 'legacy-row'`); err != nil {
		t.Fatal(err)
	}
	if err := service.CheckReady(t.Context()); err != nil {
		t.Fatalf("CheckReady() after terminal legacy disposition = %v", err)
	}

	completedKey, err := NewScheduledRunReservationKey("flow-def-completed", "flow-version-completed", "window-completed", "record-completed")
	if err != nil {
		t.Fatal(err)
	}
	completedLease, err := service.Claim(t.Context(), completedKey, testScheduledRunCanonicalInputs())
	if err != nil {
		t.Fatal(err)
	}
	if err := service.Complete(t.Context(), completedLease, ScheduledRunExecutionStatusCompleted, ""); err != nil {
		t.Fatalf("Complete() error = %v", err)
	}
	completedDuplicate, err := service.Claim(t.Context(), completedKey, testScheduledRunCanonicalInputs())
	if err != nil || completedDuplicate.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("completed duplicate Claim() = %#v, %v", completedDuplicate, err)
	}

	timedKey, err := NewScheduledRunReservationKey(
		"flow-def-timed",
		"flow-version-timed",
		"timed-window",
		"record-timed",
	)
	if err != nil {
		t.Fatal(err)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	scope := persistence.RuntimeWorkScope{
		OrgID: binding.OrgID, AppInstallationID: binding.AppInstallationID,
		RuntimeCapabilityID: binding.RuntimeCapabilityID,
	}
	if _, err := db.Exec(`INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, OccurrenceIdentityDigest, BatchIdentityDigest,
 FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey,
 State, Status, Attempt, EntryElement, DueAt,
 MetadataBinding, InputPayloadFormatVersion, InputPayload)
VALUES ('timed-reservation', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
        'ACCEPTED', 'accepted', 0, 'RunLater', UTC_TIMESTAMP(6) + INTERVAL 1 HOUR,
        ?, ?, ?)`,
		scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
		scheduledRunOccurrenceIdentityDigest(scope, timedKey),
		scheduledRunBatchIdentityDigest(scope, timedKey.FlowDefinitionID, timedKey.FlowVersionID, timedKey.WindowToken),
		timedKey.FlowDefinitionID, timedKey.FlowVersionID,
		timedKey.WindowToken, timedKey.WindowToken, timedKey.InputKey,
		encodedBinding, scheduledRunInputPayloadFormatV1, testScheduledRunCanonicalInputs(),
	); err != nil {
		t.Fatal(err)
	}
	if candidates, err := service.ListRecoverable(t.Context(), 10); err != nil ||
		slices.ContainsFunc(candidates, func(candidate ScheduledRunCandidate) bool {
			return candidate.Key == timedKey
		}) {
		t.Fatalf("future timed candidate = %#v, %v, want not due", candidates, err)
	}
	if lease, err := service.Reclaim(t.Context(), timedKey); err != nil ||
		lease.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("future timed Reclaim() = %#v, %v, want duplicate", lease, err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET DueAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = 'timed-reservation'`); err != nil {
		t.Fatal(err)
	}
	candidates, err := service.ListRecoverable(t.Context(), 10)
	if err != nil || !slices.ContainsFunc(candidates, func(candidate ScheduledRunCandidate) bool {
		return candidate.Key == timedKey && candidate.EntryElement == "RunLater"
	}) {
		t.Fatalf("due timed candidates = %#v, %v", candidates, err)
	}
	timedLease, err := service.Reclaim(t.Context(), timedKey)
	if err != nil || timedLease.Status != ScheduledRunLeaseClaimed ||
		timedLease.Attempt != 1 || timedLease.EntryElement != "RunLater" {
		t.Fatalf("due timed Reclaim() = %#v, %v", timedLease, err)
	}

	corruptKey, err := NewScheduledRunReservationKey("flow-def-corrupt", "flow-version-corrupt", "window-corrupt", "record-corrupt")
	if err != nil {
		t.Fatal(err)
	}
	corruptLease, err := service.Claim(t.Context(), corruptKey, testScheduledRunCanonicalInputs())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET State = 'COMPLETED', LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND
WHERE Id = ?`, corruptLease.ReservationID); err != nil {
		t.Fatal(err)
	}
	if _, err := service.Reclaim(t.Context(), corruptKey); err == nil ||
		!strings.Contains(err.Error(), "durable state columns disagree") {
		t.Fatalf("mixed state Claim() error = %v", err)
	}

	longKey, err := NewScheduledRunReservationKey(
		"flow-def-long",
		"flow-version-long",
		"window-long",
		"record-long",
	)
	if err != nil {
		t.Fatal(err)
	}
	longLease, err := service.Claim(t.Context(), longKey, testScheduledRunCanonicalInputs())
	if err != nil {
		t.Fatal(err)
	}
	effectStarted := make(chan struct{})
	allowEffectCommit := make(chan struct{})
	effectDone := make(chan error, 1)
	go func() {
		effectDone <- service.Execute(t.Context(), longLease, func(ctx context.Context) error {
			tx, ok := transactions.ExtractTx(ctx).(*sql.Tx)
			if !ok || tx == nil {
				return errors.New("long effect transaction is absent")
			}
			if _, execErr := tx.ExecContext(ctx, `INSERT INTO ScheduledEffect (Id) VALUES ('long-effect')`); execErr != nil {
				return execErr
			}
			close(effectStarted)
			<-allowEffectCommit
			return nil
		})
	}()
	select {
	case <-effectStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("long effect did not start")
	}
	time.Sleep(1100 * time.Millisecond)
	reclaimDone := make(chan ScheduledRunLease, 1)
	reclaimErr := make(chan error, 1)
	go func() {
		reclaimed, reclaimCallErr := service.Reclaim(t.Context(), longKey)
		reclaimDone <- reclaimed
		reclaimErr <- reclaimCallErr
	}()
	select {
	case reclaimed := <-reclaimDone:
		t.Fatalf("reclaim passed the in-flight effect lock after lease expiry: %#v", reclaimed)
	case <-time.After(200 * time.Millisecond):
	}
	close(allowEffectCommit)
	if err := <-effectDone; err != nil {
		t.Fatalf("long effect Execute() error = %v", err)
	}
	reclaimedAfterCommit := <-reclaimDone
	if err := <-reclaimErr; err != nil ||
		reclaimedAfterCommit.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("post-effect reclaim = %#v, %v, want completed duplicate", reclaimedAfterCommit, err)
	}

	secondKey, err := NewScheduledRunReservationKey("flow-def-2", "flow-version-2", "window-2", "record-2")
	if err != nil {
		t.Fatal(err)
	}
	staleLease, err := service.Claim(t.Context(), secondKey, testScheduledRunCanonicalInputs())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = 'scheduled-worker-two', OwnershipEpoch = 2, OperationId = 'activate-scheduled-two'
WHERE WorkFamily = ?`, owner.WorkFamily); err != nil {
		t.Fatal(err)
	}
	err = service.Execute(t.Context(), staleLease, func(ctx context.Context) error {
		tx, ok := transactions.ExtractTx(ctx).(*sql.Tx)
		if !ok {
			return errors.New("effect transaction is absent")
		}
		_, err := tx.ExecContext(ctx, `INSERT INTO ScheduledEffect (Id) VALUES ('stale-effect')`)
		return err
	})
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("stale owner transaction error = %v, want ownership lost", err)
	}
	if err := service.Complete(t.Context(), staleLease, ScheduledRunExecutionStatusCompleted, ""); !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("stale owner completion error = %v, want ownership lost", err)
	}
	var staleEffects int
	if err := db.QueryRow(`SELECT COUNT(*) FROM ScheduledEffect WHERE Id = 'stale-effect'`).Scan(&staleEffects); err != nil {
		t.Fatal(err)
	}
	if staleEffects != 0 {
		t.Fatalf("stale owner committed %d effects, want zero", staleEffects)
	}
}

func TestFencedScheduledRunTiDBDurablyDisposesPreparationFailures(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	owner := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeScheduledFlowWorkFamily, DeploymentID: "scheduled-preparation-failure",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("a", 64), OwnershipEpoch: 1,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-scheduled-preparation-failure')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(),
	); err != nil {
		t.Fatal(err)
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	binding := catalogsnapshot.RuntimeWorkBinding{
		OrgID: "org-preparation-failure", AppInstallationID: "installation-preparation-failure",
		RuntimeCapabilityID: "runtime.automation.execute-flow.v1",
		CatalogRevision:     7, SchemaGeneration: 2, EngineGeneration: 3,
		CatalogSnapshotDigest:     catalogsnapshot.Digest("sha256:" + strings.Repeat("b", 64)),
		RuntimeMetadataViewDigest: catalogsnapshot.Digest("sha256:" + strings.Repeat("c", 64)),
	}
	service, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}

	refreshKey, err := NewScheduledRunReservationKey(
		"flow-def-refresh", "flow-version-refresh", "window-refresh", "record-refresh",
	)
	if err != nil {
		t.Fatal(err)
	}
	refreshLease, err := service.Claim(
		t.Context(), refreshKey, []byte(`{"$Record":{"Id":"record-refresh"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`,
		refreshLease.ReservationID,
	); err != nil {
		t.Fatal(err)
	}
	refreshedLease, err := service.RefreshLease(t.Context(), refreshLease)
	if err != nil || !refreshedLease.LeaseExpiresAt.After(time.Now().UTC()) {
		t.Fatalf("RefreshLease() = %#v, %v, want fresh fenced lease", refreshedLease, err)
	}
	staleRefresh := refreshLease
	staleRefresh.LeaseToken = "stale-refresh-token"
	if _, err := service.RefreshLease(t.Context(), staleRefresh); !errors.Is(err, ErrScheduledRunLeaseLost) {
		t.Fatalf("stale RefreshLease() error = %v, want lease lost", err)
	}
	if err := service.Complete(
		t.Context(),
		refreshedLease,
		ScheduledRunExecutionStatusFailed,
		"refresh proof terminalized",
	); err != nil {
		t.Fatal(err)
	}

	retryableKey, err := NewScheduledRunReservationKey(
		"flow-def-retryable", "flow-version-retryable", "window-retryable", "record-retryable",
	)
	if err != nil {
		t.Fatal(err)
	}
	retryableLease, err := service.Claim(
		t.Context(), retryableKey, []byte(`{"$Record":{"Id":"record-retryable"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := service.RecordPreparationFailure(
		t.Context(),
		retryableLease,
		ScheduledRunPreparationFailureRetryable,
		"immutable artifact is temporarily unavailable",
	); err != nil {
		t.Fatal(err)
	}
	var retryableStatus, retryableError string
	var retryableAttempt int
	if err := db.QueryRow(`SELECT Status, Attempt, COALESCE(LastError, '')
FROM _System_ScheduledRunReservation WHERE Id = ?`, retryableLease.ReservationID).
		Scan(&retryableStatus, &retryableAttempt, &retryableError); err != nil {
		t.Fatal(err)
	}
	if retryableStatus != string(ScheduledRunExecutionStatusReserved) ||
		retryableAttempt != 1 ||
		retryableError != "immutable artifact is temporarily unavailable" {
		t.Fatalf("retryable disposition = %q/%d/%q, want reserved attempt one with durable error",
			retryableStatus, retryableAttempt, retryableError)
	}
	if recoverable, err := service.ListRecoverable(t.Context(), 10); err != nil || len(recoverable) != 0 {
		t.Fatalf("retryable active lease remained in scan window: %#v, %v", recoverable, err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`,
		retryableLease.ReservationID,
	); err != nil {
		t.Fatal(err)
	}
	retried, err := service.Reclaim(t.Context(), retryableKey)
	if err != nil || retried.Status != ScheduledRunLeaseClaimed || retried.Attempt != 2 {
		t.Fatalf("retryable preparation Reclaim() = %#v, %v, want attempt two", retried, err)
	}

	permanentKey, err := NewScheduledRunReservationKey(
		"flow-def-permanent", "flow-version-permanent", "window-permanent", "record-permanent",
	)
	if err != nil {
		t.Fatal(err)
	}
	permanentLease, err := service.Claim(
		t.Context(), permanentKey, []byte(`{"$Record":{"Id":"record-permanent"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := service.RecordPreparationFailure(
		t.Context(),
		permanentLease,
		ScheduledRunPreparationFailurePermanent,
		"immutable artifact is invalid",
	); err != nil {
		t.Fatal(err)
	}
	var permanentStatus, permanentState, permanentError string
	if err := db.QueryRow(`SELECT Status, State, COALESCE(LastError, '')
FROM _System_ScheduledRunReservation WHERE Id = ?`, permanentLease.ReservationID).
		Scan(&permanentStatus, &permanentState, &permanentError); err != nil {
		t.Fatal(err)
	}
	if permanentStatus != string(ScheduledRunExecutionStatusFailed) ||
		permanentState != "FAILED" ||
		permanentError != "immutable artifact is invalid" {
		t.Fatalf("permanent disposition = %q/%q/%q, want terminal failed authority",
			permanentStatus, permanentState, permanentError)
	}
	if duplicate, err := service.Reclaim(t.Context(), permanentKey); err != nil ||
		duplicate.Status != ScheduledRunLeaseDuplicate {
		t.Fatalf("permanent preparation Reclaim() = %#v, %v, want terminal duplicate", duplicate, err)
	}
}

func TestFencedScheduledRunTiDBUnavailableAuthorityCannotStarveHealthyRecovery(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	owner := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeScheduledFlowWorkFamily, DeploymentID: "scheduled-recovery-fairness",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("a", 64), OwnershipEpoch: 1,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-scheduled-recovery-fairness')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(),
	); err != nil {
		t.Fatal(err)
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	binding := catalogsnapshot.RuntimeWorkBinding{
		OrgID: "org-recovery-fairness", AppInstallationID: "installation-recovery-fairness",
		RuntimeCapabilityID: "runtime.automation.execute-flow.v1",
		CatalogRevision:     7, SchemaGeneration: 2, EngineGeneration: 3,
		CatalogSnapshotDigest:     catalogsnapshot.Digest("sha256:" + strings.Repeat("b", 64)),
		RuntimeMetadataViewDigest: catalogsnapshot.Digest("sha256:" + strings.Repeat("c", 64)),
	}
	service, err := NewFencedScheduledRunReservationService(
		db, scheduledRunIntegrationTransactionOwner{transactions}, owner, binding, time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	badBinding := binding
	badBinding.RuntimeMetadataViewDigest = catalogsnapshot.Digest(
		"sha256:" + strings.Repeat("d", 64),
	)
	encodedBadBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(badBinding)
	if err != nil {
		t.Fatal(err)
	}

	const workerCapacity = 2
	for index := 0; index < workerCapacity+1; index++ {
		key, err := NewScheduledRunReservationKey(
			fmt.Sprintf("bad-definition-%d", index),
			fmt.Sprintf("bad-version-%d", index),
			fmt.Sprintf("bad-window-%d", index),
			fmt.Sprintf("bad-record-%d", index),
		)
		if err != nil {
			t.Fatal(err)
		}
		lease, err := service.Claim(
			t.Context(),
			key,
			[]byte(fmt.Sprintf(`{"$Record":{"Id":"bad-record-%d"}}`, index)),
		)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET MetadataBinding = ?, LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL ? SECOND
WHERE Id = ?`, encodedBadBinding, 100-index, lease.ReservationID); err != nil {
			t.Fatal(err)
		}
	}
	definition := &models.FlowDefinition{
		Id: "healthy-definition", DeveloperName: "HealthyRecovery",
		ActiveVersionId: "healthy-version",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = definition.ActiveVersionId
	version.DefinitionId = definition.Id
	healthyKey, err := NewScheduledRunReservationKey(
		definition.Id, version.Id, "healthy-window", "healthy-record",
	)
	if err != nil {
		t.Fatal(err)
	}
	healthyLease, err := service.Claim(
		t.Context(), healthyKey, []byte(`{"$Record":{"Id":"healthy-record"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND WHERE Id = ?`,
		healthyLease.ReservationID,
	); err != nil {
		t.Fatal(err)
	}

	var healthyExecutions atomic.Int64
	scheduler := &SchedulerService{
		fencedExecutionAuthority: scheduledRunRecoveryAuthority{
			goodBinding: service.MetadataBinding(),
			plan:        testScheduledFlowExecutionPlan(definition, version),
		},
		executePreparedScheduledFlowFn: func(
			context.Context,
			ScheduledFlowExecutionPlan,
			map[string]interface{},
			*models.UserSession,
		) (*models.FlowExecutionResult, error) {
			healthyExecutions.Add(1)
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore { return service },
		scheduledFlowWorkers:     make(chan struct{}, workerCapacity),
		stopCh:                   make(chan struct{}),
	}
	for round := 0; round < 2; round++ {
		scheduler.recoverExpiredScheduledRuns()
		scheduler.wg.Wait()
	}
	if healthyExecutions.Load() != 1 {
		t.Fatalf("healthy executions after bounded bad rows = %d, want one", healthyExecutions.Load())
	}
	var healthyStatus string
	if err := db.QueryRow(`SELECT Status FROM _System_ScheduledRunReservation WHERE Id = ?`,
		healthyLease.ReservationID,
	).Scan(&healthyStatus); err != nil {
		t.Fatal(err)
	}
	if healthyStatus != string(ScheduledRunExecutionStatusCompleted) {
		t.Fatalf("healthy recovery status = %q, want completed", healthyStatus)
	}
	var durableRetryableFailures int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE FlowDefinitionId LIKE 'bad-definition-%' AND Status = 'reserved'
  AND Attempt = 2 AND LastError LIKE '%temporarily unavailable%'`).
		Scan(&durableRetryableFailures); err != nil {
		t.Fatal(err)
	}
	if durableRetryableFailures != workerCapacity+1 {
		t.Fatalf("durable retryable failures = %d, want %d",
			durableRetryableFailures, workerCapacity+1)
	}
}

func createFencedScheduledRunTables(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, statement := range []string{
		`CREATE TABLE _System_RuntimeWorkOwnership (
WorkFamily VARCHAR(64) PRIMARY KEY,
DeploymentId VARCHAR(255) NOT NULL,
WorkerReleaseDigest CHAR(71) NOT NULL,
OwnershipEpoch DECIMAL(20,0) NOT NULL,
IsActive TINYINT(1) NOT NULL,
OperationId VARCHAR(255) NOT NULL,
CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6))`,
		`CREATE TABLE _System_ScheduledRunReservation (
Id VARCHAR(255) PRIMARY KEY,
OrgId VARCHAR(191) NULL,
AppInstallationId VARCHAR(191) NULL,
RuntimeCapabilityId VARCHAR(191) NULL,
OccurrenceIdentityDigest CHAR(64) COLLATE utf8mb4_bin NULL,
BatchIdentityDigest CHAR(64) COLLATE utf8mb4_bin NULL,
FlowDefinitionId VARCHAR(191) NOT NULL,
FlowVersionId VARCHAR(191) NOT NULL,
ScheduleWindowKey VARCHAR(191) NOT NULL,
WindowToken VARCHAR(191) NULL,
InputKey VARCHAR(191) NOT NULL,
State VARCHAR(32) NOT NULL DEFAULT 'RESERVED',
Status VARCHAR(32) NULL,
Attempt INT NOT NULL DEFAULT 1,
LeaseToken VARCHAR(64) NULL,
LeaseExpiresAt DATETIME(6) NULL,
EntryElement VARCHAR(191) NULL,
DueAt DATETIME(6) NULL,
OwnershipEpoch DECIMAL(20,0) NULL,
MetadataBinding MEDIUMTEXT NULL,
InputPayloadFormatVersion BIGINT NULL,
InputPayload MEDIUMBLOB NULL,
ReservedAt DATETIME NULL,
CompletedAt DATETIME NULL,
LastError LONGTEXT NULL,
CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uq_scheduled_run_reservation_identity
  (OccurrenceIdentityDigest),
KEY idx_scheduled_run_reservation_claim (Status, LeaseExpiresAt, Id),
KEY idx_scheduled_run_reservation_due (Status, DueAt, Id),
KEY idx_scheduled_run_reservation_scope_claim
  (OrgId, AppInstallationId, RuntimeCapabilityId, Status, LeaseExpiresAt),
KEY idx_scheduled_run_reservation_scope_due
  (OrgId, AppInstallationId, RuntimeCapabilityId, Status, DueAt),
KEY idx_scheduled_run_reservation_batch
  (BatchIdentityDigest))`,
		`CREATE TABLE ScheduledEffect (Id VARCHAR(255) PRIMARY KEY)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
}

func TestFencedScheduledRunReclaimLookupUsesUniqueDigestPointGet(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	scope := persistence.RuntimeWorkScope{
		OrgID: "org-point-get", AppInstallationID: "installation-point-get",
		RuntimeCapabilityID: "runtime.automation.execute-flow.v1",
	}
	key, err := NewScheduledRunReservationKey("definition-target", "version-target", "window-target", "input-target")
	if err != nil {
		t.Fatal(err)
	}
	insert, err := db.Prepare(`INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, OccurrenceIdentityDigest,
 FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey,
 State, Status, Attempt)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'COMPLETED', 'completed', 1)`)
	if err != nil {
		t.Fatal(err)
	}
	defer insert.Close()
	for i := 0; i < 256; i++ {
		fillerKey, keyErr := NewScheduledRunReservationKey(
			fmt.Sprintf("definition-%03d", i), "version", fmt.Sprintf("window-%03d", i), fmt.Sprintf("input-%03d", i),
		)
		if keyErr != nil {
			t.Fatal(keyErr)
		}
		if _, err := insert.Exec(
			fmt.Sprintf("filler-%03d", i), scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
			scheduledRunOccurrenceIdentityDigest(scope, fillerKey),
			fillerKey.FlowDefinitionID, fillerKey.FlowVersionID, fillerKey.WindowToken, fillerKey.WindowToken, fillerKey.InputKey,
		); err != nil {
			t.Fatal(err)
		}
	}
	digest := scheduledRunOccurrenceIdentityDigest(scope, key)
	if _, err := insert.Exec(
		"target", scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, digest,
		key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.WindowToken, key.InputKey,
	); err != nil {
		t.Fatal(err)
	}

	rows, err := db.QueryContext(t.Context(), "EXPLAIN ANALYZE "+scheduledRunReclaimLookupSQL,
		digest,
		scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
		key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.InputKey,
	)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	var planLines []string
	for rows.Next() {
		values := make([]any, len(columns))
		destinations := make([]any, len(columns))
		for i := range values {
			destinations[i] = &values[i]
		}
		if err := rows.Scan(destinations...); err != nil {
			t.Fatal(err)
		}
		parts := make([]string, len(values))
		for i, value := range values {
			switch value := value.(type) {
			case nil:
				parts[i] = "<nil>"
			case []byte:
				parts[i] = string(value)
			default:
				parts[i] = fmt.Sprint(value)
			}
		}
		planLines = append(planLines, strings.Join(parts, " | "))
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	plan := strings.Join(planLines, "\n")
	if !strings.Contains(plan, "Point_Get") || strings.Contains(plan, "TableFullScan") {
		t.Fatalf("reclaim lookup is not a bounded unique point get:\n%s", plan)
	}
	t.Logf("reclaim lookup plan:\n%s", plan)

	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET OccurrenceIdentityDigest = ? WHERE Id = 'target'`, strings.Repeat("f", 64)); err != nil {
		t.Fatal(err)
	}
	missing, err := db.QueryContext(t.Context(), scheduledRunReclaimLookupSQL,
		digest,
		scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
		key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.InputKey,
	)
	if err != nil {
		t.Fatal(err)
	}
	defer missing.Close()
	if missing.Next() {
		t.Fatal("reclaim lookup matched authority columns after the unique identity digest was corrupted")
	}
	if err := missing.Err(); err != nil {
		t.Fatal(err)
	}
}

func TestFencedScheduledRunRecoveryQueriesStayInsideRuntimeScopeOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createFencedScheduledRunTables(t, db)

	insert, err := db.Prepare(`INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, BatchIdentityDigest, FlowDefinitionId, FlowVersionId,
 ScheduleWindowKey, WindowToken, InputKey, State, Status, Attempt, DueAt)
VALUES (?, 'foreign-org', 'foreign-installation', 'foreign-capability', ?, 'target-definition', 'target-version', 'target-window', 'target-window',
        '__global__', 'ACCEPTED', 'accepted', 0, UTC_TIMESTAMP(6) - INTERVAL 1 MINUTE)`)
	if err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 1000; i++ {
		id := fmt.Sprintf("foreign-%04d", i)
		foreignScope := persistence.RuntimeWorkScope{
			OrgID: "foreign-org", AppInstallationID: "foreign-installation", RuntimeCapabilityID: "foreign-capability",
		}
		if _, err := insert.Exec(id, scheduledRunBatchIdentityDigest(foreignScope, "target-definition", "target-version", "target-window")); err != nil {
			_ = insert.Close()
			t.Fatal(err)
		}
	}
	if err := insert.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ANALYZE TABLE _System_ScheduledRunReservation`); err != nil {
		t.Fatal(err)
	}

	queries := []struct {
		name string
		sql  string
		args []any
	}{
		{
			name: "accepted_batch",
			sql:  scheduledRunAcceptedBatchSQL,
			args: []any{
				scheduledRunBatchIdentityDigest(persistence.RuntimeWorkScope{
					OrgID: "target-org", AppInstallationID: "target-installation", RuntimeCapabilityID: "target-capability",
				}, "target-definition", "target-version", "target-window"),
				"target-org", "target-installation", "target-capability",
				"target-definition", "target-version", "target-window",
			},
		},
		{
			name: "accepted_hint_read",
			sql:  scheduledRunRecoverableAcceptedHintReadSQL,
			args: []any{"target-org", "target-installation", "target-capability"},
		},
		{
			name: "reserved_hint_read",
			sql:  scheduledRunRecoverableReservedHintReadSQL,
			args: []any{"target-org", "target-installation", "target-capability"},
		},
		{
			name: "list",
			sql:  scheduledRunRecoverableListSQL,
			args: []any{
				"target-org", "target-installation", "target-capability", 100,
				"target-org", "target-installation", "target-capability", 100, 100,
			},
		},
	}
	for _, query := range queries {
		rows, err := db.QueryContext(t.Context(), "EXPLAIN ANALYZE "+query.sql, query.args...)
		if err != nil {
			t.Fatalf("explain %s: %v", query.name, err)
		}
		scanned, fullScan := scheduledRunPlanScanRows(t, rows)
		if fullScan || scanned > 8 {
			t.Fatalf("%s escaped the exact Runtime scope: rows=%d full_scan=%t", query.name, scanned, fullScan)
		}
		t.Logf("plan=%s scan_rows=%d full_scan=%t", query.name, scanned, fullScan)
	}
}

func scheduledRunPlanScanRows(t *testing.T, rows *sql.Rows) (int, bool) {
	t.Helper()
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	values := make([]sql.NullString, len(columns))
	targets := make([]any, len(columns))
	for i := range values {
		targets[i] = &values[i]
	}
	total, fullScan, observed := 0, false, false
	for rows.Next() {
		if err := rows.Scan(targets...); err != nil {
			t.Fatal(err)
		}
		row := make(map[string]string, len(columns))
		for i, column := range columns {
			row[column] = values[i].String
		}
		if strings.Contains(row["id"], "Point_Get") {
			observed = true
		}
		if !strings.Contains(row["id"], "Scan") {
			continue
		}
		observed = true
		count, err := strconv.ParseFloat(row["actRows"], 64)
		if err != nil {
			t.Fatalf("parse %s scan rows %q: %v", row["id"], row["actRows"], err)
		}
		total += int(count)
		fullScan = fullScan || strings.Contains(row["id"], "FullScan")
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !observed {
		t.Fatal("plan did not expose a data-access operator; explain the inner read instead of folded EXISTS")
	}
	return total, fullScan
}

func testScheduledRunCanonicalInputs() []byte {
	return []byte(`{"$Record":{"Id":"record-1"}}`)
}