package persistence

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/nexuscrm/kernel/pkg/ports"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

const RuntimePlatformCommandExecutionWorkFamily = "runtime.platform-command.v1"

const (
	runtimePlatformCommandTerminalResultMaxBytes = 1 << 20
	// Publication commands persist a bounded semantic post-image before the
	// Catalog ref exists. Three MiB matches the largest admitted neutral
	// response while its base64 form remains well inside MEDIUMTEXT.
	runtimePlatformCommandPublicationPendingResultMaxBytes = 3 << 20
	// ResultPayload is encoded with runtimePlatformCommandResultPrefix and base64
	// into a MEDIUMTEXT/MEDIUMBLOB column. Keep configurable publication bounds
	// inside that physical representation instead of discovering the limit at
	// completion time.
	runtimePlatformCommandResultStorageMaxBytes    = (1 << 24) - 1
	runtimePlatformCommandResultRawStorageMaxBytes = (runtimePlatformCommandResultStorageMaxBytes - len(runtimePlatformCommandResultPrefix)) / 4 * 3
)

const (
	RuntimePlatformCommandProcessing = "PROCESSING"
	RuntimePlatformCommandPublishing = "PUBLISHING"
	RuntimePlatformCommandBlocked    = "BLOCKED"
	RuntimePlatformCommandCompleted  = "COMPLETED"
	RuntimePlatformCommandFailed     = "FAILED"
	RuntimePlatformCommandPoisoned   = "POISONED"
)

var (
	ErrRuntimePlatformCommandLeaseLost                 = errors.New("runtime platform command execution lease lost")
	ErrRuntimePlatformCommandOutcomeUncertain          = errors.New("runtime platform command transaction outcome is uncertain")
	ErrRuntimePlatformCommandResultInvalid             = errors.New("runtime platform command terminal result is invalid")
	ErrRuntimePlatformCommandDependencyChanged         = errors.New("runtime platform command dependency changed across exact replay")
	ErrRuntimePlatformCommandApplicationResultRequired = errors.New("runtime platform command v3 requires the ApplicationResult owner")
)

var errRuntimePlatformCommandRollbackForTerminalResult = errors.New(
	"runtime platform command application failure requires rolled-back effect",
)

const runtimePlatformCommandDependencyMaxWait = 30 * time.Minute
const runtimePlatformCommandDefaultPublicationAttempts = 3

type RuntimePlatformCommandExecutionOption func(*runtimePlatformCommandExecutionConfig) error

type runtimePlatformCommandExecutionConfig struct {
	maxPublicationAttempts            int
	publicationTerminalResultMaxBytes int
}

// WithRuntimePlatformCommandPublicationAttempts configures only the
// post-commit publication/reconciliation budget. It never changes or resets
// the business execution attempt budget.
func WithRuntimePlatformCommandPublicationAttempts(maxAttempts int) RuntimePlatformCommandExecutionOption {
	return func(config *runtimePlatformCommandExecutionConfig) error {
		if maxAttempts <= 0 || maxAttempts > 100 {
			return errors.New("runtime platform command execution: publication bounds are invalid")
		}
		config.maxPublicationAttempts = maxAttempts
		return nil
	}
}

// WithRuntimePlatformCommandPublicationTerminalResultMaxBytes declares the
// durable-envelope bound owned by the publication implementation. The generic
// state store must not infer that a verified terminal representation is the
// same size as its pending post-image.
func WithRuntimePlatformCommandPublicationTerminalResultMaxBytes(maxBytes int) RuntimePlatformCommandExecutionOption {
	return func(config *runtimePlatformCommandExecutionConfig) error {
		if maxBytes < runtimePlatformCommandPublicationPendingResultMaxBytes ||
			maxBytes > runtimePlatformCommandResultRawStorageMaxBytes {
			return errors.New("runtime platform command execution: publication terminal result bound is invalid")
		}
		config.publicationTerminalResultMaxBytes = maxBytes
		return nil
	}
}

type runtimePlatformCommandTransactionOwner interface {
	ExtractTx(context.Context) interface{}
	RunInPessimisticTransactionWithOutcome(
		context.Context,
		func(context.Context) error,
	) (ports.TransactionCompletionOutcome, error)
}

type RuntimePlatformCommandLease struct {
	Row               RuntimePlatformCommandRow
	LeaseToken        string
	Attempt           int
	AttemptsRemaining int
	LeaseExpiresAt    time.Time
	OwnershipEpoch    uint64
	ImmutableBinding  string
	CallerWorkloadID  string
	RuntimeCapability string
	AppInstallationID string
	OrganizationID    string
}

type RuntimePlatformCommandExecutionState struct {
	QueueDepth  int64
	PoisonDepth int64
	OldestAge   time.Duration
}

// RuntimePlatformCommandObservationStore owns only the diagnostic read path.
// It cannot claim, execute, retry, poison, or repair a durable command.
type RuntimePlatformCommandObservationStore struct {
	db          *sql.DB
	ownership   *RuntimeWorkOwnershipRepository
	scope       RuntimeWorkScope
	scopeDigest string
}

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

// RuntimePlatformCommandExecutionStore owns the accepted-command state
// transition from one immutable acceptance authority to one bounded,
// transactionally fenced Runtime effect. External provider effects are not
// permitted through this owner; Integration retains those effects.
type RuntimePlatformCommandExecutionStore struct {
	db                                *sql.DB
	transactions                      runtimePlatformCommandTransactionOwner
	ownership                         *RuntimeWorkOwnershipRepository
	expected                          RuntimeWorkOwnership
	scope                             RuntimeWorkScope
	scopeDigest                       string
	leaseDuration                     time.Duration
	maxAttempts                       int
	maxPublicationAttempts            int
	publicationTerminalResultMaxBytes int
	tokenFn                           func() string
}

func NewRuntimePlatformCommandExecutionStore(
	db *sql.DB,
	transactions runtimePlatformCommandTransactionOwner,
	expected RuntimeWorkOwnership,
	scope RuntimeWorkScope,
	leaseDuration time.Duration,
	maxAttempts int,
	options ...RuntimePlatformCommandExecutionOption,
) (*RuntimePlatformCommandExecutionStore, error) {
	if transactions == nil {
		return nil, errors.New("runtime platform command execution: transaction owner is required")
	}
	return newRuntimePlatformCommandExecutionStore(
		db,
		transactions,
		expected,
		scope,
		leaseDuration,
		maxAttempts,
		options...,
	)
}

// NewRuntimePlatformCommandRepairStore opens only the fenced state owner used
// by the short-lived compare-and-set repair role. It cannot execute a command
// because it has no effect transaction owner.
func NewRuntimePlatformCommandRepairStore(
	db *sql.DB,
	expected RuntimeWorkOwnership,
	scope RuntimeWorkScope,
) (*RuntimePlatformCommandExecutionStore, error) {
	return newRuntimePlatformCommandExecutionStore(
		db,
		nil,
		expected,
		scope,
		time.Second,
		1,
	)
}

func NewRuntimePlatformCommandObservationStore(
	db *sql.DB,
	expected RuntimeWorkOwnership,
	scope RuntimeWorkScope,
) (*RuntimePlatformCommandObservationStore, error) {
	if db == nil || scope.Validate() != nil {
		return nil, errors.New("runtime platform command observation: database and scope are required")
	}
	if expected.WorkFamily != RuntimePlatformCommandExecutionWorkFamily {
		return nil, fmt.Errorf("runtime platform command observation: wrong work family %q", expected.WorkFamily)
	}
	ownership, err := NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	scopeDigest, err := scope.Digest()
	if err != nil {
		return nil, err
	}
	return &RuntimePlatformCommandObservationStore{
		db: db, ownership: ownership, scope: scope, scopeDigest: scopeDigest,
	}, nil
}

