package records

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/recordmutation"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

const runtimeRecordCommandTransactionRetries = 3

var errRuntimeRecordCommandExpired = errors.New("runtime record command execution window expired")

type RuntimeRecordCommandStore interface {
	Accept(context.Context, persistence.RuntimeRecordCommandAcceptance) (persistence.RuntimeRecordCommandRow, bool, error)
	Load(context.Context, string, string) (persistence.RuntimeRecordCommandRow, error)
	LoadUnexpired(context.Context, string, string) (persistence.RuntimeRecordCommandRow, error)
	LockCommand(context.Context, string) (persistence.RuntimeRecordCommandRow, error)
	LockOperation(context.Context, string, int) (persistence.RuntimeRecordCommandOperationRow, error)
	CompleteOperation(context.Context, string, int, []byte) error
	LockOperations(context.Context, string) ([]persistence.RuntimeRecordCommandOperationRow, error)
	LoadOperations(context.Context, string) ([]persistence.RuntimeRecordCommandOperationRow, error)
	CompleteCommand(context.Context, string, []byte, time.Time) error
}

// RuntimeRecordMutationTransactions extends the existing transaction owner
// with its established transaction-bound database clock. Durable command
// windows must not depend on an API or worker process wall clock.
type RuntimeRecordMutationTransactions interface {
	ports.TransactionManager
	ReadTransactionTime(context.Context) (time.Time, error)
}

// RuntimeRecordMutationExecutionTransactions adds the Records-owned outer
// mutation boundary required only by the API business-effect path. Command
// lifecycle workers use RuntimeRecordMutationTransactions and do not depend on
// the in-process Records engine.
type RuntimeRecordMutationExecutionTransactions interface {
	RuntimeRecordMutationTransactions
	RunRecordMutationTransaction(context.Context, []string, func(context.Context) error) error
}

// RuntimeRecordMutationEngine is sealed to this package so production
// composition must use NewRuntimeRecordMutationPersistenceEngine instead of
// accidentally supplying the legacy persistence surface directly.
type RuntimeRecordMutationEngine interface {
	runtimeRecordMutationEngine()
	Insert(context.Context, string, models.Record, *models.UserSession) (models.Record, error)
	UpdateAuthorized(context.Context, string, string, models.Record, *models.UserSession, func(context.Context, models.Record) error) error
	DeleteAuthorized(context.Context, string, string, *models.UserSession, func(context.Context, models.Record) error) error
	ResolveOpaqueRecordObject(context.Context, string) (string, error)
	ResolveOpaqueRecordObjectCandidates(context.Context, []string) (map[string][]string, error)
	ReadPolicyTarget(context.Context, string, string, semantic.OperationKind, *models.UserSession) (models.Record, error)
}

// RuntimeRecordMutationMetadata is deliberately narrow. Production supplies a
// request-bound RuntimeMetadataView adapter; no repository/global fallback is
// representable at this use-case boundary.
type RuntimeRecordMutationMetadata interface {
	Describe(context.Context, string) (*models.ObjectMetadata, error)
	ResolveObjectForRecordID(context.Context, string) (string, bool)
}

// RuntimeRecordMutationActivity is the existing generic recent-activity
// mechanism. It remains best-effort, but it must execute inside the Runtime
// ownership boundary so L1 does not need a second metadata/data-side call after
// one coarse mutation command.
type RuntimeRecordMutationActivity interface {
	TrackRecordAccess(context.Context, string, string, string)
}

type RuntimeRecordMutationConfig struct {
	Engine           RuntimeRecordMutationEngine
	Metadata         RuntimeRecordMutationMetadata
	Activity         RuntimeRecordMutationActivity
	Commands         RuntimeRecordCommandStore
	Transactions     RuntimeRecordMutationExecutionTransactions
	OutcomeRetention time.Duration
}

// RuntimeRecordMutationService owns the complete mutation command boundary.
// It is intentionally separate from ContractService: this is one coarse use
// case with durable outcome authority, not a remote version of Records.
type RuntimeRecordMutationService struct {
	engine    RuntimeRecordMutationEngine
	metadata  RuntimeRecordMutationMetadata
	activity  RuntimeRecordMutationActivity
	commands  RuntimeRecordCommandStore
	tx        RuntimeRecordMutationExecutionTransactions
	retention time.Duration
}

func NewRuntimeRecordMutationService(config RuntimeRecordMutationConfig) (*RuntimeRecordMutationService, error) {
	if config.Engine == nil || config.Metadata == nil || config.Activity == nil || config.Commands == nil || config.Transactions == nil || config.OutcomeRetention.Microseconds() <= 0 {
		return nil, errors.New("runtime record mutation service: engine, request-bound metadata, activity, command store, transactions, and explicit retention are required")
	}
	return &RuntimeRecordMutationService{
		engine: config.Engine, metadata: config.Metadata, activity: config.Activity,
		commands: config.Commands, tx: config.Transactions, retention: config.OutcomeRetention,
	}, nil
}

func (s *RuntimeRecordMutationService) Mutate(ctx context.Context, scopeDigest string, request semantic.Request, receipts semantic.Receipts, user *models.UserSession) (semantic.Response, error) {
	if err := semantic.ValidateRequest(ctx, request); err != nil {
		return semantic.Response{}, err
	}
	request = semantic.NormalizeRequest(request)
	ctx, phaseProfile := withRuntimeMutationPhaseProfile(ctx, request)
	defer phaseProfile.log(ctx)
	if !validRuntimeCommandScopeDigest(scopeDigest) || user == nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	if receipts.CapabilityVersion != semantic.CapabilityName {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentCommandData}
	}
	acceptCtx, finishAccept := beginRuntimeMutationPhase(ctx, runtimeMutationPhaseAccept)
	command, replay, err := s.accept(acceptCtx, scopeDigest, request, receipts, digest)
	finishAccept()
	if err != nil {
		if errors.Is(err, persistence.ErrRuntimeRecordCommandConflict) {
			return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentCommandData}
		}
		if errors.Is(err, persistence.ErrRuntimeRecordCommandServingBindingChanged) {
			// The accepted command may already contain committed independent
			// positions. It must never resume under a different Catalog/Security
			// binding, but leaving it ACCEPTED would also strand its outcome and
			// eventually erase duplicate-prevention authority. Close it under the
			// original receipt without executing any pending operation.
			return s.closeInterruptedCommand(ctx, command, request)
		}
		return semantic.Response{}, classifyRuntimeMutationTopLevelError(err)
	}
	if replay && command.State == persistence.RuntimeRecordCommandCompleted {
		return decodeRuntimeMutationResponse(request, command.ResultPayload)
	}
	if replay && command.State == persistence.RuntimeRecordCommandAccepted {
		// Expiration belongs to the database clock. Re-read through the
		// repository's CURRENT_TIMESTAMP predicate rather than comparing an API
		// process wall clock with a durable timestamp.
		unexpired, loadErr := s.commands.LoadUnexpired(ctx, scopeDigest, request.CommandID)
		if errors.Is(loadErr, persistence.ErrRuntimeRecordCommandMissing) {
			return s.closeInterruptedCommand(ctx, command, request)
		}
		if loadErr != nil {
			return semantic.Response{}, classifyRuntimeMutationTopLevelError(loadErr)
		}
		command = unexpired
		if command.State == persistence.RuntimeRecordCommandCompleted {
			return decodeRuntimeMutationResponse(request, command.ResultPayload)
		}
	}
	prepareCtx, finishPrepare := beginRuntimeMutationPhase(ctx, runtimeMutationPhasePrepare)
	prepared, staticReady := s.prepareFreshIndependentSingle(prepareCtx, request, user, replay)
	var terminal *semantic.Response
	if !staticReady {
		prepared, terminal, err = s.prepareAcceptedCommand(prepareCtx, command, request, receipts, user)
	}
	finishPrepare()
	if err != nil {
		if errors.Is(err, errRuntimeRecordCommandExpired) {
			return s.closeInterruptedCommand(ctx, command, request)
		}
		return semantic.Response{}, classifyRuntimeMutationTopLevelError(err)
	}
	if terminal != nil {
		return *terminal, nil
	}

	var response semantic.Response
	if request.BatchMode == semantic.BatchAtomic {
		response, err = s.executeAtomic(ctx, command, request, prepared, receipts, user)
	} else {
		response, err = s.executeIndependent(ctx, command, request, prepared, receipts, user)
	}
	if err != nil {
		if errors.Is(err, errRuntimeRecordCommandExpired) {
			return s.closeInterruptedCommand(ctx, command, request)
		}
		return semantic.Response{}, classifyRuntimeMutationTopLevelError(err)
	}
	if err := semantic.ValidateResponse(request, response); err != nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	s.trackSuccessfulSingleMutation(ctx, request, response, user)
	return response, nil
}

func (s *RuntimeRecordMutationService) trackSuccessfulSingleMutation(ctx context.Context, request semantic.Request, response semantic.Response, user *models.UserSession) {
	if s == nil || s.activity == nil || user == nil || request.UseCase != semantic.MutationUseCaseSingle || len(response.Results) != 1 || !response.Results[0].Success {
		return
	}
	operation := request.Operations[0]
	if operation.Kind != semantic.OperationCreate && operation.Kind != semantic.OperationUpdate {
		return
	}
	s.activity.TrackRecordAccess(ctx, user.ID, operation.ObjectID, response.Results[0].RecordID)
}

