package records

import (
	"context"
	"encoding/json"
	"errors"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	kernelValidation "github.com/nexuscrm/kernel/pkg/validation"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

const testRuntimeMutationScope = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"

func TestRuntimeRecordMutationTypedValuesAndReplay(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})

	empty := ""
	falseValue := false
	zero := int64(0)
	decimal := semantic.Decimal{Coefficient: "123456789012345678901200", Scale: 4}
	request := semantic.Request{
		CommandID: "typed-values-1", BatchMode: semantic.BatchIndependent,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "Invoice", Fields: []semantic.FieldValue{
			{Name: "Description", Kind: semantic.ValueKindString, String: &empty},
			{Name: "Active", Kind: semantic.ValueKindBoolean, Boolean: &falseValue},
			{Name: "Count", Kind: semantic.ValueKindInteger, Integer: &zero},
			{Name: "Amount", Kind: semantic.ValueKindDecimal, Decimal: &decimal},
			{Name: "ClosedAt", Kind: semantic.ValueKindNull, Null: true},
		}}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	first, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	if len(first.Results) != 1 || !first.Results[0].Success || first.Results[0].RecordID == "" {
		t.Fatalf("Mutate() result = %#v", first)
	}
	got := engine.lastInsert
	if got["Description"] != "" || got["Active"] != false || got["Count"] != int64(0) || got["ClosedAt"] != nil {
		t.Fatalf("typed values = %#v", got)
	}
	if exact, ok := got["Amount"].(schemautil.ExactDecimal); !ok || exact.String() != "12345678901234567890.1200" {
		t.Fatalf("Amount = %#v, want exact decimal", got["Amount"])
	}

	replayed, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("replay Mutate() error = %v", err)
	}
	if !reflect.DeepEqual(replayed, first) || engine.insertCalls != 1 {
		t.Fatalf("replay = %#v, first = %#v, insert calls = %d", replayed, first, engine.insertCalls)
	}
}

func TestRuntimeRecordMutationBusinessTransactionsReceivePreparedObjectSet(t *testing.T) {
	tests := []struct {
		name string
		mode semantic.BatchMode
		want [][]string
	}{
		{name: "independent", mode: semantic.BatchIndependent, want: [][]string{{constants.TableUser}, {"Account"}}},
		{name: "atomic", mode: semantic.BatchAtomic, want: [][]string{{constants.TableUser, "Account"}}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			tx := &runtimeMutationTxStub{}
			service := newRuntimeMutationTestService(t, &runtimeMutationEngineStub{}, store, tx)
			request := semantic.Request{
				CommandID: "object-set-" + tt.name,
				BatchMode: tt.mode,
				UseCase:   semantic.MutationUseCaseCollection,
				Operations: []semantic.Operation{
					{Kind: semantic.OperationCreate, ObjectID: constants.TableUser},
					{Kind: semantic.OperationCreate, ObjectID: "Account"},
				},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			if _, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"}); err != nil {
				t.Fatal(err)
			}
			tx.mu.Lock()
			got := append([][]string(nil), tx.mutationObjectSets...)
			tx.mu.Unlock()
			if !reflect.DeepEqual(got, tt.want) {
				t.Fatalf("record mutation transaction object sets = %#v, want %#v", got, tt.want)
			}
		})
	}
}

func TestRuntimeRecordMutationOwnsSingleWriteActivityWithoutReplayDuplication(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	activity := &runtimeMutationActivityStub{}
	service := newRuntimeMutationTestServiceWithActivity(t, engine, runtimeMutationMetadataStub{}, activity, store, &runtimeMutationTxStub{})
	name := "Created through Runtime"
	request := semantic.Request{
		CommandID: "single-activity", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	first, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"}); err != nil {
		t.Fatal(err)
	}
	activity.mu.Lock()
	defer activity.mu.Unlock()
	want := [][3]string{{"user-1", "WorkItem", first.Results[0].RecordID}}
	if !reflect.DeepEqual(activity.calls, want) {
		t.Fatalf("activity calls = %#v, want %#v", activity.calls, want)
	}
}

func TestRuntimeRecordMutationInterpretsLosslessSourceValuesFromBoundMetadata(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{
		"invoice": {
			APIName: "Invoice", Fields: []models.FieldMetadata{
				{APIName: "Amount", Type: constants.FieldTypeCurrency, Createable: true},
				{APIName: "Count", Type: models.FieldType("integer"), Createable: true},
				{APIName: "Active", Type: constants.FieldTypeBoolean, Createable: true},
				{APIName: "OccurredOn", Type: constants.FieldTypeDate, Createable: true},
				{APIName: "OwnerRef", Type: constants.FieldTypeReference, Createable: true},
				{APIName: "Payload", Type: constants.FieldTypeJson, Createable: true},
				{APIName: "Tags", Type: constants.FieldTypeMultiPicklist, Createable: true},
			},
		},
	}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	amount, count, boolean := "1.2500e2", "1e2", "1.0"
	date, reference := "2026-07-19", "owner-1"
	payload, tags := `{"z":2,"a":[0,false]}`, `["A","B"]`
	request := semantic.Request{
		CommandID: "source-values", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "invoice", Fields: []semantic.FieldValue{
			{Name: "amount", Kind: semantic.ValueKindNumber, Number: &amount},
			{Name: "Count", Kind: semantic.ValueKindNumber, Number: &count},
			{Name: "ACTIVE", Kind: semantic.ValueKindNumber, Number: &boolean},
			{Name: "OccurredOn", Kind: semantic.ValueKindString, String: &date},
			{Name: "OwnerRef", Kind: semantic.ValueKindString, String: &reference},
			{Name: "Payload", Kind: semantic.ValueKindJSON, JSON: &payload},
			{Name: "Tags", Kind: semantic.ValueKindJSON, JSON: &tags},
		}}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || !response.Results[0].Success {
		t.Fatalf("Mutate() = %#v, %v", response, err)
	}
	if exact, ok := engine.lastInsert["Amount"].(schemautil.ExactDecimal); !ok || exact.String() != "125.00" {
		t.Fatalf("Amount = %#v, want exact 125.00", engine.lastInsert["Amount"])
	}
	if engine.lastInsert["Count"] != int64(100) || engine.lastInsert["Active"] != true {
		t.Fatalf("integer/boolean values = %#v", engine.lastInsert)
	}
	if want := `{"a":[0,false],"z":2}`; engine.lastInsert["Payload"] != want {
		t.Fatalf("Payload = %#v, want canonical JSON text", engine.lastInsert["Payload"])
	}
	if got, ok := engine.lastInsert["Tags"].([]any); !ok || !reflect.DeepEqual(got, []any{"A", "B"}) {
		t.Fatalf("Tags = %#v, want decoded array", engine.lastInsert["Tags"])
	}
	if _, leaked := engine.lastInsert["amount"]; leaked {
		t.Fatalf("caller field casing survived Catalog canonicalization: %#v", engine.lastInsert)
	}
}

func TestRuntimeRecordMutationPolicyIsCommandPreflight(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	deniedSchema := &models.ObjectMetadata{
		APIName: "ProtectedItem", AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceCollectionCreate}, Field: "State",
			Message: "Collection creation is closed.", ErrorCode: "INVALID_REQUEST",
		}}}},
		Fields: []models.FieldMetadata{{APIName: "State", Type: constants.FieldTypeString, Createable: true}},
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{
		"ordinaryitem":  {APIName: "OrdinaryItem"},
		"protecteditem": deniedSchema,
	}}
	tx := &runtimeMutationTxStub{}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, tx)
	state := "Active"
	request := semantic.Request{
		CommandID: "policy-preflight", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{
			{Kind: semantic.OperationCreate, ObjectID: "OrdinaryItem"},
			{Kind: semantic.OperationCreate, ObjectID: "ProtectedItem", Fields: []semantic.FieldValue{{Name: "State", Kind: semantic.ValueKindString, String: &state}}},
		},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatal(err)
	}
	if engine.insertCalls != 0 || tx.calls != 2 {
		t.Fatalf("insert calls = %d, transactions = %d; want acceptance plus one locked preflight terminalization", engine.insertCalls, tx.calls)
	}
	if len(response.Results) != 2 || response.Results[0].Errors[0].Kind != semantic.ErrorProcessingHalted ||
		response.Results[1].Errors[0].Kind != semantic.ErrorPolicyDenied ||
		response.Results[1].Errors[0].AuthoredMessage != "Collection creation is closed." {
		t.Fatalf("policy preflight results = %#v", response.Results)
	}
}

func TestRuntimeRecordMutationPreservesAllValidationRuleFailures(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	validationErr := appErrors.NewValidationRuleFailures([]appErrors.ValidationRuleFailure{
		{RuleName: "IQM_Existing_Database", Message: "Existing Database is required", Field: "Existing_Database__c"},
		{RuleName: "IQM_Completed_Date", Message: "IQM Completed Date is required", Field: "IQM_Completed_Date__c"},
	})
	engine := &runtimeMutationEngineStub{insertErrors: []error{validationErr}}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	request := semantic.Request{
		CommandID: "validation-rule-failures", BatchMode: semantic.BatchIndependent,
		UseCase:    semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "Lead"}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	want := []semantic.OperationError{
		{Kind: semantic.ErrorCustomValidation, AuthoredMessage: "Existing Database is required", Fields: []string{"Existing_Database__c"}},
		{Kind: semantic.ErrorCustomValidation, AuthoredMessage: "IQM Completed Date is required", Fields: []string{"IQM_Completed_Date__c"}},
	}
	if len(response.Results) != 1 || !reflect.DeepEqual(response.Results[0].Errors, want) {
		t.Fatalf("Mutate() errors = %#v, want %#v", response.Results, want)
	}
}

func TestRuntimeRecordMutationPreservesStaticAndRuleValidationFailures(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	validationErr := appErrors.NewClientValidationErrors(
		appErrors.NewValidationError("LastName", "is required"),
		appErrors.NewValidationError("Company", "is required"),
		appErrors.NewValidationRuleFailures([]appErrors.ValidationRuleFailure{
			{RuleName: "Industry_Required", Message: "Industry is required", Field: "Industry"},
		}),
	)
	engine := &runtimeMutationEngineStub{insertErrors: []error{validationErr}}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	request := semantic.Request{
		CommandID: "mixed-validation-failures", BatchMode: semantic.BatchIndependent,
		UseCase:    semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "Lead"}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	want := []semantic.OperationError{
		{Kind: semantic.ErrorFieldIntegrity, Fields: []string{"LastName"}},
		{Kind: semantic.ErrorFieldIntegrity, Fields: []string{"Company"}},
		{Kind: semantic.ErrorCustomValidation, AuthoredMessage: "Industry is required", Fields: []string{"Industry"}},
	}
	if len(response.Results) != 1 || !reflect.DeepEqual(response.Results[0].Errors, want) {
		t.Fatalf("Mutate() errors = %#v, want %#v", response.Results, want)
	}
}

func TestRuntimeRecordMutationPreservesNeutralValidationContracts(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	validationErr := errors.Join(
		neutralProjectionValidationError{field: "Name", message: "Name is required"},
		neutralProjectionWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable"},
	)
	engine := &runtimeMutationEngineStub{insertErrors: []error{validationErr}}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	request := semantic.Request{
		CommandID: "neutral-validation-failures", BatchMode: semantic.BatchIndependent,
		UseCase:    semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "Lead"}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	want := []semantic.OperationError{
		{Kind: semantic.ErrorFieldIntegrity, Fields: []string{"Name"}},
		{Kind: semantic.ErrorAccessDenied, Fields: []string{"Secret__c"}},
	}
	if len(response.Results) != 1 || !reflect.DeepEqual(response.Results[0].Errors, want) {
		t.Fatalf("Mutate() errors = %#v, want %#v", response.Results, want)
	}
}

