package automation

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

	"github.com/google/uuid"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
)

const RuntimeScheduledFlowWorkFamily = "runtime.scheduled-flow.v1"

const (
	ScheduledRunExecutionStatusPoisoned ScheduledRunExecutionStatus = "poisoned"
	scheduledRunInputPayloadFormatV1                                = 1
	maxScheduledRunInputPayloadBytes                                = 4 << 20
	maxScheduledRunAcceptanceBatchBytes                             = 16 << 20
)

type ScheduledRunLeaseStatus string

const (
	ScheduledRunLeaseClaimed   ScheduledRunLeaseStatus = "claimed"
	ScheduledRunLeaseDuplicate ScheduledRunLeaseStatus = "duplicate"
)

var (
	ErrScheduledRunLeaseLost        = errors.New("scheduled run lease lost")
	ErrScheduledRunOutcomeUncertain = errors.New("scheduled run transaction outcome is uncertain")
)

type ScheduledRunLease struct {
	Key             ScheduledRunReservationKey
	EntryElement    string
	Status          ScheduledRunLeaseStatus
	ReservationID   string
	LeaseToken      string
	Attempt         int
	LeaseExpiresAt  time.Time
	MetadataBinding string
	CanonicalInputs []byte
}

type ScheduledRunAcceptance struct {
	Key             ScheduledRunReservationKey
	CanonicalInputs []byte
}

type ScheduledRunBatchClaim struct {
	Leases          []ScheduledRunLease
	AlreadyAccepted bool
}

type ScheduledRunCandidate struct {
	Key             ScheduledRunReservationKey
	EntryElement    string
	MetadataBinding string
}

type scheduledRunTransactionExtractor interface {
	ExtractTx(context.Context) interface{}
}

type scheduledRunTransactionOwner interface {
	scheduledRunTransactionExtractor
	RunInPessimisticTransactionWithOutcome(
		context.Context,
		func(context.Context) error,
	) (ports.TransactionCompletionOutcome, error)
}

const scheduledRunAcceptedBatchSQL = `SELECT Id, COALESCE(OccurrenceIdentityDigest, ''), InputKey, State, Status, COALESCE(WindowToken, ''), Attempt,
       COALESCE(LeaseToken, ''), LeaseExpiresAt, COALESCE(CAST(OwnershipEpoch AS CHAR), '0'),
       COALESCE(MetadataBinding, ''), InputPayloadFormatVersion, InputPayload
FROM _System_ScheduledRunReservation USE INDEX (idx_scheduled_run_reservation_batch)
WHERE BatchIdentityDigest = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND FlowDefinitionId = ? AND FlowVersionId = ? AND ScheduleWindowKey = ?
FOR UPDATE`

// FencedScheduledRunReservationService is the independent Runtime Worker
// authority for one scheduled occurrence. It reuses the existing reservation
// identity while adding a lease, attempt, and immutable metadata binding.
// Successful DB-only Flow effects and the COMPLETED disposition commit in the
// same transaction, so an expired RESERVED row is always safe to reclaim.
type FencedScheduledRunReservationService struct {
	db            *sql.DB
	transactions  scheduledRunTransactionOwner
	ownership     *persistence.RuntimeWorkOwnershipRepository
	expected      persistence.RuntimeWorkOwnership
	scope         persistence.RuntimeWorkScope
	binding       string
	leaseDuration time.Duration
	maxAttempts   int
	idFn          func() string
	tokenFn       func() string
}

func NewFencedScheduledRunReservationService(
	db *sql.DB,
	transactions scheduledRunTransactionOwner,
	expected persistence.RuntimeWorkOwnership,
	binding catalogsnapshot.RuntimeWorkBinding,
	leaseDuration time.Duration,
	maxAttempts int,
) (*FencedScheduledRunReservationService, error) {
	if db == nil || transactions == nil {
		return nil, errors.New("scheduled run reservation: database and transaction extractor are required")
	}
	if expected.WorkFamily != RuntimeScheduledFlowWorkFamily {
		return nil, fmt.Errorf("scheduled run reservation: wrong work family %q", expected.WorkFamily)
	}
	if leaseDuration < time.Second || leaseDuration > 5*time.Minute || maxAttempts <= 0 {
		return nil, errors.New("scheduled run reservation: processing bounds are invalid")
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		return nil, fmt.Errorf("scheduled run reservation: immutable metadata binding: %w", err)
	}
	scope := persistence.RuntimeWorkScope{
		OrgID:               binding.OrgID,
		AppInstallationID:   binding.AppInstallationID,
		RuntimeCapabilityID: binding.RuntimeCapabilityID,
	}
	if err := scope.Validate(); err != nil {
		return nil, fmt.Errorf("scheduled run reservation: immutable work scope: %w", err)
	}
	ownership, err := persistence.NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	return &FencedScheduledRunReservationService{
		db: db, transactions: transactions, ownership: ownership, expected: expected,
		scope: scope, binding: encoded, leaseDuration: leaseDuration, maxAttempts: maxAttempts,
		idFn:    func() string { return uuid.NewString() },
		tokenFn: func() string { return uuid.NewString() },
	}, nil
}

func (s *FencedScheduledRunReservationService) MetadataBinding() string {
	if s == nil {
		return ""
	}
	return s.binding
}

var fencedScheduledRunRequiredColumns = map[string]string{
	"Id":                        "varchar",
	"OrgId":                     "varchar",
	"AppInstallationId":         "varchar",
	"RuntimeCapabilityId":       "varchar",
	"OccurrenceIdentityDigest":  "char",
	"BatchIdentityDigest":       "char",
	"FlowDefinitionId":          "varchar",
	"FlowVersionId":             "varchar",
	"ScheduleWindowKey":         "varchar",
	"WindowToken":               "varchar",
	"InputKey":                  "varchar",
	"State":                     "varchar",
	"Status":                    "varchar",
	"Attempt":                   "int",
	"LeaseToken":                "varchar",
	"LeaseExpiresAt":            "datetime",
	"EntryElement":              "varchar",
	"DueAt":                     "datetime",
	"OwnershipEpoch":            "decimal(20,0)",
	"MetadataBinding":           "mediumtext",
	"InputPayloadFormatVersion": "bigint",
	"InputPayload":              "mediumblob",
	"ReservedAt":                "datetime",
	"CompletedAt":               "datetime",
	"LastError":                 "longtext",
	"CreatedDate":               "datetime",
	"LastModifiedDate":          "datetime",
}