func (s *RuntimeRecordMutationService) Status(ctx context.Context, scopeDigest string, request semantic.StatusRequest) (semantic.CommandShape, semantic.StatusResponse, error) {
	if err := semantic.ValidateStatusRequest(ctx, request); err != nil {
		return semantic.CommandShape{}, semantic.StatusResponse{}, err
	}
	if !validRuntimeCommandScopeDigest(scopeDigest) {
		return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	row, err := s.commands.LoadUnexpired(ctx, scopeDigest, request.CommandID)
	if errors.Is(err, persistence.ErrRuntimeRecordCommandMissing) {
		// A nonterminal command is never disposable idempotency state. If its
		// execution window elapsed before the lifecycle worker observed it,
		// reconciliation itself closes it and makes the terminal proof visible.
		expired, loadErr := s.commands.Load(ctx, scopeDigest, request.CommandID)
		if loadErr != nil || expired.State != persistence.RuntimeRecordCommandAccepted || expired.RequestDigest != request.CommandDigest {
			return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureMissing}
		}
		response, closeErr := s.closeInterruptedCommandRow(ctx, expired)
		if closeErr != nil {
			return semantic.CommandShape{}, semantic.StatusResponse{}, closeErr
		}
		shape, shapeErr := s.commandShape(ctx, expired)
		if shapeErr != nil {
			return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
		}
		status := semantic.StatusResponse{
			CommandID: response.CommandID, CommandDigest: response.CommandDigest, State: semantic.CommandCompleted,
			Results: response.Results, Receipts: &response.Receipts, OutcomeExpiresAtUTC: response.OutcomeExpiresAtUTC,
		}
		if err := semantic.ValidateStatusResponse(shape, request, status); err != nil {
			return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
		}
		return shape, status, nil
	}
	if err != nil {
		return semantic.CommandShape{}, semantic.StatusResponse{}, classifyRuntimeMutationTopLevelError(err)
	}
	// A command ID is not sufficient reconciliation authority. The caller must
	// prove the exact semantic content it is asking about; mismatch is masked
	// together with unknown/expired state so Status is not a binding oracle.
	if row.RequestDigest != request.CommandDigest {
		return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureMissing}
	}
	shape, err := s.commandShape(ctx, row)
	if err != nil {
		return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	if row.State != persistence.RuntimeRecordCommandCompleted {
		status := semantic.StatusResponse{
			CommandID: request.CommandID, CommandDigest: row.RequestDigest, State: semantic.CommandAccepted,
		}
		if err := semantic.ValidateStatusResponse(shape, request, status); err != nil {
			return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
		}
		return shape, status, nil
	}
	var response semantic.Response
	if err := json.Unmarshal(row.ResultPayload, &response); err != nil {
		return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	var boundReceipts semantic.Receipts
	boundExpiry := row.OutcomeExpiresAt.UTC().Format(time.RFC3339Nano)
	if err := json.Unmarshal(row.ReceiptsPayload, &boundReceipts); err != nil ||
		response.CommandID != row.CommandID || response.CommandDigest != row.RequestDigest ||
		response.Receipts != boundReceipts || response.OutcomeExpiresAtUTC != boundExpiry {
		return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	status := semantic.StatusResponse{
		CommandID: response.CommandID, CommandDigest: row.RequestDigest,
		State: semantic.CommandCompleted, Results: response.Results,
		Receipts: &response.Receipts, OutcomeExpiresAtUTC: response.OutcomeExpiresAtUTC,
	}
	if err := semantic.ValidateStatusResponse(shape, request, status); err != nil {
		return semantic.CommandShape{}, semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	return shape, status, nil
}

func (s *RuntimeRecordMutationService) commandShape(ctx context.Context, row persistence.RuntimeRecordCommandRow) (semantic.CommandShape, error) {
	operations, err := s.commands.LoadOperations(ctx, row.RowID)
	if err != nil || len(operations) != row.OperationCount {
		return semantic.CommandShape{}, errors.New("runtime record command operation shape is incomplete")
	}
	return runtimeRecordCommandShape(row, operations)
}

func runtimeRecordCommandShape(row persistence.RuntimeRecordCommandRow, operations []persistence.RuntimeRecordCommandOperationRow) (semantic.CommandShape, error) {
	if len(operations) != row.OperationCount {
		return semantic.CommandShape{}, errors.New("runtime record command operation shape is incomplete")
	}
	shape := semantic.CommandShape{
		CommandID: row.CommandID, CommandDigest: row.RequestDigest,
		BatchMode: semantic.BatchMode(row.BatchMode), OperationKinds: make([]semantic.OperationKind, len(operations)),
		OperationObjectIDs: make([]string, len(operations)),
		OperationRecordIDs: make([]string, len(operations)),
	}
	for i := range operations {
		if operations[i].Index != i {
			return semantic.CommandShape{}, errors.New("runtime record command operation order is invalid")
		}
		shape.OperationKinds[i] = semantic.OperationKind(operations[i].Kind)
		shape.OperationObjectIDs[i] = operations[i].ObjectID
		if shape.OperationKinds[i] != semantic.OperationCreate {
			shape.OperationRecordIDs[i] = operations[i].BoundRecordID
		}
	}
	return shape, nil
}

func (s *RuntimeRecordMutationService) closeInterruptedCommand(ctx context.Context, command persistence.RuntimeRecordCommandRow, request semantic.Request) (semantic.Response, error) {
	response, err := s.closeInterruptedCommandRow(ctx, command)
	if err != nil {
		return semantic.Response{}, err
	}
	if err := semantic.ValidateResponse(request, response); err != nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	return response, nil
}

func (s *RuntimeRecordMutationService) closeInterruptedCommandRow(ctx context.Context, command persistence.RuntimeRecordCommandRow) (semantic.Response, error) {
	var response semantic.Response
	err := s.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		locked, err := s.commands.LockCommand(txCtx, command.RowID)
		if err != nil {
			return err
		}
		if locked.State == persistence.RuntimeRecordCommandCompleted {
			return json.Unmarshal(locked.ResultPayload, &response)
		}
		slots, err := s.commands.LockOperations(txCtx, locked.RowID)
		if err != nil {
			return err
		}
		terminalExpiry, err := s.nextOutcomeExpiry(txCtx, locked)
		if err != nil {
			return err
		}
		results, pending, err := interruptedRuntimeRecordCommandResults(locked, slots)
		if err != nil {
			return err
		}
		for _, index := range pending {
			payload, err := json.Marshal(results[index])
			if err != nil {
				return err
			}
			if err := s.commands.CompleteOperation(txCtx, locked.RowID, index, payload); err != nil {
				return err
			}
		}
		var receipts semantic.Receipts
		if err := json.Unmarshal(locked.ReceiptsPayload, &receipts); err != nil {
			return err
		}
		locked.OutcomeExpiresAt = terminalExpiry
		response = s.response(locked, locked.CommandID, receipts, results)
		payload, err := json.Marshal(response)
		if err != nil {
			return err
		}
		return s.commands.CompleteCommand(txCtx, locked.RowID, payload, terminalExpiry)
	}, runtimeRecordCommandTransactionRetries)
	if err != nil {
		return semantic.Response{}, classifyRuntimeMutationTopLevelError(err)
	}
	return response, nil
}

func interruptedRuntimeRecordCommandResults(command persistence.RuntimeRecordCommandRow, slots []persistence.RuntimeRecordCommandOperationRow) ([]semantic.OperationResult, []int, error) {
	shape, err := runtimeRecordCommandShape(command, slots)
	if err != nil {
		return nil, nil, err
	}
	results := make([]semantic.OperationResult, len(slots))
	pending := make([]int, 0, len(slots))
	done := 0
	for i, slot := range slots {
		if slot.State == persistence.RuntimeRecordOperationDone {
			if len(slot.ResultPayload) == 0 || json.Unmarshal(slot.ResultPayload, &results[i]) != nil {
				return nil, nil, errors.New("runtime record command completed slot is corrupt")
			}
			done++
			continue
		}
		if slot.State != persistence.RuntimeRecordOperationPending {
			return nil, nil, errors.New("runtime record command slot has invalid state")
		}
		if len(slot.ResultPayload) != 0 {
			return nil, nil, errors.New("runtime record command pending slot is corrupt")
		}
		pending = append(pending, i)
	}
	mode := semantic.BatchMode(command.BatchMode)
	if mode == semantic.BatchAtomic {
		if done != 0 {
			return nil, nil, errors.New("accepted atomic command contains terminal operation slots")
		}
		for i := range slots {
			kind := semantic.ErrorAtomicRolledBack
			if i == 0 {
				kind = semantic.ErrorInternal
			}
			operation := semantic.Operation{Kind: shape.OperationKinds[i], ObjectID: shape.OperationObjectIDs[i], RecordID: shape.OperationRecordIDs[i]}
			results[i] = failedRuntimeMutationResult(i, operation, shape.OperationRecordIDs[i], kind, nil)
		}
	} else if mode == semantic.BatchIndependent {
		for _, i := range pending {
			operation := semantic.Operation{Kind: shape.OperationKinds[i], ObjectID: shape.OperationObjectIDs[i], RecordID: shape.OperationRecordIDs[i]}
			results[i] = failedRuntimeMutationResult(i, operation, shape.OperationRecordIDs[i], semantic.ErrorInternal, nil)
		}
	} else {
		return nil, nil, errors.New("runtime record command has invalid batch mode")
	}
	statusRequest := semantic.StatusRequest{CommandID: command.CommandID, CommandDigest: command.RequestDigest}
	status := semantic.StatusResponse{CommandID: command.CommandID, CommandDigest: command.RequestDigest, State: semantic.CommandCompleted, Results: results}
	var receipts semantic.Receipts
	if err := json.Unmarshal(command.ReceiptsPayload, &receipts); err != nil {
		return nil, nil, err
	}
	status.Receipts = &receipts
	status.OutcomeExpiresAtUTC = command.OutcomeExpiresAt.UTC().Format(time.RFC3339Nano)
	if err := semantic.ValidateStatusResponse(shape, statusRequest, status); err != nil {
		return nil, nil, errors.New("runtime record command interrupted outcome is invalid")
	}
	return results, pending, nil
}

func (s *RuntimeRecordMutationService) accept(ctx context.Context, scopeDigest string, request semantic.Request, receipts semantic.Receipts, digest string) (persistence.RuntimeRecordCommandRow, bool, error) {
	receiptsPayload, err := json.Marshal(receipts)
	if err != nil {
		return persistence.RuntimeRecordCommandRow{}, false, err
	}
	rowID := utils.GenerateID()
	operations := make([]persistence.RuntimeRecordCommandOperationAcceptance, len(request.Operations))
	for i, operation := range request.Operations {
		operations[i] = persistence.RuntimeRecordCommandOperationAcceptance{
			RowID: fmt.Sprintf("%s:%03d", rowID, i), Index: i, Kind: string(operation.Kind), ObjectID: operation.ObjectID, BoundRecordID: operation.RecordID,
		}
		if operation.Kind == semantic.OperationCreate {
			operations[i].BoundRecordID = utils.GenerateID()
		}
	}
	input := persistence.RuntimeRecordCommandAcceptance{
		RowID: rowID, ScopeDigest: scopeDigest, CommandID: request.CommandID, RequestDigest: digest,
		BatchMode: string(request.BatchMode), ReceiptsPayload: receiptsPayload,
		OutcomeRetention: s.retention, Operations: operations,
	}
	var command persistence.RuntimeRecordCommandRow
	var replay bool
	err = s.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		var acceptErr error
		command, replay, acceptErr = s.commands.Accept(txCtx, input)
		return acceptErr
	}, runtimeRecordCommandTransactionRetries)
	return command, replay, err
}