func newRuntimePlatformCommandExecutionStore(
	db *sql.DB,
	transactions runtimePlatformCommandTransactionOwner,
	expected RuntimeWorkOwnership,
	scope RuntimeWorkScope,
	leaseDuration time.Duration,
	maxAttempts int,
	options ...RuntimePlatformCommandExecutionOption,
) (*RuntimePlatformCommandExecutionStore, error) {
	if db == nil || scope.Validate() != nil {
		return nil, errors.New("runtime platform command execution: database and scope are required")
	}
	if expected.WorkFamily != RuntimePlatformCommandExecutionWorkFamily {
		return nil, fmt.Errorf("runtime platform command execution: wrong work family %q", expected.WorkFamily)
	}
	if leaseDuration < time.Second || leaseDuration > 5*time.Minute || maxAttempts <= 0 || maxAttempts > 100 {
		return nil, errors.New("runtime platform command execution: processing bounds are invalid")
	}
	config := runtimePlatformCommandExecutionConfig{
		maxPublicationAttempts:            runtimePlatformCommandDefaultPublicationAttempts,
		publicationTerminalResultMaxBytes: runtimePlatformCommandPublicationPendingResultMaxBytes,
	}
	for _, option := range options {
		if option == nil {
			return nil, errors.New("runtime platform command execution: option is nil")
		}
		if err := option(&config); err != nil {
			return nil, err
		}
	}
	ownership, err := NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	scopeDigest, err := scope.Digest()
	if err != nil {
		return nil, err
	}
	return &RuntimePlatformCommandExecutionStore{
		db: db, transactions: transactions, ownership: ownership, expected: expected,
		scope: scope, scopeDigest: scopeDigest, leaseDuration: leaseDuration, maxAttempts: maxAttempts,
		maxPublicationAttempts:            config.maxPublicationAttempts,
		publicationTerminalResultMaxBytes: config.publicationTerminalResultMaxBytes,
		tokenFn:                           func() string { return uuid.NewString() },
	}, nil
}

func (s *RuntimePlatformCommandExecutionStore) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("runtime platform command execution: store is unavailable")
	}
	if err := s.ownership.CheckReady(ctx); err != nil {
		return err
	}
	if err := s.ownership.CheckFencesReady(ctx); err != nil {
		return err
	}
	if err := VerifyRuntimePlatformCommandSchema(ctx, s.db); err != nil {
		return err
	}
	var malformed int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_RuntimePlatformCommand
WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND (
    ExecutionScopeDigest <> ? OR MetadataBinding IS NULL OR MetadataBinding = '' OR Attempt < 0 OR PublishAttempt < 0 OR
    (State <> 'POISONED' AND RecoveryState IS NOT NULL) OR
    (State = 'POISONED' AND COALESCE(RecoveryState, 'ACCEPTED') NOT IN ('ACCEPTED', 'PUBLISHING')) OR
    (State = 'POISONED' AND RecoveryState = 'PUBLISHING' AND
      (PublicationIdentity IS NULL OR PublicationIdentity = '' OR ResultPayload IS NULL OR ResultPayload = '')) OR
    (State = 'ACCEPTED' AND
      (LeaseToken IS NOT NULL OR LeaseExpiresAt IS NOT NULL OR OwnershipEpoch IS NOT NULL OR
       DependencyKind IS NOT NULL OR DependencyObjectAPIName IS NOT NULL OR
       DependencyRecordId IS NOT NULL OR DependencyKeyHash IS NOT NULL OR
       DependencyConnectorOwnerKey IS NOT NULL OR DependencyConnectorNamespace IS NOT NULL OR
       DependencyConnectorSourceObject IS NOT NULL OR DependencyDestinationSlotId IS NOT NULL OR
       DependencyConnectorRemoteId IS NOT NULL OR DependencyBlockedAt IS NOT NULL)) OR
    (State = 'PROCESSING' AND
      (Attempt = 0 OR LeaseToken IS NULL OR LeaseToken = '' OR
       LeaseExpiresAt IS NULL OR OwnershipEpoch IS NULL OR OwnershipEpoch = 0 OR
       (DependencyKind IS NULL AND
         (DependencyObjectAPIName IS NOT NULL OR DependencyRecordId IS NOT NULL OR
          DependencyKeyHash IS NOT NULL OR DependencyConnectorOwnerKey IS NOT NULL OR
          DependencyConnectorNamespace IS NOT NULL OR DependencyConnectorSourceObject IS NOT NULL OR
          DependencyDestinationSlotId IS NOT NULL OR DependencyConnectorRemoteId IS NOT NULL OR
          DependencyBlockedAt IS NOT NULL)) OR
       (DependencyKind IS NOT NULL AND
         (DependencyObjectAPIName IS NULL OR DependencyRecordId IS NULL OR
          DependencyKeyHash IS NULL OR DependencyConnectorOwnerKey IS NULL OR
          DependencyConnectorNamespace IS NULL OR DependencyConnectorSourceObject IS NULL OR
          DependencyDestinationSlotId IS NULL OR DependencyConnectorRemoteId IS NULL OR
          DependencyBlockedAt IS NULL)))) OR
    (State = 'PUBLISHING' AND
      (Attempt = 0 OR ResultPayload IS NULL OR ResultPayload = '' OR
       PublicationIdentity IS NULL OR PublicationIdentity = '' OR RecoveryState IS NOT NULL OR
       DependencyKind IS NOT NULL OR DependencyObjectAPIName IS NOT NULL OR
       DependencyRecordId IS NOT NULL OR DependencyKeyHash IS NOT NULL OR
       DependencyConnectorOwnerKey IS NOT NULL OR DependencyConnectorNamespace IS NOT NULL OR
       DependencyConnectorSourceObject IS NOT NULL OR DependencyDestinationSlotId IS NOT NULL OR
       DependencyConnectorRemoteId IS NOT NULL OR DependencyBlockedAt IS NOT NULL OR
       ((LeaseToken IS NULL OR LeaseToken = '') AND
         (LeaseExpiresAt IS NOT NULL OR OwnershipEpoch IS NOT NULL)) OR
       (LeaseToken IS NOT NULL AND LeaseToken <> '' AND
         (LeaseExpiresAt IS NULL OR OwnershipEpoch IS NULL OR OwnershipEpoch = 0)))) OR
    (State = 'BLOCKED' AND
      (Attempt = 0 OR LeaseToken IS NOT NULL OR LeaseExpiresAt IS NOT NULL OR OwnershipEpoch IS NOT NULL OR
       ResultPayload IS NOT NULL OR DependencyKind IS NULL OR DependencyObjectAPIName IS NULL OR
       DependencyRecordId IS NULL OR DependencyKeyHash IS NULL OR DependencyConnectorOwnerKey IS NULL OR
       DependencyConnectorNamespace IS NULL OR DependencyConnectorSourceObject IS NULL OR
       DependencyDestinationSlotId IS NULL OR DependencyConnectorRemoteId IS NULL OR DependencyBlockedAt IS NULL)) OR
    (State = 'COMPLETED' AND
      (Attempt = 0 OR LeaseToken IS NULL OR LeaseToken = '' OR
       LeaseExpiresAt IS NULL OR OwnershipEpoch IS NULL OR OwnershipEpoch = 0 OR
       ResultPayload IS NULL OR ResultPayload = '' OR DependencyKind IS NOT NULL OR
       DependencyObjectAPIName IS NOT NULL OR DependencyRecordId IS NOT NULL OR
       DependencyKeyHash IS NOT NULL OR DependencyConnectorOwnerKey IS NOT NULL OR
       DependencyConnectorNamespace IS NOT NULL OR DependencyConnectorSourceObject IS NOT NULL OR
       DependencyDestinationSlotId IS NOT NULL OR DependencyConnectorRemoteId IS NOT NULL OR
       DependencyBlockedAt IS NOT NULL)) OR
    (State IN ('FAILED', 'POISONED') AND
      (LeaseToken IS NOT NULL OR LeaseExpiresAt IS NOT NULL OR OwnershipEpoch IS NOT NULL OR
       DependencyKind IS NOT NULL OR DependencyObjectAPIName IS NOT NULL OR
       DependencyRecordId IS NOT NULL OR DependencyKeyHash IS NOT NULL OR
       DependencyConnectorOwnerKey IS NOT NULL OR DependencyConnectorNamespace IS NOT NULL OR
       DependencyConnectorSourceObject IS NOT NULL OR DependencyDestinationSlotId IS NOT NULL OR
       DependencyConnectorRemoteId IS NOT NULL OR DependencyBlockedAt IS NOT NULL)) OR
    State NOT IN ('ACCEPTED', 'PROCESSING', 'PUBLISHING', 'BLOCKED', 'COMPLETED', 'FAILED', 'POISONED')
  )`,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID, s.scopeDigest,
	).Scan(&malformed); err != nil {
		return fmt.Errorf("runtime platform command execution: verify durable authority: %w", err)
	}
	if malformed != 0 {
		return errors.New("runtime platform command execution: durable rows contain missing or mixed authority")
	}
	var duplicateSerializationOwners int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM (
  SELECT SerializationKeyDigest
  FROM _System_RuntimePlatformCommand
  WHERE ExecutionScopeDigest = ?
    AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
    AND SerializationKeyDigest IS NOT NULL AND SerializationKeyDigest <> ''
    AND (
      State IN ('BLOCKED', 'PUBLISHING')
      OR (State = 'POISONED' AND RecoveryState = 'PUBLISHING')
      OR (State = 'PROCESSING' AND LeaseExpiresAt > UTC_TIMESTAMP(6))
    )
  GROUP BY SerializationKeyDigest
  HAVING COUNT(*) > 1
) duplicate_owner_keys`,
		s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(&duplicateSerializationOwners); err != nil {
		return fmt.Errorf("runtime platform command execution: verify serialization owners: %w", err)
	}
	if duplicateSerializationOwners != 0 {
		return errors.New("runtime platform command execution: multiple active owners share a serialization key")
	}
	if err := s.verifyDependencies(ctx); err != nil {
		return err
	}
	if err := verifyRuntimePlatformApplicationResults(ctx, s.db, s.scope); err != nil {
		return err
	}
	return nil
}