func (s *FencedScheduledRunReservationService) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("scheduled run reservation: service is unavailable")
	}
	if err := s.ownership.CheckReady(ctx); err != nil {
		return err
	}
	if err := verifyFencedScheduledRunSchema(ctx, s.db); err != nil {
		return err
	}
	cutoverAt, err := verifyLegacyScheduledRunCutoverQuiescence(ctx, s.db, s.expected)
	if err != nil {
		return err
	}
	var malformedActive int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_ScheduledRunReservation
WHERE (LOWER(COALESCE(Status, '')) IN ('accepted', 'reserved')
       OR UPPER(COALESCE(State, '')) IN ('ACCEPTED', 'RESERVED'))
  AND (
       (
          (OrgId IS NULL OR AppInstallationId IS NULL OR RuntimeCapabilityId IS NULL
           OR OrgId = '' OR AppInstallationId = '' OR RuntimeCapabilityId = '')
          AND (CreatedDate IS NULL OR CreatedDate >= ?)
       )
       OR (
          OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
          AND (
             OccurrenceIdentityDigest IS NULL OR OccurrenceIdentityDigest = ''
             OR BatchIdentityDigest IS NULL OR BatchIdentityDigest = ''
             OR MetadataBinding IS NULL OR MetadataBinding = ''
             OR InputPayloadFormatVersion IS NULL OR InputPayloadFormatVersion <> ?
             OR InputPayload IS NULL OR OCTET_LENGTH(InputPayload) = 0
             OR (
                LOWER(COALESCE(Status, '')) = 'accepted'
                AND (State <> 'ACCEPTED' OR Attempt <> 0
                     OR EntryElement IS NULL OR EntryElement = '' OR DueAt IS NULL
                     OR LeaseToken IS NOT NULL OR LeaseExpiresAt IS NOT NULL
                     OR OwnershipEpoch IS NOT NULL)
             )
             OR (
                LOWER(COALESCE(Status, '')) = 'reserved'
                AND (State <> 'RESERVED' OR Attempt < 1
                     OR LeaseToken IS NULL OR LeaseToken = ''
                     OR LeaseExpiresAt IS NULL OR OwnershipEpoch IS NULL OR OwnershipEpoch = 0)
             )
             OR LOWER(COALESCE(Status, '')) NOT IN ('accepted', 'reserved')
          )
       )
  )`,
		cutoverAt,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		scheduledRunInputPayloadFormatV1,
	).Scan(&malformedActive); err != nil {
		return fmt.Errorf("scheduled run reservation: verify active authority rows: %w", err)
	}
	if malformedActive != 0 {
		return errors.New("scheduled run reservation: active rows contain missing, stale, or mixed authority")
	}
	rows, err := s.db.QueryContext(ctx, `SELECT DISTINCT MetadataBinding
FROM _System_ScheduledRunReservation
WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND LOWER(COALESCE(Status, '')) IN ('accepted', 'reserved')`,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	)
	if err != nil {
		return fmt.Errorf("scheduled run reservation: list active immutable bindings: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var encoded string
		if err := rows.Scan(&encoded); err != nil {
			return fmt.Errorf("scheduled run reservation: scan active immutable binding: %w", err)
		}
		if err := validateScheduledRunBindingScope(encoded, s.scope); err != nil {
			return fmt.Errorf("scheduled run reservation: active row immutable authority: %w", err)
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("scheduled run reservation: iterate active immutable bindings: %w", err)
	}
	return nil
}

// verifyLegacyScheduledRunCutoverQuiescence binds compatibility to the durable
// Runtime ownership transfer, not to mutation of historical legacy queue rows.
// The fenced Worker never reads unscoped rows. It rejects any such row created
// at or after the transfer and starts only after one complete legacy discovery
// lookback, proving that the stopped writer remained silent.
func verifyLegacyScheduledRunCutoverQuiescence(
	ctx context.Context,
	db *sql.DB,
	expected persistence.RuntimeWorkOwnership,
) (time.Time, error) {
	var now, cutoverAt time.Time
	if err := db.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6), LastModifiedDate
FROM _System_RuntimeWorkOwnership
WHERE WorkFamily = ? AND DeploymentId = ? AND WorkerReleaseDigest = ?
  AND OwnershipEpoch = ? AND IsActive = 1`,
		expected.WorkFamily, expected.DeploymentID, expected.WorkerReleaseDigest,
		expected.EpochDecimal(),
	).Scan(&now, &cutoverAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return time.Time{}, persistence.RuntimeWorkOwnershipLost(
				errors.New("scheduled Flow ownership cutover is missing"),
			)
		}
		return time.Time{}, fmt.Errorf(
			"scheduled run reservation: verify ownership cutover time: %w", err,
		)
	}
	now = scheduledRunDatabaseUTC(now)
	cutoverAt = scheduledRunDatabaseUTC(cutoverAt)
	if !cutoverAt.Before(now.Add(-schedulerStartupLookback)) {
		return time.Time{}, errors.New(
			"scheduled run reservation: ownership cutover still overlaps the legacy startup lookback",
		)
	}
	return cutoverAt, nil
}

// TiDB/MySQL returns UTC_TIMESTAMP as a timezone-free SQL DATETIME. The core
// database DSN parses DATETIME in loc=Local, so Time.UTC would shift an already
// UTC wall clock a second time. Reinterpret the database wall components as UTC
// at the boundary instead.
func scheduledRunDatabaseUTC(value time.Time) time.Time {
	return time.Date(
		value.Year(),
		value.Month(),
		value.Day(),
		value.Hour(),
		value.Minute(),
		value.Second(),
		value.Nanosecond(),
		time.UTC,
	)
}