type runtimePreparedMutation struct {
	operation      semantic.Operation
	useCase        semantic.MutationUseCase
	schema         *models.ObjectMetadata
	objectName     string
	fields         models.Record
	requiredFields []string
	maskedMissing  bool
	preflightErr   error
}

type runtimeMutationPreflightFailure struct {
	index int
	error semantic.OperationError
}

// prepareFreshIndependentSingle removes the command-locking preparation
// transaction only when every fact needed by the business transaction is
// immutable in the request-bound Catalog view or carried by the request. The
// existing Records transaction still locks the command and operation slot,
// checks the database-clock execution window, authorizes against the DML-locked
// row, applies the effect, and persists the terminal outcome.
//
// Any replay, broader command shape, authored denial, existing-row dependency,
// or locally detectable operation failure returns to prepareAcceptedCommand.
// That locked path remains the sole owner of command-wide preflight results and
// resume semantics.
func (s *RuntimeRecordMutationService) prepareFreshIndependentSingle(
	ctx context.Context,
	request semantic.Request,
	user *models.UserSession,
	replay bool,
) ([]runtimePreparedMutation, bool) {
	if replay || request.BatchMode != semantic.BatchIndependent || request.UseCase != semantic.MutationUseCaseSingle || len(request.Operations) != 1 {
		return nil, false
	}
	operation := request.Operations[0]
	if (operation.Kind != semantic.OperationCreate && operation.Kind != semantic.OperationUpdate) || strings.TrimSpace(operation.ObjectID) == "" {
		return nil, false
	}

	schema, err := s.metadata.Describe(ctx, strings.TrimSpace(operation.ObjectID))
	if err != nil || schema == nil {
		return nil, false
	}
	fields, err := runtimeMutationFields(schema, operation.Fields)
	if err != nil {
		return nil, false
	}
	denial, needsExisting, err := s.runtimeMutationInputPolicyDenial(ctx, schema, request.UseCase, operation, fields, user)
	if err != nil {
		return nil, false
	}
	if denial != nil || needsExisting {
		return nil, false
	}

	prepared := runtimePreparedMutation{
		operation:  operation,
		useCase:    request.UseCase,
		schema:     schema,
		objectName: schema.APIName,
		fields:     fields,
	}
	if operation.Kind == semantic.OperationCreate {
		prepared.requiredFields = runtimeRequiredCreateFields(schema, fields)
		if len(prepared.requiredFields) != 0 {
			return nil, false
		}
	}
	return []runtimePreparedMutation{prepared}, true
}

func (s *RuntimeRecordMutationService) prepareAcceptedCommand(
	ctx context.Context,
	command persistence.RuntimeRecordCommandRow,
	request semantic.Request,
	receipts semantic.Receipts,
	user *models.UserSession,
) ([]runtimePreparedMutation, *semantic.Response, error) {
	var prepared []runtimePreparedMutation
	var terminal *semantic.Response
	err := s.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		prepared = nil
		terminal = nil
		locked, err := s.commands.LockCommand(txCtx, command.RowID)
		if err != nil {
			return err
		}
		if locked.State == persistence.RuntimeRecordCommandCompleted {
			response, err := decodeRuntimeMutationResponse(request, locked.ResultPayload)
			if err != nil {
				return err
			}
			terminal = &response
			return nil
		}
		if locked.State != persistence.RuntimeRecordCommandAccepted {
			return errors.New("runtime record command has invalid state")
		}
		command = locked
		slots, err := s.commands.LockOperations(txCtx, command.RowID)
		if err != nil || len(slots) != len(request.Operations) {
			if err == nil {
				err = errors.New("runtime record command operation slots are incomplete")
			}
			return err
		}
		if err := s.requireRuntimeRecordCommandExecutionWindow(txCtx, locked); err != nil {
			return err
		}
		_, pending, err := interruptedRuntimeRecordCommandResults(command, slots)
		if err != nil {
			return err
		}
		completed := make([]bool, len(slots))
		for i := range completed {
			completed[i] = true
		}
		for _, i := range pending {
			completed[i] = false
		}
		hasCompleted := len(pending) != len(slots)

		// The command-row lock is the preflight owner. No operation transaction
		// can commit while this decision is in progress because every independent
		// operation takes the same lock before its slot lock. Once any independent
		// position is durable, replay resumes only pending positions and relies on
		// their mandatory transaction-local policy check; it must not reopen the
		// zero-effect command-preflight phase.
		var preflight *runtimeMutationPreflightFailure
		prepared, preflight, err = s.prepareCommand(txCtx, request, user, !hasCompleted, completed)
		if err != nil {
			return err
		}
		if preflight == nil {
			return nil
		}
		response, err := s.persistCommandFailureLocked(txCtx, command, request, receipts, runtimeMutationPreflightResults(request, *preflight), slots)
		if err != nil {
			return err
		}
		terminal = &response
		return nil
	}, runtimeRecordCommandTransactionRetries)
	return prepared, terminal, err
}

func (s *RuntimeRecordMutationService) prepareCommand(ctx context.Context, request semantic.Request, user *models.UserSession, commandPreflight bool, completed []bool) ([]runtimePreparedMutation, *runtimeMutationPreflightFailure, error) {
	if len(completed) != len(request.Operations) {
		return nil, nil, errors.New("runtime record command completion shape is invalid")
	}
	prepared := make([]runtimePreparedMutation, len(request.Operations))
	var opaqueCandidatesByIndex map[int][]string
	var err error
	if commandPreflight {
		var candidateFailure *runtimeMutationPreflightFailure
		opaqueCandidatesByIndex, candidateFailure, err = s.preflightOpaqueCollectionDeleteCandidates(ctx, request, user)
		if err != nil || candidateFailure != nil {
			return prepared, candidateFailure, err
		}
	}
	for i, operation := range request.Operations {
		if completed[i] {
			prepared[i] = runtimePreparedMutation{operation: operation, useCase: request.UseCase}
			continue
		}
		objectName := strings.TrimSpace(operation.ObjectID)
		if operation.Kind == semantic.OperationDelete {
			candidates, candidateResolved := opaqueCandidatesByIndex[i]
			if candidateResolved && len(candidates) == 0 {
				prepared[i] = runtimePreparedMutation{operation: operation, useCase: request.UseCase, maskedMissing: true}
				continue
			}
			if candidateResolved && len(candidates) == 1 {
				objectName = candidates[0]
			} else {
				resolved, maskedMissing, err := s.resolveDeleteObject(ctx, operation)
				if err != nil {
					return nil, nil, err
				}
				if maskedMissing {
					prepared[i] = runtimePreparedMutation{operation: operation, useCase: request.UseCase, maskedMissing: true}
					continue
				}
				objectName = resolved
			}
		}
		schema, err := s.metadata.Describe(ctx, objectName)
		if err != nil || schema == nil {
			if err == nil {
				err = appErrors.NewNotFoundError("Object", "")
			}
			if runtimeMutationNotFound(err) {
				if !commandPreflight {
					return nil, nil, errors.New("runtime record command bound Catalog object is unavailable during resume")
				}
				return prepared, &runtimeMutationPreflightFailure{index: i, error: semantic.OperationError{
					Kind: semantic.ErrorObjectNotFound, Fields: []string{},
				}}, nil
			}
			return nil, nil, err
		}
		fields, err := runtimeMutationFields(schema, operation.Fields)
		if err != nil {
			prepared[i] = runtimePreparedMutation{operation: operation, useCase: request.UseCase, schema: schema, objectName: schema.APIName, preflightErr: err}
			continue
		}
		prepared[i] = runtimePreparedMutation{operation: operation, useCase: request.UseCase, schema: schema, objectName: schema.APIName, fields: fields}

		if commandPreflight {
			denial, needsExisting, err := s.runtimeMutationInputPolicyDenial(
				ctx, schema, request.UseCase, operation, fields, user,
			)
			if err != nil {
				return nil, nil, err
			}
			if denial != nil {
				policyFields := []string{}
				if field := strings.TrimSpace(denial.Field); field != "" {
					policyFields = []string{field}
				}
				return prepared, &runtimeMutationPreflightFailure{index: i, error: semantic.OperationError{
					Kind: semantic.ErrorPolicyDenied, AuthoredMessage: denial.Message, Fields: policyFields,
				}}, nil
			}
			existing, maskedMissing, err := s.runtimeMutationPolicyTarget(ctx, schema, operation, user)
			if err != nil {
				if _, ok := classifyRuntimeMutationDomainError(operation, i, err); ok {
					prepared[i].preflightErr = err
					continue
				}
				return nil, nil, err
			}
			if maskedMissing {
				prepared[i].maskedMissing = true
				continue
			}
			if needsExisting {
				denial, err = s.runtimeMutationPolicyDenial(ctx, schema, request.UseCase, operation, fields, existing, user)
				if err != nil {
					return nil, nil, err
				}
			}
			if denial != nil {
				fields := []string{}
				if field := strings.TrimSpace(denial.Field); field != "" {
					fields = []string{field}
				}
				return prepared, &runtimeMutationPreflightFailure{index: i, error: semantic.OperationError{
					Kind: semantic.ErrorPolicyDenied, AuthoredMessage: denial.Message, Fields: fields,
				}}, nil
			}
		}
		if request.UseCase == semantic.MutationUseCaseSingle && operation.Kind == semantic.OperationCreate {
			prepared[i].requiredFields = runtimeRequiredCreateFields(schema, fields)
		}
	}
	return prepared, nil, nil
}