func TestRuntimeRecordMutationProjectsURLValueValidationAsFieldIntegrity(t *testing.T) {
	schema := &models.ObjectMetadata{
		APIName: "Document",
		Fields: []models.FieldMetadata{{
			APIName: "DownloadURL", Type: constants.FieldTypeUrl, Updateable: true,
		}},
	}
	validator := metadataFeature.NewValidationService(nil)
	engine := &runtimeMutationEngineStub{
		records: map[string]models.SObject{"document-1": {constants.FieldID: "document-1"}},
		updateValidate: func(updates models.SObject) error {
			return validator.ValidateRecordStaticConstraints(updates, schema)
		},
	}
	mutate := func(commandID, value string) semantic.Response {
		t.Helper()
		ctx, cancel := context.WithTimeout(t.Context(), time.Second)
		defer cancel()
		service := newRuntimeMutationTestServiceWithMetadata(
			t, engine,
			runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"document": schema}},
			newRuntimeMutationMemoryStore(), &runtimeMutationTxStub{},
		)
		response, err := service.Mutate(
			ctx, testRuntimeMutationScope,
			semantic.Request{
				CommandID: commandID, BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
				Operations: []semantic.Operation{{
					Kind: semantic.OperationUpdate, ObjectID: "Document", RecordID: "document-1",
					Fields: []semantic.FieldValue{{Name: "DownloadURL", Kind: semantic.ValueKindString, String: &value}},
				}},
			},
			runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"},
		)
		if err != nil {
			t.Fatalf("Mutate(%q): %v", value, err)
		}
		return response
	}

	const supported = "https://example.invalid/services/data/v63.0/files/file-1/download"
	response := mutate("url-supported", supported)
	if len(response.Results) != 1 || !response.Results[0].Success || engine.updateCalls != 1 {
		t.Fatalf("supported URL response = %#v, update calls = %d", response, engine.updateCalls)
	}
	stored, err := engine.ReadPolicyTarget(context.Background(), "Document", "document-1", semantic.OperationUpdate, nil)
	if err != nil || stored["DownloadURL"] != supported {
		t.Fatalf("stored URL = %#v, %v; want %q", stored["DownloadURL"], err, supported)
	}

	response = mutate("malformed-url-rejected", "not a URL")
	if len(response.Results) != 1 || response.Results[0].Success || len(response.Results[0].Errors) != 1 ||
		response.Results[0].Errors[0].Kind != semantic.ErrorFieldIntegrity || engine.updateCalls != 1 {
		t.Fatalf("unsafe network URL response = %#v, update calls = %d", response, engine.updateCalls)
	}
}

func TestRuntimeRecordMutationProjectsUnknownFieldAsInvalidFieldForWrite(t *testing.T) {
	schema := &models.ObjectMetadata{
		APIName: "Document",
		Fields: []models.FieldMetadata{{
			APIName: "DownloadURL", Type: constants.FieldTypeUrl, Updateable: true,
		}},
	}
	validator := kernelValidation.NewSchemaValidator()
	engine := &runtimeMutationEngineStub{
		records: map[string]models.SObject{"document-1": {constants.FieldID: "document-1"}},
		updateValidate: func(updates models.SObject) error {
			return validator.ValidateSchemaCompliance(updates, schema, true, false, false)
		},
	}
	service := newRuntimeMutationTestServiceWithMetadata(
		t, engine,
		runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"document": schema}},
		newRuntimeMutationMemoryStore(), &runtimeMutationTxStub{},
	)
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	value := "must-not-write"
	response, err := service.Mutate(
		ctx, testRuntimeMutationScope,
		semantic.Request{
			CommandID: "unknown-field-rejected", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
			Operations: []semantic.Operation{{
				Kind: semantic.OperationUpdate, ObjectID: "Document", RecordID: "document-1",
				Fields: []semantic.FieldValue{{Name: "NeverAField__c", Kind: semantic.ValueKindString, String: &value}},
			}},
		},
		runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"},
	)
	if err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	want := []semantic.OperationError{{Kind: semantic.ErrorInvalidFieldForWrite, Fields: []string{"NeverAField__c"}}}
	if len(response.Results) != 1 || response.Results[0].Success || !reflect.DeepEqual(response.Results[0].Errors, want) || engine.updateCalls != 0 {
		t.Fatalf("unknown field response = %#v, update calls = %d; want errors %#v", response, engine.updateCalls, want)
	}
}

func TestRuntimeRecordMutationFreshSuccessDoesNotReopenTerminalOutcome(t *testing.T) {
	for _, test := range []struct {
		mode             semantic.BatchMode
		wantTransactions int
	}{
		{mode: semantic.BatchIndependent, wantTransactions: 2},
		{mode: semantic.BatchAtomic, wantTransactions: 3},
	} {
		t.Run(string(test.mode), func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			tx := &runtimeMutationTxStub{}
			service := newRuntimeMutationTestService(t, &runtimeMutationEngineStub{}, store, tx)
			request := semantic.Request{
				CommandID: "terminal-in-effect-" + string(test.mode), BatchMode: test.mode,
				UseCase:    semantic.MutationUseCaseSingle,
				Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem"}},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()

			response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
			if err != nil || len(response.Results) != 1 || !response.Results[0].Success {
				t.Fatalf("Mutate() = %#v, %v", response, err)
			}
			if tx.calls != test.wantTransactions {
				t.Fatalf("transactions = %d, want %d", tx.calls, test.wantTransactions)
			}
			if store.completeOperationCalls != 1 || store.completeCommandCalls != 1 || store.row.State != persistence.RuntimeRecordCommandCompleted {
				t.Fatalf("terminal writes operations=%d command=%d row=%#v", store.completeOperationCalls, store.completeCommandCalls, store.row)
			}
			if store.loadCalls != 0 {
				t.Fatalf("post-commit command reloads = %d, want zero", store.loadCalls)
			}
		})
	}
}

func TestRuntimeRecordMutationAmbiguousBusinessCommitIsReconciledFromTerminalOutcome(t *testing.T) {
	for _, test := range []struct {
		mode                semantic.BatchMode
		businessTransaction int
	}{
		{mode: semantic.BatchIndependent, businessTransaction: 2},
		{mode: semantic.BatchAtomic, businessTransaction: 3},
	} {
		t.Run(string(test.mode), func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			// Model a database COMMIT that applied business+terminal state but
			// lost its acknowledgement. Fresh independent single-record commands
			// have no separate locked preparation transaction.
			tx := &runtimeMutationTxStub{commitErrorOnCall: test.businessTransaction}
			service := newRuntimeMutationTestService(t, &runtimeMutationEngineStub{}, store, tx)
			request := semantic.Request{
				CommandID: "ambiguous-terminal-" + string(test.mode), BatchMode: test.mode,
				UseCase:    semantic.MutationUseCaseSingle,
				Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem"}},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()

			if response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"}); err == nil {
				t.Fatalf("Mutate() = %#v, nil; want ambiguous transport failure", response)
			}
			statusRequest, err := semantic.StatusRequestFor(request)
			if err != nil {
				t.Fatal(err)
			}
			_, status, err := service.Status(ctx, testRuntimeMutationScope, statusRequest)
			if err != nil || status.State != semantic.CommandCompleted || len(status.Results) != 1 || !status.Results[0].Success {
				t.Fatalf("Status() = %#v, %v; want committed terminal reconciliation", status, err)
			}
		})
	}
}

func TestRuntimeRecordMutationFreshSingleUpdatePreparesWithoutTargetRead(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{records: map[string]models.SObject{
		"work-1": {constants.FieldID: "work-1", "Description": "before"},
	}}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
		APIName: "WorkItem", Fields: []models.FieldMetadata{{APIName: "Description", Type: constants.FieldTypeString, Updateable: true}},
	}}}
	tx := &runtimeMutationTxStub{}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, tx)
	description := "after"
	request := semantic.Request{
		CommandID: "fresh-update-static-prepare", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationUpdate, ObjectID: "WorkItem", RecordID: "work-1",
			Fields: []semantic.FieldValue{{Name: "Description", Kind: semantic.ValueKindString, String: &description}}}},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || !response.Results[0].Success || tx.calls != 2 ||
		engine.retrieveCalls != 0 || engine.updateCalls != 1 || engine.updateAuthorizationCalls != 1 ||
		!reflect.DeepEqual(engine.authorizationInTx, []bool{true}) {
		t.Fatalf("fresh update = %#v, %v; transactions=%d retrieve=%d update=%d authorize=%d authorizeInTx=%v",
			response, err, tx.calls, engine.retrieveCalls, engine.updateCalls, engine.updateAuthorizationCalls, engine.authorizationInTx)
	}
}

func TestRuntimeRecordMutationAcceptedReplayKeepsLockedPreparation(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	tx := &runtimeMutationTxStub{}
	service := newRuntimeMutationTestService(t, engine, store, tx)
	request := semantic.Request{
		CommandID: "accepted-single-replay", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem"}},
	}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	receipts := runtimeMutationTestReceipts(7, 11)
	receiptsPayload, err := json.Marshal(receipts)
	if err != nil {
		t.Fatal(err)
	}
	store.row = persistence.RuntimeRecordCommandRow{
		RowID: "accepted-single-replay-row", ScopeDigest: testRuntimeMutationScope, CommandID: request.CommandID,
		RequestDigest: digest, BatchMode: string(request.BatchMode), OperationCount: 1,
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: receiptsPayload,
		OutcomeExpiresAt: store.now.Add(time.Hour), OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	}
	store.slots = []persistence.RuntimeRecordCommandOperationRow{{
		RowID: "accepted-single-replay-slot", CommandRowID: store.row.RowID, Index: 0,
		Kind: string(semantic.OperationCreate), ObjectID: "WorkItem", BoundRecordID: "reserved-work-1",
		State: persistence.RuntimeRecordOperationPending,
	}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, receipts, &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || !response.Results[0].Success ||
		response.Results[0].RecordID != "reserved-work-1" || tx.calls != 3 || engine.insertCalls != 1 {
		t.Fatalf("accepted replay = %#v, %v; transactions=%d inserts=%d", response, err, tx.calls, engine.insertCalls)
	}
}

func TestRuntimeRecordMutationMasksMissingTargetBeforeAuthoredPolicy(t *testing.T) {
	for _, kind := range []semantic.OperationKind{semantic.OperationUpdate, semantic.OperationDelete} {
		t.Run(string(kind), func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			engine := &runtimeMutationEngineStub{
				retrieveErrors: []error{appErrors.NewNotFoundError("Record", "")},
			}
			surface := models.RecordMutationSurfaceSingleUpdate
			if kind == semantic.OperationDelete {
				surface = models.RecordMutationSurfaceSingleDelete
			}
			metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
				APIName: "WorkItem",
				Fields: []models.FieldMetadata{
					{APIName: "Name", Type: constants.FieldTypeString, Updateable: true},
					{APIName: "State", Type: constants.FieldTypeString, Updateable: true},
				},
				AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
					Surfaces: []string{surface}, Message: "This authored policy must not disclose a missing record.",
					When: &models.RecordMutationCondition{FieldEquals: &models.RecordMutationFieldEqualsCondition{Field: "State", Value: "Locked"}},
				}}}},
			}}}
			service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
			operation := semantic.Operation{Kind: kind, ObjectID: "WorkItem", RecordID: "a00-missing"}
			if kind == semantic.OperationUpdate {
				name := "changed"
				operation.Fields = []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}
			}
			request := semantic.Request{
				CommandID: "missing-before-policy-" + string(kind), BatchMode: semantic.BatchIndependent,
				UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{operation},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
			if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 ||
				response.Results[0].Errors[0].Kind != semantic.ErrorMissingOrDenied ||
				response.Results[0].Errors[0].AuthoredMessage != "" ||
				engine.retrieveCalls != 1 || engine.updateCalls != 0 || engine.deleteCalls != 0 ||
				engine.opaqueResolutionCalls != 0 {
				t.Fatalf("missing target = %#v, %v; retrieve=%d update=%d delete=%d opaque=%d",
					response, err, engine.retrieveCalls, engine.updateCalls, engine.deleteCalls, engine.opaqueResolutionCalls)
			}
		})
	}
}