func verifyFencedScheduledRunSchema(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return errors.New("scheduled run reservation: database is required")
	}
	if err := persistence.VerifyRequiredTableColumns(
		ctx,
		db,
		"_System_ScheduledRunReservation",
		fencedScheduledRunRequiredColumns,
	); err != nil {
		return fmt.Errorf("scheduled run reservation: %w", err)
	}
	if err := persistence.VerifyExactTableIndex(
		ctx,
		db,
		"_System_ScheduledRunReservation",
		"uq_scheduled_run_reservation_identity",
		true,
		[]string{"OccurrenceIdentityDigest"},
	); err != nil {
		return fmt.Errorf("scheduled run reservation: %w", err)
	}
	if err := persistence.VerifyExactTableIndex(
		ctx,
		db,
		"_System_ScheduledRunReservation",
		"idx_scheduled_run_reservation_claim",
		false,
		[]string{"Status", "LeaseExpiresAt", "Id"},
	); err != nil {
		return fmt.Errorf("scheduled run reservation: %w", err)
	}
	if err := persistence.VerifyExactTableIndex(
		ctx,
		db,
		"_System_ScheduledRunReservation",
		"idx_scheduled_run_reservation_due",
		false,
		[]string{"Status", "DueAt", "Id"},
	); err != nil {
		return fmt.Errorf("scheduled run reservation: %w", err)
	}
	for name, columns := range map[string][]string{
		"idx_scheduled_run_reservation_scope_claim": {"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Status", "LeaseExpiresAt"},
		"idx_scheduled_run_reservation_scope_due":   {"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Status", "DueAt"},
		"idx_scheduled_run_reservation_batch":       {"BatchIdentityDigest"},
	} {
		if err := persistence.VerifyExactTableIndex(
			ctx, db, "_System_ScheduledRunReservation", name, false, columns,
		); err != nil {
			return fmt.Errorf("scheduled run reservation: %w", err)
		}
	}
	return nil
}

func (s *FencedScheduledRunReservationService) Claim(
	ctx context.Context,
	key ScheduledRunReservationKey,
	canonicalInputs []byte,
) (ScheduledRunLease, error) {
	batch, err := s.ClaimBatch(ctx, []ScheduledRunAcceptance{{
		Key: key, CanonicalInputs: canonicalInputs,
	}})
	if err != nil {
		return ScheduledRunLease{}, err
	}
	key, err = normalizeScheduledRunReservationKey(key)
	if err != nil {
		return ScheduledRunLease{}, err
	}
	for _, lease := range batch.Leases {
		if lease.Key == key {
			return lease, nil
		}
	}
	return ScheduledRunLease{}, errors.New("scheduled run reservation: occurrence batch was already accepted without the requested input")
}

func validateScheduledRunCanonicalInputs(canonicalInputs []byte) error {
	if len(canonicalInputs) == 0 {
		return errors.New("scheduled run reservation: canonical inputs are required")
	}
	if len(canonicalInputs) > maxScheduledRunInputPayloadBytes {
		return errors.New("scheduled run reservation: canonical inputs exceed the bounded payload limit")
	}
	if !json.Valid(canonicalInputs) {
		return errors.New("scheduled run reservation: canonical inputs are not valid JSON")
	}
	return nil
}