// preflightOpaqueCollectionDeleteCandidates closes the ambiguity gap for
// objectless collection deletes whose opaque IDs can exist in more than one
// physical object table. One batch discovery preserves every candidate, then
// the request-bound Catalog policy is evaluated for each candidate before any
// position can commit. Prefix-addressed and explicit-object deletes already
// have one Catalog owner and never enter this compatibility path.
//
// Candidate discovery is deliberately command-preflight-only. An accepted
// independent replay with durable positions must not reopen a zero-effect
// command decision; its pending positions retain the existing transaction-
// local policy recheck and execution semantics.
func (s *RuntimeRecordMutationService) preflightOpaqueCollectionDeleteCandidates(
	ctx context.Context,
	request semantic.Request,
	user *models.UserSession,
) (map[int][]string, *runtimeMutationPreflightFailure, error) {
	if request.UseCase != semantic.MutationUseCaseCollection {
		return nil, nil, nil
	}
	indicesByID := make(map[string][]int)
	ids := make([]string, 0, len(request.Operations))
	for i, operation := range request.Operations {
		if operation.Kind != semantic.OperationDelete || strings.TrimSpace(operation.ObjectID) != "" {
			continue
		}
		if _, resolved := s.metadata.ResolveObjectForRecordID(ctx, operation.RecordID); resolved {
			continue
		}
		id := strings.TrimSpace(operation.RecordID)
		if _, exists := indicesByID[id]; !exists {
			ids = append(ids, id)
		}
		indicesByID[id] = append(indicesByID[id], i)
	}
	if len(ids) == 0 {
		return nil, nil, nil
	}
	// Physical candidate discovery is a read-only compatibility snapshot and
	// cannot use the command transaction's single connection concurrently.
	// Detaching only this catalog walk lets the repository use its bounded pool;
	// the command lock, policy target reads, DML locks, and durable outcome stay
	// on the owning transaction below.
	candidatesByID, err := s.engine.ResolveOpaqueRecordObjectCandidates(s.tx.DetachTx(ctx), ids)
	if err != nil {
		return nil, nil, err
	}
	resolvedByIndex := make(map[int][]string, len(indicesByID))
	for _, id := range ids {
		candidates := append([]string(nil), candidatesByID[id]...)
		sort.SliceStable(candidates, func(i, j int) bool {
			left, right := strings.ToLower(candidates[i]), strings.ToLower(candidates[j])
			if left == right {
				return candidates[i] < candidates[j]
			}
			return left < right
		})
		seen := make(map[string]struct{}, len(candidates))
		normalized := make([]string, 0, len(candidates))
		for _, objectName := range candidates {
			objectName = strings.TrimSpace(objectName)
			key := strings.ToLower(objectName)
			if objectName == "" {
				return nil, nil, errors.New("runtime record mutation opaque candidate has empty object identity")
			}
			if _, duplicate := seen[key]; duplicate {
				continue
			}
			seen[key] = struct{}{}
			normalized = append(normalized, objectName)
			schema, err := s.metadata.Describe(ctx, objectName)
			if err != nil || schema == nil || strings.TrimSpace(schema.APIName) == "" {
				if err == nil {
					err = errors.New("runtime record mutation opaque candidate is absent from the request-bound Catalog")
				}
				return nil, nil, err
			}
			for _, index := range indicesByID[id] {
				operation := request.Operations[index]
				fields := models.Record{}
				denial, needsExisting, err := s.runtimeMutationInputPolicyDenial(ctx, schema, request.UseCase, operation, fields, user)
				if err != nil {
					return nil, nil, err
				}
				if denial == nil && needsExisting {
					existing, readErr := s.engine.ReadPolicyTarget(ctx, schema.APIName, operation.RecordID, operation.Kind, user)
					if readErr != nil {
						if runtimeMutationNotFound(readErr) {
							continue
						}
						return nil, nil, readErr
					}
					denial, err = s.runtimeMutationPolicyDenial(ctx, schema, request.UseCase, operation, fields, contract.Record(existing), user)
					if err != nil {
						return nil, nil, err
					}
				}
				if denial != nil {
					policyFields := []string{}
					if field := strings.TrimSpace(denial.Field); field != "" {
						policyFields = []string{field}
					}
					return nil, &runtimeMutationPreflightFailure{index: index, error: semantic.OperationError{
						Kind: semantic.ErrorPolicyDenied, AuthoredMessage: denial.Message, Fields: policyFields,
					}}, nil
				}
			}
		}
		for _, index := range indicesByID[id] {
			resolvedByIndex[index] = append([]string(nil), normalized...)
		}
	}
	return resolvedByIndex, nil, nil
}

// runtimeMutationInputPolicyDenial evaluates the part of a Catalog policy
// whose truth is fully determined by the request. An unconditional object
// guard is safe to return before looking up an update/delete target because it
// is identical for every record and therefore discloses no row existence.
// Policies that depend on the stored row still defer to the access-aware target
// read and locked transactional recheck below.
func (s *RuntimeRecordMutationService) runtimeMutationInputPolicyDenial(
	ctx context.Context,
	schema *models.ObjectMetadata,
	useCase semantic.MutationUseCase,
	operation semantic.Operation,
	fields models.Record,
	user *models.UserSession,
) (*models.RecordMutationDenial, bool, error) {
	record := recordmutation.WithRecordID(fields, operation.RecordID)
	needsExisting := false
	for _, surface := range runtimeMutationPolicySurfaces(useCase, operation.Kind) {
		denial, err := recordmutation.Denial(ctx, recordmutation.Evaluator{}, schema, surface, record, user)
		if err != nil || denial != nil {
			return denial, false, err
		}
		if operation.RecordID != "" && recordmutation.MayNeedExisting(schema, surface, record) {
			needsExisting = true
		}
	}
	return nil, needsExisting, nil
}

// resolveDeleteObject binds a delete to the request-bound Catalog identity
// without enumerating every physical object table. An explicit ObjectID is
// already the operation's object boundary; DeleteAuthorized later proves
// existence and authorization against that table under the mutation lock.
// Deletes without ObjectID use the same KeyPrefix owner. Only opaque/UUID IDs,
// which intentionally have no Catalog prefix identity, retain the legacy
// physical reverse lookup.
func (s *RuntimeRecordMutationService) resolveDeleteObject(ctx context.Context, operation semantic.Operation) (string, bool, error) {
	if err := ctx.Err(); err != nil {
		return "", false, err
	}
	requested := strings.TrimSpace(operation.ObjectID)
	inferred, inferredByMetadata := s.metadata.ResolveObjectForRecordID(ctx, operation.RecordID)
	if requested != "" {
		if inferredByMetadata && !strings.EqualFold(requested, inferred) {
			return "", true, nil
		}
		return requested, false, nil
	}
	if inferredByMetadata {
		return inferred, false, nil
	}
	resolved, err := s.engine.ResolveOpaqueRecordObject(ctx, operation.RecordID)
	if err != nil || strings.TrimSpace(resolved) == "" {
		if err == nil {
			err = appErrors.NewNotFoundError("Record", "")
		}
		if runtimeMutationNotFound(err) {
			// Let the regular DML path produce the position-correlated,
			// existence-masked result without exposing metadata facts.
			return "", true, nil
		}
		return "", false, err
	}
	return resolved, false, nil
}

// runtimeMutationPolicyTarget establishes the visible row identity before any
// authored update/delete policy can produce a distinguishable response. The
// object boundary is already known, so this is one access-aware read from that
// object—not physical object discovery. The returned row is reused by
// conditional policy evaluation; locked DML remains the final authority.
func (s *RuntimeRecordMutationService) runtimeMutationPolicyTarget(
	ctx context.Context,
	schema *models.ObjectMetadata,
	operation semantic.Operation,
	user *models.UserSession,
) (contract.Record, bool, error) {
	if operation.Kind != semantic.OperationUpdate && operation.Kind != semantic.OperationDelete {
		return nil, false, nil
	}
	loaded, err := s.engine.ReadPolicyTarget(ctx, schema.APIName, operation.RecordID, operation.Kind, user)
	if err != nil {
		if runtimeMutationNotFound(err) {
			return nil, true, nil
		}
		return nil, false, err
	}
	if loaded == nil {
		return nil, true, nil
	}
	return contract.Record(loaded), false, nil
}

func (s *RuntimeRecordMutationService) runtimeMutationPolicyDenial(
	ctx context.Context,
	schema *models.ObjectMetadata,
	useCase semantic.MutationUseCase,
	operation semantic.Operation,
	fields models.Record,
	existing contract.Record,
	user *models.UserSession,
) (*models.RecordMutationDenial, error) {
	surfaces := runtimeMutationPolicySurfaces(useCase, operation.Kind)
	if len(surfaces) == 0 {
		return nil, appErrors.NewValidationError("operation", "unsupported mutation policy use case")
	}
	record := recordmutation.WithRecordID(fields, operation.RecordID)
	for _, surface := range surfaces {
		denial, err := recordmutation.Denial(ctx, recordmutation.Evaluator{}, schema, surface, record, user)
		if err != nil || denial != nil {
			return denial, err
		}
		if operation.RecordID == "" || !recordmutation.MayNeedExisting(schema, surface, record) {
			continue
		}
		if existing == nil {
			return nil, errors.New("runtime record mutation policy target was not loaded")
		}
		denial, err = recordmutation.Denial(ctx, recordmutation.Evaluator{}, schema, surface, recordmutation.MergeRecord(existing, fields), user)
		if err != nil || denial != nil {
			return denial, err
		}
	}
	return nil, nil
}

// runtimeMutationPolicyDenialFromLockedRecord is the only authoritative
// conditional-policy decision for update and delete. The persistence engine
// supplies the row image after acquiring its normal mutation lock, and the
// same locked image remains authoritative through DML. Related-record
// conditions deliberately have no generic Query here: an unconstrained query
// cannot prove serialization with concurrent child inserts in TiDB, so those
// policies fail closed until Catalog can supply an explicit coordination
// protocol.
func (s *RuntimeRecordMutationService) runtimeMutationPolicyDenialFromLockedRecord(
	ctx context.Context,
	schema *models.ObjectMetadata,
	useCase semantic.MutationUseCase,
	operation semantic.Operation,
	fields models.Record,
	locked models.Record,
	user *models.UserSession,
) (*models.RecordMutationDenial, error) {
	record := recordmutation.MergeRecord(contract.Record(locked), recordmutation.WithRecordID(fields, operation.RecordID))
	for _, surface := range runtimeMutationPolicySurfaces(useCase, operation.Kind) {
		denial, err := recordmutation.Denial(ctx, recordmutation.Evaluator{}, schema, surface, record, user)
		if err != nil || denial != nil {
			return denial, err
		}
	}
	return nil, nil
}