func TestRuntimeRecordMutationReturnsUnconditionalObjectGuardBeforeMissingTarget(t *testing.T) {
	for _, kind := range []semantic.OperationKind{semantic.OperationUpdate, semantic.OperationDelete} {
		t.Run(string(kind), func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			engine := &runtimeMutationEngineStub{
				retrieveErrors: []error{appErrors.NewNotFoundError("Record", "")},
			}
			surface := models.RecordMutationSurfaceSingleUpdate
			if kind == semantic.OperationDelete {
				surface = models.RecordMutationSurfaceSingleDelete
			}
			metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"processitem": {
				APIName: "ProcessItem",
				Fields:  []models.FieldMetadata{{APIName: "Name", Type: constants.FieldTypeString, Updateable: true}},
				AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
					Surfaces: []string{surface}, Message: "Public writes are closed.",
				}}}},
			}}}
			tx := &runtimeMutationTxStub{}
			service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, tx)
			operation := semantic.Operation{Kind: kind, ObjectID: "ProcessItem", RecordID: "a00-missing"}
			if kind == semantic.OperationUpdate {
				name := "changed"
				operation.Fields = []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}
			}
			request := semantic.Request{
				CommandID: "unconditional-policy-" + string(kind), BatchMode: semantic.BatchIndependent,
				UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{operation},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			response, err := service.Mutate(
				ctx, testRuntimeMutationScope, request,
				runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"},
			)
			if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 ||
				response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied ||
				response.Results[0].Errors[0].AuthoredMessage != "Public writes are closed." ||
				engine.retrieveCalls != 0 || tx.calls != 2 {
				t.Fatalf("unconditional guard = %#v, %v; retrieve=%d transactions=%d", response, err, engine.retrieveCalls, tx.calls)
			}
		})
	}
}

func TestRuntimeRecordMutationTargetAccessFailureDoesNotHaltIndependentSibling(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{
		retrieveErrors: []error{appErrors.NewPermissionError("edit", "WorkItem")},
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{
		"workitem": {
			APIName: "WorkItem",
			Fields:  []models.FieldMetadata{{APIName: "Name", Type: constants.FieldTypeString, Createable: true, Updateable: true}},
		},
	}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	name := "changed"
	request := semantic.Request{
		CommandID: "target-access-independent", BatchMode: semantic.BatchIndependent,
		UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{
			{Kind: semantic.OperationUpdate, ObjectID: "WorkItem", RecordID: "a00-denied", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}},
			{Kind: semantic.OperationCreate, ObjectID: "WorkItem", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}},
		},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 2 ||
		response.Results[0].Success || response.Results[0].Errors[0].Kind != semantic.ErrorAccessDenied ||
		!response.Results[1].Success || engine.updateCalls != 0 || engine.insertCalls != 1 {
		t.Fatalf("independent target access = %#v, %v; update=%d insert=%d", response, err, engine.updateCalls, engine.insertCalls)
	}
}

func TestRuntimeRecordMutationReplaySkipsCommandPreflightAndPreservesCommittedIndependentResults(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	deniedSchema := &models.ObjectMetadata{
		APIName: "ProtectedItem", AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceCollectionCreate}, Field: "State",
			Message: "Collection creation is now closed.", ErrorCode: "INVALID_REQUEST",
		}}}},
		Fields: []models.FieldMetadata{{APIName: "State", Type: constants.FieldTypeString, Createable: true}},
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{
		"ordinaryitem":  {APIName: "OrdinaryItem"},
		"protecteditem": deniedSchema,
	}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	state := "Active"
	request := semantic.Request{
		CommandID: "partial-policy-replay", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{
			{Kind: semantic.OperationCreate, ObjectID: "OrdinaryItem"},
			{Kind: semantic.OperationCreate, ObjectID: "ProtectedItem", Fields: []semantic.FieldValue{{Name: "State", Kind: semantic.ValueKindString, String: &state}}},
			{Kind: semantic.OperationUpdate, ObjectID: "OrdinaryItem", RecordID: "ordinary-2"},
		},
	}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	receipts := runtimeMutationTestReceipts(7, 11)
	receiptsPayload, _ := json.Marshal(receipts)
	committed := semantic.OperationResult{Index: 0, Kind: semantic.OperationCreate, RecordID: "ordinary-1", Success: true, Created: boolPointer(true), Errors: []semantic.OperationError{}}
	committedPayload, _ := json.Marshal(committed)
	store.row = persistence.RuntimeRecordCommandRow{
		RowID: "row-partial-policy", ScopeDigest: testRuntimeMutationScope, CommandID: request.CommandID,
		RequestDigest: digest, BatchMode: string(request.BatchMode), OperationCount: 3,
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: receiptsPayload,
		OutcomeExpiresAt: store.now.Add(time.Hour), OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	}
	store.slots = []persistence.RuntimeRecordCommandOperationRow{
		{RowID: "slot-0", CommandRowID: store.row.RowID, Index: 0, Kind: string(semantic.OperationCreate), ObjectID: "OrdinaryItem", BoundRecordID: "ordinary-1", State: persistence.RuntimeRecordOperationDone, ResultPayload: committedPayload},
		{RowID: "slot-1", CommandRowID: store.row.RowID, Index: 1, Kind: string(semantic.OperationCreate), ObjectID: "ProtectedItem", BoundRecordID: "protected-1", State: persistence.RuntimeRecordOperationPending},
		{RowID: "slot-2", CommandRowID: store.row.RowID, Index: 2, Kind: string(semantic.OperationUpdate), ObjectID: "OrdinaryItem", BoundRecordID: "ordinary-2", State: persistence.RuntimeRecordOperationPending},
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, receipts, &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("Mutate() error = %#v; response=%#v; slots=%#v command=%#v", err, response, store.slots, store.row)
	}
	if len(response.Results) != 3 || !reflect.DeepEqual(response.Results[0], committed) ||
		response.Results[1].Success || response.Results[1].Errors[0].Kind != semantic.ErrorPolicyDenied ||
		!response.Results[2].Success {
		t.Fatalf("partial policy replay = %#v", response.Results)
	}
	if engine.insertCalls != 0 || engine.updateCalls != 1 || store.completeOperationCalls != 2 || store.completeCommandCalls != 1 ||
		store.row.State != persistence.RuntimeRecordCommandCompleted {
		t.Fatalf("effects insert=%d update=%d slot completions=%d command completions=%d command=%#v", engine.insertCalls, engine.updateCalls, store.completeOperationCalls, store.completeCommandCalls, store.row)
	}
}

func TestRuntimeRecordMutationRejectsCorruptDurableSlotBeforeTerminalWrites(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "corrupt-replay", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{
		{Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "issue-1"},
		{Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "issue-2"},
	}}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	receipts := runtimeMutationTestReceipts(7, 11)
	receiptsPayload, _ := json.Marshal(receipts)
	command := persistence.RuntimeRecordCommandRow{
		RowID: "row-corrupt", ScopeDigest: testRuntimeMutationScope, CommandID: request.CommandID,
		RequestDigest: digest, BatchMode: string(request.BatchMode), OperationCount: 2,
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: receiptsPayload, OutcomeExpiresAt: store.now.Add(time.Hour),
		OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	}
	// The JSON is syntactically valid but claims the wrong position. Terminal
	// response validation, not decoding alone, must reject it before any durable
	// PENDING -> DONE or ACCEPTED -> COMPLETED transition.
	corruptPayload, _ := json.Marshal(semantic.OperationResult{
		Index: 1, Kind: semantic.OperationDelete, RecordID: "issue-1", Success: false,
		Errors: []semantic.OperationError{{Kind: semantic.ErrorInternal, Fields: []string{}}},
	})
	store.row = command
	store.slots = []persistence.RuntimeRecordCommandOperationRow{
		{RowID: "slot-corrupt", CommandRowID: command.RowID, Index: 0, Kind: string(semantic.OperationDelete), ObjectID: "Issue", BoundRecordID: "issue-1", State: persistence.RuntimeRecordOperationDone, ResultPayload: corruptPayload},
		{RowID: "slot-pending", CommandRowID: command.RowID, Index: 1, Kind: string(semantic.OperationDelete), ObjectID: "Issue", BoundRecordID: "issue-2", State: persistence.RuntimeRecordOperationPending},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = service.Mutate(ctx, testRuntimeMutationScope, request, receipts, &models.UserSession{ID: "user-1"})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal ||
		store.completeOperationCalls != 0 || store.completeCommandCalls != 0 ||
		store.slots[1].State != persistence.RuntimeRecordOperationPending || store.row.State != persistence.RuntimeRecordCommandAccepted ||
		engine.updateCalls != 0 || engine.deleteCalls != 0 {
		t.Fatalf("corrupt replay err=%v effects=%d/%d operation completions=%d command completions=%d slots=%#v command=%#v", err, engine.updateCalls, engine.deleteCalls, store.completeOperationCalls, store.completeCommandCalls, store.slots, store.row)
	}
}

func TestRuntimeRecordMutationPersistsObjectNotFoundAsCommandPreflight(t *testing.T) {
	for _, mode := range []semantic.BatchMode{semantic.BatchIndependent, semantic.BatchAtomic} {
		t.Run(string(mode), func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			engine := &runtimeMutationEngineStub{}
			metadata := runtimeMutationMetadataStub{
				schemas: map[string]*models.ObjectMetadata{"knownitem": {APIName: "KnownItem"}},
				missing: map[string]bool{"unknownitem": true},
			}
			service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
			request := semantic.Request{
				CommandID: "object-preflight-" + strings.ToLower(string(mode)), BatchMode: mode, UseCase: semantic.MutationUseCaseCollection,
				Operations: []semantic.Operation{
					{Kind: semantic.OperationCreate, ObjectID: "KnownItem"},
					{Kind: semantic.OperationCreate, ObjectID: "UnknownItem"},
				},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			first, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
			if err != nil {
				t.Fatal(err)
			}
			if engine.insertCalls != 0 || len(first.Results) != 2 ||
				first.Results[0].Errors[0].Kind != semantic.ErrorProcessingHalted ||
				first.Results[1].Errors[0].Kind != semantic.ErrorObjectNotFound ||
				store.row.State != persistence.RuntimeRecordCommandCompleted {
				t.Fatalf("object preflight = %#v; insert calls=%d command=%#v", first.Results, engine.insertCalls, store.row)
			}

			replayed, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
			if err != nil || !reflect.DeepEqual(replayed, first) || engine.insertCalls != 0 || store.completeCommandCalls != 1 {
				t.Fatalf("replay = %#v, %v; insert calls=%d command completions=%d", replayed, err, engine.insertCalls, store.completeCommandCalls)
			}
			statusRequest, err := semantic.StatusRequestFor(request)
			if err != nil {
				t.Fatal(err)
			}
			_, status, err := service.Status(ctx, testRuntimeMutationScope, statusRequest)
			if err != nil || status.State != semantic.CommandCompleted || !reflect.DeepEqual(status.Results, first.Results) {
				t.Fatalf("status = %#v, %v", status, err)
			}
		})
	}
}