// ClaimBatch atomically accepts every input selected for one scheduled
// occurrence. The discovery cursor may advance only after this transaction
// commits, so a process crash cannot leave a partially durable occurrence whose
// remaining inputs exist only in process memory.
func (s *FencedScheduledRunReservationService) ClaimBatch(
	ctx context.Context,
	acceptances []ScheduledRunAcceptance,
) (ScheduledRunBatchClaim, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return ScheduledRunBatchClaim{}, errors.New("scheduled run reservation: service is unavailable")
	}
	if len(acceptances) == 0 || len(acceptances) > scheduledFlowMaxMatched {
		return ScheduledRunBatchClaim{}, errors.New("scheduled run reservation: acceptance batch size is outside the supported bound")
	}
	normalized := make([]ScheduledRunAcceptance, len(acceptances))
	seen := make(map[string]struct{}, len(acceptances))
	totalPayloadBytes := 0
	var batchDefinitionID, batchVersionID, batchWindowToken string
	for i, acceptance := range acceptances {
		key, err := normalizeScheduledRunReservationKey(acceptance.Key)
		if err != nil {
			return ScheduledRunBatchClaim{}, err
		}
		if i == 0 {
			batchDefinitionID = key.FlowDefinitionID
			batchVersionID = key.FlowVersionID
			batchWindowToken = key.WindowToken
		} else if key.FlowDefinitionID != batchDefinitionID ||
			key.FlowVersionID != batchVersionID || key.WindowToken != batchWindowToken {
			return ScheduledRunBatchClaim{}, errors.New("scheduled run reservation: acceptance batch spans multiple schedule occurrences")
		}
		if _, duplicate := seen[key.CanonicalID()]; duplicate {
			return ScheduledRunBatchClaim{}, errors.New("scheduled run reservation: acceptance batch contains duplicate occurrence identity")
		}
		seen[key.CanonicalID()] = struct{}{}
		if err := validateScheduledRunCanonicalInputs(acceptance.CanonicalInputs); err != nil {
			return ScheduledRunBatchClaim{}, err
		}
		totalPayloadBytes += len(acceptance.CanonicalInputs)
		if totalPayloadBytes > maxScheduledRunAcceptanceBatchBytes {
			return ScheduledRunBatchClaim{}, errors.New("scheduled run reservation: acceptance batch payload exceeds the supported bound")
		}
		normalized[i] = ScheduledRunAcceptance{
			Key: key, CanonicalInputs: append([]byte(nil), acceptance.CanonicalInputs...),
		}
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return ScheduledRunBatchClaim{}, fmt.Errorf("scheduled run reservation: begin batch claim: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (ScheduledRunBatchClaim, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return ScheduledRunBatchClaim{}, errors.Join(cause, rollbackErr)
		}
		return ScheduledRunBatchClaim{}, cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: read batch claim time: %w", err))
	}
	databaseNow := now
	now = scheduledRunDatabaseUTC(databaseNow)
	rows, err := tx.QueryContext(ctx, scheduledRunAcceptedBatchSQL,
		scheduledRunBatchIdentityDigest(s.scope, batchDefinitionID, batchVersionID, batchWindowToken),
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		batchDefinitionID, batchVersionID, batchWindowToken,
	)
	if err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: load accepted batch: %w", err))
	}
	existingLeases := make([]ScheduledRunLease, 0, len(normalized))
	// The first durable acceptance, not the process's current Catalog view,
	// owns the exact execution binding for this occurrence.
	var acceptedBatchBinding string
	for rows.Next() {
		var rowLease ScheduledRunLease
		var identityDigest, inputKey, state, status, windowToken, ownershipEpoch string
		var payloadFormat sql.NullInt64
		var persistedInputs []byte
		var leaseExpiresAt sql.NullTime
		if err := rows.Scan(
			&rowLease.ReservationID, &identityDigest, &inputKey, &state, &status, &windowToken, &rowLease.Attempt,
			&rowLease.LeaseToken, &leaseExpiresAt, &ownershipEpoch, &rowLease.MetadataBinding,
			&payloadFormat, &persistedInputs,
		); err != nil {
			rows.Close()
			return rollback(fmt.Errorf("scheduled run reservation: scan accepted batch: %w", err))
		}
		if err := validateScheduledRunBindingScope(rowLease.MetadataBinding, s.scope); err != nil {
			rows.Close()
			return rollback(fmt.Errorf("scheduled run reservation: accepted occurrence immutable authority: %w", err))
		}
		if acceptedBatchBinding == "" {
			acceptedBatchBinding = rowLease.MetadataBinding
		} else if rowLease.MetadataBinding != acceptedBatchBinding {
			rows.Close()
			return rollback(errors.New("scheduled run reservation: accepted batch spans multiple immutable metadata bindings"))
		}
		if windowToken != batchWindowToken {
			rows.Close()
			return rollback(errors.New("scheduled run reservation: occurrence identity columns disagree"))
		}
		persistedEpoch, err := strconv.ParseUint(ownershipEpoch, 10, 64)
		if err != nil || persistedEpoch == 0 {
			rows.Close()
			return rollback(errors.New("scheduled run reservation: ownership epoch is absent or invalid"))
		}
		if state != legacyScheduledRunExecutionState(ScheduledRunExecutionStatus(status)) {
			rows.Close()
			return rollback(errors.New("scheduled run reservation: durable state columns disagree"))
		}
		if leaseExpiresAt.Valid {
			rowLease.LeaseExpiresAt = scheduledRunDatabaseUTC(leaseExpiresAt.Time)
		}
		switch ScheduledRunExecutionStatus(status) {
		case ScheduledRunExecutionStatusReserved:
			if !payloadFormat.Valid || payloadFormat.Int64 != scheduledRunInputPayloadFormatV1 ||
				len(persistedInputs) == 0 || len(persistedInputs) > maxScheduledRunInputPayloadBytes ||
				!json.Valid(persistedInputs) {
				rows.Close()
				return rollback(errors.New("scheduled run reservation: active occurrence has missing or unsupported canonical inputs"))
			}
			rowLease.Status = ScheduledRunLeaseDuplicate
			rowLease.CanonicalInputs = append([]byte(nil), persistedInputs...)
		case ScheduledRunExecutionStatusCompleted, ScheduledRunExecutionStatusFailed,
			ScheduledRunExecutionStatusPoisoned:
			rowLease.Status = ScheduledRunLeaseDuplicate
		default:
			rows.Close()
			return rollback(fmt.Errorf("scheduled run reservation: invalid durable status %q", status))
		}
		// The persisted batch is the acceptance authority. A changed re-scan
		// must not append new inputs, but every stored input must still have a
		// valid bounded identity.
		if _, err := NewScheduledRunReservationKey(
			batchDefinitionID, batchVersionID, batchWindowToken, inputKey,
		); err != nil {
			rows.Close()
			return rollback(err)
		}
		rowLease.Key = ScheduledRunReservationKey{
			FlowDefinitionID: batchDefinitionID,
			FlowVersionID:    batchVersionID,
			WindowToken:      batchWindowToken,
			InputKey:         inputKey,
		}
		if identityDigest != scheduledRunOccurrenceIdentityDigest(s.scope, rowLease.Key) {
			rows.Close()
			return rollback(errors.New("scheduled run reservation: occurrence identity digest disagrees with authority columns"))
		}
		existingLeases = append(existingLeases, rowLease)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return rollback(fmt.Errorf("scheduled run reservation: iterate accepted batch: %w", err))
	}
	if err := rows.Close(); err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: close accepted batch rows: %w", err))
	}
	if len(existingLeases) != 0 {
		if err := tx.Commit(); err != nil {
			return ScheduledRunBatchClaim{}, fmt.Errorf("scheduled run reservation: existing batch claim commit outcome unknown: %w", err)
		}
		return ScheduledRunBatchClaim{Leases: existingLeases, AlreadyAccepted: true}, nil
	}

	expiresAt := now.Add(s.leaseDuration)
	databaseExpiresAt := databaseNow.Add(s.leaseDuration)
	leases := make([]ScheduledRunLease, len(normalized))
	for i, acceptance := range normalized {
		key := acceptance.Key
		id, token := s.idFn(), s.tokenFn()
		identityDigest := scheduledRunOccurrenceIdentityDigest(s.scope, key)
		if _, err := tx.ExecContext(ctx, `INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId,
 OccurrenceIdentityDigest, BatchIdentityDigest,
 FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey,
 State, Status, Attempt, LeaseToken, LeaseExpiresAt, OwnershipEpoch,
 MetadataBinding, InputPayloadFormatVersion, InputPayload, ReservedAt)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'RESERVED', 'reserved', 1, ?, ?, ?, ?, ?, ?, ?)`,
			id, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			identityDigest, scheduledRunBatchIdentityDigest(s.scope, key.FlowDefinitionID, key.FlowVersionID, key.WindowToken),
			key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.WindowToken, key.InputKey,
			token, databaseExpiresAt, s.expected.EpochDecimal(), s.binding,
			scheduledRunInputPayloadFormatV1, acceptance.CanonicalInputs, databaseNow,
		); err != nil {
			return rollback(fmt.Errorf("scheduled run reservation: insert batch claim: %w", err))
		}
		leases[i] = ScheduledRunLease{
			Key: key, Status: ScheduledRunLeaseClaimed, ReservationID: id, LeaseToken: token,
			Attempt: 1, LeaseExpiresAt: expiresAt,
			MetadataBinding: s.binding,
			CanonicalInputs: append([]byte(nil), acceptance.CanonicalInputs...),
		}
	}
	if err := tx.Commit(); err != nil {
		return ScheduledRunBatchClaim{}, fmt.Errorf("scheduled run reservation: batch claim commit outcome unknown: %w", err)
	}
	return ScheduledRunBatchClaim{Leases: leases}, nil
}

func scheduledRunOccurrenceIdentityDigest(
	scope persistence.RuntimeWorkScope,
	key ScheduledRunReservationKey,
) string {
	return scheduledRunIdentityDigest(
		"scoped",
		scope.OrgID,
		scope.AppInstallationID,
		scope.RuntimeCapabilityID,
		key.FlowDefinitionID,
		key.FlowVersionID,
		key.WindowToken,
		key.InputKey,
	)
}