func (s *RuntimePlatformCommandExecutionStore) verifyDependencies(ctx context.Context) error {
	rows, err := s.db.QueryContext(ctx, `SELECT DependencyKind, DependencyObjectAPIName,
       DependencyRecordId, DependencyKeyHash, DependencyConnectorOwnerKey,
       DependencyConnectorNamespace, DependencyConnectorSourceObject,
       DependencyDestinationSlotId, DependencyConnectorRemoteId
FROM _System_RuntimePlatformCommand
WHERE ExecutionScopeDigest = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND DependencyKind IS NOT NULL`,
		s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	)
	if err != nil {
		return fmt.Errorf("runtime platform command execution: verify durable dependencies: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var dependency contractmodels.FlowExecutionDependency
		if err := rows.Scan(
			&dependency.Kind, &dependency.ObjectAPIName, &dependency.RecordID,
			&dependency.KeyHash, &dependency.ConnectorOwnerKey,
			&dependency.ConnectorNamespace, &dependency.ConnectorSourceObject,
			&dependency.DestinationSlotID, &dependency.ConnectorRemoteID,
		); err != nil {
			return fmt.Errorf("runtime platform command execution: scan durable dependency: %w", err)
		}
		if err := dependency.Validate(); err != nil {
			return errors.New("runtime platform command execution: durable dependency is invalid")
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("runtime platform command execution: verify durable dependencies: %w", err)
	}
	return nil
}

// HasWork is a negative-only admission hint over the same scope and due states
// as Claim. Claim remains the sole durable authority for ordering, ownership,
// lease recovery, dependency expiry, and state transition.
func (s *RuntimePlatformCommandExecutionStore) HasWork(ctx context.Context) (bool, error) {
	if s == nil || s.db == nil {
		return false, errors.New("runtime platform command execution: store is unavailable")
	}
	var exists bool
	if err := s.db.QueryRowContext(ctx, `SELECT EXISTS(
  SELECT 1
  FROM _System_RuntimePlatformCommand USE INDEX (idx_runtime_platform_command_claim)
  WHERE ExecutionScopeDigest = ?
    AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
    AND ((State = 'ACCEPTED' AND AvailableAt <= UTC_TIMESTAMP(6))
         OR (State = 'PROCESSING' AND LeaseExpiresAt <= UTC_TIMESTAMP(6))
         OR (State = 'BLOCKED' AND AvailableAt <= UTC_TIMESTAMP(6)))
  LIMIT 1
)`, s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID).Scan(&exists); err != nil {
		return false, fmt.Errorf("runtime platform command execution: hint: %w", err)
	}
	return exists, nil
}

func (s *RuntimePlatformCommandObservationStore) Observe(ctx context.Context) (RuntimePlatformCommandExecutionState, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return RuntimePlatformCommandExecutionState{}, errors.New("runtime platform command observation: store is unavailable")
	}
	if err := s.ownership.VerifyActive(ctx); err != nil {
		return RuntimePlatformCommandExecutionState{}, err
	}
	var state RuntimePlatformCommandExecutionState
	var oldestMicros int64
	if err := s.db.QueryRowContext(ctx, `SELECT
  COALESCE(SUM(CASE WHEN State IN ('ACCEPTED', 'PROCESSING', 'PUBLISHING', 'BLOCKED') THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN State = 'POISONED' THEN 1 ELSE 0 END), 0),
  COALESCE(TIMESTAMPDIFF(MICROSECOND,
    MIN(CASE WHEN State IN ('ACCEPTED', 'PROCESSING', 'PUBLISHING', 'BLOCKED') THEN CreatedDate END),
    UTC_TIMESTAMP(6)), 0)
FROM _System_RuntimePlatformCommand
WHERE ExecutionScopeDigest = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State IN ('ACCEPTED', 'PROCESSING', 'PUBLISHING', 'BLOCKED', 'POISONED')`,
		s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(&state.QueueDepth, &state.PoisonDepth, &oldestMicros); err != nil {
		return RuntimePlatformCommandExecutionState{}, fmt.Errorf("runtime platform command observation: observe: %w", err)
	}
	if state.QueueDepth < 0 || state.PoisonDepth < 0 || oldestMicros < 0 {
		return RuntimePlatformCommandExecutionState{}, errors.New("runtime platform command observation: invalid queue aggregate")
	}
	state.OldestAge = time.Duration(oldestMicros) * time.Microsecond
	return state, nil
}

const runtimePlatformCommandClaimSQL = `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest, CommandId, IdempotencyKey,
       CommandKind, COALESCE(PartitionKey, ''), COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion, RequestDigest,
       Payload, Receipts, State, Attempt, COALESCE(ResultPayload, ''),
       COALESCE(DependencyKind, ''), COALESCE(DependencyObjectAPIName, ''),
       COALESCE(DependencyRecordId, ''), COALESCE(DependencyKeyHash, ''),
       COALESCE(DependencyConnectorOwnerKey, ''), COALESCE(DependencyConnectorNamespace, ''),
       COALESCE(DependencyConnectorSourceObject, ''), COALESCE(DependencyDestinationSlotId, ''),
       COALESCE(DependencyConnectorRemoteId, ''), DependencyBlockedAt
FROM _System_RuntimePlatformCommand candidate
     USE INDEX (idx_runtime_platform_command_claim)
WHERE candidate.ExecutionScopeDigest = ?
  AND candidate.OrgId = ? AND candidate.AppInstallationId = ? AND candidate.RuntimeCapabilityId = ?
  AND ((candidate.State = 'ACCEPTED' AND candidate.AvailableAt <= ?)
       OR (candidate.State = 'PROCESSING' AND candidate.LeaseExpiresAt <= ?)
       OR (candidate.State = 'BLOCKED' AND candidate.AvailableAt <= ?))
  AND (
      COALESCE(candidate.PartitionKey, '') = ''
      OR NOT EXISTS (
          SELECT 1
            FROM _System_RuntimePlatformCommand active_source
                 USE INDEX (idx_runtime_platform_command_partition)
           WHERE active_source.ExecutionScopeDigest = candidate.ExecutionScopeDigest
             AND active_source.OrgId = candidate.OrgId
             AND active_source.AppInstallationId = candidate.AppInstallationId
             AND active_source.RuntimeCapabilityId = candidate.RuntimeCapabilityId
             AND active_source.PartitionKey = candidate.PartitionKey
             AND active_source.Id <> candidate.Id
             AND (
                 active_source.State IN ('BLOCKED', 'PUBLISHING')
                 OR (
                     active_source.State = 'POISONED'
                     AND active_source.RecoveryState = 'PUBLISHING'
                 )
                 OR (
                     active_source.State = 'PROCESSING'
                     AND active_source.LeaseExpiresAt > ?
                 )
             )
      )
  )
  AND (
      COALESCE(candidate.SerializationKeyDigest, '') = ''
      OR NOT EXISTS (
          SELECT 1
            FROM _System_RuntimePlatformCommand active_aggregate
                 USE INDEX (idx_runtime_platform_command_serialization)
           WHERE active_aggregate.ExecutionScopeDigest = candidate.ExecutionScopeDigest
             AND active_aggregate.OrgId = candidate.OrgId
             AND active_aggregate.AppInstallationId = candidate.AppInstallationId
             AND active_aggregate.RuntimeCapabilityId = candidate.RuntimeCapabilityId
             AND active_aggregate.SerializationKeyDigest = candidate.SerializationKeyDigest
             AND active_aggregate.Id <> candidate.Id
             AND (
                 active_aggregate.State IN ('BLOCKED', 'PUBLISHING')
                 OR (
                     active_aggregate.State = 'POISONED'
                     AND active_aggregate.RecoveryState = 'PUBLISHING'
                 )
                 OR (
                     active_aggregate.State = 'PROCESSING'
                     AND active_aggregate.LeaseExpiresAt > ?
                 )
             )
      )
  )
ORDER BY CASE candidate.State WHEN 'PROCESSING' THEN 0 WHEN 'BLOCKED' THEN 1 ELSE 2 END,
         candidate.AvailableAt, candidate.Id
LIMIT 1 FOR UPDATE`

// Claim selects only the configured AppInstallation/capability partition.
// A non-empty declared PartitionKey is single-flight across command execution,
// dependency blocking, and post-commit publication. It owns source ordering,
// not business identity. Independently, a trusted SerializationKeyDigest keeps
// one aggregate single-flight across those states. A command must satisfy both
// constraints before it can be claimed.
// An expired PROCESSING row is safe to reclaim because the only supported
// effect shape commits the Runtime effect and COMPLETED transition in one TiDB
// transaction.
func (s *RuntimePlatformCommandExecutionStore) Claim(
	ctx context.Context,
) (RuntimePlatformCommandLease, bool, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return RuntimePlatformCommandLease{}, false, errors.New("runtime platform command execution: store is unavailable")
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return RuntimePlatformCommandLease{}, false, fmt.Errorf("runtime platform command execution: begin claim: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (RuntimePlatformCommandLease, bool, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return RuntimePlatformCommandLease{}, false, errors.Join(cause, rollbackErr)
		}
		return RuntimePlatformCommandLease{}, false, 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("runtime platform command execution: read claim time: %w", err))
	}
	databaseNow := now
	now = runtimePlatformCommandDatabaseUTC(databaseNow)
	var row RuntimePlatformCommandRow
	var attempt int
	var state string
	var dependency contractmodels.FlowExecutionDependency
	var dependencyBlockedAt sql.NullTime
	err = tx.QueryRowContext(ctx, runtimePlatformCommandClaimSQL,
		s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		databaseNow, databaseNow, databaseNow, databaseNow, databaseNow,
	).Scan(
		&row.RowID, &row.OrgID, &row.AppInstallationID, &row.CallerWorkloadID,
		&row.RuntimeCapabilityID, &row.MetadataBinding, &row.ExecutionScopeDigest, &row.ScopeDigest, &row.CommandID,
		&row.IdempotencyKey, &row.CommandKind, &row.PartitionKey, &row.SerializationKeyDigest, &row.PayloadFormatVersion,
		&row.RequestDigest, &row.Payload, &row.Receipts, &state, &attempt, &row.ResultPayload,
		&dependency.Kind, &dependency.ObjectAPIName, &dependency.RecordID, &dependency.KeyHash,
		&dependency.ConnectorOwnerKey, &dependency.ConnectorNamespace,
		&dependency.ConnectorSourceObject, &dependency.DestinationSlotID,
		&dependency.ConnectorRemoteID, &dependencyBlockedAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		if err := tx.Commit(); err != nil {
			return RuntimePlatformCommandLease{}, false, fmt.Errorf("runtime platform command execution: empty claim commit: %w", err)
		}
		return RuntimePlatformCommandLease{}, false, nil
	}
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: select claim: %w", err))
	}
	if row.OrgID != s.scope.OrgID || row.AppInstallationID != s.scope.AppInstallationID ||
		row.RuntimeCapabilityID != s.scope.RuntimeCapabilityID || row.MetadataBinding == "" ||
		row.ExecutionScopeDigest != s.scopeDigest ||
		(state != RuntimePlatformCommandAccepted && state != RuntimePlatformCommandProcessing &&
			state != RuntimePlatformCommandBlocked) {
		return rollback(errors.New("runtime platform command execution: selected row authority is invalid"))
	}
	if dependency.Kind != "" {
		if !dependencyBlockedAt.Valid || dependency.Validate() != nil {
			return rollback(errors.New("runtime platform command execution: selected dependency is invalid"))
		}
		row.Dependency = &dependency
		row.DependencyBlockedAt = runtimePlatformCommandDatabaseUTC(dependencyBlockedAt.Time)
	} else if dependencyBlockedAt.Valid || state == RuntimePlatformCommandBlocked {
		return rollback(errors.New("runtime platform command execution: selected dependency state is incomplete"))
	}
	if state == RuntimePlatformCommandBlocked &&
		!row.DependencyBlockedAt.Add(runtimePlatformCommandDependencyMaxWait).After(now) {
		result, updateErr := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'FAILED', AvailableAt = UTC_TIMESTAMP(6),
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL, LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'BLOCKED' AND Attempt = ? AND DependencyKeyHash = ?`,
			"identity convergence dependency did not become authoritative before the bounded wait expired",
			row.RowID, attempt, dependency.KeyHash,
		)
		if updateErr != nil {
			return rollback(fmt.Errorf("runtime platform command execution: expire dependency wait: %w", updateErr))
		}
		affected, updateErr := result.RowsAffected()
		if updateErr != nil || affected != 1 {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		if err := tx.Commit(); err != nil {
			return RuntimePlatformCommandLease{}, false, fmt.Errorf("runtime platform command execution: dependency expiry commit outcome unknown: %w", err)
		}
		return RuntimePlatformCommandLease{}, false, nil
	}
	nextAttempt := attempt + 1
	if state == RuntimePlatformCommandBlocked {
		nextAttempt = attempt
	}
	if nextAttempt > s.maxAttempts {
		lastError := "bounded execution attempts exhausted"
		result, updateErr := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'POISONED', LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    RecoveryState = 'ACCEPTED',
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL,
    LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = ? AND Attempt = ?`,
			lastError, row.RowID, state, attempt,
		)
		if updateErr != nil {
			return rollback(fmt.Errorf("runtime platform command execution: poison exhausted row: %w", updateErr))
		}
		affected, updateErr := result.RowsAffected()
		if updateErr != nil || affected != 1 {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		if err := tx.Commit(); err != nil {
			return RuntimePlatformCommandLease{}, false, fmt.Errorf("runtime platform command execution: poison commit outcome unknown: %w", err)
		}
		return RuntimePlatformCommandLease{}, false, nil
	}
	token := s.tokenFn()
	expiresAt := now.Add(s.leaseDuration)
	databaseExpiresAt := databaseNow.Add(s.leaseDuration)
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'PROCESSING', Attempt = ?, LeaseToken = ?, LeaseExpiresAt = ?,
    OwnershipEpoch = ?, LastError = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = ? AND Attempt = ?`,
		nextAttempt, token, databaseExpiresAt, s.expected.EpochDecimal(), row.RowID, state, attempt,
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: claim row: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return RuntimePlatformCommandLease{}, false, fmt.Errorf("runtime platform command execution: claim commit outcome unknown: %w", err)
	}
	row.State = RuntimePlatformCommandProcessing
	return RuntimePlatformCommandLease{
		Row: row, LeaseToken: token, Attempt: nextAttempt,
		AttemptsRemaining: s.maxAttempts - nextAttempt, LeaseExpiresAt: expiresAt,
		OwnershipEpoch: s.expected.OwnershipEpoch, ImmutableBinding: row.MetadataBinding,
		CallerWorkloadID: row.CallerWorkloadID, RuntimeCapability: row.RuntimeCapabilityID,
		AppInstallationID: row.AppInstallationID, OrganizationID: row.OrgID,
	}, true, nil
}

// Execute owns the complete command transaction. It locks the work-family
// authority and exact lease attempt, invokes the bounded Runtime effect, and
// writes COMPLETED with its result before the same transaction commits.
// Reclaim follows the same lock order, so it cannot race an in-flight effect
// after the wall-clock lease expires.
func (s *RuntimePlatformCommandExecutionStore) Execute(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	effect func(context.Context) ([]byte, error),
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if effect == nil {
		return errors.New("runtime platform command execution: effect callback is required")
	}
	completion, err := s.executeTransition(ctx, lease, false, RuntimePlatformCommandCompleted, effect)
	return runtimePlatformCommandTransactionResult(completion, err)
}

// ExecuteApplication owns the complete v3 application transaction. Applied
// and explicitly proved no-op outcomes commit their canonical terminal
// envelope with the bounded Runtime effect. Rejected and failed outcomes force
// that effect transaction to roll back first, then publish their envelope in a
// separately fenced terminal transaction. An ordinary callback error never
// mints an ApplicationResult.
func (s *RuntimePlatformCommandExecutionStore) ExecuteApplication(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	effect func(context.Context) (RuntimePlatformCommandApplicationCompletion, error),
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if effect == nil {
		return errors.New("runtime platform command application execution: effect callback is required")
	}
	var rolledBackTerminal []byte
	completion, err := s.executeTransition(ctx, lease, true, RuntimePlatformCommandCompleted, func(txCtx context.Context) ([]byte, error) {
		terminal, effectErr := effect(txCtx)
		if effectErr != nil {
			return nil, effectErr
		}
		encoded, encodeErr := encodeRuntimePlatformCommandApplicationCompletion(terminal)
		if encodeErr != nil {
			return nil, fmt.Errorf("%w: %v", ErrRuntimePlatformCommandResultInvalid, encodeErr)
		}
		switch terminal.Outcome {
		case applicationresultsemantic.OutcomeApplied, applicationresultsemantic.OutcomeNoop:
			return encoded, nil
		case applicationresultsemantic.OutcomeRejected, applicationresultsemantic.OutcomeFailed:
			rolledBackTerminal = append(rolledBackTerminal[:0], encoded...)
			return nil, errRuntimePlatformCommandRollbackForTerminalResult
		default:
			return nil, ErrRuntimePlatformCommandResultInvalid
		}
	})
	if !errors.Is(err, errRuntimePlatformCommandRollbackForTerminalResult) {
		return runtimePlatformCommandTransactionResult(completion, err)
	}
	if completion != ports.TransactionCompletionRolledBack {
		return fmt.Errorf(
			"%w: application failure rollback was not proved",
			ErrRuntimePlatformCommandOutcomeUncertain,
		)
	}
	return s.commitRolledBackApplicationResult(ctx, lease, rolledBackTerminal)
}

func runtimePlatformCommandEffectFenceKey(lease RuntimePlatformCommandLease) string {
	return "command:" + lease.Row.RowID
}

func (s *RuntimePlatformCommandExecutionStore) executeTransition(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	applicationResult bool,
	targetState string,
	effect func(context.Context) ([]byte, error),
) (ports.TransactionCompletionOutcome, error) {
	if targetState != RuntimePlatformCommandCompleted && targetState != RuntimePlatformCommandPublishing {
		return ports.TransactionCompletionRolledBack, errors.New("runtime platform command execution: transition target is invalid")
	}
	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("runtime platform command execution: effect transaction is required")
		}
		if err := s.ownership.LockActiveFence(
			txCtx,
			tx,
			runtimePlatformCommandEffectFenceKey(lease),
		); err != nil {
			return err
		}
		var locked RuntimePlatformCommandRow
		var state, token, epoch string
		var attempt int
		var expiresAt time.Time
		if err := tx.QueryRowContext(txCtx, `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest,
       CommandId, IdempotencyKey, CommandKind, COALESCE(PartitionKey, ''),
       COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion, RequestDigest,
       Payload, Receipts, State, COALESCE(LeaseToken, ''), Attempt,
       COALESCE(CAST(OwnershipEpoch AS CHAR), '0'), LeaseExpiresAt