func TestRuntimeRecordMutationDoesNotPersistTransientMetadataFailureAsObjectMissing(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	metadata := runtimeMutationMetadataStub{describeErrors: map[string]error{"workitem": errors.New("metadata unavailable")}}
	tx := &runtimeMutationTxStub{}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, tx)
	request := semantic.Request{CommandID: "metadata-transient", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem"}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal ||
		store.row.State != persistence.RuntimeRecordCommandAccepted || store.completeCommandCalls != 0 || engine.insertCalls != 0 || tx.calls != 2 {
		t.Fatalf("transient metadata = %#v; command=%#v completions=%d inserts=%d transactions=%d", err, store.row, store.completeCommandCalls, engine.insertCalls, tx.calls)
	}
}

func TestRuntimeRecordMutationUnspecifiedUseCaseCannotBypassPolicy(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"protecteditem": {
		APIName: "ProtectedItem", AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceCollectionCreate}, Message: "Closed.",
		}}}},
	}}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "n-minus-one-policy", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "ProtectedItem"}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied || engine.insertCalls != 0 {
		t.Fatalf("N-1 policy = %#v, %v; insert calls %d", response, err, engine.insertCalls)
	}
}

func TestRuntimeCreateDemandCanBeSatisfiedByCompoundInput(t *testing.T) {
	schema := &models.ObjectMetadata{Fields: []models.FieldMetadata{
		{APIName: "Name", Type: "text", Required: true, Createable: true, IsCompound: true, ComposedFrom: []string{"First", "Last"}},
	}}
	for _, test := range []struct {
		name    string
		fields  models.Record
		missing bool
	}{
		{"absent", models.Record{}, true},
		{"blank component", models.Record{"First": "  "}, true},
		{"component supplied", models.Record{"Last": "Example"}, false},
		{"parent supplied", models.Record{"Name": "Example"}, false},
	} {
		t.Run(test.name, func(t *testing.T) {
			got := runtimeRequiredCreateFields(schema, test.fields)
			if test.missing && (len(got) != 1 || got[0] != "Name") || !test.missing && len(got) != 0 {
				t.Fatalf("missing fields = %v, want missing=%v", got, test.missing)
			}
		})
	}
}

func TestRuntimeRecordMutationRequiredFieldsComeFromBoundMetadata(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
		APIName: "WorkItem", Fields: []models.FieldMetadata{
			{APIName: "Name", Type: constants.FieldTypeString, Required: true, Createable: true},
			{APIName: "ServerValue", Type: constants.FieldTypeString, Required: true, Createable: false},
		},
	}}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "required", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem"}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatal(err)
	}
	if engine.insertCalls != 0 || len(response.Results) != 1 || response.Results[0].Errors[0].Kind != semantic.ErrorRequiredFieldMissing ||
		!reflect.DeepEqual(response.Results[0].Errors[0].Fields, []string{"Name"}) {
		t.Fatalf("required result = %#v; insert calls %d", response.Results, engine.insertCalls)
	}
}

func TestRuntimeRecordMutationPolicyReadsExistingRecordInsideRuntime(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{records: map[string]models.SObject{"work-1": {constants.FieldID: "work-1", "State": "Locked"}}}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
		APIName: "WorkItem", Fields: []models.FieldMetadata{
			{APIName: "State", Type: constants.FieldTypeString, Updateable: true},
			{APIName: "Description", Type: constants.FieldTypeString, Updateable: true},
		},
		AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceSingleUpdate}, Field: "State", Message: "Locked records cannot be updated.",
			When: &models.RecordMutationCondition{FieldEquals: &models.RecordMutationFieldEqualsCondition{Field: "State", Value: "Locked"}},
		}}}},
	}}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	description := "changed"
	request := semantic.Request{CommandID: "existing-policy", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{
		Kind: semantic.OperationUpdate, ObjectID: "WorkItem", RecordID: "work-1",
		Fields: []semantic.FieldValue{{Name: "Description", Kind: semantic.ValueKindString, String: &description}},
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied || engine.updateCalls != 0 {
		t.Fatalf("existing-record policy = %#v, %v; update calls %d", response, err, engine.updateCalls)
	}
}

func TestRuntimeRecordMutationOpaqueCollectionDeleteChecksEveryCandidateOnce(t *testing.T) {
	const (
		sharedID  = "7c8b6800-4c25-46f1-8071-11d9d6305bb8"
		missingID = "00000000-0000-4000-8000-000000000000"
	)
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{
		opaqueCandidates: map[string][]string{
			sharedID: []string{"OrdinaryNode", "ProtectedNode"},
		},
		recordType: "OrdinaryNode",
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{
		"ordinarynode": {APIName: "OrdinaryNode"},
		"protectednode": {
			APIName: "ProtectedNode",
			AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
				Surfaces: []string{models.RecordMutationSurfaceCollectionDelete}, Message: "Metadata-owned rows are closed.",
			}}}},
		},
	}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	request := semantic.Request{
		CommandID: "opaque-candidate-policy", BatchMode: semantic.BatchAtomic, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{
			{Kind: semantic.OperationDelete, RecordID: sharedID},
			{Kind: semantic.OperationDelete, RecordID: missingID},
		},
	}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 2 || len(response.Results[0].Errors) != 1 ||
		response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied ||
		response.Results[0].Errors[0].AuthoredMessage != "Metadata-owned rows are closed." ||
		response.Results[1].Errors[0].Kind != semantic.ErrorProcessingHalted {
		t.Fatalf("candidate policy response = %#v, %v", response, err)
	}
	if engine.opaqueCandidateCalls != 1 || !reflect.DeepEqual(engine.opaqueCandidateIDs, []string{sharedID, missingID}) ||
		!reflect.DeepEqual(engine.opaqueCandidateInTx, []bool{false}) || engine.opaqueResolutionCalls != 0 ||
		engine.deleteCalls != 0 || engine.deleteAuthorizationCalls != 0 {
		t.Fatalf("candidate discovery calls=%d ids=%#v inTx=%v legacy=%d delete=%d authorize=%d",
			engine.opaqueCandidateCalls, engine.opaqueCandidateIDs, engine.opaqueCandidateInTx,
			engine.opaqueResolutionCalls, engine.deleteCalls, engine.deleteAuthorizationCalls)
	}
}

func TestRuntimeRecordMutationOpaqueCollectionDeleteEvaluatesVisibleStoredCandidate(t *testing.T) {
	const sharedID = "7c8b6800-4c25-46f1-8071-11d9d6305bb8"
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{
		opaqueCandidates: map[string][]string{sharedID: []string{"ProtectedNode"}},
		retrieveResults:  []models.SObject{{constants.FieldID: sharedID, "State": "Locked"}},
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{
		"protectednode": {
			APIName: "ProtectedNode", Fields: []models.FieldMetadata{{APIName: "State", Type: constants.FieldTypeString}},
			AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{
				Deny: []models.RecordMutationDenial{{
					Surfaces: []string{models.RecordMutationSurfaceCollectionDelete}, Field: "State", Message: "Locked rows are closed.",
					When: &models.RecordMutationCondition{FieldEquals: &models.RecordMutationFieldEqualsCondition{Field: "State", Value: "Locked"}},
				}},
			}},
		},
	}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "opaque-candidate-stored-policy", BatchMode: semantic.BatchAtomic, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{{Kind: semantic.OperationDelete, RecordID: sharedID}}}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied ||
		response.Results[0].Errors[0].AuthoredMessage != "Locked rows are closed." || engine.retrieveCalls != 1 ||
		engine.deleteCalls != 0 || engine.opaqueResolutionCalls != 0 || !reflect.DeepEqual(engine.retrieveInTx, []bool{true}) {
		t.Fatalf("stored candidate response=%#v err=%v retrieve=%d delete=%d legacy=%d inTx=%v",
			response, err, engine.retrieveCalls, engine.deleteCalls, engine.opaqueResolutionCalls, engine.retrieveInTx)
	}
}

func TestRuntimeRecordMutationOpaqueCollectionDeleteReusesSingleCandidate(t *testing.T) {
	const recordID = "4a6db77a-c60d-47c4-b25b-da4ef2cd20bc"
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{
		opaqueCandidates: map[string][]string{recordID: []string{"Issue"}},
		records:          map[string]models.SObject{recordID: {constants.FieldID: recordID}},
	}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, runtimeMutationMetadataStub{}, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "opaque-candidate-reuse", BatchMode: semantic.BatchAtomic, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{{Kind: semantic.OperationDelete, RecordID: recordID}}}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || !response.Results[0].Success || engine.opaqueCandidateCalls != 1 ||
		engine.opaqueResolutionCalls != 0 || engine.deleteCalls != 1 || engine.deleteAuthorizationCalls != 1 {
		t.Fatalf("candidate reuse response=%#v err=%v candidate=%d legacy=%d delete=%d authorize=%d",
			response, err, engine.opaqueCandidateCalls, engine.opaqueResolutionCalls, engine.deleteCalls, engine.deleteAuthorizationCalls)
	}
}

func TestRuntimeRecordMutationOpaqueCollectionDeleteCandidateFailureFailsClosed(t *testing.T) {
	const recordID = "4a6db77a-c60d-47c4-b25b-da4ef2cd20bc"
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{opaqueCandidateErr: errors.New("candidate probe unavailable")}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, runtimeMutationMetadataStub{}, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "opaque-candidate-failure", BatchMode: semantic.BatchAtomic, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{{Kind: semantic.OperationDelete, RecordID: recordID}}}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	_, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal ||
		engine.opaqueCandidateCalls != 1 || engine.opaqueResolutionCalls != 0 || engine.deleteCalls != 0 ||
		store.row.State != persistence.RuntimeRecordCommandAccepted || store.completeCommandCalls != 0 {
		t.Fatalf("candidate failure=%#v candidate=%d legacy=%d delete=%d command=%#v completions=%d",
			err, engine.opaqueCandidateCalls, engine.opaqueResolutionCalls, engine.deleteCalls, store.row, store.completeCommandCalls)
	}
}

func TestRuntimeRecordMutationCollectionDeleteKeepsExplicitAndPrefixCatalogPaths(t *testing.T) {
	const (
		explicitID = "4a6db77a-c60d-47c4-b25b-da4ef2cd20bc"
		prefixID   = "a01000000000001AAA"
	)
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{records: map[string]models.SObject{
		explicitID: {constants.FieldID: explicitID},
		prefixID:   {constants.FieldID: prefixID},
	}}
	metadata := runtimeMutationMetadataStub{objectByRecordID: map[string]string{prefixID: "Issue"}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "catalog-delete-paths", BatchMode: semantic.BatchAtomic, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{
			{Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: explicitID},
			{Kind: semantic.OperationDelete, RecordID: prefixID},
		}}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 2 || !response.Results[0].Success || !response.Results[1].Success ||
		engine.opaqueCandidateCalls != 0 || engine.opaqueResolutionCalls != 0 || engine.deleteCalls != 2 {
		t.Fatalf("Catalog delete paths response=%#v err=%v candidate=%d legacy=%d delete=%d",
			response, err, engine.opaqueCandidateCalls, engine.opaqueResolutionCalls, engine.deleteCalls)
	}
}