func scheduledRunBatchIdentityDigest(
	scope persistence.RuntimeWorkScope,
	flowDefinitionID, flowVersionID, windowToken string,
) string {
	return scheduledRunIdentityDigest(
		"scoped-batch",
		scope.OrgID,
		scope.AppInstallationID,
		scope.RuntimeCapabilityID,
		flowDefinitionID,
		flowVersionID,
		windowToken,
	)
}

const scheduledRunReclaimLookupSQL = `SELECT Id, COALESCE(OccurrenceIdentityDigest, ''), State, Status, COALESCE(WindowToken, ''), Attempt,
       COALESCE(LeaseToken, ''), LeaseExpiresAt, COALESCE(CAST(OwnershipEpoch AS CHAR), '0'),
       COALESCE(MetadataBinding, ''), InputPayloadFormatVersion, InputPayload,
       COALESCE(EntryElement, ''), DueAt
FROM _System_ScheduledRunReservation
WHERE OccurrenceIdentityDigest = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND FlowDefinitionId = ? AND FlowVersionId = ? AND ScheduleWindowKey = ? AND InputKey = ?
FOR UPDATE`

const scheduledRunRecoverableListSQL = `SELECT FlowDefinitionId, FlowVersionId, WindowToken, InputKey,
       OccurrenceIdentityDigest, EntryElement, MetadataBinding
FROM (
  (SELECT FlowDefinitionId, FlowVersionId, WindowToken, InputKey,
          COALESCE(OccurrenceIdentityDigest, '') AS OccurrenceIdentityDigest,
          COALESCE(EntryElement, '') AS EntryElement,
          COALESCE(MetadataBinding, '') AS MetadataBinding,
          DueAt AS WakeAt, Id
   FROM _System_ScheduledRunReservation USE INDEX (idx_scheduled_run_reservation_scope_due)
   WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
     AND Status = 'accepted' AND DueAt <= UTC_TIMESTAMP(6)
   ORDER BY DueAt, Id
   LIMIT ?)
  UNION ALL
  (SELECT FlowDefinitionId, FlowVersionId, WindowToken, InputKey,
          COALESCE(OccurrenceIdentityDigest, ''), COALESCE(EntryElement, ''),
          COALESCE(MetadataBinding, ''), LeaseExpiresAt AS WakeAt, Id
   FROM _System_ScheduledRunReservation USE INDEX (idx_scheduled_run_reservation_scope_claim)
   WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
     AND Status = 'reserved' AND LeaseExpiresAt <= UTC_TIMESTAMP(6)
   ORDER BY LeaseExpiresAt, Id
   LIMIT ?)
) AS recoverable
ORDER BY WakeAt, Id
LIMIT ?`

const scheduledRunRecoverableAcceptedHintReadSQL = `SELECT 1
    FROM _System_ScheduledRunReservation USE INDEX (idx_scheduled_run_reservation_scope_due)
    WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
      AND Status = 'accepted' AND DueAt <= UTC_TIMESTAMP(6)
    LIMIT 1`

const scheduledRunRecoverableReservedHintReadSQL = `SELECT 1
    FROM _System_ScheduledRunReservation USE INDEX (idx_scheduled_run_reservation_scope_claim)
    WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
      AND Status = 'reserved' AND LeaseExpiresAt <= UTC_TIMESTAMP(6)
    LIMIT 1`

const scheduledRunRecoverableHintSQL = `SELECT EXISTS(` + scheduledRunRecoverableAcceptedHintReadSQL +
	`) OR EXISTS(` + scheduledRunRecoverableReservedHintReadSQL + `)`

// Reclaim resumes an already accepted occurrence. It can never create a new
// occurrence: discovery owns acceptance, while restart recovery owns only rows
// durably accepted before the prior process stopped.
func (s *FencedScheduledRunReservationService) Reclaim(
	ctx context.Context,
	key ScheduledRunReservationKey,
) (ScheduledRunLease, error) {
	return s.reclaim(ctx, key)
}