FROM _System_RuntimePlatformCommand
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`,
			lease.Row.RowID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		).Scan(
			&locked.RowID, &locked.OrgID, &locked.AppInstallationID, &locked.CallerWorkloadID,
			&locked.RuntimeCapabilityID, &locked.MetadataBinding, &locked.ExecutionScopeDigest,
			&locked.ScopeDigest, &locked.CommandID, &locked.IdempotencyKey, &locked.CommandKind,
			&locked.PartitionKey, &locked.SerializationKeyDigest, &locked.PayloadFormatVersion,
			&locked.RequestDigest, &locked.Payload, &locked.Receipts,
			&state, &token, &attempt, &epoch, &expiresAt,
		); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return ErrRuntimePlatformCommandLeaseLost
			}
			return fmt.Errorf("runtime platform command execution: lock effect attempt: %w", err)
		}
		if state != RuntimePlatformCommandProcessing || token != lease.LeaseToken ||
			attempt != lease.Attempt || epoch != s.expected.EpochDecimal() {
			return ErrRuntimePlatformCommandLeaseLost
		}
		if !runtimePlatformCommandSameImmutableAuthority(locked, lease.Row) {
			return ErrRuntimePlatformCommandLeaseLost
		}
		isApplicationCommand := locked.CommandKind == flowcommandsemanticv3.CommandKind &&
			locked.PayloadFormatVersion == flowcommandsemanticv3.PayloadFormatVersion
		if applicationResult != isApplicationCommand {
			if isApplicationCommand {
				return ErrRuntimePlatformCommandApplicationResultRequired
			}
			return errors.New("runtime platform command application execution requires a v3 Flow command")
		}
		if applicationResult {
			command, decodeErr := flowcommandsemanticv3.Decode(locked.Payload)
			if decodeErr != nil {
				return fmt.Errorf("runtime platform command application execution: decode persisted command: %w", decodeErr)
			}
			if _, digestErr := applicationresultsemantic.ExecutionRefDigest(command.ExecutionRef); digestErr != nil {
				return fmt.Errorf("runtime platform command application execution: bind persisted execution ref: %w", digestErr)
			}
		}
		var now time.Time
		if err := tx.QueryRowContext(txCtx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
			return fmt.Errorf("runtime platform command execution: read effect time: %w", err)
		}
		now = runtimePlatformCommandDatabaseUTC(now)
		expiresAt = runtimePlatformCommandDatabaseUTC(expiresAt)
		if !expiresAt.After(now) {
			return ErrRuntimePlatformCommandLeaseLost
		}
		resultPayload, err := effect(txCtx)
		if err != nil {
			return err
		}
		if err := txCtx.Err(); err != nil {
			return err
		}
		resultMaximum := runtimePlatformCommandTerminalResultMaxBytes
		if targetState == RuntimePlatformCommandPublishing {
			resultMaximum = runtimePlatformCommandPublicationPendingResultMaxBytes
		}
		if len(resultPayload) == 0 || len(resultPayload) > resultMaximum {
			return fmt.Errorf(
				"%w: result must contain 1 through %d bytes",
				ErrRuntimePlatformCommandResultInvalid,
				resultMaximum,
			)
		}
		if targetState == RuntimePlatformCommandPublishing {
			publicationIdentity, identityErr := runtimePlatformCommandPublicationIdentity(
				locked.ScopeDigest, locked.CommandID, locked.RequestDigest,
			)
			if identityErr != nil {
				return identityErr
			}
			result, updateErr := tx.ExecContext(txCtx, `UPDATE _System_RuntimePlatformCommand