func TestRuntimeRecordMutationRechecksConditionalPolicyInsideDMLTransaction(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{
		retrieveResults: []models.SObject{{constants.FieldID: "work-1", "State": "Open"}},
		records:         map[string]models.SObject{"work-1": {constants.FieldID: "work-1", "State": "Locked"}},
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
		APIName: "WorkItem", Fields: []models.FieldMetadata{
			{APIName: "State", Type: constants.FieldTypeString, Updateable: true},
			{APIName: "Description", Type: constants.FieldTypeString, Updateable: true},
		},
		AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceSingleUpdate}, Field: "State", Message: "Locked records cannot be updated.",
			When: &models.RecordMutationCondition{FieldEquals: &models.RecordMutationFieldEqualsCondition{Field: "State", Value: "Locked"}},
		}}}},
	}}}
	tx := &runtimeMutationTxStub{}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, tx)
	description := "changed"
	request := semantic.Request{CommandID: "policy-toctou-recheck", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{
		Kind: semantic.OperationUpdate, ObjectID: "WorkItem", RecordID: "work-1",
		Fields: []semantic.FieldValue{{Name: "Description", Kind: semantic.ValueKindString, String: &description}},
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := semantic.ValidateRequest(ctx, request); err != nil {
		t.Fatalf("invalid TOCTOU fixture: %#v", err)
	}
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 ||
		response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied || engine.updateCalls != 0 || engine.retrieveCalls != 1 ||
		engine.updateAuthorizationCalls != 1 || tx.calls != 5 || !reflect.DeepEqual(engine.retrieveInTx, []bool{true}) ||
		!reflect.DeepEqual(engine.authorizationInTx, []bool{true}) {
		t.Fatalf("TOCTOU recheck = %#v, %v; transactions=%d update=%d authorize=%d retrieve=%d retrieveInTx=%v authorizeInTx=%v", response, err, tx.calls, engine.updateCalls, engine.updateAuthorizationCalls, engine.retrieveCalls, engine.retrieveInTx, engine.authorizationInTx)
	}
}

func TestRuntimeRecordMutationDeleteAuthorizesLockedRow(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{
		retrieveResults: []models.SObject{{constants.FieldID: "work-1", "State": "Open"}},
		records:         map[string]models.SObject{"work-1": {constants.FieldID: "work-1", "State": "Locked"}},
	}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"issue": {
		APIName: "Issue", Fields: []models.FieldMetadata{{APIName: "State", Type: constants.FieldTypeString}},
		AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceSingleDelete}, Field: "State", Message: "Locked records cannot be deleted.",
			When: &models.RecordMutationCondition{FieldEquals: &models.RecordMutationFieldEqualsCondition{Field: "State", Value: "Locked"}},
		}}}},
	}}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "delete-policy-toctou", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{
		Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "work-1",
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 ||
		response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied || engine.deleteCalls != 0 ||
		engine.deleteAuthorizationCalls != 1 || engine.retrieveCalls != 1 ||
		engine.opaqueResolutionCalls != 0 || !reflect.DeepEqual(engine.authorizationInTx, []bool{true}) {
		t.Fatalf("delete TOCTOU recheck = %#v, %v; delete=%d authorize=%d retrieve=%d opaqueResolution=%d authorizeInTx=%v", response, err, engine.deleteCalls, engine.deleteAuthorizationCalls, engine.retrieveCalls, engine.opaqueResolutionCalls, engine.authorizationInTx)
	}
}

func TestRuntimeRecordMutationRelatedExistsFailsClosedWithoutSerializedEvidence(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{records: map[string]models.SObject{"work-1": {constants.FieldID: "work-1", "State": "Open"}}}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
		APIName: "WorkItem", Fields: []models.FieldMetadata{{APIName: "State", Type: constants.FieldTypeString, Updateable: true}},
		AccessPolicy: &models.ObjectAccessPolicy{RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{
			Surfaces: []string{models.RecordMutationSurfaceSingleUpdate}, Message: "Related work blocks mutation.",
			When: &models.RecordMutationCondition{RelatedExists: &models.RecordMutationRelatedExistsCondition{
				ObjectAPIName: "WorkLog", ParentField: "WorkItemId",
			}},
		}}}},
	}}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	state := "Closed"
	request := semantic.Request{CommandID: "related-policy-fail-closed", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{
		Kind: semantic.OperationUpdate, ObjectID: "WorkItem", RecordID: "work-1", Fields: []semantic.FieldValue{{Name: "State", Kind: semantic.ValueKindString, String: &state}},
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 ||
		response.Results[0].Errors[0].Kind != semantic.ErrorPolicyDenied || engine.updateCalls != 0 {
		t.Fatalf("relatedExists fail closed = %#v, %v; update=%d", response, err, engine.updateCalls)
	}
}

func TestRuntimeRecordMutationRejectsPostCatalogDuplicateField(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	metadata := runtimeMutationMetadataStub{schemas: map[string]*models.ObjectMetadata{"workitem": {
		APIName: "WorkItem", Fields: []models.FieldMetadata{{APIName: "Name", Type: constants.FieldTypeString, Createable: true}},
	}}}
	service := newRuntimeMutationTestServiceWithMetadata(t, engine, metadata, store, &runtimeMutationTxStub{})
	first, second := "one", "two"
	request := semantic.Request{CommandID: "canonical-duplicate", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{
		Kind: semantic.OperationCreate, ObjectID: "WorkItem", Fields: []semantic.FieldValue{
			{Name: "Name", Kind: semantic.ValueKindString, String: &first},
			{Name: "name", Kind: semantic.ValueKindString, String: &second},
		},
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || response.Results[0].Errors[0].Kind != semantic.ErrorInvalidFieldForWrite || engine.insertCalls != 0 {
		t.Fatalf("canonical duplicate = %#v, %v; insert calls %d", response, err, engine.insertCalls)
	}
}

func TestRuntimeRecordMutationClassifiesProtectedGuardAsInvalidFieldForWrite(t *testing.T) {
	err := rejectProtectedGuardFieldWrites(constants.TableProfile, models.SObject{
		constants.FieldProfile_IsSystem: true,
	})
	operation := semantic.Operation{Kind: semantic.OperationUpdate, ObjectID: constants.TableProfile, RecordID: "profile-1"}
	result, ok := classifyRuntimeMutationDomainError(operation, 0, err)
	if !ok || len(result.Errors) != 1 || result.Errors[0].Kind != semantic.ErrorInvalidFieldForWrite ||
		!reflect.DeepEqual(result.Errors[0].Fields, []string{constants.FieldProfile_IsSystem}) {
		t.Fatalf("protected guard Runtime projection = %#v, %t; want invalid-field-for-write", result, ok)
	}
}

func TestRuntimeExactJSONNumberRejectsExponentExpansion(t *testing.T) {
	for raw, want := range map[string]string{
		"1.2500e2": "125.00",
		"1e-2":     "0.01",
		"0e999999": "0",
	} {
		got, err := runtimeExactJSONNumber(raw)
		if err != nil || got.String() != want {
			t.Fatalf("runtimeExactJSONNumber(%q) = %q, %v; want %q", raw, got, err, want)
		}
	}
	for _, raw := range []string{"1e999999", "1e9223372036854775807", "1e-9223372036854775808"} {
		if _, err := runtimeExactJSONNumber(raw); err == nil {
			t.Fatalf("runtimeExactJSONNumber(%q) accepted expansion bomb", raw)
		}
	}
}

func TestRuntimeRecordMutationIndependentDomainFailureDoesNotAbortSibling(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{insertErrors: []error{
		appErrors.NewPermissionError("create", "Denied"), nil,
	}}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	name := "ok"
	request := semantic.Request{CommandID: "partial-1", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{
		{Kind: semantic.OperationCreate, ObjectID: "Denied", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}},
		{Kind: semantic.OperationCreate, ObjectID: "Allowed", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}},
	}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	if len(response.Results) != 2 || response.Results[0].Success || response.Results[0].Errors[0].Kind != semantic.ErrorAccessDenied || !response.Results[1].Success {
		t.Fatalf("Mutate() results = %#v", response.Results)
	}
}

func TestRuntimeRecordMutationPersistsCreateFromSourceGuardForStatus(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{insertErrors: []error{
		appErrors.NewCreateFromSourceGuardError("TargetObject", []string{"createTarget"}),
	}}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "guard-status-1", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{{
		Kind: semantic.OperationCreate, ObjectID: "TargetObject",
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 {
		t.Fatalf("Mutate() = %#v, %v", response, err)
	}
	want := &semantic.CreateFromSourceGuard{ObjectID: "TargetObject", ActionIDs: []string{"createTarget"}}
	if !reflect.DeepEqual(response.Results[0].Errors[0].CreateFromSourceGuard, want) {
		t.Fatalf("Mutate() guard = %#v, want %#v", response.Results[0].Errors[0].CreateFromSourceGuard, want)
	}

	query, err := semantic.StatusRequestFor(request)
	if err != nil {
		t.Fatal(err)
	}
	shape, status, err := service.Status(ctx, testRuntimeMutationScope, query)
	if err != nil || len(shape.OperationObjectIDs) != 1 || shape.OperationObjectIDs[0] != "TargetObject" || len(status.Results) != 1 {
		t.Fatalf("Status() shape=%#v response=%#v err=%v", shape, status, err)
	}
	if !reflect.DeepEqual(status.Results[0].Errors[0].CreateFromSourceGuard, want) {
		t.Fatalf("Status() guard = %#v, want %#v", status.Results[0].Errors[0].CreateFromSourceGuard, want)
	}
}

func TestRuntimeRecordMutationPersistsDeleteConflictForStatusAndReplay(t *testing.T) {
	engine := &runtimeMutationEngineStub{deleteError: appErrors.NewConflictErrorWithMessage("Profile", "Profile has referencing users")}
	service := newRuntimeMutationTestService(t, engine, newRuntimeMutationMemoryStore(), &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "delete-conflict-status", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{{
		Kind: semantic.OperationDelete, ObjectID: "Profile", RecordID: "profile-1",
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	actor := &models.UserSession{ID: "user-1"}
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), actor)
	if err != nil || len(response.Results) != 1 || len(response.Results[0].Errors) != 1 || response.Results[0].Success || response.Results[0].Errors[0].Kind != semantic.ErrorInvalidOperation {
		t.Fatalf("delete conflict must be a terminal domain result, got %#v, %v", response, err)
	}
	query, err := semantic.StatusRequestFor(request)
	if err != nil {
		t.Fatal(err)
	}
	_, status, err := service.Status(ctx, testRuntimeMutationScope, query)
	if err != nil || status.State != semantic.CommandCompleted || !reflect.DeepEqual(status.Results, response.Results) || status.CommandDigest != response.CommandDigest || status.OutcomeExpiresAtUTC != response.OutcomeExpiresAtUTC {
		t.Fatalf("Status() = %#v, %v; want %#v", status, err, response)
	}
	replay, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), actor)
	if err != nil || !reflect.DeepEqual(replay, response) || engine.deleteCalls != 1 {
		t.Fatalf("replay = %#v, %v; delete calls = %d", replay, err, engine.deleteCalls)
	}
}

func TestRuntimeRecordMutationDoesNotClaimDomainFailureWithoutDurableRollbackProof(t *testing.T) {
	for _, test := range []struct {
		name           string
		commandID      string
		useCase        semantic.MutationUseCase
		rollbackOnCall int
	}{
		{name: "legacy locked prepare", commandID: "rollback-proof-legacy", useCase: semantic.MutationUseCaseUnspecified, rollbackOnCall: 3},
		{name: "fresh static prepare", commandID: "rollback-proof-static", useCase: semantic.MutationUseCaseSingle, rollbackOnCall: 2},
	} {
		t.Run(test.name, func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			engine := &runtimeMutationEngineStub{insertErrors: []error{appErrors.NewValidationRuleError("BlockIssue", "blocked")}}
			tx := &runtimeMutationTxStub{rollbackErrorOnCall: test.rollbackOnCall}
			service := newRuntimeMutationTestService(t, engine, store, tx)
			name := "blocked"
			request := semantic.Request{CommandID: test.commandID, BatchMode: semantic.BatchIndependent, UseCase: test.useCase, Operations: []semantic.Operation{{
				Kind: semantic.OperationCreate, ObjectID: "Issue", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}},
			}}}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()

			_, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
			var failure *semantic.Failure
			if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal {
				t.Fatalf("Mutate() error = %#v, want INTERNAL_FAILURE", err)
			}
			if store.completeOperationCalls != 0 || store.completeCommandCalls != 0 {
				t.Fatalf("durable terminal writes = operations %d command %d, want zero", store.completeOperationCalls, store.completeCommandCalls)
			}
		})
	}
}