func (s *FencedScheduledRunReservationService) reclaim(
	ctx context.Context,
	key ScheduledRunReservationKey,
) (ScheduledRunLease, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return ScheduledRunLease{}, errors.New("scheduled run reservation: service is unavailable")
	}
	key, err := normalizeScheduledRunReservationKey(key)
	if err != nil {
		return ScheduledRunLease{}, err
	}
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return ScheduledRunLease{}, fmt.Errorf("scheduled run reservation: begin claim: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (ScheduledRunLease, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return ScheduledRunLease{}, errors.Join(cause, rollbackErr)
		}
		return ScheduledRunLease{}, cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: read claim time: %w", err))
	}
	databaseNow := now
	now = scheduledRunDatabaseUTC(databaseNow)
	var existing ScheduledRunLease
	var state, status, windowToken, identityDigest string
	var payloadFormat sql.NullInt64
	var persistedInputs []byte
	var ownershipEpoch string
	var leaseExpiresAt, dueAt sql.NullTime
	expectedIdentityDigest := scheduledRunOccurrenceIdentityDigest(s.scope, key)
	err = tx.QueryRowContext(ctx, scheduledRunReclaimLookupSQL,
		expectedIdentityDigest,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		key.FlowDefinitionID, key.FlowVersionID, key.WindowToken, key.InputKey,
	).Scan(
		&existing.ReservationID, &identityDigest, &state, &status, &windowToken, &existing.Attempt, &existing.LeaseToken,
		&leaseExpiresAt, &ownershipEpoch, &existing.MetadataBinding,
		&payloadFormat, &persistedInputs, &existing.EntryElement, &dueAt,
	)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return rollback(errors.New("scheduled run reservation: recoverable occurrence no longer exists"))
		}
		return rollback(fmt.Errorf("scheduled run reservation: load reclaim: %w", err))
	}
	if leaseExpiresAt.Valid {
		existing.LeaseExpiresAt = scheduledRunDatabaseUTC(leaseExpiresAt.Time)
	}
	if dueAt.Valid {
		dueAt.Time = scheduledRunDatabaseUTC(dueAt.Time)
	}
	existing.Key = key
	if identityDigest != expectedIdentityDigest {
		return rollback(errors.New("scheduled run reservation: occurrence identity digest disagrees with authority columns"))
	}
	if err := validateScheduledRunBindingScope(existing.MetadataBinding, s.scope); err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: occurrence immutable authority: %w", err))
	}
	if windowToken != key.WindowToken {
		return rollback(errors.New("scheduled run reservation: occurrence identity columns disagree"))
	}
	persistedEpoch, err := strconv.ParseUint(ownershipEpoch, 10, 64)
	if err != nil {
		return rollback(errors.New("scheduled run reservation: ownership epoch is absent or invalid"))
	}
	if state != legacyScheduledRunExecutionState(ScheduledRunExecutionStatus(status)) {
		return rollback(errors.New("scheduled run reservation: durable state columns disagree"))
	}
	switch ScheduledRunExecutionStatus(status) {
	case ScheduledRunExecutionStatusAccepted:
		if persistedEpoch != 0 || existing.Attempt != 0 || existing.LeaseToken != "" ||
			leaseExpiresAt.Valid || !dueAt.Valid || strings.TrimSpace(existing.EntryElement) == "" {
			return rollback(errors.New("scheduled run reservation: accepted occurrence has invalid due-state authority"))
		}
	case ScheduledRunExecutionStatusReserved:
		if persistedEpoch == 0 {
			return rollback(errors.New("scheduled run reservation: reserved occurrence has no ownership epoch"))
		}
	}
	if ScheduledRunExecutionStatus(status) == ScheduledRunExecutionStatusAccepted ||
		ScheduledRunExecutionStatus(status) == ScheduledRunExecutionStatusReserved {
		if !payloadFormat.Valid || payloadFormat.Int64 != scheduledRunInputPayloadFormatV1 ||
			len(persistedInputs) == 0 || len(persistedInputs) > maxScheduledRunInputPayloadBytes {
			return rollback(errors.New("scheduled run reservation: active occurrence has missing or unsupported canonical inputs"))
		}
		if !json.Valid(persistedInputs) {
			return rollback(errors.New("scheduled run reservation: active occurrence has malformed canonical inputs"))
		}
		existing.CanonicalInputs = append([]byte(nil), persistedInputs...)
	}
	switch ScheduledRunExecutionStatus(status) {
	case ScheduledRunExecutionStatusCompleted, ScheduledRunExecutionStatusFailed,
		ScheduledRunExecutionStatusPoisoned:
		existing.Status = ScheduledRunLeaseDuplicate
	case ScheduledRunExecutionStatusAccepted:
		if dueAt.Time.After(now) {
			existing.Status = ScheduledRunLeaseDuplicate
			break
		}
		nextToken := s.tokenFn()
		nextExpiry := now.Add(s.leaseDuration)
		databaseNextExpiry := databaseNow.Add(s.leaseDuration)
		update, err := tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET State = 'RESERVED', Status = 'reserved', Attempt = 1, LeaseToken = ?,
    LeaseExpiresAt = ?, OwnershipEpoch = ?, ReservedAt = ?,
    LastError = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'accepted' AND Attempt = 0 AND DueAt <= ?
  AND LeaseToken IS NULL AND LeaseExpiresAt IS NULL AND OwnershipEpoch IS NULL`,
			nextToken, databaseNextExpiry, s.expected.EpochDecimal(), databaseNow,
			existing.ReservationID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			databaseNow,
		)
		if err != nil {
			return rollback(fmt.Errorf("scheduled run reservation: claim due occurrence: %w", err))
		}
		updated, err := update.RowsAffected()
		if err != nil || updated != 1 {
			return rollback(ErrScheduledRunLeaseLost)
		}
		existing.Status = ScheduledRunLeaseClaimed
		existing.Attempt = 1
		existing.LeaseToken = nextToken
		existing.LeaseExpiresAt = nextExpiry
	case ScheduledRunExecutionStatusReserved:
		if existing.LeaseExpiresAt.After(now) {
			existing.Status = ScheduledRunLeaseDuplicate
			break
		}
		if existing.Attempt >= s.maxAttempts {
			update, err := tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET State = 'POISONED', Status = 'poisoned', LeaseToken = NULL, LeaseExpiresAt = NULL,
    LastError = 'safe reclaim attempts exhausted', LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND LeaseToken = ? AND Attempt = ?`,
				existing.ReservationID,
				s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
				existing.LeaseToken, existing.Attempt,
			)
			if err != nil {
				return rollback(fmt.Errorf("scheduled run reservation: poison exhausted claim: %w", err))
			}
			updated, err := update.RowsAffected()
			if err != nil || updated != 1 {
				return rollback(ErrScheduledRunLeaseLost)
			}
			existing.Status = ScheduledRunLeaseDuplicate
			break
		}
		nextAttempt := existing.Attempt + 1
		nextToken := s.tokenFn()
		nextExpiry := now.Add(s.leaseDuration)
		databaseNextExpiry := databaseNow.Add(s.leaseDuration)
		update, err := tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET State = 'RESERVED', Status = 'reserved', Attempt = ?, LeaseToken = ?,
    LeaseExpiresAt = ?, OwnershipEpoch = ?, ReservedAt = ?,
    LastError = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND LeaseToken = ? AND Attempt = ?
  AND LeaseExpiresAt <= ?`,
			nextAttempt, nextToken, databaseNextExpiry, s.expected.EpochDecimal(), databaseNow,
			existing.ReservationID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			existing.LeaseToken, existing.Attempt, databaseNow,
		)
		if err != nil {
			return rollback(fmt.Errorf("scheduled run reservation: reclaim: %w", err))
		}
		updated, err := update.RowsAffected()
		if err != nil || updated != 1 {
			return rollback(ErrScheduledRunLeaseLost)
		}
		existing.Status = ScheduledRunLeaseClaimed
		existing.Attempt = nextAttempt
		existing.LeaseToken = nextToken
		existing.LeaseExpiresAt = nextExpiry
	default:
		return rollback(fmt.Errorf("scheduled run reservation: invalid durable status %q", status))
	}
	if err := tx.Commit(); err != nil {
		return ScheduledRunLease{}, fmt.Errorf("scheduled run reservation: duplicate claim commit outcome unknown: %w", err)
	}
	return existing, nil
}

// ListRecoverable returns due ACCEPTED timed paths and expired RESERVED
// attempts in this work scope. Every candidate carries its own immutable
// Runtime binding; recovery reopens that exact revision rather than selecting
// the binding that happened to be current when this worker process started.
// Claiming remains a separate fenced step so every preparation failure belongs
// to one durable attempt and can receive a bounded retryable or terminal
// disposition before the scan moves on.
func (s *FencedScheduledRunReservationService) ListRecoverable(
	ctx context.Context,
	limit int,
) ([]ScheduledRunCandidate, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return nil, errors.New("scheduled run reservation: service is unavailable")
	}
	if limit <= 0 || limit > 100 {
		return nil, errors.New("scheduled run reservation: recovery limit must be between 1 and 100")
	}
	// This is a bounded negative hint, not an ownership or claim decision. A
	// positive result still enters the unchanged pessimistic transaction below,
	// locks the exact active deployment, and rechecks the same eligibility
	// predicate. A failed hint deliberately falls through to that authoritative
	// path. A row racing a successful zero waits only until the next Worker poll.
	if pending, hintErr := s.hasRecoverableHint(ctx); hintErr == nil && !pending {
		return nil, nil
	}
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return nil, fmt.Errorf("scheduled run reservation: begin recovery scan: %w", err)
	}
	defer func() { _ = cleanup() }()
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		_ = tx.Rollback()
		return nil, err
	}
	rows, err := tx.QueryContext(ctx, scheduledRunRecoverableListSQL,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		limit,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		limit, limit,
	)
	if err != nil {
		_ = tx.Rollback()
		return nil, fmt.Errorf("scheduled run reservation: list recoverable occurrences: %w", err)
	}
	defer rows.Close()
	candidates := make([]ScheduledRunCandidate, 0, limit)
	for rows.Next() {
		var candidate ScheduledRunCandidate
		var identityDigest string
		if err := rows.Scan(
			&candidate.Key.FlowDefinitionID, &candidate.Key.FlowVersionID,
			&candidate.Key.WindowToken, &candidate.Key.InputKey,
			&identityDigest, &candidate.EntryElement, &candidate.MetadataBinding,
		); err != nil {
			_ = tx.Rollback()
			return nil, fmt.Errorf("scheduled run reservation: scan recoverable occurrence: %w", err)
		}
		candidate.Key, err = normalizeScheduledRunReservationKey(candidate.Key)
		if err != nil {
			_ = tx.Rollback()
			return nil, fmt.Errorf("scheduled run reservation: invalid recoverable occurrence: %w", err)
		}
		if identityDigest != scheduledRunOccurrenceIdentityDigest(s.scope, candidate.Key) {
			_ = tx.Rollback()
			return nil, errors.New("scheduled run reservation: recoverable occurrence identity digest disagrees with authority columns")
		}
		if err := validateScheduledRunBindingScope(candidate.MetadataBinding, s.scope); err != nil {
			_ = tx.Rollback()
			return nil, fmt.Errorf("scheduled run reservation: recoverable occurrence immutable authority: %w", err)
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		_ = tx.Rollback()
		return nil, fmt.Errorf("scheduled run reservation: iterate recoverable occurrences: %w", err)
	}
	if err := rows.Close(); err != nil {
		_ = tx.Rollback()
		return nil, fmt.Errorf("scheduled run reservation: close recoverable occurrence rows: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("scheduled run reservation: recovery scan commit outcome unknown: %w", err)
	}
	return candidates, nil
}

func (s *FencedScheduledRunReservationService) hasRecoverableHint(ctx context.Context) (bool, error) {
	if s == nil || s.db == nil {
		return false, errors.New("scheduled run reservation: service is unavailable")
	}
	var present int
	err := s.db.QueryRowContext(ctx, scheduledRunRecoverableHintSQL,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(&present)
	if err != nil {
		return false, fmt.Errorf("scheduled run reservation: recoverable hint: %w", err)
	}
	return present != 0, nil
}

// RefreshLease gives a locally queued claimed occurrence a fresh execution
// window immediately before it enters the effect transaction. The same
// ownership and lease-token CAS used by Execute prevents an expired former
// owner from reviving work already reclaimed elsewhere.
func (s *FencedScheduledRunReservationService) RefreshLease(
	ctx context.Context,
	lease ScheduledRunLease,
) (ScheduledRunLease, error) {
	if err := s.validateLease(lease); err != nil {
		return ScheduledRunLease{}, err
	}
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return ScheduledRunLease{}, fmt.Errorf("scheduled run reservation: begin lease refresh: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (ScheduledRunLease, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return ScheduledRunLease{}, errors.Join(cause, rollbackErr)
		}
		return ScheduledRunLease{}, cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: read lease refresh time: %w", err))
	}
	databaseNow := now
	now = scheduledRunDatabaseUTC(databaseNow)
	refreshedExpiry := now.Add(s.leaseDuration)
	databaseRefreshedExpiry := databaseNow.Add(s.leaseDuration)
	result, err := tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND Attempt = ? AND LeaseToken = ?
  AND OwnershipEpoch = ?`,
		databaseRefreshedExpiry,
		lease.ReservationID,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		lease.Attempt, lease.LeaseToken, s.expected.EpochDecimal(),
	)
	if err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: refresh lease: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrScheduledRunLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return ScheduledRunLease{}, fmt.Errorf("scheduled run reservation: lease refresh outcome unknown: %w", err)
	}
	lease.LeaseExpiresAt = refreshedExpiry
	return lease, nil
}