func runtimeMutationPolicySurfaces(useCase semantic.MutationUseCase, kind semantic.OperationKind) []string {
	single, collection := "", ""
	switch kind {
	case semantic.OperationCreate:
		single, collection = models.RecordMutationSurfaceSingleCreate, models.RecordMutationSurfaceCollectionCreate
	case semantic.OperationUpdate:
		single, collection = models.RecordMutationSurfaceSingleUpdate, models.RecordMutationSurfaceCollectionUpdate
	case semantic.OperationDelete:
		single, collection = models.RecordMutationSurfaceSingleDelete, models.RecordMutationSurfaceCollectionDelete
	}
	switch useCase {
	case semantic.MutationUseCaseSingle:
		return []string{single}
	case semantic.MutationUseCaseCollection:
		return []string{collection}
	case semantic.MutationUseCaseUnspecified:
		// N-1 callers do not get to bypass a newly projected policy. Evaluate
		// both existing policy families; never infer from operation count.
		return []string{single, collection}
	default:
		return nil
	}
}

func runtimeMutationPreflightResults(request semantic.Request, failure runtimeMutationPreflightFailure) []semantic.OperationResult {
	results := make([]semantic.OperationResult, len(request.Operations))
	for i, operation := range request.Operations {
		operationError := semantic.OperationError{Kind: semantic.ErrorProcessingHalted, Fields: []string{}}
		if i == failure.index {
			operationError = failure.error
			operationError.Fields = append([]string(nil), operationError.Fields...)
		}
		results[i] = semantic.OperationResult{Index: uint32(i), Kind: operation.Kind, RecordID: operation.RecordID,
			Success: false, Errors: []semantic.OperationError{operationError}}
	}
	return results
}

func (s *RuntimeRecordMutationService) executeIndependent(ctx context.Context, command persistence.RuntimeRecordCommandRow, request semantic.Request, prepared []runtimePreparedMutation, receipts semantic.Receipts, user *models.UserSession) (semantic.Response, error) {
	var committed *semantic.Response
	for i := range prepared {
		finalizeSingle := len(prepared) == 1
		response, err := s.executeIndependentOperation(ctx, command, request, receipts, prepared[i], i, user, finalizeSingle)
		if err != nil {
			return semantic.Response{}, err
		}
		if response != nil {
			committed = response
		}
	}
	if committed != nil {
		return *committed, nil
	}
	finalizeCtx, finishFinalize := beginRuntimeMutationPhase(ctx, runtimeMutationPhaseFinalize)
	response, err := s.finalizeIfComplete(finalizeCtx, command, request, receipts)
	finishFinalize()
	return response, err
}

func (s *RuntimeRecordMutationService) executeIndependentOperation(
	ctx context.Context,
	command persistence.RuntimeRecordCommandRow,
	request semantic.Request,
	receipts semantic.Receipts,
	prepared runtimePreparedMutation,
	index int,
	user *models.UserSession,
	finalizeSingle bool,
) (*semantic.Response, error) {
	var domain *runtimeMutationDomainFailure
	var completion ports.TransactionCompletionOutcome
	var committed *semantic.Response
	err := s.tx.RunRecordMutationTransaction(ctx, []string{prepared.objectName}, func(txCtx context.Context) error {
		ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
			completion = outcome
		})
		guardCtx, finishGuard := beginRuntimeMutationPhase(txCtx, runtimeMutationPhaseOperationGuard)
		var slot persistence.RuntimeRecordCommandOperationRow
		skipOperation := false
		guardErr := func() error {
			locked, err := s.commands.LockCommand(guardCtx, command.RowID)
			if err != nil {
				return err
			}
			if locked.State == persistence.RuntimeRecordCommandCompleted {
				skipOperation = true
				return nil
			}
			if locked.State != persistence.RuntimeRecordCommandAccepted {
				return errors.New("runtime record command has invalid state")
			}
			command = locked
			slot, err = s.commands.LockOperation(guardCtx, command.RowID, index)
			if err != nil {
				return err
			}
			if slot.State == persistence.RuntimeRecordOperationDone {
				skipOperation = true
				return nil
			}
			if slot.State != persistence.RuntimeRecordOperationPending || len(slot.ResultPayload) != 0 {
				return errors.New("runtime record command operation has invalid state")
			}
			return s.requireRuntimeRecordCommandExecutionWindow(guardCtx, locked)
		}()
		finishGuard()
		if guardErr != nil {
			return guardErr
		}
		if skipOperation {
			return nil
		}

		effectCtx, finishEffect := beginRuntimeMutationPhase(txCtx, runtimeMutationPhaseBusinessEffect)
		result, err := s.executeOperation(effectCtx, prepared, index, slot.BoundRecordID, user)
		finishEffect()
		if err != nil {
			if classified, ok := classifyRuntimeMutationDomainError(prepared.operation, index, err); ok {
				domain = &runtimeMutationDomainFailure{result: classified, cause: err}
				return domain
			}
			return err
		}
		payload, err := json.Marshal(result)
		if err != nil {
			return err
		}
		completeCtx, finishComplete := beginRuntimeMutationPhase(txCtx, runtimeMutationPhaseOperationComplete)
		completeErr := s.commands.CompleteOperation(completeCtx, command.RowID, index, payload)
		if completeErr == nil && finalizeSingle {
			command.OutcomeExpiresAt, completeErr = s.nextOutcomeExpiry(completeCtx, command)
			if completeErr == nil {
				response := s.response(command, request.CommandID, receipts, []semantic.OperationResult{result})
				if completeErr = validateRuntimeMutationResponseForPersistence(request, response); completeErr == nil {
					var responsePayload []byte
					responsePayload, completeErr = json.Marshal(response)
					if completeErr == nil {
						completeErr = s.commands.CompleteCommand(completeCtx, command.RowID, responsePayload, command.OutcomeExpiresAt)
					}
				}
				if completeErr == nil {
					committed = &response
				}
			}
		}
		finishComplete()
		return completeErr
	})
	if err == nil {
		return committed, nil
	}
	if domain == nil || !errors.As(err, &domain) || completion != ports.TransactionCompletionRolledBack {
		return nil, err
	}
	// The domain failure transaction is known rolled back. Persist only its
	// terminal result in a fresh transaction; no business effect can accompany it.
	err = s.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		locked, lockErr := s.commands.LockCommand(txCtx, command.RowID)
		if lockErr != nil {
			return lockErr
		}
		if locked.State == persistence.RuntimeRecordCommandCompleted {
			return nil
		}
		if locked.State != persistence.RuntimeRecordCommandAccepted {
			return errors.New("runtime record command has invalid state")
		}
		slot, lockErr := s.commands.LockOperation(txCtx, command.RowID, index)
		if lockErr != nil {
			return lockErr
		}
		if slot.State == persistence.RuntimeRecordOperationDone {
			return nil
		}
		payload, marshalErr := json.Marshal(domain.result)
		if marshalErr != nil {
			return marshalErr
		}
		return s.commands.CompleteOperation(txCtx, command.RowID, index, payload)
	}, runtimeRecordCommandTransactionRetries)
	return nil, err
}

func (s *RuntimeRecordMutationService) executeAtomic(ctx context.Context, command persistence.RuntimeRecordCommandRow, request semantic.Request, prepared []runtimePreparedMutation, receipts semantic.Receipts, user *models.UserSession) (semantic.Response, error) {
	results := make([]semantic.OperationResult, len(request.Operations))
	var committed *semantic.Response
	objectNames := make([]string, len(prepared))
	for i := range prepared {
		objectNames[i] = prepared[i].objectName
	}
	var domain *runtimeMutationDomainFailure
	var completion ports.TransactionCompletionOutcome
	err := s.tx.RunRecordMutationTransaction(ctx, objectNames, func(txCtx context.Context) error {
		ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
			completion = outcome
		})
		guardCtx, finishGuard := beginRuntimeMutationPhase(txCtx, runtimeMutationPhaseOperationGuard)
		var slots []persistence.RuntimeRecordCommandOperationRow
		skipCommand := false
		guardErr := func() error {
			locked, err := s.commands.LockCommand(guardCtx, command.RowID)
			if err != nil {
				return err
			}
			if locked.State == persistence.RuntimeRecordCommandCompleted {
				response, decodeErr := decodeRuntimeMutationResponse(request, locked.ResultPayload)
				if decodeErr != nil {
					return decodeErr
				}
				committed = &response
				skipCommand = true
				return nil
			}
			if locked.State != persistence.RuntimeRecordCommandAccepted {
				return errors.New("runtime record command has invalid state")
			}
			command = locked
			slots, err = s.commands.LockOperations(guardCtx, command.RowID)
			if err != nil || len(slots) != len(request.Operations) {
				if err == nil {
					err = errors.New("runtime record command operation slots are incomplete")
				}
				return err
			}
			for _, slot := range slots {
				if slot.State != persistence.RuntimeRecordOperationPending || len(slot.ResultPayload) != 0 {
					return errors.New("accepted atomic command contains invalid operation state")
				}
			}
			if err := s.requireRuntimeRecordCommandExecutionWindow(guardCtx, locked); err != nil {
				return err
			}
			command.OutcomeExpiresAt, err = s.nextOutcomeExpiry(guardCtx, locked)
			return err
		}()
		finishGuard()
		if guardErr != nil || skipCommand {
			return guardErr
		}
		for i := range prepared {
			operation := prepared[i].operation
			effectCtx, finishEffect := beginRuntimeMutationPhase(txCtx, runtimeMutationPhaseBusinessEffect)
			result, operationErr := s.executeOperation(effectCtx, prepared[i], i, slots[i].BoundRecordID, user)
			finishEffect()
			if operationErr != nil {
				if classified, ok := classifyRuntimeMutationDomainError(operation, i, operationErr); ok {
					domain = &runtimeMutationDomainFailure{result: classified, cause: operationErr}
					return domain
				}
				return operationErr
			}
			results[i] = result
		}
		finalizeCtx, finishFinalize := beginRuntimeMutationPhase(txCtx, runtimeMutationPhaseFinalize)
		finalizeErr := func() error {
			response := s.response(command, request.CommandID, receipts, results)
			if err := validateRuntimeMutationResponseForPersistence(request, response); err != nil {
				return err
			}
			payload, err := json.Marshal(response)
			if err != nil {
				return err
			}
			for i := range results {
				resultPayload, marshalErr := json.Marshal(results[i])
				if marshalErr != nil {
					return marshalErr
				}
				if err := s.commands.CompleteOperation(finalizeCtx, command.RowID, i, resultPayload); err != nil {
					return err
				}
			}
			if err := s.commands.CompleteCommand(finalizeCtx, command.RowID, payload, command.OutcomeExpiresAt); err != nil {
				return err
			}
			committed = &response
			return nil
		}()
		finishFinalize()
		return finalizeErr
	})
	if err == nil {
		if committed == nil {
			return semantic.Response{}, errors.New("runtime record command committed without a terminal response")
		}
		return *committed, nil
	}
	if domain == nil || !errors.As(err, &domain) || completion != ports.TransactionCompletionRolledBack {
		return semantic.Response{}, err
	}
	for i, operation := range request.Operations {
		if i == int(domain.result.Index) {
			results[i] = domain.result
			continue
		}
		results[i] = failedRuntimeMutationResult(i, operation, operation.RecordID, semantic.ErrorAtomicRolledBack, nil)
	}
	return s.persistCommandFailure(ctx, command, request, receipts, results)
}