func TestRuntimeRecordMutationClosesAcceptedCommandWhenServingRevisionChanges(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	name := "first"
	request := semantic.Request{CommandID: "revision-1", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{{
		Kind: semantic.OperationCreate, ObjectID: "Issue", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}},
	}}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	// Leave a durable accepted command as if the API stopped after acceptance.
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	receiptsPayload, _ := json.Marshal(runtimeMutationTestReceipts(7, 11))
	store.row = persistence.RuntimeRecordCommandRow{
		RowID: "row-1", ScopeDigest: testRuntimeMutationScope, CommandID: request.CommandID,
		RequestDigest: digest, BatchMode: string(request.BatchMode), OperationCount: 1,
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: receiptsPayload,
		OutcomeExpiresAt: time.Now().Add(time.Hour), OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	}
	store.slots = []persistence.RuntimeRecordCommandOperationRow{{RowID: "slot-1", CommandRowID: "row-1", Index: 0, Kind: string(semantic.OperationCreate), ObjectID: "Issue", BoundRecordID: "reserved-record-1", State: persistence.RuntimeRecordOperationPending}}
	if results, pending, buildErr := interruptedRuntimeRecordCommandResults(store.row, store.slots); buildErr != nil {
		t.Fatalf("interrupted outcome fixture = %#v pending=%v err=%v", results, pending, buildErr)
	}

	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(8, 11), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 1 || response.Results[0].Success ||
		len(response.Results[0].Errors) != 1 || response.Results[0].Errors[0].Kind != semantic.ErrorInternal {
		t.Fatalf("Mutate() = %#v, %v; want durable terminal internal result", response, err)
	}
	if response.Receipts.CatalogRevision != 7 || response.Receipts.SecurityEpoch != 11 ||
		store.row.State != persistence.RuntimeRecordCommandCompleted || engine.insertCalls != 0 {
		t.Fatalf("terminal binding = %#v command=%#v inserts=%d", response.Receipts, store.row, engine.insertCalls)
	}
	statusRequest, err := semantic.StatusRequestFor(request)
	if err != nil {
		t.Fatal(err)
	}
	_, status, err := service.Status(ctx, testRuntimeMutationScope, statusRequest)
	if err != nil || status.State != semantic.CommandCompleted || !reflect.DeepEqual(status.Results, response.Results) {
		t.Fatalf("Status() = %#v, %v; want the same terminal outcome", status, err)
	}
}

func TestRuntimeRecordMutationClosesPartialIndependentCommandWithoutRepeatingCommittedSlot(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	name := "first"
	request := semantic.Request{CommandID: "partial-revision-change", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{
		{Kind: semantic.OperationCreate, ObjectID: "Issue", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}},
		{Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "issue-2"},
	}}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	receiptsPayload, _ := json.Marshal(runtimeMutationTestReceipts(7, 11))
	store.row = persistence.RuntimeRecordCommandRow{
		RowID: "row-partial", ScopeDigest: testRuntimeMutationScope, CommandID: request.CommandID,
		RequestDigest: digest, BatchMode: string(request.BatchMode), OperationCount: 2,
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: receiptsPayload,
		OutcomeExpiresAt: time.Now().Add(time.Hour), OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	}
	committed := semantic.OperationResult{Index: 0, Kind: semantic.OperationCreate, RecordID: "issue-1", Success: true, Created: boolPointer(true), Errors: []semantic.OperationError{}}
	committedPayload, _ := json.Marshal(committed)
	store.slots = []persistence.RuntimeRecordCommandOperationRow{
		{RowID: "slot-1", CommandRowID: "row-partial", Index: 0, Kind: string(semantic.OperationCreate), ObjectID: "Issue", BoundRecordID: "issue-1", State: persistence.RuntimeRecordOperationDone, ResultPayload: committedPayload},
		{RowID: "slot-2", CommandRowID: "row-partial", Index: 1, Kind: string(semantic.OperationDelete), ObjectID: "Issue", BoundRecordID: "issue-2", State: persistence.RuntimeRecordOperationPending},
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(8, 12), &models.UserSession{ID: "user-1"})
	if err != nil || len(response.Results) != 2 || !reflect.DeepEqual(response.Results[0], committed) ||
		response.Results[1].Success || response.Results[1].Errors[0].Kind != semantic.ErrorInternal {
		t.Fatalf("partial closure = %#v, %v", response, err)
	}
	if engine.insertCalls != 0 || engine.deleteCalls != 0 || store.completeOperationCalls != 1 || store.completeCommandCalls != 1 {
		t.Fatalf("effects insert=%d delete=%d slot completions=%d command completions=%d", engine.insertCalls, engine.deleteCalls, store.completeOperationCalls, store.completeCommandCalls)
	}
}

func TestRuntimeRecordMutationDoesNotExecuteExpiredAcceptedCommand(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	engine := &runtimeMutationEngineStub{}
	service := newRuntimeMutationTestService(t, engine, store, &runtimeMutationTxStub{})
	request := semantic.Request{CommandID: "expired-accepted", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "Issue"}}}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	receipts := runtimeMutationTestReceipts(7, 11)
	receiptsPayload, _ := json.Marshal(receipts)
	store.row = persistence.RuntimeRecordCommandRow{
		RowID: "row-expired-accepted", ScopeDigest: testRuntimeMutationScope, CommandID: request.CommandID,
		RequestDigest: digest, BatchMode: string(request.BatchMode), OperationCount: 1,
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: receiptsPayload,
		OutcomeExpiresAt: time.Date(2026, 7, 19, 11, 0, 0, 0, time.UTC), OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	}
	store.slots = []persistence.RuntimeRecordCommandOperationRow{{
		RowID: "slot-expired", CommandRowID: store.row.RowID, Index: 0, Kind: string(semantic.OperationCreate),
		ObjectID: "Issue", BoundRecordID: "reserved-expired", State: persistence.RuntimeRecordOperationPending,
	}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := service.Mutate(ctx, testRuntimeMutationScope, request, receipts, &models.UserSession{ID: "user-1"})
	if err != nil || response.Results[0].Errors[0].Kind != semantic.ErrorInternal || engine.insertCalls != 0 ||
		response.OutcomeExpiresAtUTC != "2026-07-19T13:00:00Z" || store.row.State != persistence.RuntimeRecordCommandCompleted {
		t.Fatalf("expired replay = %#v, %v; inserts=%d command=%#v", response, err, engine.insertCalls, store.row)
	}
}

func TestRuntimeRecordMutationRechecksExecutionWindowUnderCommandLock(t *testing.T) {
	acceptedAt := time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC)
	expiresAt := acceptedAt.Add(time.Hour)
	for _, tc := range []struct {
		name        string
		batchMode   semantic.BatchMode
		wantInserts int
		wantErrors  []semantic.ErrorKind
		times       []time.Time
	}{
		{
			name:      "independent command preserves committed prefix and stops at expiry",
			batchMode: semantic.BatchIndependent, wantInserts: 1,
			wantErrors: []semantic.ErrorKind{"", semantic.ErrorInternal},
			times:      []time.Time{expiresAt.Add(-time.Minute), expiresAt.Add(-time.Microsecond), expiresAt, expiresAt},
		},
		{
			name:      "atomic command performs no DML when execution starts at expiry",
			batchMode: semantic.BatchAtomic, wantInserts: 0,
			wantErrors: []semantic.ErrorKind{semantic.ErrorInternal, semantic.ErrorAtomicRolledBack},
			times:      []time.Time{expiresAt.Add(-time.Minute), expiresAt, expiresAt},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newRuntimeMutationMemoryStore()
			store.now = acceptedAt
			engine := &runtimeMutationEngineStub{}
			tx := &runtimeMutationTxStub{times: append([]time.Time(nil), tc.times...)}
			service := newRuntimeMutationTestService(t, engine, store, tx)
			request := semantic.Request{
				CommandID: "expiry-under-lock", BatchMode: tc.batchMode,
				Operations: []semantic.Operation{
					{Kind: semantic.OperationCreate, ObjectID: "Issue"},
					{Kind: semantic.OperationCreate, ObjectID: "Issue"},
				},
			}
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			response, err := service.Mutate(ctx, testRuntimeMutationScope, request,
				runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"})
			if err != nil {
				t.Fatal(err)
			}
			if engine.insertCalls != tc.wantInserts || store.row.State != persistence.RuntimeRecordCommandCompleted ||
				response.OutcomeExpiresAtUTC != expiresAt.Add(time.Hour).Format(time.RFC3339Nano) {
				t.Fatalf("response=%#v inserts=%d command=%#v", response, engine.insertCalls, store.row)
			}
			if len(response.Results) != len(tc.wantErrors) {
				t.Fatalf("results=%#v", response.Results)
			}
			for i, want := range tc.wantErrors {
				if want == "" {
					if !response.Results[i].Success {
						t.Fatalf("result[%d]=%#v, want success", i, response.Results[i])
					}
					continue
				}
				if response.Results[i].Success || len(response.Results[i].Errors) != 1 || response.Results[i].Errors[0].Kind != want {
					t.Fatalf("result[%d]=%#v, want %s", i, response.Results[i], want)
				}
			}
			if len(tx.times) != 0 {
				t.Fatalf("unused database-time observations = %v", tx.times)
			}
		})
	}
}

func boolPointer(value bool) *bool { return &value }