// RecordPreparationFailure durably disposes a claimed occurrence that could
// not reach the effect transaction. A retryable failure keeps the exact
// attempt RESERVED until its existing lease expires, which removes it from
// the recovery scan window while preserving bounded retry/poison behavior.
// A permanent failure terminalizes the occurrence as FAILED immediately.
func (s *FencedScheduledRunReservationService) RecordPreparationFailure(
	ctx context.Context,
	lease ScheduledRunLease,
	failureClass ScheduledRunPreparationFailureClass,
	lastError string,
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	switch failureClass {
	case ScheduledRunPreparationFailureRetryable, ScheduledRunPreparationFailurePermanent:
	default:
		return fmt.Errorf("scheduled run reservation: invalid preparation failure class %q", failureClass)
	}
	lastError = strings.TrimSpace(lastError)
	if lastError == "" {
		return errors.New("scheduled run reservation: preparation failure detail is required")
	}
	const maxPreparationFailureBytes = 64 << 10
	if len(lastError) > maxPreparationFailureBytes {
		lastError = lastError[:maxPreparationFailureBytes]
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("scheduled run reservation: begin preparation failure disposition: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) error {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return errors.Join(cause, rollbackErr)
		}
		return cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}

	var result sql.Result
	switch failureClass {
	case ScheduledRunPreparationFailureRetryable:
		result, err = tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND Attempt = ? AND LeaseToken = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
			lastError,
			lease.ReservationID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			lease.Attempt, lease.LeaseToken, s.expected.EpochDecimal(),
		)
	case ScheduledRunPreparationFailurePermanent:
		result, err = tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET State = 'FAILED', Status = 'failed', CompletedAt = UTC_TIMESTAMP(6),
    LeaseToken = NULL, LeaseExpiresAt = NULL, LastError = ?,
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND Attempt = ? AND LeaseToken = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
			lastError,
			lease.ReservationID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			lease.Attempt, lease.LeaseToken, s.expected.EpochDecimal(),
		)
	}
	if err != nil {
		return rollback(fmt.Errorf("scheduled run reservation: record preparation failure: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrScheduledRunLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("scheduled run reservation: preparation failure disposition outcome unknown: %w", err)
	}
	return nil
}

// Execute owns the complete effect transaction for one claimed occurrence.
// It locks the work-family authority and exact reservation attempt before
// invoking the existing Flow engine, then marks the occurrence COMPLETED in
// the same transaction. Reclaim uses the same lock order, so it cannot race a
// long-running effect after the initial lease expires; a process crash rolls
// back both the effect and terminal disposition before the row becomes
// reclaimable.
func (s *FencedScheduledRunReservationService) Execute(
	ctx context.Context,
	lease ScheduledRunLease,
	effect func(context.Context) error,
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if effect == nil {
		return errors.New("scheduled run reservation: effect callback is required")
	}
	completion, err := s.transactions.RunInPessimisticTransactionWithOutcome(ctx, func(txCtx context.Context) error {
		tx, ok := s.transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok || tx == nil {
			return errors.New("scheduled run reservation: effect transaction is required")
		}
		if err := s.ownership.LockActive(txCtx, tx); err != nil {
			return err
		}
		var status, token, epoch string
		var attempt int
		var expiresAt time.Time
		if err := tx.QueryRowContext(txCtx, `SELECT Status, Attempt, COALESCE(LeaseToken, ''),
       LeaseExpiresAt, COALESCE(CAST(OwnershipEpoch AS CHAR), '0')
FROM _System_ScheduledRunReservation
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`,
			lease.ReservationID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		).Scan(&status, &attempt, &token, &expiresAt, &epoch); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrScheduledRunLeaseLost
			}
			return fmt.Errorf("scheduled run reservation: lock effect attempt: %w", err)
		}
		if status != string(ScheduledRunExecutionStatusReserved) ||
			attempt != lease.Attempt || token != lease.LeaseToken ||
			epoch != s.expected.EpochDecimal() {
			return ErrScheduledRunLeaseLost
		}
		var now time.Time
		if err := tx.QueryRowContext(txCtx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
			return fmt.Errorf("scheduled run reservation: read effect time: %w", err)
		}
		now = scheduledRunDatabaseUTC(now)
		expiresAt = scheduledRunDatabaseUTC(expiresAt)
		if !expiresAt.After(now) {
			return ErrScheduledRunLeaseLost
		}
		if err := effect(txCtx); err != nil {
			return err
		}
		if err := txCtx.Err(); err != nil {
			return err
		}
		result, err := tx.ExecContext(txCtx, `UPDATE _System_ScheduledRunReservation
SET State = 'COMPLETED', Status = 'completed', CompletedAt = UTC_TIMESTAMP(6),
    LeaseToken = NULL, LeaseExpiresAt = NULL, LastError = NULL,
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND Attempt = ? AND LeaseToken = ? AND OwnershipEpoch = ?`,
			lease.ReservationID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			lease.Attempt, lease.LeaseToken, s.expected.EpochDecimal(),
		)
		if err != nil {
			return fmt.Errorf("scheduled run reservation: complete effect: %w", err)
		}
		affected, err := result.RowsAffected()
		if err != nil || affected != 1 {
			return ErrScheduledRunLeaseLost
		}
		return nil
	})
	if err != nil && completion != ports.TransactionCompletionRolledBack {
		return fmt.Errorf("%w: %v", ErrScheduledRunOutcomeUncertain, err)
	}
	return err
}