func (s *RuntimeRecordMutationService) persistCommandFailure(ctx context.Context, command persistence.RuntimeRecordCommandRow, request semantic.Request, receipts semantic.Receipts, results []semantic.OperationResult) (semantic.Response, error) {
	err := s.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		locked, lockErr := s.commands.LockCommand(txCtx, command.RowID)
		if lockErr != nil {
			return lockErr
		}
		if locked.State == persistence.RuntimeRecordCommandCompleted {
			return nil
		}
		command = locked
		slots, lockErr := s.commands.LockOperations(txCtx, command.RowID)
		if lockErr != nil || len(slots) != len(results) {
			if lockErr == nil {
				lockErr = errors.New("runtime record command operation slots are incomplete")
			}
			return lockErr
		}
		_, lockErr = s.persistCommandFailureLocked(txCtx, command, request, receipts, results, slots)
		return lockErr
	}, runtimeRecordCommandTransactionRetries)
	if err != nil {
		return semantic.Response{}, err
	}
	loaded, err := s.commands.Load(ctx, command.ScopeDigest, command.CommandID)
	if err != nil {
		return semantic.Response{}, err
	}
	return decodeRuntimeMutationResponse(request, loaded.ResultPayload)
}

func (s *RuntimeRecordMutationService) persistCommandFailureLocked(
	txCtx context.Context,
	command persistence.RuntimeRecordCommandRow,
	request semantic.Request,
	receipts semantic.Receipts,
	results []semantic.OperationResult,
	slots []persistence.RuntimeRecordCommandOperationRow,
) (semantic.Response, error) {
	if len(slots) != len(results) {
		return semantic.Response{}, errors.New("runtime record command operation slots are incomplete")
	}
	var err error
	command.OutcomeExpiresAt, err = s.nextOutcomeExpiry(txCtx, command)
	if err != nil {
		return semantic.Response{}, err
	}
	for i, slot := range slots {
		switch slot.State {
		case persistence.RuntimeRecordOperationDone:
			// Durable operation outcomes are immutable. A terminal command must
			// preserve them rather than overwrite committed effects with a later
			// failure proposal.
			if len(slot.ResultPayload) == 0 || json.Unmarshal(slot.ResultPayload, &results[i]) != nil {
				return semantic.Response{}, errors.New("runtime record command completed slot is corrupt")
			}
		case persistence.RuntimeRecordOperationPending:
			if len(slot.ResultPayload) != 0 {
				return semantic.Response{}, errors.New("runtime record command pending slot is corrupt")
			}
		default:
			return semantic.Response{}, errors.New("runtime record command slot has invalid state")
		}
	}
	response := s.response(command, request.CommandID, receipts, results)
	if err := validateRuntimeMutationResponseForPersistence(request, response); err != nil {
		return semantic.Response{}, err
	}
	payload, err := json.Marshal(response)
	if err != nil {
		return semantic.Response{}, err
	}
	for i, slot := range slots {
		if slot.State == persistence.RuntimeRecordOperationDone {
			continue
		}
		resultPayload, err := json.Marshal(results[i])
		if err != nil {
			return semantic.Response{}, err
		}
		if err := s.commands.CompleteOperation(txCtx, command.RowID, i, resultPayload); err != nil {
			return semantic.Response{}, err
		}
	}
	if err := s.commands.CompleteCommand(txCtx, command.RowID, payload, command.OutcomeExpiresAt); err != nil {
		return semantic.Response{}, err
	}
	return response, nil
}

func (s *RuntimeRecordMutationService) finalizeIfComplete(ctx context.Context, command persistence.RuntimeRecordCommandRow, request semantic.Request, receipts semantic.Receipts) (semantic.Response, error) {
	var response semantic.Response
	err := s.tx.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		locked, err := s.commands.LockCommand(txCtx, command.RowID)
		if err != nil {
			return err
		}
		if locked.State == persistence.RuntimeRecordCommandCompleted {
			return json.Unmarshal(locked.ResultPayload, &response)
		}
		command = locked
		command.OutcomeExpiresAt, err = s.nextOutcomeExpiry(txCtx, locked)
		if err != nil {
			return err
		}
		slots, err := s.commands.LockOperations(txCtx, command.RowID)
		if err != nil || len(slots) != len(request.Operations) {
			if err == nil {
				err = errors.New("runtime record command operation slots are incomplete")
			}
			return err
		}
		results := make([]semantic.OperationResult, len(slots))
		for i, slot := range slots {
			if slot.State != persistence.RuntimeRecordOperationDone || len(slot.ResultPayload) == 0 {
				return errors.New("runtime record command has a nonterminal operation")
			}
			if err := json.Unmarshal(slot.ResultPayload, &results[i]); err != nil {
				return err
			}
		}
		response = s.response(command, request.CommandID, receipts, results)
		if err := validateRuntimeMutationResponseForPersistence(request, response); err != nil {
			return err
		}
		payload, err := json.Marshal(response)
		if err != nil {
			return err
		}
		return s.commands.CompleteCommand(txCtx, command.RowID, payload, command.OutcomeExpiresAt)
	}, runtimeRecordCommandTransactionRetries)
	return response, err
}

func (s *RuntimeRecordMutationService) executeOperation(ctx context.Context, prepared runtimePreparedMutation, index int, boundRecordID string, user *models.UserSession) (semantic.OperationResult, error) {
	operation := prepared.operation
	fields := prepared.fields
	if prepared.maskedMissing {
		return semantic.OperationResult{}, appErrors.NewNotFoundError("Record", "")
	}
	if prepared.preflightErr != nil {
		return semantic.OperationResult{}, prepared.preflightErr
	}
	if len(prepared.requiredFields) != 0 {
		return semantic.OperationResult{}, runtimeRequiredFieldsError{fields: prepared.requiredFields}
	}
	// The earlier command preflight preserves the public collection behavior
	// (a known policy denial prevents any sibling write), but it is not
	// authorization authority. Update and delete pass policy evaluation into the
	// persistence mutation so it runs against the row image held by that DML's
	// lock. Merely reading through the same transaction context is insufficient:
	// TiDB snapshot reads and later current reads can observe different versions.
	switch operation.Kind {
	case semantic.OperationCreate:
		denial, err := s.runtimeMutationPolicyDenial(ctx, prepared.schema, prepared.useCase, operation, fields, nil, user)
		if err != nil {
			return semantic.OperationResult{}, err
		}
		if denial != nil {
			return semantic.OperationResult{}, runtimeMutationPolicyError{denial: *denial}
		}
		created, err := s.engine.Insert(withRuntimeReservedCreateID(ctx, boundRecordID), prepared.objectName, fields, user)
		if err != nil {
			return semantic.OperationResult{}, err
		}
		id, _ := created[constants.FieldID].(string)
		if id == "" || id != boundRecordID {
			return semantic.OperationResult{}, errors.New("runtime record mutation create did not preserve reserved identifier")
		}
		createdFlag := true
		return semantic.OperationResult{Index: uint32(index), Kind: operation.Kind, RecordID: id, Success: true, Created: &createdFlag, Errors: []semantic.OperationError{}}, nil
	case semantic.OperationUpdate:
		if err := s.engine.UpdateAuthorized(ctx, prepared.objectName, operation.RecordID, fields, user, s.runtimeMutationAuthorizer(prepared, user)); err != nil {
			return semantic.OperationResult{}, err
		}
		return semantic.OperationResult{Index: uint32(index), Kind: operation.Kind, RecordID: operation.RecordID, Success: true, Errors: []semantic.OperationError{}}, nil
	case semantic.OperationDelete:
		if err := s.engine.DeleteAuthorized(ctx, prepared.objectName, operation.RecordID, user, s.runtimeMutationAuthorizer(prepared, user)); err != nil {
			return semantic.OperationResult{}, err
		}
		return semantic.OperationResult{Index: uint32(index), Kind: operation.Kind, RecordID: operation.RecordID, Success: true, Errors: []semantic.OperationError{}}, nil
	default:
		return semantic.OperationResult{}, appErrors.NewValidationError("operation", "unsupported mutation operation")
	}
}

func (s *RuntimeRecordMutationService) runtimeMutationAuthorizer(prepared runtimePreparedMutation, user *models.UserSession) func(context.Context, models.Record) error {
	return func(ctx context.Context, locked models.Record) error {
		denial, err := s.runtimeMutationPolicyDenialFromLockedRecord(ctx, prepared.schema, prepared.useCase, prepared.operation, prepared.fields, locked, user)
		if err != nil {
			return err
		}
		if denial != nil {
			return runtimeMutationPolicyError{denial: *denial}
		}
		return nil
	}
}