func TestRuntimeRecordMutationDeleteUsesCatalogIdentityBeforePhysicalFallback(t *testing.T) {
	tests := []struct {
		name             string
		operation        semantic.Operation
		metadataObjects  map[string]string
		engineObject     string
		wantObject       string
		wantMasked       bool
		wantPhysicalCall int
	}{
		{
			name: "explicit object is already the operation boundary",
			operation: semantic.Operation{
				Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "opaque-record-id",
			},
			wantObject: "Issue",
		},
		{
			name: "matching catalog prefix avoids physical discovery",
			operation: semantic.Operation{
				Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "a01000000000001AAA",
			},
			metadataObjects: map[string]string{"a01000000000001AAA": "Issue"},
			wantObject:      "Issue",
		},
		{
			name: "catalog prefix mismatch is existence masked",
			operation: semantic.Operation{
				Kind: semantic.OperationDelete, ObjectID: "Account", RecordID: "a01000000000001AAA",
			},
			metadataObjects: map[string]string{"a01000000000001AAA": "Issue"},
			wantMasked:      true,
		},
		{
			name: "objectless prefixed delete uses catalog",
			operation: semantic.Operation{
				Kind: semantic.OperationDelete, RecordID: "a01000000000001AAA",
			},
			metadataObjects: map[string]string{"a01000000000001AAA": "Issue"},
			wantObject:      "Issue",
		},
		{
			name: "objectless opaque delete preserves physical compatibility",
			operation: semantic.Operation{
				Kind: semantic.OperationDelete, RecordID: "4a6db77a-c60d-47c4-b25b-da4ef2cd20bc",
			},
			engineObject:     "Issue",
			wantObject:       "Issue",
			wantPhysicalCall: 1,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			engine := &runtimeMutationEngineStub{recordType: test.engineObject}
			service := &RuntimeRecordMutationService{
				engine: engine,
				metadata: runtimeMutationMetadataStub{
					objectByRecordID: test.metadataObjects,
				},
			}
			objectName, masked, err := service.resolveDeleteObject(t.Context(), test.operation)
			if err != nil {
				t.Fatalf("resolveDeleteObject() error = %v", err)
			}
			if objectName != test.wantObject || masked != test.wantMasked {
				t.Fatalf("resolveDeleteObject() = (%q, %v), want (%q, %v)", objectName, masked, test.wantObject, test.wantMasked)
			}
			if engine.opaqueResolutionCalls != test.wantPhysicalCall {
				t.Fatalf("physical reverse lookups = %d, want %d", engine.opaqueResolutionCalls, test.wantPhysicalCall)
			}
		})
	}
}

func TestRuntimeRecordMutationDeleteObjectResolutionPreservesCancellation(t *testing.T) {
	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	engine := &runtimeMutationEngineStub{}
	service := &RuntimeRecordMutationService{engine: engine, metadata: runtimeMutationMetadataStub{}}
	_, _, err := service.resolveDeleteObject(ctx, semantic.Operation{
		Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "a01000000000001AAA",
	})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("resolveDeleteObject() error = %v, want context.Canceled", err)
	}
	if engine.opaqueResolutionCalls != 0 {
		t.Fatalf("canceled resolution performed %d physical lookups", engine.opaqueResolutionCalls)
	}
}

func TestRuntimeRecordMutationDeleteExecutionDoesNotRepeatObjectDiscovery(t *testing.T) {
	engine := &runtimeMutationEngineStub{records: map[string]models.SObject{
		"a01000000000001AAA": {constants.FieldID: "a01000000000001AAA"},
	}}
	service := &RuntimeRecordMutationService{engine: engine}
	operation := semantic.Operation{
		Kind: semantic.OperationDelete, ObjectID: "Issue", RecordID: "a01000000000001AAA",
	}
	result, err := service.executeOperation(t.Context(), runtimePreparedMutation{
		operation:  operation,
		schema:     &models.ObjectMetadata{APIName: "Issue"},
		objectName: "Issue",
		useCase:    semantic.MutationUseCaseSingle,
	}, 0, "", &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("executeOperation() error = %v", err)
	}
	if !result.Success || engine.deleteCalls != 1 {
		t.Fatalf("executeOperation() result = %#v, delete calls = %d", result, engine.deleteCalls)
	}
	if engine.opaqueResolutionCalls != 0 {
		t.Fatalf("execution repeated physical object discovery %d times", engine.opaqueResolutionCalls)
	}
}

func TestRuntimeRecordMutationOutcomeWindowMatchesTiDBPrecision(t *testing.T) {
	tx := &runtimeMutationTxStub{now: time.Date(2026, 7, 19, 12, 0, 0, 123456789, time.UTC)}
	service := newRuntimeMutationTestService(t, &runtimeMutationEngineStub{}, newRuntimeMutationMemoryStore(), tx)
	// Current process policy may differ after a restart. The accepted command's
	// persisted lineage remains authoritative for the terminal proof window.
	service.retention = 5 * time.Minute
	want := time.Date(2026, 7, 19, 13, 0, 0, 123456000, time.UTC)
	got, err := service.nextOutcomeExpiry(context.Background(), persistence.RuntimeRecordCommandRow{
		OutcomeRetentionMicros: int64(time.Hour / time.Microsecond),
	})
	if err != nil || got != want {
		t.Fatalf("nextOutcomeExpiry() = %s, want DATETIME(6) precision %s", got, want)
	}
}

func TestRuntimeRecordMutationStatusBindsContentAndDurableShape(t *testing.T) {
	store := newRuntimeMutationMemoryStore()
	service := newRuntimeMutationTestService(t, &runtimeMutationEngineStub{}, store, &runtimeMutationTxStub{})
	name := "updated"
	request := semantic.Request{CommandID: "status-binding-1", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{
		{Kind: semantic.OperationUpdate, ObjectID: "Issue", RecordID: "issue-1", Fields: []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &name}}},
		{Kind: semantic.OperationDelete, RecordID: "issue-2"},
	}}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if _, err := service.Mutate(ctx, testRuntimeMutationScope, request, runtimeMutationTestReceipts(7, 11), &models.UserSession{ID: "user-1"}); err != nil {
		t.Fatalf("Mutate() error = %v", err)
	}
	query, err := semantic.StatusRequestFor(request)
	if err != nil {
		t.Fatal(err)
	}
	shape, status, err := service.Status(ctx, testRuntimeMutationScope, query)
	if err != nil || status.State != semantic.CommandCompleted || len(status.Results) != len(request.Operations) {
		t.Fatalf("Status() = %#v, %v", status, err)
	}
	if len(shape.OperationKinds) != 2 || shape.OperationKinds[0] != semantic.OperationUpdate || shape.OperationObjectIDs[0] != "Issue" || shape.OperationObjectIDs[1] != "" || shape.OperationRecordIDs[0] != "issue-1" || shape.OperationRecordIDs[1] != "issue-2" {
		t.Fatalf("Status() shape = %#v", shape)
	}

	wrongBinding := query
	wrongBinding.CommandDigest = "sha256:" + strings.Repeat("0", 64)
	_, _, err = service.Status(ctx, testRuntimeMutationScope, wrongBinding)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureMissing {
		t.Fatalf("wrong-binding Status() error = %#v, want MISSING", err)
	}

	var persisted semantic.Response
	if err := json.Unmarshal(store.row.ResultPayload, &persisted); err != nil {
		t.Fatal(err)
	}
	persisted.Results = persisted.Results[:1]
	store.row.ResultPayload, _ = json.Marshal(persisted)
	_, _, err = service.Status(ctx, testRuntimeMutationScope, query)
	if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal {
		t.Fatalf("truncated durable Status() error = %#v, want INTERNAL_FAILURE", err)
	}
}

func newRuntimeMutationTestService(t *testing.T, engine RuntimeRecordMutationEngine, store RuntimeRecordCommandStore, tx RuntimeRecordMutationExecutionTransactions) *RuntimeRecordMutationService {
	return newRuntimeMutationTestServiceWithMetadata(t, engine, runtimeMutationMetadataStub{}, store, tx)
}

func newRuntimeMutationTestServiceWithMetadata(t *testing.T, engine RuntimeRecordMutationEngine, metadata RuntimeRecordMutationMetadata, store RuntimeRecordCommandStore, tx RuntimeRecordMutationExecutionTransactions) *RuntimeRecordMutationService {
	return newRuntimeMutationTestServiceWithActivity(t, engine, metadata, &runtimeMutationActivityStub{}, store, tx)
}

func newRuntimeMutationTestServiceWithActivity(t *testing.T, engine RuntimeRecordMutationEngine, metadata RuntimeRecordMutationMetadata, activity RuntimeRecordMutationActivity, store RuntimeRecordCommandStore, tx RuntimeRecordMutationExecutionTransactions) *RuntimeRecordMutationService {
	t.Helper()
	service, err := NewRuntimeRecordMutationService(RuntimeRecordMutationConfig{
		Engine: engine, Metadata: metadata, Activity: activity, Commands: store, Transactions: tx, OutcomeRetention: time.Hour,
	})
	if err != nil {
		t.Fatal(err)
	}
	return service
}

type runtimeMutationActivityStub struct {
	mu    sync.Mutex
	calls [][3]string
}

func (a *runtimeMutationActivityStub) TrackRecordAccess(_ context.Context, userID, objectAPIName, recordID string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.calls = append(a.calls, [3]string{userID, objectAPIName, recordID})
}

func runtimeMutationTestReceipts(catalog, security uint64) semantic.Receipts {
	return semantic.Receipts{
		ProtocolVersion: "1.0", CatalogRevision: catalog, SchemaGeneration: 2,
		EngineGeneration: 3, SecurityEpoch: security, CapabilityVersion: semantic.CapabilityName,
	}
}

type runtimeMutationTxStub struct {
	mu                  sync.Mutex
	calls               int
	commitErrorOnCall   int
	rollbackErrorOnCall int
	now                 time.Time
	times               []time.Time
	mutationObjectSets  [][]string
}

type runtimeMutationTestTxKey struct{}

func (t *runtimeMutationTxStub) ReadTransactionTime(context.Context) (time.Time, error) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if len(t.times) != 0 {
		now := t.times[0]
		t.times = t.times[1:]
		return now, nil
	}
	if !t.now.IsZero() {
		return t.now, nil
	}
	return time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC), nil
}

func (t *runtimeMutationTxStub) WithPessimisticRetry(ctx context.Context, fn func(context.Context) error, _ int) error {
	t.mu.Lock()
	t.calls++
	call := t.calls
	t.mu.Unlock()
	txCtx := ports.WithTransactionCompletionHooks(context.WithValue(ctx, runtimeMutationTestTxKey{}, true))
	err := fn(txCtx)
	if err == nil {
		if call == t.commitErrorOnCall {
			ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitError)
			return appErrors.NewInternalError("commit outcome unknown", errors.New("commit acknowledgement lost"))
		}
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionCommitted)
		return nil
	}
	if call == t.rollbackErrorOnCall {
		ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRollbackError)
		return appErrors.NewInternalError("rollback outcome unknown", err)
	}
	ports.RunTransactionCompletionHooks(txCtx, ports.TransactionCompletionRolledBack)
	return err
}

func (t *runtimeMutationTxStub) RunRecordMutationTransaction(ctx context.Context, objectNames []string, fn func(context.Context) error) error {
	t.mu.Lock()
	t.mutationObjectSets = append(t.mutationObjectSets, append([]string(nil), objectNames...))
	t.mu.Unlock()
	return t.WithPessimisticRetry(ctx, fn, runtimeRecordCommandTransactionRetries)
}