SET State = 'PUBLISHING', ResultPayload = ?, PublicationIdentity = ?,
    PublishAttempt = 0, RecoveryState = NULL,
    AvailableAt = UTC_TIMESTAMP(6), LeaseToken = NULL, LeaseExpiresAt = NULL,
    OwnershipEpoch = NULL, LastError = NULL,
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ? AND OwnershipEpoch = ?`,
				encodeRuntimePlatformCommandResult(resultPayload), publicationIdentity, lease.Row.RowID,
				s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
				lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
			)
			if updateErr != nil {
				return fmt.Errorf("runtime platform command execution: prepare publication: %w", updateErr)
			}
			affected, updateErr := result.RowsAffected()
			if updateErr != nil || affected != 1 {
				return ErrRuntimePlatformCommandLeaseLost
			}
			return nil
		}
		result, err := tx.ExecContext(txCtx, `UPDATE _System_RuntimePlatformCommand
SET State = 'COMPLETED', ResultPayload = ?, LastError = NULL,
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL,
    LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ? AND OwnershipEpoch = ?`,
			encodeRuntimePlatformCommandResult(resultPayload), lease.Row.RowID,
			s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
			lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
		)
		if err != nil {
			return fmt.Errorf("runtime platform command execution: complete effect: %w", err)
		}
		affected, err := result.RowsAffected()
		if err != nil || affected != 1 {
			return ErrRuntimePlatformCommandLeaseLost
		}
		return nil
	})
	return completion, err
}

func runtimePlatformCommandTransactionResult(
	completion ports.TransactionCompletionOutcome,
	err error,
) error {
	if err != nil && completion != ports.TransactionCompletionRolledBack {
		return fmt.Errorf("%w: %v", ErrRuntimePlatformCommandOutcomeUncertain, err)
	}
	return err
}

func runtimePlatformCommandSameImmutableAuthority(
	locked RuntimePlatformCommandRow,
	claimed RuntimePlatformCommandRow,
) bool {
	return locked.RowID == claimed.RowID &&
		locked.OrgID == claimed.OrgID &&
		locked.AppInstallationID == claimed.AppInstallationID &&
		locked.CallerWorkloadID == claimed.CallerWorkloadID &&
		locked.RuntimeCapabilityID == claimed.RuntimeCapabilityID &&
		locked.MetadataBinding == claimed.MetadataBinding &&
		locked.ExecutionScopeDigest == claimed.ExecutionScopeDigest &&
		locked.ScopeDigest == claimed.ScopeDigest &&
		locked.CommandID == claimed.CommandID &&
		locked.IdempotencyKey == claimed.IdempotencyKey &&
		locked.CommandKind == claimed.CommandKind &&
		locked.PartitionKey == claimed.PartitionKey &&
		locked.SerializationKeyDigest == claimed.SerializationKeyDigest &&
		locked.PayloadFormatVersion == claimed.PayloadFormatVersion &&
		locked.RequestDigest == claimed.RequestDigest &&
		bytes.Equal(locked.Payload, claimed.Payload) &&
		bytes.Equal(locked.Receipts, claimed.Receipts)
}

func (s *RuntimePlatformCommandExecutionStore) commitRolledBackApplicationResult(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	encoded []byte,
) error {
	terminal, err := decodeRuntimePlatformCommandApplicationCompletion(encoded)
	if err != nil || (terminal.Outcome != applicationresultsemantic.OutcomeRejected &&
		terminal.Outcome != applicationresultsemantic.OutcomeFailed) {
		return ErrRuntimePlatformCommandResultInvalid
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command application execution: begin failed terminal result: %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 commandID, commandKind, token, epoch string
	var payloadFormatVersion uint32
	var payload []byte
	var attempt int
	var expiresAt time.Time
	if err := tx.QueryRowContext(ctx, `SELECT CommandId, CommandKind, PayloadFormatVersion, Payload,
       COALESCE(LeaseToken, ''), Attempt, COALESCE(CAST(OwnershipEpoch AS CHAR), '0'), LeaseExpiresAt