func (s *FencedScheduledRunReservationService) Complete(
	ctx context.Context,
	lease ScheduledRunLease,
	status ScheduledRunExecutionStatus,
	lastError string,
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if status != ScheduledRunExecutionStatusCompleted && status != ScheduledRunExecutionStatusFailed {
		return fmt.Errorf("scheduled run reservation: invalid terminal status %q", status)
	}
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return err
	}
	defer func() { _ = cleanup() }()
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		_ = tx.Rollback()
		return err
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_ScheduledRunReservation
SET State = ?, Status = ?, CompletedAt = UTC_TIMESTAMP(6), LastError = ?,
    LeaseToken = NULL, LeaseExpiresAt = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status = 'reserved' AND Attempt = ? AND LeaseToken = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
		legacyScheduledRunExecutionState(status), string(status), nullableScheduledRunLastError(lastError),
		lease.ReservationID,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		lease.Attempt, lease.LeaseToken, s.expected.EpochDecimal(),
	)
	if err != nil {
		_ = tx.Rollback()
		return fmt.Errorf("scheduled run reservation: complete: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		_ = tx.Rollback()
		return ErrScheduledRunLeaseLost
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("scheduled run reservation: completion outcome unknown: %w", err)
	}
	return nil
}

func (s *FencedScheduledRunReservationService) validateLease(lease ScheduledRunLease) error {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("scheduled run reservation: service is unavailable")
	}
	if err := validateScheduledRunLease(lease); err != nil {
		return err
	}
	if err := validateScheduledRunBindingScope(lease.MetadataBinding, s.scope); err != nil {
		return fmt.Errorf("scheduled run reservation: lease immutable authority: %w", err)
	}
	return nil
}

func validateScheduledRunBindingScope(
	encoded string,
	scope persistence.RuntimeWorkScope,
) error {
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(encoded)
	if err != nil {
		return fmt.Errorf("open immutable metadata binding: %w", err)
	}
	if binding.OrgID != scope.OrgID ||
		binding.AppInstallationID != scope.AppInstallationID ||
		binding.RuntimeCapabilityID != scope.RuntimeCapabilityID {
		return errors.New("immutable metadata binding scope does not match the scheduled work scope")
	}
	return nil
}

func validateScheduledRunLease(lease ScheduledRunLease) error {
	if lease.Status != ScheduledRunLeaseClaimed || strings.TrimSpace(lease.ReservationID) == "" ||
		strings.TrimSpace(lease.LeaseToken) == "" || lease.Attempt <= 0 ||
		lease.LeaseExpiresAt.IsZero() ||
		strings.TrimSpace(lease.MetadataBinding) == "" {
		return errors.New("scheduled run reservation: active lease is incomplete")
	}
	return nil
}