func runtimeMutationFields(schema *models.ObjectMetadata, fields []semantic.FieldValue) (models.Record, error) {
	result := make(models.Record, len(fields))
	for _, field := range fields {
		fieldMetadata := models.FindField(schema, field.Name)
		name := field.Name
		if fieldMetadata != nil {
			name = fieldMetadata.APIName
		}
		if _, duplicate := result[name]; duplicate {
			return nil, appErrors.NewInvalidFieldForWriteError([]string{name}, "field is assigned more than once after Catalog canonicalization")
		}
		if field.Null {
			result[name] = nil
			continue
		}
		var value any
		switch field.Kind {
		case semantic.ValueKindString:
			value = *field.String
		case semantic.ValueKindBoolean:
			value = *field.Boolean
		case semantic.ValueKindInteger:
			value = *field.Integer
		case semantic.ValueKindDecimal:
			decimal, err := runtimeExactDecimal(*field.Decimal)
			if err != nil {
				return nil, err
			}
			value = decimal
		case semantic.ValueKindDate:
			value = *field.Date
		case semantic.ValueKindDateTime:
			value = *field.DateTime
		case semantic.ValueKindReference:
			value = *field.Reference
		case semantic.ValueKindNumber:
			converted, err := runtimeJSONNumberValue(*field.Number, fieldMetadata)
			if err != nil {
				return nil, appErrors.NewValidationError(name, "invalid numeric field value")
			}
			value = converted
		case semantic.ValueKindJSON:
			converted, err := runtimeJSONContainerValue(*field.JSON, fieldMetadata)
			if err != nil {
				return nil, appErrors.NewValidationError(name, "invalid JSON field value")
			}
			value = converted
		default:
			return nil, appErrors.NewValidationError(name, "unsupported field value")
		}
		result[name] = value
	}
	return result, nil
}

func runtimeJSONNumberValue(raw string, field *models.FieldMetadata) (any, error) {
	exact, err := runtimeExactJSONNumber(raw)
	if err != nil {
		return nil, err
	}
	if field == nil {
		return exact, nil
	}
	switch constants.NormalizeFieldType(string(field.Type)) {
	case constants.FieldTypeBoolean:
		value, ok := new(big.Rat).SetString(exact.String())
		if !ok {
			return nil, errors.New("invalid exact boolean number")
		}
		if value.Sign() == 0 {
			return false, nil
		}
		if value.Cmp(big.NewRat(1, 1)) == 0 {
			return true, nil
		}
		return nil, errors.New("boolean number must be zero or one")
	case constants.FieldTypeInt, constants.SchemaFieldType("integer"), constants.SchemaFieldType("long"):
		value, ok := new(big.Rat).SetString(exact.String())
		if !ok || !value.IsInt() || !value.Num().IsInt64() {
			return nil, errors.New("integer is outside int64 or has a fractional component")
		}
		return value.Num().Int64(), nil
	case constants.FieldTypeDouble, constants.FieldTypeCurrency, constants.FieldTypePercent:
		return exact, nil
	default:
		// Preserve the exact source carrier. The schema validator owns the
		// eventual type mismatch; conversion must not make a JSON number look
		// like an accepted string field.
		return exact, nil
	}
}

func runtimeJSONContainerValue(raw string, field *models.FieldMetadata) (any, error) {
	if field != nil && constants.NormalizeFieldType(string(field.Type)) == constants.FieldTypeJson {
		return raw, nil
	}
	decoder := json.NewDecoder(strings.NewReader(raw))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		return nil, err
	}
	return value, nil
}

func runtimeExactJSONNumber(raw string) (schemautil.ExactDecimal, error) {
	negative := strings.HasPrefix(raw, "-")
	unsigned := strings.TrimPrefix(raw, "-")
	mantissa := unsigned
	exponent := int64(0)
	if index := strings.IndexAny(unsigned, "eE"); index >= 0 {
		mantissa = unsigned[:index]
		exponentText := unsigned[index+1:]
		var err error
		exponent, err = strconv.ParseInt(exponentText, 10, 64)
		if err != nil {
			// A zero coefficient never needs exponent expansion and remains
			// exactly zero even when the lexical exponent is enormous.
			if strings.Trim(strings.ReplaceAll(mantissa, ".", ""), "0") == "" {
				return schemautil.ExactDecimal("0"), nil
			}
			return "", err
		}
	}
	whole, fraction, _ := strings.Cut(mantissa, ".")
	digits := whole + fraction
	if strings.Trim(digits, "0") == "" {
		return schemautil.ExactDecimal("0"), nil
	}
	if exponent > semantic.MaxDecimalPrecision || exponent < -semantic.MaxDecimalScale-int64(len(digits)) {
		return "", errors.New("numeric exponent exceeds Runtime decimal storage envelope")
	}
	point := int64(len(whole)) + exponent
	if point > semantic.MaxDecimalPrecision || point < -semantic.MaxDecimalScale-int64(len(digits)) {
		return "", errors.New("numeric exponent exceeds Runtime decimal storage envelope")
	}
	var integerPart, fractionalPart string
	switch {
	case point <= 0:
		integerPart = "0"
		fractionalPart = strings.Repeat("0", int(-point)) + digits
	case point >= int64(len(digits)):
		integerPart = digits + strings.Repeat("0", int(point)-len(digits))
	default:
		integerPart = digits[:point]
		fractionalPart = digits[point:]
	}
	integerPart = strings.TrimLeft(integerPart, "0")
	if integerPart == "" {
		integerPart = "0"
	}
	integerDigits := len(integerPart)
	if integerPart == "0" {
		integerDigits = 0
	}
	if len(fractionalPart) > semantic.MaxDecimalScale || integerDigits+len(fractionalPart) > semantic.MaxDecimalPrecision {
		return "", errors.New("number exceeds Runtime decimal storage envelope")
	}
	value := integerPart
	if fractionalPart != "" {
		value += "." + fractionalPart
	}
	if negative {
		value = "-" + value
	}
	exact := schemautil.ExactDecimal(value)
	if _, _, ok := exact.CoefficientAndScale(); !ok {
		return "", errors.New("invalid exact JSON number")
	}
	return exact, nil
}

func runtimeRequiredCreateFields(schema *models.ObjectMetadata, fields models.Record) []string {
	missing := make([]string, 0)
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !schemautil.FieldRequiresCreateInput(field) {
			continue
		}
		if runtimeMutationValuePresent(fields[field.APIName]) {
			continue
		}
		if field.IsCompound {
			satisfied := false
			for _, component := range field.ComposedFrom {
				if runtimeMutationValuePresent(fields[component]) {
					satisfied = true
					break
				}
			}
			if satisfied {
				continue
			}
		}
		missing = append(missing, field.APIName)
	}
	sort.Strings(missing)
	return missing
}

func runtimeMutationValuePresent(value any) bool {
	if value == nil {
		return false
	}
	if text, ok := value.(string); ok {
		return strings.TrimSpace(text) != ""
	}
	return true
}

func runtimeExactDecimal(value semantic.Decimal) (schemautil.ExactDecimal, error) {
	coefficient := value.Coefficient
	negative := strings.HasPrefix(coefficient, "-")
	digits := strings.TrimPrefix(coefficient, "-")
	if value.Scale > uint32(len(digits)) {
		digits = strings.Repeat("0", int(value.Scale)-len(digits)+1) + digits
	}
	if value.Scale > 0 {
		position := len(digits) - int(value.Scale)
		digits = digits[:position] + "." + digits[position:]
	}
	if negative && digits != "0" {
		digits = "-" + digits
	}
	exact := schemautil.ExactDecimal(digits)
	if _, _, ok := exact.CoefficientAndScale(); !ok {
		return "", appErrors.NewValidationError("decimal", "invalid exact decimal")
	}
	return exact, nil
}

type runtimeMutationDomainFailure struct {
	result semantic.OperationResult
	cause  error
}

func (e *runtimeMutationDomainFailure) Error() string { return e.cause.Error() }
func (e *runtimeMutationDomainFailure) Unwrap() error { return e.cause }

type runtimeRequiredFieldsError struct{ fields []string }

func (e runtimeRequiredFieldsError) Error() string { return "required Catalog fields are missing" }
func (e runtimeRequiredFieldsError) RequiredFields() []string {
	return append([]string(nil), e.fields...)
}

type runtimeMutationPolicyError struct{ denial models.RecordMutationDenial }

func (e runtimeMutationPolicyError) Error() string { return "record mutation denied by Catalog policy" }