FROM _System_RuntimePlatformCommand
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PROCESSING'
FOR UPDATE`,
		lease.Row.RowID, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(
		&commandID, &commandKind, &payloadFormatVersion, &payload,
		&token, &attempt, &epoch, &expiresAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		return rollback(fmt.Errorf("runtime platform command application execution: lock failed terminal result: %w", err))
	}
	if commandID != lease.Row.CommandID || commandKind != flowcommandsemanticv3.CommandKind ||
		payloadFormatVersion != flowcommandsemanticv3.PayloadFormatVersion ||
		!bytes.Equal(payload, lease.Row.Payload) || token != lease.LeaseToken ||
		attempt != lease.Attempt || epoch != s.expected.EpochDecimal() {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if _, err := flowcommandsemanticv3.Decode(payload); err != nil {
		return rollback(fmt.Errorf("runtime platform command application execution: decode failed terminal command: %w", err))
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("runtime platform command application execution: read failed terminal time: %w", err))
	}
	if !runtimePlatformCommandDatabaseUTC(expiresAt).After(runtimePlatformCommandDatabaseUTC(now)) {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'FAILED', ResultPayload = ?, AvailableAt = UTC_TIMESTAMP(6),
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL, LastError = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ? AND OwnershipEpoch = ?`,
		encodeRuntimePlatformCommandResult(encoded), lease.Row.RowID,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command application execution: write failed terminal result: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("%w: failed terminal result commit: %v", ErrRuntimePlatformCommandOutcomeUncertain, err)
	}
	return nil
}