func (t *runtimeMutationTxStub) WithTransaction(ctx context.Context, fn func(context.Context) error) error {
	return t.WithPessimisticRetry(ctx, fn, 1)
}
func (t *runtimeMutationTxStub) WithPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	return t.WithPessimisticRetry(ctx, fn, 1)
}
func (t *runtimeMutationTxStub) WithRetry(ctx context.Context, fn func(context.Context) error, _ int) error {
	return t.WithPessimisticRetry(ctx, fn, 1)
}
func (*runtimeMutationTxStub) IsTransactionActive(context.Context) bool { return true }
func (*runtimeMutationTxStub) IsPessimisticTx(context.Context) bool     { return true }
func (*runtimeMutationTxStub) ExtractTx(context.Context) interface{}    { return struct{}{} }
func (*runtimeMutationTxStub) DetachTx(ctx context.Context) context.Context {
	return context.WithValue(ctx, runtimeMutationTestTxKey{}, false)
}

type runtimeMutationEngineStub struct {
	insertCalls              int
	insertErrors             []error
	lastInsert               models.SObject
	updateCalls              int
	updateAuthorizationCalls int
	lastUpdate               models.SObject
	updateValidate           func(models.SObject) error
	deleteCalls              int
	deleteError              error
	deleteAuthorizationCalls int
	records                  map[string]models.SObject
	retrieveCalls            int
	retrieveResults          []models.SObject
	retrieveErrors           []error
	retrieveInTx             []bool
	authorizationInTx        []bool
	opaqueResolutionCalls    int
	opaqueCandidateCalls     int
	opaqueCandidateIDs       []string
	opaqueCandidates         map[string][]string
	opaqueCandidateErr       error
	opaqueCandidateInTx      []bool
	recordType               string
	recordTypeErr            error
}

func (*runtimeMutationEngineStub) runtimeRecordMutationEngine() {}

func (e *runtimeMutationEngineStub) Insert(ctx context.Context, _ string, data models.SObject, _ *models.UserSession) (models.SObject, error) {
	errIndex := e.insertCalls
	e.insertCalls++
	if errIndex < len(e.insertErrors) && e.insertErrors[errIndex] != nil {
		return nil, e.insertErrors[errIndex]
	}
	e.lastInsert = cloneSObject(data)
	result := cloneSObject(data)
	if id, _ := result[constants.FieldID].(string); id == "" {
		result[constants.FieldID] = runtimeReservedCreateID(ctx)
	}
	return result, nil
}

func (e *runtimeMutationEngineStub) UpdateAuthorized(ctx context.Context, _ string, id string, updates models.SObject, _ *models.UserSession, authorize func(context.Context, models.Record) error) error {
	e.updateAuthorizationCalls++
	e.authorizationInTx = append(e.authorizationInTx, ctx.Value(runtimeMutationTestTxKey{}) == true)
	locked := cloneSObject(e.records[id])
	if locked == nil {
		locked = models.SObject{constants.FieldID: id}
	}
	if err := authorize(ctx, models.Record(locked)); err != nil {
		return err
	}
	if e.updateValidate != nil {
		if err := e.updateValidate(updates); err != nil {
			return err
		}
		for field, value := range updates {
			locked[field] = value
		}
		e.records[id] = locked
	}
	e.updateCalls++
	e.lastUpdate = cloneSObject(updates)
	return nil
}

func (e *runtimeMutationEngineStub) DeleteAuthorized(ctx context.Context, _ string, id string, _ *models.UserSession, authorize func(context.Context, models.Record) error) error {
	e.deleteAuthorizationCalls++
	e.authorizationInTx = append(e.authorizationInTx, ctx.Value(runtimeMutationTestTxKey{}) == true)
	locked := cloneSObject(e.records[id])
	if locked == nil {
		locked = models.SObject{constants.FieldID: id}
	}
	if err := authorize(ctx, models.Record(locked)); err != nil {
		return err
	}
	e.deleteCalls++
	return e.deleteError
}
func (e *runtimeMutationEngineStub) ResolveOpaqueRecordObject(context.Context, string) (string, error) {
	e.opaqueResolutionCalls++
	if e.recordTypeErr != nil {
		return "", e.recordTypeErr
	}
	if e.recordType != "" {
		return e.recordType, nil
	}
	return "Issue", nil
}
func (e *runtimeMutationEngineStub) ResolveOpaqueRecordObjectCandidates(ctx context.Context, ids []string) (map[string][]string, error) {
	e.opaqueCandidateCalls++
	e.opaqueCandidateIDs = append([]string(nil), ids...)
	e.opaqueCandidateInTx = append(e.opaqueCandidateInTx, ctx.Value(runtimeMutationTestTxKey{}) == true)
	if e.opaqueCandidateErr != nil {
		return nil, e.opaqueCandidateErr
	}
	result := make(map[string][]string, len(e.opaqueCandidates))
	for id, candidates := range e.opaqueCandidates {
		result[id] = append([]string(nil), candidates...)
	}
	return result, nil
}
func (e *runtimeMutationEngineStub) ReadPolicyTarget(ctx context.Context, _ string, id string, _ semantic.OperationKind, _ *models.UserSession) (models.SObject, error) {
	e.retrieveInTx = append(e.retrieveInTx, ctx.Value(runtimeMutationTestTxKey{}) == true)
	if e.retrieveCalls < len(e.retrieveErrors) && e.retrieveErrors[e.retrieveCalls] != nil {
		err := e.retrieveErrors[e.retrieveCalls]
		e.retrieveCalls++
		return nil, err
	}
	if e.retrieveCalls < len(e.retrieveResults) {
		result := cloneSObject(e.retrieveResults[e.retrieveCalls])
		e.retrieveCalls++
		return result, nil
	}
	e.retrieveCalls++
	if record := e.records[id]; record != nil {
		return cloneSObject(record), nil
	}
	return models.SObject{constants.FieldID: id}, nil
}

type runtimeMutationMetadataStub struct {
	schemas          map[string]*models.ObjectMetadata
	missing          map[string]bool
	describeErrors   map[string]error
	objectByRecordID map[string]string
}

func (m runtimeMutationMetadataStub) Describe(_ context.Context, objectName string) (*models.ObjectMetadata, error) {
	key := strings.ToLower(objectName)
	if err := m.describeErrors[key]; err != nil {
		return nil, err
	}
	if m.missing[key] {
		return nil, appErrors.NewNotFoundError("Object", objectName)
	}
	if schema := m.schemas[key]; schema != nil {
		return schema, nil
	}
	return &models.ObjectMetadata{APIName: objectName}, nil
}

func (m runtimeMutationMetadataStub) ResolveObjectForRecordID(_ context.Context, recordID string) (string, bool) {
	objectName, ok := m.objectByRecordID[recordID]
	return objectName, ok
}

type runtimeMutationMemoryStore struct {
	row                    persistence.RuntimeRecordCommandRow
	slots                  []persistence.RuntimeRecordCommandOperationRow
	loadCalls              int
	completeOperationCalls int
	completeCommandCalls   int
	now                    time.Time
}

func newRuntimeMutationMemoryStore() *runtimeMutationMemoryStore {
	return &runtimeMutationMemoryStore{now: time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC)}
}

func (s *runtimeMutationMemoryStore) Accept(_ context.Context, input persistence.RuntimeRecordCommandAcceptance) (persistence.RuntimeRecordCommandRow, bool, error) {
	if s.row.RowID != "" {
		if s.row.RequestDigest != input.RequestDigest || s.row.BatchMode != input.BatchMode || s.row.OperationCount != len(input.Operations) {
			return s.row, true, persistence.ErrRuntimeRecordCommandConflict
		}
		if !reflect.DeepEqual(s.row.ReceiptsPayload, input.ReceiptsPayload) {
			return s.row, true, persistence.ErrRuntimeRecordCommandServingBindingChanged
		}
		return s.row, true, nil
	}
	s.row = persistence.RuntimeRecordCommandRow{
		RowID: input.RowID, ScopeDigest: input.ScopeDigest, CommandID: input.CommandID,
		RequestDigest: input.RequestDigest, BatchMode: input.BatchMode, OperationCount: len(input.Operations),
		State: persistence.RuntimeRecordCommandAccepted, ReceiptsPayload: append([]byte(nil), input.ReceiptsPayload...),
		OutcomeExpiresAt:       s.now.Add(input.OutcomeRetention).Truncate(time.Microsecond),
		OutcomeRetentionMicros: input.OutcomeRetention.Microseconds(),
	}
	s.slots = make([]persistence.RuntimeRecordCommandOperationRow, len(input.Operations))
	for i, operation := range input.Operations {
		s.slots[i] = persistence.RuntimeRecordCommandOperationRow{
			RowID: operation.RowID, CommandRowID: input.RowID, Index: operation.Index,
			Kind: operation.Kind, ObjectID: operation.ObjectID, BoundRecordID: operation.BoundRecordID, State: persistence.RuntimeRecordOperationPending,
		}
	}
	return s.row, false, nil
}

func (s *runtimeMutationMemoryStore) Load(_ context.Context, scopeDigest, commandID string) (persistence.RuntimeRecordCommandRow, error) {
	s.loadCalls++
	if s.row.ScopeDigest != scopeDigest || s.row.CommandID != commandID {
		return persistence.RuntimeRecordCommandRow{}, persistence.ErrRuntimeRecordCommandMissing
	}
	return s.row, nil
}
func (s *runtimeMutationMemoryStore) LoadUnexpired(ctx context.Context, scopeDigest, commandID string) (persistence.RuntimeRecordCommandRow, error) {
	row, err := s.Load(ctx, scopeDigest, commandID)
	if err != nil || !row.OutcomeExpiresAt.After(s.now) {
		return persistence.RuntimeRecordCommandRow{}, persistence.ErrRuntimeRecordCommandMissing
	}
	return row, nil
}
func (s *runtimeMutationMemoryStore) ListExpiredAccepted(_ context.Context, limit int) ([]string, error) {
	if limit <= 0 || s.row.State != persistence.RuntimeRecordCommandAccepted || s.row.OutcomeExpiresAt.After(s.now) {
		return []string{}, nil
	}
	return []string{s.row.RowID}, nil
}
func (s *runtimeMutationMemoryStore) LockCommand(context.Context, string) (persistence.RuntimeRecordCommandRow, error) {
	return s.row, nil
}
func (s *runtimeMutationMemoryStore) LockOperation(_ context.Context, _ string, index int) (persistence.RuntimeRecordCommandOperationRow, error) {
	return s.slots[index], nil
}
func (s *runtimeMutationMemoryStore) CompleteOperation(_ context.Context, _ string, index int, payload []byte) error {
	if index < 0 || index >= len(s.slots) || s.slots[index].State != persistence.RuntimeRecordOperationPending {
		return errors.New("complete runtime record command operation lost authority")
	}
	s.completeOperationCalls++
	s.slots[index].State = persistence.RuntimeRecordOperationDone
	s.slots[index].ResultPayload = append([]byte(nil), payload...)
	return nil
}
func (s *runtimeMutationMemoryStore) LockOperations(context.Context, string) ([]persistence.RuntimeRecordCommandOperationRow, error) {
	return append([]persistence.RuntimeRecordCommandOperationRow(nil), s.slots...), nil
}

func (s *runtimeMutationMemoryStore) LoadOperations(context.Context, string) ([]persistence.RuntimeRecordCommandOperationRow, error) {
	return append([]persistence.RuntimeRecordCommandOperationRow(nil), s.slots...), nil
}
func (s *runtimeMutationMemoryStore) CompleteCommand(_ context.Context, _ string, payload []byte, outcomeExpiresAt time.Time) error {
	s.completeCommandCalls++
	s.row.State = persistence.RuntimeRecordCommandCompleted
	s.row.ResultPayload = append([]byte(nil), payload...)
	s.row.OutcomeExpiresAt = outcomeExpiresAt
	return nil
}