func classifyRuntimeMutationDomainError(operation semantic.Operation, index int, err error) (semantic.OperationResult, bool) {
	if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return semantic.OperationResult{}, false
	}
	var internal interface{ InternalCondition() bool }
	if errors.As(err, &internal) && internal.InternalCondition() {
		return semantic.OperationResult{}, false
	}
	if sharedErrors.IsRetriable(err) || sharedErrors.IsUnauthenticated(err) {
		return semantic.OperationResult{}, false
	}
	var domain *appErrors.DomainWireError
	if errors.As(err, &domain) && domain != nil {
		kind, ok := runtimeMutationSemanticDomainErrorKind(domain.Kind)
		if !ok {
			return semantic.OperationResult{}, false
		}
		result := failedRuntimeMutationResult(index, operation, operation.RecordID, kind, canonicalRuntimeMutationErrorFields(domain.Fields))
		if domain.CreateFromSourceGuard != nil {
			guard := runtimeMutationCreateFromSourceGuard(operation, domain)
			if guard == nil {
				return semantic.OperationResult{}, false
			}
			result.Errors[0].CreateFromSourceGuard = guard
		}
		return result, true
	}
	fields := []string{}
	var policy runtimeMutationPolicyError
	if errors.As(err, &policy) {
		if field := strings.TrimSpace(policy.denial.Field); field != "" {
			fields = []string{field}
		}
		result := failedRuntimeMutationResult(index, operation, operation.RecordID, semantic.ErrorPolicyDenied, fields)
		result.Errors[0].AuthoredMessage = policy.denial.Message
		return result, true
	}
	var required interface{ RequiredFields() []string }
	if errors.As(err, &required) {
		fields = required.RequiredFields()
		sort.Strings(fields)
		return failedRuntimeMutationResult(index, operation, operation.RecordID, semantic.ErrorRequiredFieldMissing, fields), true
	}
	var notFound interface{ NotFoundCondition() bool }
	if errors.As(err, &notFound) && notFound.NotFoundCondition() {
		return failedRuntimeMutationResult(index, operation, operation.RecordID, semantic.ErrorMissingOrDenied, nil), true
	}
	if validationErrors, ok := appErrors.AsClientValidationErrors(err); ok {
		result := semantic.OperationResult{
			Index: uint32(index), Kind: operation.Kind, RecordID: operation.RecordID, Success: false,
			Errors: make([]semantic.OperationError, 0, len(validationErrors)),
		}
		for _, validationErr := range validationErrors {
			if failures, ok := appErrors.AsValidationRuleFailures(validationErr); ok {
				for _, failure := range failures {
					result.Errors = append(result.Errors, semantic.OperationError{
						Kind:            semantic.ErrorCustomValidation,
						AuthoredMessage: failure.Message,
						Fields:          canonicalRuntimeMutationErrorFields([]string{failure.Field}),
					})
				}
				continue
			}
			if deniedFields, _, ok := sharedErrors.FieldWriteDeniedDetails(validationErr); ok {
				result.Errors = append(result.Errors, semantic.OperationError{
					Kind:   semantic.ErrorAccessDenied,
					Fields: canonicalRuntimeMutationErrorFields(deniedFields),
				})
				continue
			}
			if fields, _, ok := sharedErrors.InvalidFieldForWriteDetails(validationErr); ok {
				result.Errors = append(result.Errors, semantic.OperationError{
					Kind:   semantic.ErrorInvalidFieldForWrite,
					Fields: canonicalRuntimeMutationErrorFields(fields),
				})
				continue
			}
			if field, _, ok := sharedErrors.ValidationDetails(validationErr); ok {
				result.Errors = append(result.Errors, semantic.OperationError{
					Kind:   semantic.ErrorFieldIntegrity,
					Fields: canonicalRuntimeMutationErrorFields([]string{field}),
				})
			}
		}
		if len(result.Errors) == 0 {
			return semantic.OperationResult{}, false
		}
		return result, true
	}
	if sharedErrors.IsPermissionDenied(err) {
		return failedRuntimeMutationResult(index, operation, operation.RecordID, semantic.ErrorAccessDenied, nil), true
	}
	if sharedErrors.IsConflict(err) {
		// A business conflict rejects this operation. PROCESSING_HALTED is
		// reserved for siblings skipped by a preflight failure, and cannot
		// represent (or be journaled as) a single operation's domain outcome.
		return failedRuntimeMutationResult(index, operation, operation.RecordID, semantic.ErrorInvalidOperation, nil), true
	}
	return semantic.OperationResult{}, false
}

func runtimeMutationSemanticDomainErrorKind(kind appErrors.WireConditionKind) (semantic.ErrorKind, bool) {
	switch kind {
	case appErrors.WireInvalidReference:
		return semantic.ErrorInvalidReference, true
	case appErrors.WireFieldIntegrity:
		return semantic.ErrorFieldIntegrity, true
	case appErrors.WireInvalidOperation:
		return semantic.ErrorInvalidOperation, true
	case appErrors.WireInvalidField:
		return semantic.ErrorInvalidField, true
	case appErrors.WireNotFound:
		return semantic.ErrorMissingOrDenied, true
	default:
		return "", false
	}
}

func canonicalRuntimeMutationErrorFields(fields []string) []string {
	if len(fields) == 0 {
		return []string{}
	}
	seen := make(map[string]struct{}, len(fields))
	result := make([]string, 0, len(fields))
	for _, field := range fields {
		field = strings.TrimSpace(field)
		if !runtimeMutationAPIIdentifier(field, semantic.MaxFieldIdentifierBytes) {
			continue
		}
		if _, ok := seen[field]; ok {
			continue
		}
		seen[field] = struct{}{}
		result = append(result, field)
	}
	sort.Strings(result)
	return result
}

func runtimeMutationCreateFromSourceGuard(operation semantic.Operation, domain *appErrors.DomainWireError) *semantic.CreateFromSourceGuard {
	if operation.Kind != semantic.OperationCreate || domain == nil || domain.Kind != appErrors.WireInvalidOperation || domain.CreateFromSourceGuard == nil {
		return nil
	}
	objectID := strings.TrimSpace(domain.CreateFromSourceGuard.ObjectName)
	if objectID != operation.ObjectID || !runtimeMutationAPIIdentifier(objectID, semantic.MaxObjectIdentifierBytes) {
		return nil
	}
	actionIDs := canonicalRuntimeMutationErrorFields(domain.CreateFromSourceGuard.ActionNames)
	if len(actionIDs) == 0 {
		return nil
	}
	return &semantic.CreateFromSourceGuard{ObjectID: objectID, ActionIDs: actionIDs}
}

func runtimeMutationAPIIdentifier(value string, maximum int) bool {
	if len(value) == 0 || len(value) > maximum {
		return false
	}
	for i := 0; i < len(value); i++ {
		char := value[i]
		if (char >= 'A' && char <= 'Z') || (char >= 'a' && char <= 'z') || char == '_' || (i > 0 && char >= '0' && char <= '9') {
			continue
		}
		return false
	}
	return true
}

func runtimeMutationNotFound(err error) bool {
	var notFound interface{ NotFoundCondition() bool }
	if errors.As(err, &notFound) && notFound.NotFoundCondition() {
		return true
	}
	var status interface{ HTTPStatus() int }
	return errors.As(err, &status) && status.HTTPStatus() == http.StatusNotFound
}

func failedRuntimeMutationResult(index int, operation semantic.Operation, recordID string, kind semantic.ErrorKind, fields []string) semantic.OperationResult {
	if fields == nil {
		fields = []string{}
	}
	sort.Strings(fields)
	return semantic.OperationResult{Index: uint32(index), Kind: operation.Kind, RecordID: recordID, Success: false,
		Errors: []semantic.OperationError{{Kind: kind, Fields: fields}}}
}

func classifyRuntimeMutationTopLevelError(err error) error {
	if err == nil {
		return nil
	}
	var failure *semantic.Failure
	if errors.As(err, &failure) {
		return failure
	}
	if errors.Is(err, context.Canceled) {
		return &semantic.Failure{Category: semantic.FailureCanceled}
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return &semantic.Failure{Category: semantic.FailureDeadlineExceeded}
	}
	infrastructure := sharedErrors.ClassifyInfrastructure(err)
	if infrastructure == sharedErrors.InfrastructureConditionInternal {
		return &semantic.Failure{Category: semantic.FailureInternal}
	}
	if sharedErrors.IsUnauthenticated(err) {
		return &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	if sharedErrors.IsPermissionDenied(err) {
		return &semantic.Failure{Category: semantic.FailureForbidden}
	}
	if infrastructure == sharedErrors.InfrastructureConditionRetriable {
		return &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	return &semantic.Failure{Category: semantic.FailureInternal}
}

func (s *RuntimeRecordMutationService) response(command persistence.RuntimeRecordCommandRow, commandID string, receipts semantic.Receipts, results []semantic.OperationResult) semantic.Response {
	return semantic.Response{
		CommandID: commandID, CommandDigest: command.RequestDigest,
		Results: results, Receipts: receipts, OutcomeExpiresAtUTC: command.OutcomeExpiresAt.UTC().Format(time.RFC3339Nano),
	}
}

func (s *RuntimeRecordMutationService) nextOutcomeExpiry(ctx context.Context, command persistence.RuntimeRecordCommandRow) (time.Time, error) {
	// _System_RuntimeRecordCommand.OutcomeExpiresAt is DATETIME(6). The same
	// instant is embedded in the canonical terminal response and compared on
	// reconciliation. The duration is immutable command lineage, not current
	// process policy: a restarted process must not shorten an already accepted
	// command's duplicate-prevention proof. Derive the instant from the
	// transaction-bound database clock and normalize once to the storage
	// authority's precision.
	retention, err := runtimeRecordCommandOutcomeRetention(command)
	if err != nil {
		return time.Time{}, err
	}
	now, err := s.tx.ReadTransactionTime(ctx)
	if err != nil {
		return time.Time{}, err
	}
	return now.UTC().Add(retention).Truncate(time.Microsecond), nil
}

// requireRuntimeRecordCommandExecutionWindow is called only after the command
// row is locked by the transaction that is about to execute business DML. The
// lifecycle worker takes the same lock, so exactly one side wins the boundary:
// execution admitted before expiry may commit, while a waiter that acquires the
// row at or after expiry performs no business effect and lets terminal closure
// preserve any already-completed independent positions.
func (s *RuntimeRecordMutationService) requireRuntimeRecordCommandExecutionWindow(ctx context.Context, command persistence.RuntimeRecordCommandRow) error {
	if command.State != persistence.RuntimeRecordCommandAccepted || command.OutcomeExpiresAt.IsZero() {
		return errors.New("runtime record command has invalid execution window")
	}
	now, err := s.tx.ReadTransactionTime(ctx)
	if err != nil {
		return err
	}
	if !command.OutcomeExpiresAt.After(now) {
		return errRuntimeRecordCommandExpired
	}
	return nil
}

func runtimeRecordCommandOutcomeRetention(command persistence.RuntimeRecordCommandRow) (time.Duration, error) {
	micros := command.OutcomeRetentionMicros
	retention := time.Duration(micros) * time.Microsecond
	if micros <= 0 || retention <= 0 || retention/time.Microsecond != time.Duration(micros) {
		return 0, errors.New("runtime record command has invalid retention lineage")
	}
	return retention, nil
}

func decodeRuntimeMutationResponse(request semantic.Request, payload []byte) (semantic.Response, error) {
	var response semantic.Response
	if len(payload) == 0 || json.Unmarshal(payload, &response) != nil || semantic.ValidateResponse(request, response) != nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	return response, nil
}

func validateRuntimeMutationResponseForPersistence(request semantic.Request, response semantic.Response) error {
	if err := semantic.ValidateResponse(request, response); err != nil {
		return errors.New("runtime record command terminal response is invalid")
	}
	return nil
}

func validRuntimeCommandScopeDigest(value string) bool {
	if len(value) != sha256.Size*2 {
		return false
	}
	decoded, err := hex.DecodeString(value)
	return err == nil && len(decoded) == sha256.Size && value == strings.ToLower(value)
}