// Block persists one fully bound dependency only after Execute has proved that
// the attempted effect transaction rolled back. The same command identity and
// immutable metadata binding remain authoritative; the normal Worker reclaims
// the row after AvailableAt and replays the exact bounded Flow.
func (s *RuntimePlatformCommandExecutionStore) Block(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	dependency contractmodels.FlowExecutionDependency,
	retryAfter time.Duration,
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if runtimePlatformCommandRequiresApplicationResult(lease.Row) {
		return ErrRuntimePlatformCommandApplicationResultRequired
	}
	if err := dependency.Validate(); err != nil || retryAfter < time.Second || retryAfter > 5*time.Minute {
		return errors.New("runtime platform command execution: dependency disposition is incomplete")
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command execution: begin dependency 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 existingKey string
	if err := tx.QueryRowContext(ctx, `SELECT COALESCE(DependencyKeyHash, '')
FROM _System_RuntimePlatformCommand
WHERE Id = ? AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)
FOR UPDATE`,
		lease.Row.RowID, lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
	).Scan(&existingKey); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		return rollback(fmt.Errorf("runtime platform command execution: lock dependency disposition: %w", err))
	}
	if existingKey != "" && existingKey != dependency.KeyHash {
		return rollback(ErrRuntimePlatformCommandDependencyChanged)
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'BLOCKED', AvailableAt = TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)),
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    DependencyKind = ?, DependencyObjectAPIName = ?, DependencyRecordId = ?,
    DependencyKeyHash = ?, DependencyConnectorOwnerKey = ?,
    DependencyConnectorNamespace = ?, DependencyConnectorSourceObject = ?,
    DependencyDestinationSlotId = ?, DependencyConnectorRemoteId = ?,
    DependencyBlockedAt = COALESCE(DependencyBlockedAt, UTC_TIMESTAMP(6)),
    LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
		retryAfter.Microseconds(), dependency.Kind, dependency.ObjectAPIName,
		dependency.RecordID, dependency.KeyHash, dependency.ConnectorOwnerKey,
		dependency.ConnectorNamespace, dependency.ConnectorSourceObject,
		dependency.DestinationSlotID, dependency.ConnectorRemoteID,
		"blocked on identity convergence dependency key_hash="+dependency.KeyHash,
		lease.Row.RowID, lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: block on dependency: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("runtime platform command execution: dependency disposition commit outcome unknown: %w", err)
	}
	return nil
}

func (s *RuntimePlatformCommandExecutionStore) Retry(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	cause error,
	backoff time.Duration,
) error {
	return s.disposition(ctx, lease, cause, backoff, "")
}

// Replay releases a processing lease into the next exact execution attempt.
// Unlike Retry, it deliberately clears any durable dependency carried by the
// lease: the caller has already resolved that dependency in a separately owned
// transaction and the same immutable command must now re-run immediately.
func (s *RuntimePlatformCommandExecutionStore) Replay(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	cause error,
) error {
	if lease.AttemptsRemaining <= 0 {
		return errors.New("runtime platform command execution: replay attempt is unavailable")
	}
	return s.disposition(ctx, lease, cause, 0, RuntimePlatformCommandAccepted)
}

func (s *RuntimePlatformCommandExecutionStore) Fail(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	cause error,
) error {
	return s.disposition(ctx, lease, cause, 0, RuntimePlatformCommandFailed)
}

// FailWithResult persists one typed business rejection only after the caller's
// effect transaction has provably rolled back. It is the generic counterpart
// of the V3 Flow ApplicationResult rejection seam; Runtime recovery/internal
// failures must continue through Retry/Poison and may not use this method.
func (s *RuntimePlatformCommandExecutionStore) FailWithResult(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	encoded []byte,
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if runtimePlatformCommandRequiresApplicationResult(lease.Row) ||
		len(encoded) == 0 || len(encoded) > runtimePlatformCommandTerminalResultMaxBytes {
		return ErrRuntimePlatformCommandResultInvalid
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command execution: begin typed failure: %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 locked RuntimePlatformCommandRow
	var state, token, epoch string
	var attempt int
	var expiresAt time.Time
	if err := tx.QueryRowContext(ctx, `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest,
       CommandId, IdempotencyKey, CommandKind, COALESCE(PartitionKey, ''),
       COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion, RequestDigest,
       Payload, Receipts, State, COALESCE(LeaseToken, ''), Attempt,
       COALESCE(CAST(OwnershipEpoch AS CHAR), '0'), LeaseExpiresAt
FROM _System_RuntimePlatformCommand
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`,
		lease.Row.RowID, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(
		&locked.RowID, &locked.OrgID, &locked.AppInstallationID, &locked.CallerWorkloadID,
		&locked.RuntimeCapabilityID, &locked.MetadataBinding, &locked.ExecutionScopeDigest,
		&locked.ScopeDigest, &locked.CommandID, &locked.IdempotencyKey, &locked.CommandKind,
		&locked.PartitionKey, &locked.SerializationKeyDigest, &locked.PayloadFormatVersion,
		&locked.RequestDigest, &locked.Payload, &locked.Receipts,
		&state, &token, &attempt, &epoch, &expiresAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		return rollback(fmt.Errorf("runtime platform command execution: lock typed failure: %w", err))
	}
	if state != RuntimePlatformCommandProcessing || token != lease.LeaseToken ||
		attempt != lease.Attempt || epoch != s.expected.EpochDecimal() ||
		!runtimePlatformCommandSameImmutableAuthority(locked, lease.Row) {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: read typed failure time: %w", err))
	}
	if !runtimePlatformCommandDatabaseUTC(expiresAt).After(runtimePlatformCommandDatabaseUTC(now)) {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'FAILED', ResultPayload = ?, AvailableAt = UTC_TIMESTAMP(6),
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL, LastError = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ? AND OwnershipEpoch = ?`,
		encodeRuntimePlatformCommandResult(encoded), lease.Row.RowID,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: write typed failure: %w", err))
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("%w: typed failure commit: %v", ErrRuntimePlatformCommandOutcomeUncertain, err)
	}
	return nil
}

func (s *RuntimePlatformCommandExecutionStore) Poison(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	cause error,
) error {
	return s.disposition(ctx, lease, cause, 0, RuntimePlatformCommandPoisoned)
}

func (s *RuntimePlatformCommandExecutionStore) disposition(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	cause error,
	backoff time.Duration,
	requestedState string,
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if cause == nil || backoff < 0 || backoff > 5*time.Minute {
		return errors.New("runtime platform command execution: disposition is incomplete")
	}
	if runtimePlatformCommandRequiresApplicationResult(lease.Row) &&
		(requestedState == RuntimePlatformCommandFailed ||
			(requestedState == "" && lease.Row.Dependency != nil)) {
		return ErrRuntimePlatformCommandApplicationResultRequired
	}
	message := strings.TrimSpace(cause.Error())
	if message == "" {
		message = "command execution failed"
	}
	if len(message) > 4096 {
		message = message[:4096]
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command execution: begin 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)
	}
	nextState := requestedState
	if nextState == "" {
		if lease.Row.Dependency != nil {
			nextState = RuntimePlatformCommandBlocked
			if backoff < time.Minute {
				backoff = time.Minute
			}
		} else if lease.Attempt >= s.maxAttempts {
			nextState = RuntimePlatformCommandPoisoned
		} else {
			nextState = RuntimePlatformCommandAccepted
		}
	}
	if nextState != RuntimePlatformCommandAccepted &&
		nextState != RuntimePlatformCommandBlocked &&
		nextState != RuntimePlatformCommandFailed &&
		nextState != RuntimePlatformCommandPoisoned {
		return rollback(errors.New("runtime platform command execution: disposition state is invalid"))
	}
	var result sql.Result
	if nextState == RuntimePlatformCommandBlocked {
		result, err = tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'BLOCKED',
    AvailableAt = TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)),
    LeaseToken = NULL, LeaseExpiresAt = NULL,
    OwnershipEpoch = NULL, LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)
  AND DependencyKeyHash = ?`,
			backoff.Microseconds(), message, lease.Row.RowID, lease.LeaseToken,
			lease.Attempt, s.expected.EpochDecimal(), lease.Row.Dependency.KeyHash,
		)
	} else {
		result, err = tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = ?,
    AvailableAt = TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)),
    LeaseToken = NULL, LeaseExpiresAt = NULL,
    OwnershipEpoch = NULL, RecoveryState = CASE WHEN ? = 'POISONED' THEN 'ACCEPTED' ELSE NULL END,
    DependencyKind = NULL, DependencyObjectAPIName = NULL, DependencyRecordId = NULL,
    DependencyKeyHash = NULL, DependencyConnectorOwnerKey = NULL,
    DependencyConnectorNamespace = NULL, DependencyConnectorSourceObject = NULL,
    DependencyDestinationSlotId = NULL, DependencyConnectorRemoteId = NULL,
    DependencyBlockedAt = NULL,
    LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PROCESSING' AND LeaseToken = ? AND Attempt = ?
  AND OwnershipEpoch = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
			nextState, backoff.Microseconds(), nextState, message, lease.Row.RowID, lease.LeaseToken,
			lease.Attempt, s.expected.EpochDecimal(),
		)
	}
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: disposition: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("runtime platform command execution: disposition commit outcome unknown: %w", err)
	}
	return nil
}

func runtimePlatformCommandRequiresApplicationResult(row RuntimePlatformCommandRow) bool {
	return row.CommandKind == flowcommandsemanticv3.CommandKind &&
		row.PayloadFormatVersion == flowcommandsemanticv3.PayloadFormatVersion
}

func (s *RuntimePlatformCommandExecutionStore) RepairPoison(
	ctx context.Context,
	repair RuntimePoisonRepair,
) error {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("runtime platform command execution: store is unavailable")
	}
	if err := repair.Validate(); err != nil {
		return err
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command execution: begin repair: %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 attempt, publishAttempt int
	var recoveryState string
	var lastError string
	if err := tx.QueryRowContext(ctx, `SELECT Attempt, PublishAttempt,
       COALESCE(RecoveryState, 'ACCEPTED'), COALESCE(LastError, '')
FROM _System_RuntimePlatformCommand
WHERE Id = ? AND State = 'POISONED'
  AND ExecutionScopeDigest = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`,
		repair.RowID, s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(&attempt, &publishAttempt, &recoveryState, &lastError); err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: lock poison row: %w", err))
	}
	retryCount := attempt
	if recoveryState == RuntimePlatformCommandPublishing {
		retryCount = publishAttempt
	} else if recoveryState != RuntimePlatformCommandAccepted {
		return rollback(errors.New("runtime platform command execution: poison recovery state is invalid"))
	}
	if !repair.Matches(retryCount, lastError) {
		return rollback(errors.New("runtime platform command execution: poison observation changed"))
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = ?,
    Attempt = CASE WHEN ? = 'ACCEPTED' THEN 0 ELSE Attempt END,
    PublishAttempt = CASE WHEN ? = 'PUBLISHING' THEN 0 ELSE PublishAttempt END,
    RecoveryState = NULL, AvailableAt = UTC_TIMESTAMP(6),
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    LastError = NULL, RepairId = ?, RepairReason = ?, RepairCount = RepairCount + 1,
    LastRepairAt = UTC_TIMESTAMP(6), LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'POISONED' AND Attempt = ? AND PublishAttempt = ?
  AND COALESCE(RecoveryState, 'ACCEPTED') = ?`,
		recoveryState, recoveryState, recoveryState,
		repair.RepairID, strings.TrimSpace(repair.Reason), repair.RowID,
		attempt, publishAttempt, recoveryState,
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command execution: requeue poison row: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(errors.New("runtime platform command execution: poison row changed before replay"))
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("runtime platform command execution: repair commit outcome unknown: %w", err)
	}
	return nil
}

func (s *RuntimePlatformCommandExecutionStore) validateLease(lease RuntimePlatformCommandLease) error {
	if lease.Row.Dependency != nil &&
		(lease.Row.Dependency.Validate() != nil || lease.Row.DependencyBlockedAt.IsZero()) {
		return errors.New("runtime platform command execution: lease dependency is invalid")
	}
	if s == nil || s.db == nil || s.transactions == nil || s.ownership == nil ||
		lease.Row.RowID == "" || lease.LeaseToken == "" || lease.Attempt <= 0 ||
		lease.AttemptsRemaining != s.maxAttempts-lease.Attempt || lease.AttemptsRemaining < 0 ||
		lease.LeaseExpiresAt.IsZero() || lease.Row.State != RuntimePlatformCommandProcessing ||
		lease.OwnershipEpoch != s.expected.OwnershipEpoch ||
		lease.OrganizationID != s.scope.OrgID ||
		lease.AppInstallationID != s.scope.AppInstallationID ||
		lease.RuntimeCapability != s.scope.RuntimeCapabilityID ||
		lease.ImmutableBinding == "" ||
		lease.Row.OrgID != lease.OrganizationID ||
		lease.Row.AppInstallationID != lease.AppInstallationID ||
		lease.Row.RuntimeCapabilityID != lease.RuntimeCapability ||
		lease.Row.CallerWorkloadID != lease.CallerWorkloadID ||
		lease.Row.MetadataBinding != lease.ImmutableBinding ||
		lease.Row.ExecutionScopeDigest != s.scopeDigest ||
		!hexDigestPattern.MatchString(lease.Row.ScopeDigest) ||
		(lease.Row.SerializationKeyDigest != "" && !hexDigestPattern.MatchString(lease.Row.SerializationKeyDigest)) ||
		lease.Row.RequestDigest == "" {
		return errors.New("runtime platform command execution: lease is incomplete or outside configured authority")
	}
	return nil
}