package automation

// flow_preview_test.go — coverage for issue #11
// (generic no-DML record-create preview path for Flow-driven defaults).
//
// Pins the PreviewRecordCreateFields contract:
//
//  1. Captures every RecordCreate's resolved field map without ever calling
//     PersistenceProvider.Insert on the wrapped (real) provider — i.e. zero
//     DML reaches the DB.
//  2. Routes RecordUpdate / RecordDelete through fault path (or surfaces them
//     as flow-runtime errors when no fault connector is wired) so the preview
//     never mutates state via any code path.
//  3. Shares the full executor pipeline with live runs — InputAssignment
//     literal resolution, ObjectFieldMapping, and stage-3 ActionFieldDefault
//     all run inside the preview, so callers see the same defaulted values a
//     subsequent real run would write.
//  4. Honors authorizeProtectedStandardFlow so a caller without permission to
//     run a protected/standard flow gets the same denial they would on a real
//     run.

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	seedrows "github.com/pingcap-inc/nexuscrm-contract/pkg/seedrows"
)

// recordingPersistence captures every PersistenceProvider call so a preview
// run can be asserted against a strict zero-write contract on the inner
// (wrapped) provider.
type recordingPersistence struct {
	insertCalls        int
	previewInsertCalls int
	updateCalls        int
	previewUpdateCalls int
	deleteCalls        int
	previewInsertFn    func(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error)
	previewUpdateFn    func(ctx context.Context, tableName string, id string, updates models.SObject, internalStamps models.SObject, user *models.UserSession) (*models.PreparedRecordUpdate, error)

	lastPreviewUpdateObject string
	lastPreviewUpdateID     string
	lastPreviewUpdates      models.SObject
	lastPreviewInternal     models.SObject
	previewUpdatePrepared   *models.PreparedRecordUpdate
	lastCommitObject        string
	lastCommitID            string
	lastCommitExtraUpdates  models.SObject
}

func (r *recordingPersistence) Insert(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
	r.insertCalls++
	out := models.SObject{}
	for k, v := range record {
		out[k] = v
	}
	out["Id"] = "should-not-be-used"
	return out, nil
}
func (r *recordingPersistence) PreviewBeforeInsert(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
	r.previewInsertCalls++
	if r.previewInsertFn != nil {
		return r.previewInsertFn(ctx, tableName, record, user)
	}
	out := models.SObject{}
	for k, v := range record {
		out[k] = v
	}
	return out, nil
}
func (r *recordingPersistence) PrelockUpdateParents(ctx context.Context, tableName string, id string, updates models.SObject, user *models.UserSession) error {
	r.previewUpdateCalls++
	r.lastPreviewUpdateObject = tableName
	r.lastPreviewUpdateID = id
	r.lastPreviewUpdates = updates
	return nil
}
func (r *recordingPersistence) Update(ctx context.Context, tableName string, id string, updates models.SObject, user *models.UserSession) error {
	r.updateCalls++
	return nil
}
func (r *recordingPersistence) PreviewBeforeUpdate(ctx context.Context, tableName string, id string, updates models.SObject, internalStamps models.SObject, user *models.UserSession) (*models.PreparedRecordUpdate, error) {
	r.previewUpdateCalls++
	if r.previewUpdateFn != nil {
		return r.previewUpdateFn(ctx, tableName, id, updates, internalStamps, user)
	}
	r.lastPreviewUpdateObject = tableName
	r.lastPreviewUpdateID = id
	r.lastPreviewUpdates = updates
	r.lastPreviewInternal = internalStamps
	if r.previewUpdatePrepared != nil {
		return r.previewUpdatePrepared, nil
	}
	return &models.PreparedRecordUpdate{}, nil
}
func (r *recordingPersistence) CommitPreparedUpdate(ctx context.Context, tableName string, id string, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject, user *models.UserSession) error {
	r.updateCalls++
	r.lastCommitObject = tableName
	r.lastCommitID = id
	r.lastCommitExtraUpdates = extraUpdates
	return nil
}
func (r *recordingPersistence) Delete(ctx context.Context, tableName string, id string, user *models.UserSession) error {
	r.deleteCalls++
	return nil
}
func (r *recordingPersistence) GetRaw(ctx context.Context, objectName string, id string) (models.SObject, error) {
	return nil, nil
}
func (r *recordingPersistence) GetRecordType(ctx context.Context, id string) (string, error) {
	return "", nil
}
func (r *recordingPersistence) GetLock(ctx context.Context, tableName string, id string) (models.SObject, error) {
	return nil, nil
}
func (r *recordingPersistence) RunInTransaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}
func (r *recordingPersistence) IsTransactionActive(context.Context) bool { return false }
func (r *recordingPersistence) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}
func (r *recordingPersistence) DetachTx(ctx context.Context) context.Context { return ctx }
func (r *recordingPersistence) RecomputeActiveFormulaProjection(_ context.Context, _ *models.UserSession, _, _ string, _ []string) (models.SObject, error) {
	return nil, nil
}

// stringValue wraps a literal in the Salesforce typed-value envelope used by
// flow InputAssignment — same shape resolveValueCtx already understands.
func stringValue(v string) interface{} { return map[string]interface{}{"stringValue": v} }

func elementReferenceValue(ref string) interface{} {
	return map[string]interface{}{"elementReference": ref}
}

// previewExecutor builds an in-memory FlowExecutor + recordingPersistence
// pair ready to host a single PreviewRecordCreateFields call against
// flowApiName="PreviewFlow". The executor's flowVersionFn returns flowMeta
// directly so tests need no QueryService wiring.
func previewExecutor(t *testing.T, flowMeta models.FlowMetadata) (*FlowExecutor, *recordingPersistence) {
	t.Helper()
	rp := &recordingPersistence{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:          rp,
		query:                nil,
		fieldMappings:        nil,
		logger:               logging.New(context.Background()),
		formulaEngine:        eng,
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(eng),
		beforeCreatePreviewSafeFn: func(ctx context.Context, objectAPIName string, user *models.UserSession) (bool, error) {
			return true, nil
		},
	}
	exec.interviews = NewFlowInterviewService(rp, nil)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       "custom",
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}
	exec.RegisterFlowActionCalls(testRuntimeFlowActionCallRegistrations()...)
	return exec, rp
}

func newPreviewUser() *models.UserSession {
	return &models.UserSession{ID: "preview-user-1"}
}

// flowMetaWithSingleRecordCreate builds a minimal flow shape:
//
//	Start → "create_thing" (RecordCreate on `object`) → end
//
// with InputAssignments stamping (field=>value) literal pairs.
func flowMetaWithSingleRecordCreate(object string, fields map[string]string) models.FlowMetadata {
	assigns := make([]models.FlowInputAssignment, 0, len(fields))
	for k, v := range fields {
		assigns = append(assigns, models.FlowInputAssignment{Field: k, Value: stringValue(v)})
	}
	return models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_thing"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:             "create_thing",
				Object:           object,
				InputAssignments: assigns,
			},
		},
	}
}

// ---------------------------------------------------------------------------
// AC.1 — RecordCreate is captured; zero DML reaches the inner persistence.
// ---------------------------------------------------------------------------

func TestPreviewRecordCreateFields_CapturesIntent_NoDML(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"FieldOne": "alpha",
		"FieldTwo": "beta",
	})
	exec, rp := previewExecutor(t, flowMeta)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; errors=%+v", result.Errors)
	}
	if got := len(result.Creates); got != 1 {
		t.Fatalf("len(Creates) = %d; want 1 (Creates=%+v)", got, result.Creates)
	}
	create := result.Creates[0]
	if create.Object != "PreviewObj" {
		t.Errorf("create.Object = %q; want %q", create.Object, "PreviewObj")
	}
	if got := create.Fields["FieldOne"]; got != "alpha" {
		t.Errorf("Fields[FieldOne] = %v; want alpha", got)
	}
	if got := create.Fields["FieldTwo"]; got != "beta" {
		t.Errorf("Fields[FieldTwo] = %v; want beta", got)
	}
	if rp.previewInsertCalls != 1 {
		t.Errorf("inner persistence.PreviewBeforeInsert calls = %d; want 1", rp.previewInsertCalls)
	}

	// The whole point: not a single Insert/Update/Delete reaches the wrapped
	// persistence. If any does, the preview path is leaking DML.
	if rp.insertCalls != 0 {
		t.Errorf("inner persistence.Insert calls = %d; want 0 (preview must perform no DML)", rp.insertCalls)
	}
	if rp.updateCalls != 0 {
		t.Errorf("inner persistence.Update calls = %d; want 0", rp.updateCalls)
	}
	if rp.deleteCalls != 0 {
		t.Errorf("inner persistence.Delete calls = %d; want 0", rp.deleteCalls)
	}
}

func TestPreparedStandardActionPreviewUsesExactClosureInsteadOfLiveVersion(t *testing.T) {
	preparedMetadata := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"Authority": "prepared",
	})
	exec, _ := previewExecutor(t, flowMetaWithSingleRecordCreate(
		"PreviewObj", map[string]string{"Authority": "live"},
	))
	liveReads := 0
	exec.flowVersionFn = func(
		context.Context,
		string,
		*models.UserSession,
	) (*models.FlowVersion, error) {
		liveReads++
		return &models.FlowVersion{
			Id: "live-version", DefinitionId: "live-definition",
			Status: models.FlowStatusActive,
			Metadata: flowMetaWithSingleRecordCreate(
				"PreviewObj", map[string]string{"Authority": "live"},
			),
		}, nil
	}
	closure := map[string]*models.FlowVersion{
		"PreviewFlow": {
			Id: "catalog-version", DefinitionId: "catalog-definition",
			Source: flowSourceStandard, Status: models.FlowStatusActive,
			Metadata: preparedMetadata,
		},
	}
	ctx := constants.WithStandardActionExecutionRootFlow(
		t.Context(), "standard-action-1", "PreviewFlow",
	)

	supported, err := exec.PreparedStandardActionCreatePreviewSupported(
		ctx, "PreviewFlow", closure, newPreviewUser(),
	)
	if err != nil || !supported {
		t.Fatalf("prepared support = %v, %v", supported, err)
	}
	result, err := exec.PreviewPreparedStandardActionRecordCreateFields(
		ctx, "PreviewFlow", closure, map[string]interface{}{}, newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("prepared preview: %v", err)
	}
	if liveReads != 0 {
		t.Fatalf("live active-version reads = %d, want 0", liveReads)
	}
	if result == nil || len(result.Creates) != 1 ||
		result.Creates[0].Fields["Authority"] != "prepared" {
		t.Fatalf("prepared preview result = %#v", result)
	}
}

func TestPreparedStandardActionPreviewRejectsMissingChildClosure(t *testing.T) {
	metadata := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "child"},
		},
		Subflows: []models.FlowSubflow{{
			Name: "child", FlowName: "MissingChild",
		}},
	}
	exec, _ := previewExecutor(t, metadata)
	closure := map[string]*models.FlowVersion{
		"PreviewFlow": {
			Id: "catalog-version", DefinitionId: "catalog-definition",
			Status: models.FlowStatusActive, Metadata: metadata,
		},
	}
	result, err := exec.PreviewPreparedStandardActionRecordCreateFields(
		t.Context(), "PreviewFlow", closure, map[string]interface{}{}, newPreviewUser(),
	)
	if err != nil || result == nil || result.IsSuccess || len(result.Errors) == 0 ||
		!strings.Contains(result.Errors[0].Message, ErrPreparedFlowClosureMissing.Error()) {
		t.Fatalf("missing prepared child result = %#v, error = %v", result, err)
	}
}

func TestPreparedStandardActionOutputPreviewAuthorizesProtectedRoot(t *testing.T) {
	metadata := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "preview_source"},
		},
		ActionCalls: []models.FlowActionCall{{
			Name:       "preview_source",
			ActionType: "previewRecordUpdate",
			InputParameters: []models.FlowActionParameter{
				{Name: "object", Value: stringValue("Lead")},
				{Name: "recordId", Value: stringValue("lead-1")},
				{Name: "updates", Value: map[string]interface{}{
					"Status": stringValue("Closed - Converted"),
				}},
			},
		}},
	}
	exec, persistence := previewExecutor(t, metadata)
	closure := map[string]*models.FlowVersion{
		"PreviewFlow": {
			Id: "catalog-version", DefinitionId: "catalog-definition",
			Source: flowSourceStandard, Status: models.FlowStatusActive,
			Metadata: metadata,
		},
	}
	ctx := constants.WithStandardActionExecutionRootFlow(
		t.Context(), "standard-action-1", "PreviewFlow",
	)

	result, err := exec.PreviewPreparedFlowOutputs(
		ctx, "PreviewFlow", closure, map[string]interface{}{}, newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("prepared output preview: %v", err)
	}
	if result == nil || !result.IsSuccess || len(result.Errors) != 0 {
		t.Fatalf("prepared output preview result = %#v, want success", result)
	}
	if persistence.previewUpdateCalls != 1 {
		t.Fatalf("PreviewBeforeUpdate calls = %d, want 1", persistence.previewUpdateCalls)
	}
	if persistence.insertCalls+persistence.updateCalls+persistence.deleteCalls != 0 {
		t.Fatalf(
			"live persistence saw %d Insert / %d Update / %d Delete, want 0/0/0",
			persistence.insertCalls, persistence.updateCalls, persistence.deleteCalls,
		)
	}
}

func TestPreparedStandardActionOutputPreviewRejectsProtectedRootWithoutActionAuthority(t *testing.T) {
	metadata := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "preview_source"},
		},
		ActionCalls: []models.FlowActionCall{{
			Name:       "preview_source",
			ActionType: "previewRecordUpdate",
		}},
	}
	exec, persistence := previewExecutor(t, metadata)
	closure := map[string]*models.FlowVersion{
		"PreviewFlow": {
			Id: "catalog-version", DefinitionId: "catalog-definition",
			Source: flowSourceStandard, Status: models.FlowStatusActive,
			Metadata: metadata,
		},
	}

	result, err := exec.PreviewPreparedFlowOutputs(
		t.Context(), "PreviewFlow", closure, map[string]interface{}{}, newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("prepared output preview: %v", err)
	}
	if result == nil || result.IsSuccess || len(result.Errors) != 1 ||
		result.Errors[0].StatusCode != "FLOW_EXECUTION_ERROR" {
		t.Fatalf("prepared output preview result = %#v, want protected-root denial", result)
	}
	if persistence.previewUpdateCalls != 0 ||
		persistence.insertCalls+persistence.updateCalls+persistence.deleteCalls != 0 {
		t.Fatalf(
			"denied preview reached persistence: preview=%d insert=%d update=%d delete=%d",
			persistence.previewUpdateCalls, persistence.insertCalls,
			persistence.updateCalls, persistence.deleteCalls,
		)
	}
}

func TestPreparedStandardActionPreviewSupportUsesExactChildClosureWithoutRepository(t *testing.T) {
	rootMetadata := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "child"},
		},
		Subflows: []models.FlowSubflow{{
			Name: "child", FlowName: "PreparedChild",
		}},
	}
	childMetadata := flowMetaWithSingleRecordCreate(
		"PreviewObj", map[string]string{"Authority": "prepared-child"},
	)
	exec, _ := previewExecutor(t, rootMetadata)
	exec.flowVersionFn = nil
	closure := map[string]*models.FlowVersion{
		"PreviewFlow": {
			Id: "catalog-root", DefinitionId: "catalog-root-definition",
			Status: models.FlowStatusActive, Metadata: rootMetadata,
		},
		"PreparedChild": {
			Id: "catalog-child", DefinitionId: "catalog-child-definition",
			Status: models.FlowStatusActive, Metadata: childMetadata,
		},
	}
	supported, err := exec.PreparedStandardActionCreatePreviewSupported(
		t.Context(), "PreviewFlow", closure, newPreviewUser(),
	)
	if err != nil || !supported {
		t.Fatalf("prepared child support = %v, %v", supported, err)
	}
}

func TestPreviewRecordCreateFields_AllowsAtomicHistoryPrelockMetadata(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{"FieldOne": "alpha"})
	flowMeta.RunAtomically = true
	flowMeta.AtomicParentPrelocks = []models.FlowAtomicParentPrelock{{
		Name:         "AccountRows",
		ParentObject: "Account",
		RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "account-1"}},
	}}
	flowMeta.AtomicHistoryPrelocks = []models.FlowAtomicHistoryPrelock{{
		Object: "Account",
		Fields: []string{"Name"},
	}}
	exec, rp := previewExecutor(t, flowMeta)
	exec.query = &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	exec.fieldMetaLookup = func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{ID: "fld-account-name", APIName: "Name"}
	}

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result == nil || !result.IsSuccess || len(result.Creates) != 1 {
		t.Fatalf("preview result = %#v, want one successful no-DML create", result)
	}
	if rp.insertCalls != 0 || rp.updateCalls != 0 || rp.deleteCalls != 0 {
		t.Fatalf("preview leaked DML: insert=%d update=%d delete=%d", rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_AllowsDurableLookupBeforeCreate(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "lookup_source"},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:               "lookup_source",
			Object:             "SourceObj",
			GetFirstRecordOnly: true,
			OutputReference:    "source",
			Filters: []models.FlowRecordFilter{{
				Field:    "Id",
				Operator: "EqualTo",
				Value:    elementReferenceValue("sourceRecordId"),
			}},
			Connector: &models.FlowConnector{TargetReference: "create_target"},
		}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:   "create_target",
			Object: "TargetObj",
			InputAssignments: []models.FlowInputAssignment{
				{Field: "Name", Value: elementReferenceValue("source.Name")},
			},
		}},
	}
	exec, rp := previewExecutor(t, flowMeta)
	query := &limitCapturingQuery{result: &models.QueryResult{
		TotalSize: 1,
		Done:      true,
		Records: []models.SObject{{
			constants.FieldID: "source-1",
			"Name":            "Source Name",
		}},
	}}
	exec.query = query
	exec.interviews = NewFlowInterviewService(rp, query)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", map[string]interface{}{
		"sourceRecordId": "source-1",
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; errors=%+v", result.Errors)
	}
	if got := len(result.Creates); got != 1 {
		t.Fatalf("len(Creates) = %d; want 1", got)
	}
	if got := result.Creates[0].Fields["Name"]; got != "Source Name" {
		t.Fatalf("preview Name = %v; want Source Name from lookup-before-create", got)
	}
	if rp.insertCalls != 0 {
		t.Errorf("inner persistence.Insert calls = %d; want 0", rp.insertCalls)
	}
}

func TestPreviewRecordCreateFields_UsesNoDMLInsertPreparedImage(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"FieldOne": "raw",
	})
	exec, rp := previewExecutor(t, flowMeta)
	rp.previewInsertFn = func(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
		out := models.SObject{}
		for k, v := range record {
			out[k] = v
		}
		out[constants.FieldID] = "real-looking-id"
		out["FieldOne"] = "prepared"
		out["DefaultedByPersistence"] = "schema-default"
		return out, nil
	}

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; errors=%+v", result.Errors)
	}
	if got := len(result.Creates); got != 1 {
		t.Fatalf("len(Creates) = %d; want 1", got)
	}
	create := result.Creates[0]
	if got := create.Fields["FieldOne"]; got != "prepared" {
		t.Errorf("Fields[FieldOne] = %v; want prepared image from PreviewBeforeInsert", got)
	}
	if got := create.Fields["DefaultedByPersistence"]; got != "schema-default" {
		t.Errorf("Fields[DefaultedByPersistence] = %v; want schema-default", got)
	}
	if _, ok := create.Fields[constants.FieldID]; ok {
		t.Errorf("Fields unexpectedly exposed durable Id: %+v", create.Fields[constants.FieldID])
	}
	if rp.insertCalls != 0 {
		t.Errorf("inner persistence.Insert calls = %d; want 0", rp.insertCalls)
	}
	if rp.previewInsertCalls != 1 {
		t.Errorf("inner persistence.PreviewBeforeInsert calls = %d; want 1", rp.previewInsertCalls)
	}
}

func TestPreviewRecordCreateFields_SyntheticParentVisibleToChildPrepare(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Parent")},
				},
				Connector: &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("parentId")},
					{Field: "Name", Value: stringValue("Preview Child")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)
	// Preview IDs are unguessable (crypto/rand hex), so the test asserts the
	// shape and the same-run propagation, not a deterministic literal. The
	// child's ParentId must equal the parent's freshly minted preview ID,
	// discovered by reading the active synthetic-record registry from ctx
	// at child-prepare time.
	rp.previewInsertFn = func(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
		if tableName == "ChildObj" {
			parentID, _ := record["ParentId"].(string)
			if !strings.HasPrefix(parentID, "preview-") {
				t.Fatalf("child ParentId = %q; want preview-* synthetic", parentID)
			}
			parent, ok := constants.PreviewSyntheticRecordFromCtx(ctx, parentID)
			if !ok {
				t.Fatalf("synthetic parent %q was not registered for child prepare", parentID)
			}
			if parent.ObjectName != "ParentObj" {
				t.Fatalf("synthetic parent object = %q; want ParentObj", parent.ObjectName)
			}
		}
		out := models.SObject{}
		for k, v := range record {
			out[k] = v
		}
		return out, nil
	}

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("result success = %#v; want success", result)
	}
	if got := len(result.Creates); got != 2 {
		t.Fatalf("len(Creates) = %d; want parent and child", got)
	}
	parentPreviewID := result.Creates[0].PreviewID
	if !strings.HasPrefix(parentPreviewID, "preview-") || len(parentPreviewID) != len("preview-")+32 {
		t.Fatalf("parent PreviewID = %q; want preview-<32 hex chars>", parentPreviewID)
	}
	if got := result.Creates[1].Fields["ParentId"]; got != parentPreviewID {
		t.Fatalf("child preview ParentId = %v; want parent's preview ID %q", got, parentPreviewID)
	}
}

// ---------------------------------------------------------------------------
// AC.2 — RecordUpdate inside the previewed flow surfaces as a flow error;
//        any prior RecordCreate is still captured; inner persistence is
//        never touched.
// ---------------------------------------------------------------------------

func TestPreviewRecordCreateFields_RecordUpdate_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_thing"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_thing",
				Object: "PreviewObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("captured-before-update")},
				},
				Connector: &models.FlowConnector{TargetReference: "update_thing"},
			},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:   "update_thing",
				Object: "PreviewObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("would-have-overwritten")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	// Provide a $Record so executeRecordUpdate's implicit-$Record path
	// reaches persistence.Update — without this the element is a no-op
	// (no inputReference, no filters, no $Record in context).
	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow",
		map[string]interface{}{
			"$Record": models.SObject{"Id": "fake-target-1", "attributes": map[string]interface{}{"type": "PreviewObj"}},
		},
		newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because flow contains a recordUpdate")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error in result.Errors")
	}
	joined := result.Errors[0].Message
	if !strings.Contains(joined, "recordUpdate is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'recordUpdate is not allowed'", joined)
	}
	if got := len(result.Creates); got != 1 {
		t.Errorf("len(Creates) = %d; want 1 (the create-before-update intent should still be captured)", got)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewStandardActionRecordCreateFields_PreparesPreviewRecordUpdateAction(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_account",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Account")},
				},
				Connector: &models.FlowConnector{TargetReference: "preview_source"},
			},
			{
				Name:   "create_contact",
				Object: "Contact",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "LastName", Value: stringValue("Preview Contact")},
				},
			},
		},
		ActionCalls: []models.FlowActionCall{
			{
				Name:       "preview_source",
				ActionType: "previewRecordUpdate",
				InputParameters: []models.FlowActionParameter{
					{Name: "object", Value: stringValue("Account")},
					{Name: "recordId", Value: stringValue("existing-account-1")},
					{Name: "updates", Value: map[string]interface{}{
						"Description": stringValue("prepared through preview action"),
					}},
				},
				Connector: &models.FlowConnector{TargetReference: "create_contact"},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}
	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")

	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; errors=%#v", result.Errors)
	}
	if len(result.Errors) != 0 {
		t.Fatalf("errors = %#v, want none", result.Errors)
	}
	if got := len(result.Creates); got != 2 {
		t.Fatalf("len(Creates) = %d; want both creates around previewRecordUpdate", got)
	}
	if rp.previewUpdateCalls != 1 {
		t.Fatalf("PreviewBeforeUpdate calls = %d; want 1", rp.previewUpdateCalls)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewStandardActionRecordCreateFields_ProtectedFinalizerActionFailsClosed(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_account",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Account")},
				},
				Connector: &models.FlowConnector{TargetReference: "stamp_source"},
			},
			{
				Name:   "create_contact",
				Object: "Contact",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "LastName", Value: stringValue("Preview Contact")},
				},
			},
		},
		ActionCalls: []models.FlowActionCall{
			{
				Name:       "stamp_source",
				ActionType: "stampConvertedAnchors",
				Connector:  &models.FlowConnector{TargetReference: "create_contact"},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}
	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")

	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatalf("IsSuccess=true; want false because unsupported protected finalizers must not be silently skipped")
	}
	if len(result.Errors) == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "actionCall is not allowed") {
		t.Fatalf("Errors[0].Message = %q; want actionCall preview rejection", result.Errors[0].Message)
	}
	if got := len(result.Creates); got != 1 {
		t.Fatalf("len(Creates) = %d; want only the create before the rejected finalizer", got)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFieldsRejectsEmailIOActionCallsBeforeProviderSend(t *testing.T) {
	for _, actionType := range []string{flowSendEmailActionType, "emailAlert", "emailSimple"} {
		t.Run(actionType, func(t *testing.T) {
			exec, sender := previewExecutorWithRecordingEmailSender(t, previewEmailActionFlow(actionType, false))

			result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
			if err != nil {
				t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
			}
			if result == nil {
				t.Fatal("result was nil")
			}
			if result.IsSuccess {
				t.Fatalf("IsSuccess=true; want false because preview must block %s notification side effects", actionType)
			}
			if len(result.Errors) == 0 {
				t.Fatal("expected at least one preview email-I/O error")
			}
			message := result.Errors[0].Message
			if !strings.Contains(message, actionType) {
				t.Fatalf("Errors[0].Message = %q; want rejection for %s", message, actionType)
			}
			if strings.EqualFold(actionType, "emailAlert") {
				for _, want := range []string{"WorkflowAlert runtime dispatch is disabled", "Flow sendEmail", "remove it"} {
					if !strings.Contains(message, want) {
						t.Fatalf("Errors[0].Message = %q; want emailAlert cutover guidance mentioning %q", message, want)
					}
				}
			} else if !strings.Contains(message, "email provider I/O") {
				t.Fatalf("Errors[0].Message = %q; want preview email-I/O rejection for %s", message, actionType)
			}
			if strings.Contains(message, "person@example.com") || strings.Contains(message, "Preview Subject") {
				t.Fatalf("Errors[0].Message = %q; must not include recipient or rendered content", message)
			}
			if len(sender.messages) != 0 {
				t.Fatalf("email sender calls = %d; want 0", len(sender.messages))
			}
		})
	}
}

func TestPreviewRecordCreateFieldsRoutesEmailIOFaultConnectorBeforeProviderSend(t *testing.T) {
	for _, actionType := range []string{flowSendEmailActionType, "emailAlert", "emailSimple"} {
		t.Run(actionType, func(t *testing.T) {
			exec, sender := previewExecutorWithRecordingEmailSender(t, previewEmailActionFlow(actionType, true))

			result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
			if err != nil {
				t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
			}
			if result == nil {
				t.Fatal("result was nil")
			}
			if !result.IsSuccess {
				t.Fatalf("IsSuccess=false; want fault connector to route preview email-I/O rejection successfully: %#v", result.Errors)
			}
			if len(result.Errors) != 0 {
				t.Fatalf("Errors = %#v; want none after fault connector handles preview email-I/O rejection", result.Errors)
			}
			if len(sender.messages) != 0 {
				t.Fatalf("email sender calls = %d; want 0", len(sender.messages))
			}
		})
	}
}

func TestPreviewRecordCreateFieldsRejectsEmailIOInsideAllowedSubflowBeforeProviderSend(t *testing.T) {
	parentMeta := models.FlowMetadata{
		Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "call_child"}},
		ActionCalls: []models.FlowActionCall{{
			Name:       "call_child",
			ActionName: "ChildEmailFlow",
			ActionType: "flow",
		}},
	}
	childMeta := previewEmailActionFlow(flowSendEmailActionType, false)
	exec, sender := previewExecutorWithRecordingEmailSender(t, parentMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		meta := parentMeta
		if apiName == "ChildEmailFlow" {
			meta = childMeta
		}
		return &models.FlowVersion{
			Id:           "preview-flow-version-" + apiName,
			DefinitionId: "preview-flow-def-" + apiName,
			Source:       "custom",
			Status:       "Active",
			Metadata:     meta,
		}, nil
	}

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatal("IsSuccess=true; want child sendEmail to be blocked under preview context")
	}
	if len(result.Errors) == 0 {
		t.Fatal("expected child preview email-I/O error")
	}
	if message := result.Errors[0].Message; !strings.Contains(message, "email provider I/O") {
		t.Fatalf("Errors[0].Message = %q; want child preview email-I/O rejection", message)
	}
	if len(sender.messages) != 0 {
		t.Fatalf("email sender calls = %d; want 0", len(sender.messages))
	}
}

func previewExecutorWithRecordingEmailSender(t *testing.T, flowMeta models.FlowMetadata) (*FlowExecutor, *recordingEmailSender) {
	t.Helper()
	exec, _ := previewExecutor(t, flowMeta)
	sender := &recordingEmailSender{}
	exec.actions = &ActionService{
		persistence: exec.persistence,
		emailSender: sender,
		logger:      logging.New(context.Background()),
	}
	return exec, sender
}

func previewEmailActionFlow(actionType string, withFaultConnector bool) models.FlowMetadata {
	action := previewEmailActionCall(actionType)
	if withFaultConnector {
		action.FaultConnector = &models.FlowConnector{TargetReference: "fault_done"}
	}
	meta := models.FlowMetadata{
		Start:       &models.FlowStart{Connector: &models.FlowConnector{TargetReference: action.Name}},
		ActionCalls: []models.FlowActionCall{action},
	}
	if withFaultConnector {
		meta.Assignments = []models.FlowAssignment{{Name: "fault_done"}}
	}
	return meta
}

func previewEmailActionCall(actionType string) models.FlowActionCall {
	actionName := "PreviewEmail"
	if strings.EqualFold(actionType, "emailAlert") {
		actionName = "Account.PreviewEmail"
	}
	return models.FlowActionCall{
		Name:       "send_preview_email",
		ActionName: actionName,
		ActionType: actionType,
		InputParameters: []models.FlowActionParameter{
			{Name: flowSendEmailTemplateIDParam, Value: "tmpl-preview"},
			{Name: constants.ConfigTo, Value: "person@example.com"},
			{Name: constants.ConfigVariables, Value: map[string]interface{}{"Subject": "Preview Subject"}},
			{Name: "emailTemplate", Value: "tmpl-preview"},
			{Name: "recipient", Value: "person@example.com"},
		},
	}
}

func TestStandardActionCreatePreviewSupportedRejectsUnsupportedProtectedFinalizer(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{{
			Name:      "create_account",
			Object:    "Account",
			Connector: &models.FlowConnector{TargetReference: "stamp_source"},
		}},
		ActionCalls: []models.FlowActionCall{{
			Name:       "stamp_source",
			ActionType: "stampConvertedAnchors",
		}},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for unsupported protected finalizer")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsRecordUpdate(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "update_existing"},
		},
		RecordUpdates: []models.FlowRecordUpdate{{
			Name:   "update_existing",
			Object: "Account",
			InputAssignments: []models.FlowInputAssignment{
				{Field: "Description", Value: stringValue("not previewable yet")},
			},
		}},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for recordUpdate")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsExternalIDUpsert(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("Lead", map[string]string{
		"Salesforce_Identity_Key__c": "salesforce:00D000000000001AAA:Lead:00Q000000000001AAA",
	})
	flowMeta.RecordCreates[0].UpsertByExternalIDField = "Salesforce_Identity_Key__c"
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for atomic external-ID upsert")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsPreviewRecordUpdate(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "preview_source"},
		},
		ActionCalls: []models.FlowActionCall{{
			Name:       "preview_source",
			ActionType: "previewRecordUpdate",
		}},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for previewRecordUpdate")
	}
}

func TestStandardActionCreatePreviewSupportedAcceptsCreateOnlyFlow(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("Account", map[string]string{"Name": "Preview Account"})
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if !ok {
		t.Fatal("StandardActionCreatePreviewSupported = false, want true for create-only flow")
	}
}

func TestStandardActionCreatePreviewSupportedAcceptsLookupBeforeCreate(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "lookup_source"},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:               "lookup_source",
			Object:             "SourceObj",
			GetFirstRecordOnly: true,
			OutputReference:    "source",
			Filters: []models.FlowRecordFilter{{
				Field:    "Id",
				Operator: "EqualTo",
				Value:    elementReferenceValue("sourceRecordId"),
			}},
			Connector: &models.FlowConnector{TargetReference: "create_target"},
		}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:   "create_target",
			Object: "TargetObj",
			InputAssignments: []models.FlowInputAssignment{
				{Field: "Name", Value: elementReferenceValue("source.Name")},
			},
		}},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if !ok {
		t.Fatal("StandardActionCreatePreviewSupported = false, want true for lookup-before-create graph")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsLookupAfterCreate(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_account",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Account")},
				},
				Connector: &models.FlowConnector{TargetReference: "lookup_account"},
			},
			{
				Name:   "create_contact",
				Object: "Contact",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "AccountId", Value: elementReferenceValue("foundAccount.Id")},
					{Field: "LastName", Value: stringValue("Preview Contact")},
				},
			},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:               "lookup_account",
			Object:             "Account",
			GetFirstRecordOnly: true,
			OutputReference:    "foundAccount",
			Filters: []models.FlowRecordFilter{{
				Field:    "Name",
				Operator: "EqualTo",
				Value:    stringValue("Preview Account"),
			}},
			Connector: &models.FlowConnector{TargetReference: "create_contact"},
		}},
	}
	exec, rp := previewExecutor(t, flowMeta)
	query := &limitCapturingQuery{result: &models.QueryResult{TotalSize: 0, Done: true}}
	exec.query = query
	exec.interviews = NewFlowInterviewService(rp, query)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for lookup-after-create graph")
	}
}

type previewProjectingQuery struct {
	limitCapturingQuery
	projectCalls int
	projectFn    func(ctx context.Context, objectName string, records []models.SObject, currentUser *models.UserSession) ([]models.SObject, error)
}

func (q *previewProjectingQuery) ProjectPreviewSyntheticRecords(ctx context.Context, objectName string, records []models.SObject, currentUser *models.UserSession) ([]models.SObject, error) {
	q.projectCalls++
	if q.projectFn != nil {
		return q.projectFn(ctx, objectName, records, currentUser)
	}
	out := make([]models.SObject, len(records))
	for i, record := range records {
		row := make(models.SObject, len(record))
		for k, v := range record {
			row[k] = v
		}
		out[i] = row
	}
	return out, nil
}

func TestMergePreviewSyntheticLookupRecords_ReplacesExistingSyntheticRow(t *testing.T) {
	ctx := constants.WithPreviewSyntheticRecordRegistry(context.Background())
	constants.RegisterPreviewSyntheticRecord(ctx, "preview-Account-1", "Account", map[string]interface{}{
		constants.FieldID: "preview-Account-1",
		"Name":            "Fresh Account",
	})

	query := &previewProjectingQuery{
		projectFn: func(ctx context.Context, objectName string, records []models.SObject, currentUser *models.UserSession) ([]models.SObject, error) {
			if objectName != "Account" {
				t.Fatalf("ProjectPreviewSyntheticRecords objectName = %q, want Account", objectName)
			}
			if len(records) != 1 || records[0][constants.FieldID] != "preview-Account-1" {
				t.Fatalf("ProjectPreviewSyntheticRecords records = %#v, want preview-Account-1", records)
			}
			return []models.SObject{{
				constants.FieldID: "preview-Account-1",
				"Name":            "Fresh Account",
				"Projected__c":    "yes",
			}}, nil
		},
	}
	exec := &FlowExecutor{query: query}
	result := &models.QueryResult{
		TotalSize: 1,
		Done:      true,
		Records: []models.SObject{{
			constants.FieldID: "preview-Account-1",
			"Name":            "Stale Account",
			"Projected__c":    "no",
		}},
	}
	lookup := &models.FlowRecordLookup{
		Object: "Account",
		Filters: []models.FlowRecordFilter{{
			Field:    "Name",
			Operator: "EqualTo",
			Value:    stringValue("Fresh Account"),
		}},
	}

	got, err := exec.mergePreviewSyntheticLookupRecords(ctx, nil, lookup, newPreviewUser(), result, 0)
	if err != nil {
		t.Fatalf("mergePreviewSyntheticLookupRecords() error = %v", err)
	}
	if got == nil || len(got.Records) != 1 {
		t.Fatalf("merged records = %#v, want one replaced row", got)
	}
	if got.Records[0]["Name"] != "Fresh Account" || got.Records[0]["Projected__c"] != "yes" {
		t.Fatalf("merged row = %#v, want projected synthetic replacement", got.Records[0])
	}
	if got.TotalSize != 1 {
		t.Fatalf("TotalSize = %d, want replacement to preserve total size 1", got.TotalSize)
	}
}

func TestPreviewRecordCreateFields_RecordLookupSeesProjectedSyntheticCreateByNaturalKey(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_account",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Account")},
				},
				Connector: &models.FlowConnector{TargetReference: "lookup_account"},
			},
			{
				Name:   "create_contact",
				Object: "Contact",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "AccountId", Value: elementReferenceValue("foundAccount.Id")},
					{Field: "LastName", Value: stringValue("Preview Contact")},
				},
			},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:               "lookup_account",
			Object:             "Account",
			GetFirstRecordOnly: true,
			OutputReference:    "foundAccount",
			Filters: []models.FlowRecordFilter{{
				Field:    "Name",
				Operator: "EqualTo",
				Value:    stringValue("Preview Account"),
			}},
			Connector: &models.FlowConnector{TargetReference: "create_contact"},
		}},
	}
	exec, rp := previewExecutor(t, flowMeta)
	query := &previewProjectingQuery{
		limitCapturingQuery: limitCapturingQuery{result: &models.QueryResult{TotalSize: 0, Done: true}},
	}
	exec.query = query
	exec.interviews = NewFlowInterviewService(rp, query)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if !ok {
		t.Fatal("StandardActionCreatePreviewSupported = false, want true when lookup-after-create has projected overlay support")
	}

	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")
	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("result success = %#v; want success", result)
	}
	if got := len(result.Creates); got != 2 {
		t.Fatalf("len(Creates) = %d; want account and contact", got)
	}
	parentID := result.Creates[0].PreviewID
	if parentID == "" {
		t.Fatalf("parent preview id empty: %#v", result.Creates[0])
	}
	if got := result.Creates[1].Fields["AccountId"]; got != parentID {
		t.Fatalf("contact AccountId = %v; want synthetic account Id %q", got, parentID)
	}
	if query.projectCalls != 1 {
		t.Fatalf("ProjectPreviewSyntheticRecords calls = %d; want 1", query.projectCalls)
	}
	if query.capturedLimit != 1 {
		t.Fatalf("recordLookup query limit = %d; want 1", query.capturedLimit)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_RecordLookupCannotMatchProjectedOutSyntheticField(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_account",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Account")},
					{Field: "Secret__c", Value: stringValue("hidden")},
				},
				Connector: &models.FlowConnector{TargetReference: "lookup_account"},
			},
			{
				Name:   "create_contact",
				Object: "Contact",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "AccountId", Value: elementReferenceValue("foundAccount.Id")},
					{Field: "LastName", Value: stringValue("Preview Contact")},
				},
			},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:               "lookup_account",
			Object:             "Account",
			GetFirstRecordOnly: true,
			OutputReference:    "foundAccount",
			Filters: []models.FlowRecordFilter{{
				Field:    "Secret__c",
				Operator: "EqualTo",
				Value:    stringValue("hidden"),
			}},
			Connector: &models.FlowConnector{TargetReference: "create_contact"},
		}},
	}
	exec, rp := previewExecutor(t, flowMeta)
	query := &previewProjectingQuery{
		limitCapturingQuery: limitCapturingQuery{result: &models.QueryResult{TotalSize: 0, Done: true}},
		projectFn: func(ctx context.Context, objectName string, records []models.SObject, currentUser *models.UserSession) ([]models.SObject, error) {
			out := make([]models.SObject, len(records))
			for i, record := range records {
				out[i] = models.SObject{
					constants.FieldID: record[constants.FieldID],
					"Name":            record["Name"],
				}
			}
			return out, nil
		},
	}
	exec.query = query
	exec.interviews = NewFlowInterviewService(rp, query)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")
	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("result success = %#v; want success", result)
	}
	if got := len(result.Creates); got != 2 {
		t.Fatalf("len(Creates) = %d; want account and contact", got)
	}
	if got := result.Creates[1].Fields["AccountId"]; got != nil {
		t.Fatalf("contact AccountId = %v; want nil because projected-out lookup field must not match", got)
	}
}

func TestPreviewSyntheticLookupFilterMatches_MultiPicklistCompatibility(t *testing.T) {
	ctx := context.Background()
	executor := &FlowExecutor{}

	tests := []struct {
		name          string
		record        models.SObject
		filter        models.FlowRecordFilter
		resolvedValue interface{}
		want          bool
	}{
		{
			name:   "typed Includes keeps any-token literal semantics after resolution",
			record: models.SObject{"Tags__c": "A"},
			filter: models.FlowRecordFilter{
				Field:    "Tags__c",
				Operator: "Includes",
				Value:    map[string]interface{}{"stringValue": "A;B"},
			},
			resolvedValue: "A;B",
			want:          true,
		},
		{
			name:   "typed Excludes keeps none-of literal semantics after resolution",
			record: models.SObject{"Tags__c": "A"},
			filter: models.FlowRecordFilter{
				Field:    "Tags__c",
				Operator: "Excludes",
				Value:    map[string]interface{}{"stringValue": "A;B"},
			},
			resolvedValue: "A;B",
			want:          false,
		},
		{
			name:   "legacy bare string keeps any-token semantics",
			record: models.SObject{"Tags__c": "A"},
			filter: models.FlowRecordFilter{
				Field:    "Tags__c",
				Operator: "Includes",
				Value:    "A;B",
			},
			resolvedValue: "A;B",
			want:          true,
		},
		{
			name:   "resolved elementReference string keeps tokenized semantics",
			record: models.SObject{"Tags__c": "A;B"},
			filter: models.FlowRecordFilter{
				Field:    "Tags__c",
				Operator: "Includes",
				Value:    map[string]interface{}{"elementReference": "wanted"},
			},
			resolvedValue: "A;B",
			want:          true,
		},
		{
			name:   "structured token container keeps any-token semantics",
			record: models.SObject{"Tags__c": "A"},
			filter: models.FlowRecordFilter{
				Field:    "Tags__c",
				Operator: "Includes",
				Value:    map[string]interface{}{"elementReference": "wanted"},
			},
			resolvedValue: []interface{}{"A", "B"},
			want:          true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := previewSyntheticLookupFilterMatches(ctx, executor, tt.record, tt.filter, tt.resolvedValue)
			if got != tt.want {
				t.Fatalf("previewSyntheticLookupFilterMatches() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestPreviewRecordCreateFields_CustomChildSyntheticLookupDoesNotInheritInternalReadScope(t *testing.T) {
	rootMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{{
			Name:   "create_account",
			Object: "Account",
			InputAssignments: []models.FlowInputAssignment{
				{Field: "Name", Value: stringValue("Preview Account")},
			},
			Connector: &models.FlowConnector{TargetReference: "run_custom_child"},
		}},
		ActionCalls: []models.FlowActionCall{{
			Name:       "run_custom_child",
			ActionType: "flow",
			ActionName: "CustomChild",
		}},
	}
	childMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "lookup_account"},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:               "lookup_account",
			Object:             "Account",
			GetFirstRecordOnly: true,
			OutputReference:    "foundAccount",
			Filters: []models.FlowRecordFilter{{
				Field:    "Name",
				Operator: "EqualTo",
				Value:    stringValue("Preview Account"),
			}},
		}},
	}

	exec, rp := previewExecutor(t, rootMeta)
	sawScopedRead := true
	query := &previewProjectingQuery{
		limitCapturingQuery: limitCapturingQuery{result: &models.QueryResult{TotalSize: 0, Done: true}},
		projectFn: func(ctx context.Context, objectName string, records []models.SObject, currentUser *models.UserSession) ([]models.SObject, error) {
			if objectName == "Account" {
				sawScopedRead = constants.IsScopedInternalAccess(ctx, "Account", constants.PermRead)
			}
			out := make([]models.SObject, len(records))
			for i, record := range records {
				row := make(models.SObject, len(record))
				for k, v := range record {
					row[k] = v
				}
				out[i] = row
			}
			return out, nil
		},
	}
	exec.query = query
	exec.interviews = NewFlowInterviewService(rp, query)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		switch apiName {
		case "PreviewFlow":
			return &models.FlowVersion{
				Id:           "preview-flow-version-1",
				DefinitionId: "preview-flow-def-1",
				Source:       flowSourceStandard,
				Status:       "Active",
				Metadata:     rootMeta,
			}, nil
		case "CustomChild":
			return &models.FlowVersion{
				Id:           "custom-child-version-1",
				DefinitionId: "custom-child-def-1",
				Source:       "custom",
				Status:       "Active",
				Metadata:     childMeta,
			}, nil
		default:
			return nil, nil
		}
	}
	exec.flowPackOwnedFn = func(ctx context.Context, apiName string, user *models.UserSession) (bool, error) {
		return apiName == "PreviewFlow", nil
	}

	ctx := constants.WithStandardActionInternalAccessScopes(context.Background(), "standard-action-1", []constants.StandardActionInternalAccessScope{{
		StandardActionID: "standard-action-1",
		ObjectAPIName:    "Account",
		Operation:        constants.PermRead,
	}})
	ctx = constants.WithStandardActionExecutionRootFlow(ctx, "standard-action-1", "PreviewFlow")
	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("result success = %#v; want success", result)
	}
	if query.projectCalls != 1 {
		t.Fatalf("ProjectPreviewSyntheticRecords calls = %d; want 1", query.projectCalls)
	}
	if sawScopedRead {
		t.Fatal("custom child synthetic lookup inherited standard-action internal read scope")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsUnsupportedSyntheticLookupFilterLogic(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{{
			Name:      "create_account",
			Object:    "Account",
			Connector: &models.FlowConnector{TargetReference: "lookup_account"},
		}},
		RecordLookups: []models.FlowRecordLookup{{
			Name:        "lookup_account",
			Object:      "Account",
			FilterLogic: "NOT 1",
			Filters: []models.FlowRecordFilter{{
				Field:    "Name",
				Operator: "EqualTo",
				Value:    stringValue("Preview Account"),
			}},
		}},
	}
	exec, rp := previewExecutor(t, flowMeta)
	query := &previewProjectingQuery{
		limitCapturingQuery: limitCapturingQuery{result: &models.QueryResult{TotalSize: 0, Done: true}},
	}
	exec.query = query
	exec.interviews = NewFlowInterviewService(rp, query)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for unsupported synthetic lookup filter logic")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsSyntheticCreateReference(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("parentId")},
				},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for synthetic parent lookup graph")
	}
}

func TestStandardActionCreatePreviewSupportedAcceptsSyntheticParentReferenceAssignment(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("parentId")},
				},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.fieldMetaLookup = func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
		if objectAPIName == "ChildObj" && fieldAPIName == "ParentId" {
			return &models.FieldMetadata{
				APIName:     "ParentId",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"ParentObj"},
			}
		}
		return nil
	}
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if !ok {
		t.Fatal("StandardActionCreatePreviewSupported = false, want true for synthetic parent Id assigned to metadata-proven child lookup")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsSyntheticParentOutsideWritableReferenceTargets(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("parentId")},
				},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.fieldMetaLookup = func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
		if objectAPIName == "ChildObj" && fieldAPIName == "ParentId" {
			return &models.FieldMetadata{
				APIName:             "ParentId",
				Type:                constants.FieldTypeReference,
				ReferenceTo:         []string{"ParentObj", "OtherParentObj"},
				WritableReferenceTo: []string{"OtherParentObj"},
			}
		}
		return nil
	}
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false when writable reference targets exclude synthetic parent")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsSyntheticParentDottedReferenceAssignment(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("parentId.Id")},
				},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.fieldMetaLookup = func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
		if objectAPIName == "ChildObj" && fieldAPIName == "ParentId" {
			return &models.FieldMetadata{
				APIName:     "ParentId",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"ParentObj"},
			}
		}
		return nil
	}
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for dotted synthetic Id reference")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsSyntheticCreateReferenceThroughAssignment(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "assign_temp"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("secondTempParentId")},
				},
			},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "assign_temp",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "tempParentId",
					Operator:          "Assign",
					Value:             elementReferenceValue("parentId"),
				}},
				Connector: &models.FlowConnector{TargetReference: "assign_second_temp"},
			},
			{
				Name: "assign_second_temp",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "secondTempParentId",
					Operator:          "Assign",
					Value:             elementReferenceValue("tempParentId"),
				}},
				Connector: &models.FlowConnector{TargetReference: "create_child"},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for assignment-mediated synthetic parent lookup graph")
	}
}

func TestStandardActionCreatePreviewSupportedAcceptsSyntheticParentReferenceThroughAssignment(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "assign_temp"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("secondTempParentId")},
				},
			},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "assign_temp",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "tempParentId",
					Operator:          "Assign",
					Value:             elementReferenceValue("parentId"),
				}},
				Connector: &models.FlowConnector{TargetReference: "assign_second_temp"},
			},
			{
				Name: "assign_second_temp",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "secondTempParentId",
					Operator:          "Assign",
					Value:             elementReferenceValue("tempParentId"),
				}},
				Connector: &models.FlowConnector{TargetReference: "create_child"},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.fieldMetaLookup = func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
		if objectAPIName == "ChildObj" && fieldAPIName == "ParentId" {
			return &models.FieldMetadata{
				APIName:     "ParentId",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"ParentObj"},
			}
		}
		return nil
	}
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if !ok {
		t.Fatal("StandardActionCreatePreviewSupported = false, want true for assignment-mediated synthetic parent Id assigned to child lookup")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsAmbiguousSyntheticParentReference(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_other_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_other_parent",
				Object:                    "OtherParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "create_parent"},
			},
			{
				Name:                      "create_parent",
				Object:                    "ParentObj",
				AssignRecordIdToReference: "parentId",
				Connector:                 &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentId", Value: elementReferenceValue("parentId")},
				},
			},
		},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.fieldMetaLookup = func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
		if objectAPIName == "ChildObj" && fieldAPIName == "ParentId" {
			return &models.FieldMetadata{
				APIName:     "ParentId",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"ParentObj"},
			}
		}
		return nil
	}
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false when one Flow reference can carry multiple synthetic object types")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsSyntheticCreateReferenceFromChildFlowOutput(t *testing.T) {
	parentMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "call_child"},
		},
		ActionCalls: []models.FlowActionCall{{
			Name:       "call_child",
			ActionType: "flow",
			ActionName: "ChildCreateParent",
			OutputParameters: []models.FlowActionParameter{{
				Name:  "createdParentId",
				Value: elementReferenceValue("parentId"),
			}},
			Connector: &models.FlowConnector{TargetReference: "create_child"},
		}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:   "create_child",
			Object: "ChildObj",
			InputAssignments: []models.FlowInputAssignment{
				{Field: "ParentId", Value: elementReferenceValue("parentId")},
			},
		}},
	}
	childMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		Variables: []models.FlowVariable{{
			Name:     "createdParentId",
			DataType: "String",
			IsOutput: true,
		}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:                      "create_parent",
			Object:                    "ParentObj",
			AssignRecordIdToReference: "createdParentId",
		}},
	}
	exec, _ := previewExecutor(t, parentMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		meta := parentMeta
		if apiName == "ChildCreateParent" {
			meta = childMeta
		}
		return &models.FlowVersion{
			Id:           "preview-flow-version-" + apiName,
			DefinitionId: "preview-flow-def-" + apiName,
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     meta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for child-flow synthetic output reference")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsSyntheticCreateReferenceFromSubflowOutput(t *testing.T) {
	parentMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "run_child"},
		},
		Subflows: []models.FlowSubflow{{
			Name:     "run_child",
			FlowName: "ChildCreateParent",
			OutputAssignments: []models.FlowSubflowOutput{{
				Name:              "createdParentId",
				AssignToReference: "parentId",
			}},
			Connector: &models.FlowConnector{TargetReference: "create_child"},
		}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:   "create_child",
			Object: "ChildObj",
			InputAssignments: []models.FlowInputAssignment{
				{Field: "ParentId", Value: elementReferenceValue("parentId")},
			},
		}},
	}
	childMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		Variables: []models.FlowVariable{{
			Name:     "createdParentId",
			DataType: "String",
			IsOutput: true,
		}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:                      "create_parent",
			Object:                    "ParentObj",
			AssignRecordIdToReference: "createdParentId",
		}},
	}
	exec, _ := previewExecutor(t, parentMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		meta := parentMeta
		if apiName == "ChildCreateParent" {
			meta = childMeta
		}
		return &models.FlowVersion{
			Id:           "preview-flow-version-" + apiName,
			DefinitionId: "preview-flow-def-" + apiName,
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     meta,
		}, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false for subflow synthetic output reference")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsUnsafeBeforeCreateFlow(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("Account", map[string]string{"Name": "Preview Account"})
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}
	exec.beforeCreatePreviewSafeFn = func(ctx context.Context, objectAPIName string, user *models.UserSession) (bool, error) {
		if objectAPIName != "Account" {
			t.Fatalf("before-create safety check object = %q, want Account", objectAPIName)
		}
		return false, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false when before-create flow safety is unproved")
	}
}

func TestStandardActionCreatePreviewSupportedRejectsBeforeCreateRecordLookup(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_parent",
				Object: "ParentObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ExternalKey", Value: stringValue("parent-key")},
				},
				Connector: &models.FlowConnector{TargetReference: "create_child"},
			},
			{
				Name:   "create_child",
				Object: "ChildObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "ParentExternalKey", Value: stringValue("parent-key")},
				},
			},
		},
	}
	beforeCreateMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Object:            "ChildObj",
			TriggerType:       "RecordBeforeSave",
			RecordTriggerType: "Create",
			Connector:         &models.FlowConnector{TargetReference: "find_parent"},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:   "find_parent",
			Object: "ParentObj",
			Filters: []models.FlowRecordFilter{{
				Field:    "ExternalKey",
				Operator: "EqualTo",
				Value:    elementReferenceValue("$Record.ParentExternalKey"),
			}},
		}},
	}
	exec, _ := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}
	exec.beforeCreatePreviewSafeFn = func(ctx context.Context, objectAPIName string, user *models.UserSession) (bool, error) {
		if objectAPIName == "ChildObj" {
			return beforeCreateFlowPreviewSupported(&beforeCreateMeta), nil
		}
		return true, nil
	}

	ok, err := exec.StandardActionCreatePreviewSupported(context.Background(), "PreviewFlow", newPreviewUser())
	if err != nil {
		t.Fatalf("StandardActionCreatePreviewSupported returned error: %v", err)
	}
	if ok {
		t.Fatal("StandardActionCreatePreviewSupported = true, want false when before-create flow has recordLookup")
	}
}

func TestPreviewStandardActionRecordCreateFields_PackRecordUpdateFailsClosed(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_account"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_account",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("Preview Account")},
				},
				Connector: &models.FlowConnector{TargetReference: "update_existing"},
			},
			{
				Name:   "create_contact",
				Object: "Contact",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "LastName", Value: stringValue("Preview Contact")},
				},
			},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:   "update_existing",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Description", Value: stringValue("must not be silently skipped")},
				},
				Connector: &models.FlowConnector{TargetReference: "create_contact"},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}
	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")

	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", map[string]interface{}{
		"$Record": models.SObject{"Id": "existing-account-1", "attributes": map[string]interface{}{"type": "Account"}},
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatalf("IsSuccess=true; want false because recordUpdate is not preview-safe until update validation parity lands")
	}
	if len(result.Errors) == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "recordUpdate is not allowed") {
		t.Fatalf("Errors[0].Message = %q; want recordUpdate preview rejection", result.Errors[0].Message)
	}
	if got := len(result.Creates); got != 1 {
		t.Fatalf("len(Creates) = %d; want only the create before rejected recordUpdate", got)
	}
	if rp.previewUpdateCalls != 0 {
		t.Fatalf("PreviewBeforeUpdate calls = %d; want 0", rp.previewUpdateCalls)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewStandardActionRecordCreateFields_CustomRootDoesNotNoopUpdates(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "update_housekeeping"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:   "update_housekeeping",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Description", Value: stringValue("must fail closed in customized root preview")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:                     "preview-flow-version-1",
			DefinitionId:           "preview-flow-def-1",
			Source:                 flowSourceStandard,
			Status:                 "Active",
			DefinitionIsCustomized: true,
			Metadata:               flowMeta,
		}, nil
	}
	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")

	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatalf("IsSuccess=true; want false because customized standard-action root must keep the generic preview fail-closed contract")
	}
	if len(result.Errors) == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "recordUpdate is not allowed") {
		t.Fatalf("Errors[0].Message = %q; want recordUpdate preview rejection", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewStandardActionRecordCreateFields_CustomSubflowDoesNotInheritNoopUpdates(t *testing.T) {
	rootMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "call_custom"},
		},
		ActionCalls: []models.FlowActionCall{
			{
				Name:       "call_custom",
				ActionType: "flow",
				ActionName: "CustomSubflow",
			},
		},
	}
	customMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "custom_update"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:   "custom_update",
				Object: "Account",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("must fail closed in preview")},
				},
			},
		},
	}

	exec, rp := previewExecutor(t, rootMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		switch apiName {
		case "PreviewFlow":
			return &models.FlowVersion{
				Id:           "preview-flow-version-1",
				DefinitionId: "preview-flow-def-1",
				Source:       flowSourceStandard,
				Status:       "Active",
				Metadata:     rootMeta,
			}, nil
		case "CustomSubflow":
			return &models.FlowVersion{
				Id:           "custom-subflow-version-1",
				DefinitionId: "custom-subflow-def-1",
				Source:       "custom",
				Status:       "Active",
				Metadata:     customMeta,
			}, nil
		default:
			return nil, nil
		}
	}
	exec.flowPackOwnedFn = func(ctx context.Context, apiName string, user *models.UserSession) (bool, error) {
		return apiName == "PreviewFlow", nil
	}
	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "PreviewFlow")

	result, err := exec.PreviewStandardActionRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewStandardActionRecordCreateFields returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatalf("IsSuccess=true; want false because non-pack-owned subflow recordUpdate must keep the generic preview fail-closed contract")
	}
	if len(result.Errors) == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "recordUpdate is not allowed") {
		t.Fatalf("Errors[0].Message = %q; want recordUpdate preview rejection", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// ---------------------------------------------------------------------------
// AC.3 — RecordDelete inside the previewed flow surfaces as a flow error and
//        never reaches the inner persistence.
// ---------------------------------------------------------------------------

func TestPreviewRecordCreateFields_RecordDelete_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "delete_thing"},
		},
		RecordDeletes: []models.FlowRecordDelete{
			{
				Name:           "delete_thing",
				Object:         "PreviewObj",
				InputReference: "$Record",
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	ctx := context.WithValue(context.Background(), flowRuntimeContextKey{}, NewFlowRuntimeContext(context.Background(), nil))

	result, err := exec.PreviewRecordCreateFields(ctx, "PreviewFlow", map[string]interface{}{
		"$Record": models.SObject{"Id": "fake-id-001"},
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because flow contains a recordDelete")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error in result.Errors")
	}
	if !strings.Contains(result.Errors[0].Message, "recordDelete is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'recordDelete is not allowed'", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// ---------------------------------------------------------------------------
// AC.4 — Stage-3 ActionFieldDefault rows fire under preview the same way they
//        fire under live execution. Validates "uses the same Flow metadata
//        and expression semantics as real execution".
// ---------------------------------------------------------------------------

func TestPreviewRecordCreateFields_ActionFieldDefaults_AppliesInPreview(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"AuthorWritten": "explicit",
	})
	exec, rp := previewExecutor(t, flowMeta)

	// Stage 3 default: fills DefaultFromAction when AuthorWritten is absent.
	// Here the field is absent (the flow only authored AuthorWritten), so
	// the default's literal-formula value should land on the captured row.
	defaultRows := []seedrows.ActionFieldDefaultRow{
		{
			TargetObject: "PreviewObj",
			FieldAPIName: "DefaultFromAction",
			ValueFormula: `"actiondefault"`,
			Priority:     10,
		},
	}
	ctx := WithActionFieldDefaults(context.Background(), defaultRows)
	result, err := exec.PreviewRecordCreateFields(ctx, "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; errors=%+v", result.Errors)
	}
	if got := len(result.Creates); got != 1 {
		t.Fatalf("len(Creates) = %d; want 1", got)
	}
	create := result.Creates[0]
	if got := create.Fields["AuthorWritten"]; got != "explicit" {
		t.Errorf("AuthorWritten = %v; want explicit (flow author should win)", got)
	}
	if got := create.Fields["DefaultFromAction"]; got != "actiondefault" {
		t.Errorf("DefaultFromAction = %v; want actiondefault (stage-3 must run in preview)", got)
	}
	if rp.insertCalls != 0 {
		t.Errorf("inner persistence.Insert calls = %d; want 0", rp.insertCalls)
	}
}

// ---------------------------------------------------------------------------
// AC.5 — Permission gate honored: a protected standard flow without the
//        ctx marker comes back as IsSuccess=false with FLOW_EXECUTION_ERROR,
//        identical to ExecuteFlow's denial shape.
// ---------------------------------------------------------------------------

func TestPreviewRecordCreateFields_ProtectedStandardFlow_DeniedWithoutMarker(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"Field": "ignored-because-denied",
	})
	exec, rp := previewExecutor(t, flowMeta)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		// flip Source to standard so authorizeProtectedStandardFlow gates it
		return &models.FlowVersion{
			Id:           "preview-flow-version-1",
			DefinitionId: "preview-flow-def-1",
			Source:       flowSourceStandard,
			Status:       "Active",
			Metadata:     flowMeta,
		}, nil
	}

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false (protected standard flow without ctx marker must be denied)")
	}
	if got := len(result.Errors); got != 1 || result.Errors[0].StatusCode != "FLOW_EXECUTION_ERROR" {
		t.Errorf("Errors = %+v; want exactly one FLOW_EXECUTION_ERROR", result.Errors)
	}
	if got := len(result.Creates); got != 0 {
		t.Errorf("len(Creates) = %d; want 0 (denied flow never enters element loop)", got)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// ---------------------------------------------------------------------------
// AC.6 — Constructor guards: nil receiver / unwired persistence / unwired
//        interview service all surface a clean internal error rather than a
//        nil-pointer panic.
// ---------------------------------------------------------------------------

func TestPreviewRecordCreateFields_GuardsAgainstUnwiredExecutor(t *testing.T) {
	t.Run("nil_receiver", func(t *testing.T) {
		var exec *FlowExecutor
		_, err := exec.PreviewRecordCreateFields(context.Background(), "X", nil, newPreviewUser())
		if err == nil {
			t.Fatal("nil receiver: expected error, got nil")
		}
	})
	t.Run("missing_persistence", func(t *testing.T) {
		exec := &FlowExecutor{interviews: NewFlowInterviewService(nil, nil)}
		_, err := exec.PreviewRecordCreateFields(context.Background(), "X", nil, newPreviewUser())
		if err == nil {
			t.Fatal("missing persistence: expected error, got nil")
		}
	})
	t.Run("missing_interview_service", func(t *testing.T) {
		rp := &recordingPersistence{}
		exec := &FlowExecutor{persistence: rp}
		_, err := exec.PreviewRecordCreateFields(context.Background(), "X", nil, newPreviewUser())
		if err == nil {
			t.Fatal("missing interview service: expected error, got nil")
		}
	})
}

// ---------------------------------------------------------------------------
// AC.7 — ActionCall side-effect gate. Non-"flow" actionCalls reach
//        ActionService / ApprovalService, both of which carry their own
//        live persistence + emailSender refs. They MUST be rejected by the
//        preview gate or a previewed flow would mutate real records, send
//        real email, or fire real webhooks via that side channel.
//        Subflow ("flow") actionCalls are preview-safe (they recurse into
//        the same wrapped executor) and stay allowed; they are covered
//        elsewhere by the broader RecordCreate-capture path.
// ---------------------------------------------------------------------------

// flowMetaWithRecordCreateThenActionCall builds:
//
//	Start → "create_thing" (RecordCreate) → "side_effect_action" (ActionCall) → end
//
// so the test can prove (a) the create-before-action intent is still
// captured for the caller, and (b) the actionCall is rejected before it
// can reach the live ActionService.
func flowMetaWithRecordCreateThenActionCall(actionType, actionName string) models.FlowMetadata {
	return models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_thing"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:   "create_thing",
				Object: "PreviewObj",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("captured-before-action")},
				},
				Connector: &models.FlowConnector{TargetReference: "side_effect_action"},
			},
		},
		ActionCalls: []models.FlowActionCall{
			{
				Name:       "side_effect_action",
				ActionName: actionName,
				ActionType: actionType,
			},
		},
	}
}

func TestPreviewRecordCreateFields_EmailAlertActionCall_Rejected(t *testing.T) {
	flowMeta := flowMetaWithRecordCreateThenActionCall("emailAlert", "PreviewObj.SomeAlert")
	exec, rp := previewExecutor(t, flowMeta)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because flow contains a retired emailAlert actionCall")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error in result.Errors")
	}
	for _, want := range []string{"WorkflowAlert runtime dispatch is disabled", "Flow sendEmail", "remove it"} {
		if !strings.Contains(result.Errors[0].Message, want) {
			t.Errorf("Errors[0].Message = %q; want it to mention %q", result.Errors[0].Message, want)
		}
	}
	if got := len(result.Creates); got != 1 {
		t.Errorf("len(Creates) = %d; want 1 (the create-before-action intent should still be captured)", got)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_MetadataActionCall_Rejected(t *testing.T) {
	// actionType="" / unknown drives the metadata-action lookup branch in
	// executeActionCall, which would hit ActionService.ExecuteActionDirect
	// and run real DML / webhooks via ActionService.persistence.
	flowMeta := flowMetaWithRecordCreateThenActionCall("", "SomeMetadataAction")
	exec, rp := previewExecutor(t, flowMeta)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because metadata actionCall would invoke live ActionService")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error in result.Errors")
	}
	if !strings.Contains(result.Errors[0].Message, "actionCall is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'actionCall is not allowed'", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_StandardActionCall_Rejected(t *testing.T) {
	// emailSimple is intercepted by the no-DML preview guard
	// (flowActionCallSendsEmail -> previewEmailIOAttemptError) before any
	// dispatch; in a live run it routes through executeSendEmailAction (the
	// canonical sendEmail path) and would call ActionService.emailSender to send
	// mail.
	flowMeta := flowMetaWithRecordCreateThenActionCall("emailSimple", "Send_Notice")
	exec, rp := previewExecutor(t, flowMeta)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because standard actionCall would invoke live ActionService.emailSender")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error in result.Errors")
	}
	if !strings.Contains(result.Errors[0].Message, "actionCall is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'actionCall is not allowed'", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewMode_AllowsProtectedPreviewRecordUpdatePrepare(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})
	captured := []FlowRecordCreatePreview{}
	exec.persistence = &previewPersistence{inner: rp, captured: &captured}
	rp.previewUpdatePrepared = &models.PreparedRecordUpdate{
		Record: models.SObject{
			"Id":     "lead-1",
			"Status": "Closed - Converted",
		},
		EffectiveUpdates: models.SObject{
			"Status": "Closed - Converted",
		},
	}

	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "Std_ConvertLead")
	ctx = constants.WithStandardActionProtectedFlowExecution(ctx, "Std_ConvertLead")
	ctx = WithPreviewMode(ctx)
	interview := &models.FlowInterview{
		Id:          "preview-interview-1",
		ContextData: map[string]interface{}{},
	}

	next, err := exec.executeActionCall(ctx, interview, &models.FlowMetadata{}, &models.FlowActionCall{
		Name:       "PreviewLead",
		ActionType: "previewRecordUpdate",
		InputParameters: []models.FlowActionParameter{
			{Name: "object", Value: "Lead"},
			{Name: "recordId", Value: "lead-1"},
			{Name: "updates", Value: map[string]interface{}{"Status": "Closed - Converted"}},
			{Name: "outputReference", Value: "preparedLead"},
		},
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("executeActionCall(previewRecordUpdate) error = %v", err)
	}
	if next != "" {
		t.Fatalf("next = %q, want empty without connector", next)
	}
	if rp.previewUpdateCalls != 1 {
		t.Fatalf("PreviewBeforeUpdate calls = %d, want 1", rp.previewUpdateCalls)
	}
	if rp.updateCalls != 0 || rp.insertCalls != 0 || rp.deleteCalls != 0 {
		t.Fatalf("inner DML calls Insert/Update/Delete = %d/%d/%d, want 0/0/0", rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
	if rp.lastPreviewUpdateObject != "Lead" || rp.lastPreviewUpdateID != "lead-1" {
		t.Fatalf("PreviewBeforeUpdate target = %s/%s, want Lead/lead-1", rp.lastPreviewUpdateObject, rp.lastPreviewUpdateID)
	}
	if got := rp.lastPreviewUpdates["Status"]; got != "Closed - Converted" {
		t.Fatalf("PreviewBeforeUpdate Status = %v, want Closed - Converted", got)
	}
	preparedLead, _ := interview.ContextData["preparedLead"].(models.SObject)
	if got := preparedLead["Status"]; got != "Closed - Converted" {
		t.Fatalf("preparedLead.Status = %v, want Closed - Converted", got)
	}
	if got := models.RecordObjectType(preparedLead); got != "Lead" {
		t.Fatalf("preparedLead object type = %q, want Lead", got)
	}
}

func TestProtectedPreviewRecordUpdateAllowsInternalOnlyExplicitNull(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})
	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "StandardRoot")
	ctx = constants.WithStandardActionProtectedFlowExecution(ctx, "StandardRoot")
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"contactId": "contact-1",
	}}

	next, err := exec.executeActionCall(ctx, interview, &models.FlowMetadata{}, &models.FlowActionCall{
		Name:       "PrepareInternalRelink",
		ActionType: "previewRecordUpdate",
		InputParameters: []models.FlowActionParameter{
			{Name: "object", Value: "Membership"},
			{Name: "recordId", Value: "membership-1"},
			{Name: "internalUpdates", Value: map[string]interface{}{
				"NewParentId": map[string]interface{}{"elementReference": "contactId"},
				"OldParentId": nil,
			}},
		},
		Connector: &models.FlowConnector{TargetReference: "CommitInternalRelink"},
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("executeActionCall(previewRecordUpdate) error = %v", err)
	}
	if next != "CommitInternalRelink" {
		t.Fatalf("next = %q, want CommitInternalRelink", next)
	}
	if rp.previewUpdateCalls != 1 {
		t.Fatalf("PreviewBeforeUpdate calls = %d, want 1", rp.previewUpdateCalls)
	}
	if len(rp.lastPreviewUpdates) != 0 {
		t.Fatalf("public updates = %#v, want none", rp.lastPreviewUpdates)
	}
	if got := rp.lastPreviewInternal["NewParentId"]; got != "contact-1" {
		t.Fatalf("NewParentId = %#v, want contact-1", got)
	}
	if got, present := rp.lastPreviewInternal["OldParentId"]; !present || got != nil {
		t.Fatalf("OldParentId = (%#v, %t), want explicit null present", got, present)
	}
}

// ---------------------------------------------------------------------------
// Round 7 regressions — IANTHEREAL gate-merge P1 blockers (PR #273)
//
// These four tests pin the additional preview guards added in round 7. Each
// targets one of the three side-effect channels that previewPersistence
// alone could not intercept:
//
//   - Blocker 1: pkgAuditWriter writes _System_FlowAuditLog under preview
//   - Blocker 2: executeRecordUpdate's in-place $Record branch returns
//     before persistence.Update, so previewPersistence.Update never sees it
//   - Blocker 3: Wait pause writes a FlowInterview row that previewPersistence
//     would otherwise capture as a user RecordCreate intent
// ---------------------------------------------------------------------------

// memoryAuditStore is a thread-safe AuditStore for tests that asserts on the
// row set written during a single test. Simplest implementation that lets us
// prove "zero rows written by a preview run".
type memoryAuditStore struct {
	mu   sync.Mutex
	rows []Row
}

func (s *memoryAuditStore) InsertRows(ctx context.Context, rows []Row) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.rows = append(s.rows, rows...)
	return nil
}

func (s *memoryAuditStore) QueryRows(ctx context.Context, filter QueryFilter) ([]Row, error) {
	return nil, nil
}

func (s *memoryAuditStore) DeleteBefore(ctx context.Context, before time.Time) (int64, error) {
	return 0, nil
}

func (s *memoryAuditStore) CountRows(ctx context.Context) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return int64(len(s.rows)), nil
}

func (s *memoryAuditStore) DeleteOldest(ctx context.Context, n int64) (int64, error) {
	return 0, nil
}

func (s *memoryAuditStore) RowCount() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.rows)
}

// TestPreviewRecordCreateFields_AuditWriter_NotEmittedDuringPreview pins
// IANTHEREAL Blocker 1: with a real Writer wired (matching production
// startup wiring), a preview run must produce zero audit rows. Before round
// 7 every previewed element enqueued a row to _System_FlowAuditLog because
// pkgAuditWriter is package-level and previewPersistence cannot intercept
// it. emitFlowAudit suppresses on isPreviewMode(ctx).
func TestPreviewRecordCreateFields_AuditWriter_NotEmittedDuringPreview(t *testing.T) {
	store := &memoryAuditStore{}
	w := NewWriter(store)
	prevWriter := GetAuditWriter()
	SetAuditWriter(w)
	t.Cleanup(func() {
		SetAuditWriter(prevWriter)
	})

	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"FieldOne": "alpha",
	})
	exec, _ := previewExecutor(t, flowMeta)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if !result.IsSuccess {
		t.Fatalf("preview must succeed; errors=%+v", result.Errors)
	}

	// Drain async writer so any pending rows reach the store before assertion.
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	if err := w.Shutdown(shutdownCtx); err != nil {
		t.Fatalf("audit writer Shutdown: %v", err)
	}

	if got := store.RowCount(); got != 0 {
		t.Errorf("audit store rows = %d; want 0 (preview path must not emit audit rows)", got)
	}
}

// TestPreviewRecordCreateFields_RecordUpdate_InPlaceNoId_Rejected pins
// IANTHEREAL Blocker 2: a RecordUpdate on $Record with no Id takes the
// in-place branch in executeRecordUpdate and returns BEFORE reaching
// persistence.Update. Without the function-entry guard, previewPersistence
// never sees the call and the preview reports IsSuccess despite an
// effective state mutation. The entry-level guard rejects the element
// before any branch, so the preview must surface the rejection error.
func TestPreviewRecordCreateFields_RecordUpdate_InPlaceNoId_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "update_thing"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name: "update_thing",
				// Object intentionally empty: triggers the implicit-$Record
				// fallback (objectName resolved from $Record.attributes.type).
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("would-mutate-in-place")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	// $Record has NO Id — recordId == "" makes isInPlace=true under live
	// execution (and the in-place branch returns before persistence).
	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow",
		map[string]interface{}{
			"$Record": models.SObject{"attributes": map[string]interface{}{"type": "PreviewObj"}},
		},
		newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because RecordUpdate must be rejected even on the in-place no-Id path")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "recordUpdate is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'recordUpdate is not allowed'", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// TestPreviewRecordCreateFields_RecordUpdate_BeforeSaveFlow_Rejected pins
// the second leg of Blocker 2: in a WithBeforeSaveFlow context the in-place
// path is taken even when $Record.Id is set. The function-entry guard must
// fire before isBeforeSaveFlow(ctx) is consulted.
func TestPreviewRecordCreateFields_RecordUpdate_BeforeSaveFlow_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Object:    "PreviewObj",
			Connector: &models.FlowConnector{TargetReference: "update_thing"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:           "update_thing",
				Object:         "PreviewObj",
				InputReference: "$Record",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("would-mutate-via-beforesave")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	// Wrap ctx with WithBeforeSaveFlow so the in-place branch would be
	// taken (recordId set + isBeforeSaveFlow(ctx)==true) under live
	// execution.
	ctx := WithBeforeSaveFlow(context.Background())

	result, err := exec.PreviewRecordCreateFields(ctx, "PreviewFlow",
		map[string]interface{}{
			"$Record": models.SObject{"Id": "fake-id-1", "attributes": map[string]interface{}{"type": "PreviewObj"}},
		},
		newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because RecordUpdate must be rejected under WithBeforeSaveFlow context")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "recordUpdate is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'recordUpdate is not allowed'", result.Errors[0].Message)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_RecordUpdate_InsertPreparationAllowsInMemoryBeforeSaveMutation(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Object:    "PreviewObj",
			Connector: &models.FlowConnector{TargetReference: "update_thing"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:           "update_thing",
				Object:         "PreviewObj",
				InputReference: "$Record",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("prepared-by-before-save-flow")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	ctx := constants.WithInsertPreviewPreparation(WithBeforeSaveFlow(context.Background()))
	result, err := exec.PreviewRecordCreateFields(ctx, "PreviewFlow",
		map[string]interface{}{
			"$Record": models.SObject{"Id": "fake-id-1", "attributes": map[string]interface{}{"type": "PreviewObj"}},
		},
		newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; want true because insert preparation may apply same-record before-save mutations: errors=%+v", result.Errors)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_RecordUpdate_UpdatePreparationAllowsInMemoryBeforeSaveMutation(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Object:    "PreviewObj",
			Connector: &models.FlowConnector{TargetReference: "update_thing"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:           "update_thing",
				Object:         "PreviewObj",
				InputReference: "$Record",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Field", Value: stringValue("prepared-by-before-update-flow")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	ctx := constants.WithUpdatePreviewPreparation(WithBeforeSaveFlow(context.Background()))
	result, err := exec.PreviewRecordCreateFields(ctx, "PreviewFlow",
		map[string]interface{}{
			"$Record": models.SObject{"Id": "fake-id-1", "attributes": map[string]interface{}{"type": "PreviewObj"}},
		},
		newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; want true because update preparation may apply same-record before-save mutations: errors=%+v", result.Errors)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_RecordDelete_LiveExecutorPreviewModeRejected(t *testing.T) {
	rp := &recordingPersistence{}
	exec := &FlowExecutor{persistence: rp}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": models.SObject{
				constants.FieldID: "fake-id-1",
				"attributes":      map[string]interface{}{"type": "PreviewObj"},
			},
		},
	}
	del := &models.FlowRecordDelete{
		Name:           "delete_thing",
		Object:         "PreviewObj",
		InputReference: "$Record",
	}

	nextElem, err := exec.executeRecordDelete(WithPreviewMode(context.Background()), interview, del, newPreviewUser())
	if err == nil {
		t.Fatalf("executeRecordDelete() err = nil, nextElem=%q; want preview delete rejection", nextElem)
	}
	if !strings.Contains(err.Error(), "recordDelete is not allowed") {
		t.Fatalf("executeRecordDelete() err = %q; want preview delete rejection", err.Error())
	}
	if rp.deleteCalls != 0 {
		t.Fatalf("inner persistence.Delete calls = %d; want 0", rp.deleteCalls)
	}
}

func TestPreviewRecordCreateFields_ActionCall_LiveExecutorPreviewModeRejected(t *testing.T) {
	rp := &recordingPersistence{}
	exec := &FlowExecutor{persistence: rp}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": models.SObject{
				constants.FieldID: "fake-id-1",
				"attributes":      map[string]interface{}{"type": "PreviewObj"},
			},
		},
	}
	action := &models.FlowActionCall{
		Name:       "send_email",
		ActionType: "emailSimple",
	}

	nextElem, err := exec.executeActionCall(WithPreviewMode(context.Background()), interview, &models.FlowMetadata{}, action, newPreviewUser())
	if err == nil {
		t.Fatalf("executeActionCall() err = nil, nextElem=%q; want preview actionCall rejection", nextElem)
	}
	if !strings.Contains(err.Error(), "actionCall is not allowed") {
		t.Fatalf("executeActionCall() err = %q; want preview actionCall rejection", err.Error())
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Fatalf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// TestPreviewRecordCreateFields_Wait_Rejected pins IANTHEREAL Blocker 3:
// a previewed flow containing a Wait element must reject before
// pauseForWait can write a FlowInterview row through previewPersistence
// (which would surface as a bogus FlowInterview entry in Creates[]).
// Wait has no preview semantics — there is no clock running — so we reject
// the element early with a clear error message.
func TestPreviewRecordCreateFields_Wait_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "wait_for_alarm"},
		},
		Waits: []models.FlowWait{
			{
				Name: "wait_for_alarm",
				WaitEvents: []models.FlowWaitEvent{
					{
						Name:      "alarm",
						EventType: "AlarmEvent",
					},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	result, err := exec.PreviewRecordCreateFields(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewRecordCreateFields returned error: %v", err)
	}
	if result.IsSuccess {
		t.Errorf("IsSuccess=true; want false because Wait must be rejected in preview")
	}
	if got := len(result.Errors); got == 0 {
		t.Fatal("expected at least one error")
	}
	if !strings.Contains(result.Errors[0].Message, "Wait is not allowed") {
		t.Errorf("Errors[0].Message = %q; want it to mention 'Wait is not allowed'", result.Errors[0].Message)
	}
	// No FlowInterview row leaked into Creates: previewPersistence.Insert
	// rejects engine-internal inserts, and the early Wait reject means
	// pauseForWait never even runs.
	if got := len(result.Creates); got != 0 {
		t.Errorf("len(Creates) = %d; want 0 (no FlowInterview row should leak into Creates from a Wait pause)", got)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// ---------------------------------------------------------------------------
// PR #1008 — Codex HIGH finding: triggered-flow side-effect channels.
//
// previewPersistence.PreviewBeforeUpdate now delegates into the live
// persistence service so the engine reuses the real RecordBeforeUpdate
// publish + validation pipeline. The published event flows synchronously into
// FlowTriggerListener, whose executor is the LIVE (unwrapped) FlowExecutor.
// Without ctx-keyed guards, a triggered before-save flow on the live executor
// could fire RecordCreate / RecordDelete / non-flow ActionCall elements and
// commit real DML / send real email from a "no-DML preview" path — even
// though e.persistence is not *previewPersistence on that executor.
//
// The tests below pin the second-line defense: an executor whose persistence
// is NOT the preview wrapper, running with isPreviewMode(ctx)==true, must
// reject each of the three side-effect channels at element entry.
// ---------------------------------------------------------------------------

// TestPreviewMode_TriggeredFlow_RecordCreate_Rejected pins the RecordCreate
// guard for the triggered-flow path: executing a RecordCreate element on an
// executor whose persistence is the LIVE service (not previewPersistence)
// while ctx is preview-marked must NOT call persistence.Insert.
func TestPreviewMode_TriggeredFlow_RecordCreate_Rejected(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})
	// Intentionally leave exec.persistence == rp (the recording stub), NOT
	// the preview wrapper — simulating the trigger-listener's live executor.

	ctx := WithPreviewMode(context.Background())
	interview := &models.FlowInterview{
		Id:          "triggered-interview-1",
		ContextData: map[string]interface{}{},
	}

	next, err := exec.executeRecordCreate(ctx, interview, nil, &models.FlowRecordCreate{
		Name:   "create_thing",
		Object: "PreviewObj",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Field", Value: stringValue("would-have-inserted")},
		},
	}, newPreviewUser())
	if err == nil {
		t.Fatal("executeRecordCreate error = nil, want errPreviewRecordCreateAttempt")
	}
	if !strings.Contains(err.Error(), "recordCreate is not allowed") {
		t.Errorf("err = %q, want it to mention 'recordCreate is not allowed'", err.Error())
	}
	if next != "" {
		t.Errorf("next = %q, want empty (guard returns no next element)", next)
	}
	if rp.insertCalls != 0 {
		t.Errorf("live persistence saw %d Insert calls, want 0 (triggered-flow RecordCreate must not reach the inner provider under preview ctx)", rp.insertCalls)
	}
}

// TestPreviewMode_TriggeredFlow_RecordDelete_Rejected pins the RecordDelete
// guard for the triggered-flow path: a RecordDelete element on the live
// trigger-listener executor under preview ctx must not call persistence.Delete.
func TestPreviewMode_TriggeredFlow_RecordDelete_Rejected(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})

	ctx := WithPreviewMode(context.Background())
	interview := &models.FlowInterview{
		Id: "triggered-interview-1",
		ContextData: map[string]interface{}{
			"someRecord": models.SObject{"Id": "rec-1", "attributes": map[string]interface{}{"type": "PreviewObj"}},
		},
	}

	next, err := exec.executeRecordDelete(ctx, interview, &models.FlowRecordDelete{
		Name:           "delete_thing",
		Object:         "PreviewObj",
		InputReference: "someRecord",
	}, newPreviewUser())
	if err == nil {
		t.Fatal("executeRecordDelete error = nil, want errPreviewRecordDeleteAttempt")
	}
	if !strings.Contains(err.Error(), "recordDelete is not allowed") {
		t.Errorf("err = %q, want it to mention 'recordDelete is not allowed'", err.Error())
	}
	if next != "" {
		t.Errorf("next = %q, want empty", next)
	}
	if rp.deleteCalls != 0 {
		t.Errorf("live persistence saw %d Delete calls, want 0 (triggered-flow RecordDelete must not reach the inner provider under preview ctx)", rp.deleteCalls)
	}
}

// TestPreviewMode_TriggeredFlow_ActionCall_Rejected pins the ActionCall guard
// for the triggered-flow path: a non-flow actionCall element on the live
// trigger-listener executor under preview ctx must reject even though
// e.persistence is not *previewPersistence. The pre-existing
// persistence-identity check only covered the top-level preview executor; the
// added isPreviewMode(ctx) condition closes the second channel.
func TestPreviewMode_TriggeredFlow_ActionCall_Rejected(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})

	ctx := WithPreviewMode(context.Background())
	interview := &models.FlowInterview{
		Id:          "triggered-interview-1",
		ContextData: map[string]interface{}{},
	}

	next, err := exec.executeActionCall(ctx, interview, &models.FlowMetadata{}, &models.FlowActionCall{
		Name:       "send_email",
		ActionName: "PreviewEmail",
		ActionType: "emailSimple",
	}, newPreviewUser())
	if err == nil {
		t.Fatal("executeActionCall error = nil, want errPreviewActionCallAttempt")
	}
	if !strings.Contains(err.Error(), "actionCall is not allowed") {
		t.Errorf("err = %q, want it to mention 'actionCall is not allowed'", err.Error())
	}
	if next != "" {
		t.Errorf("next = %q, want empty", next)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("live persistence saw %d Insert / %d Update / %d Delete; want 0/0/0 (triggered-flow ActionCall must not reach any DML path under preview ctx)",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// TestPreviewMode_TriggeredFlow_PreviewRecordUpdateAction_Rejected pins that
// the previewRecordUpdate special-case action is NOT honored on the
// triggered-flow path even if the protected standard-action context happens
// to be carried through ctx. Honoring it would re-enter the live
// PersistenceService.PreviewBeforeUpdate, recursively re-publishing
// RecordBeforeUpdate and risking double-fire / unbounded recursion.
func TestPreviewMode_TriggeredFlow_PreviewRecordUpdateAction_Rejected(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})

	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "Std_ConvertLead")
	ctx = constants.WithStandardActionProtectedFlowExecution(ctx, "Std_ConvertLead")
	ctx = WithPreviewMode(ctx)
	interview := &models.FlowInterview{
		Id:          "triggered-interview-1",
		ContextData: map[string]interface{}{},
	}

	next, err := exec.executeActionCall(ctx, interview, &models.FlowMetadata{}, &models.FlowActionCall{
		Name:       "PreviewLead",
		ActionType: "previewRecordUpdate",
		InputParameters: []models.FlowActionParameter{
			{Name: "object", Value: "Lead"},
			{Name: "recordId", Value: "lead-1"},
			{Name: "updates", Value: map[string]interface{}{"Status": "Closed - Converted"}},
		},
	}, newPreviewUser())
	if err == nil {
		t.Fatal("executeActionCall(previewRecordUpdate) error = nil, want errPreviewActionCallAttempt on triggered-flow path")
	}
	if !strings.Contains(err.Error(), "actionCall is not allowed") {
		t.Errorf("err = %q, want it to mention 'actionCall is not allowed' (live executor must NOT re-enter PreviewBeforeUpdate)", err.Error())
	}
	if next != "" {
		t.Errorf("next = %q, want empty", next)
	}
	if rp.previewUpdateCalls != 0 {
		t.Errorf("live persistence saw %d PreviewBeforeUpdate calls, want 0 (recursion risk: triggered-flow path must not re-enter PreviewBeforeUpdate)", rp.previewUpdateCalls)
	}
}

func TestPreviewMode_AllowedRegisteredActionUsesRegistryPreviewPolicy(t *testing.T) {
	exec, rp := previewExecutor(t, models.FlowMetadata{})
	captured := []FlowRecordCreatePreview{}
	exec.persistence = &previewPersistence{inner: rp, captured: &captured}
	called := false
	exec.RegisterFlowActionCalls(FlowActionCallRegistration{
		ActionType:      "syntheticPreviewAction",
		AtomicPolicy:    FlowActionCallAtomicPolicySafe,
		PreviewPolicy:   FlowActionCallPreviewPolicyAllowWithPreviewPersistence,
		RequiresContext: FlowActionCallContextStandardActionProtected,
		Handler: func(_ context.Context, _ *FlowExecutor, _ *models.FlowInterview, _ *models.FlowMetadata, _ *models.FlowActionCall, _ *models.UserSession) (string, error) {
			called = true
			return "", nil
		},
	})

	ctx := constants.WithStandardActionExecutionRootFlow(context.Background(), "standard-action-1", "Std_Test")
	ctx = constants.WithStandardActionProtectedFlowExecution(ctx, "Std_Test")
	ctx = WithPreviewMode(ctx)
	interview := &models.FlowInterview{
		Id:          "preview-interview-1",
		ContextData: map[string]interface{}{},
	}

	next, err := exec.executeActionCall(ctx, interview, &models.FlowMetadata{}, &models.FlowActionCall{
		Name:       "SyntheticPreview",
		ActionType: "syntheticPreviewAction",
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("executeActionCall(syntheticPreviewAction) error = %v", err)
	}
	if next != "" {
		t.Fatalf("next = %q, want empty", next)
	}
	if !called {
		t.Fatal("registered preview-allowed handler was not called")
	}
}

// ---------------------------------------------------------------------------
// PreviewFlowOutputs — output-only preview producer (issue #945)
// ---------------------------------------------------------------------------

func TestPreviewFlowOutputs_HappyPath_ReturnsOutputValues(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "assign_out"},
		},
		Variables: []models.FlowVariable{
			{
				Name:     "campaignName",
				DataType: "String",
				IsOutput: true,
			},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "assign_out",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "campaignName", Operator: "Assign", Value: stringValue("Summer Campaign")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	result, err := exec.PreviewFlowOutputs(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewFlowOutputs returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if !result.IsSuccess {
		t.Fatalf("IsSuccess=false; errors=%+v", result.Errors)
	}
	if got, ok := result.OutputValues["campaignName"]; !ok || got != "Summer Campaign" {
		t.Errorf("OutputValues[campaignName] = %v; want %q", got, "Summer Campaign")
	}
	// No DML should reach the inner persistence.
	if rp.insertCalls != 0 {
		t.Errorf("inner persistence.Insert calls = %d; want 0", rp.insertCalls)
	}
	if rp.updateCalls != 0 {
		t.Errorf("inner persistence.Update calls = %d; want 0", rp.updateCalls)
	}
}

func TestPreviewFlowOutputs_RecordCreate_Rejected(t *testing.T) {
	flowMeta := flowMetaWithSingleRecordCreate("PreviewObj", map[string]string{
		"FieldOne": "alpha",
	})
	exec, _ := previewExecutor(t, flowMeta)

	result, err := exec.PreviewFlowOutputs(context.Background(), "PreviewFlow", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewFlowOutputs returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatal("IsSuccess=true; output-only producer must reject RecordCreate")
	}
	// The rejection may come from either the outputOnlyPreviewPersistence sentinel
	// (direct Insert path) or the triggered before-save flow sentinel (if the
	// execution path re-enters under preview ctx). Both signal RecordCreate rejection.
	found := false
	for _, e := range result.Errors {
		if strings.Contains(e.Message, "recordCreate is not allowed") {
			found = true
			break
		}
	}
	if !found {
		t.Fatalf("expected RecordCreate rejection in errors; got %+v", result.Errors)
	}
}

func TestPreviewFlowOutputs_RecordUpdate_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "update_thing"},
		},
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:   "update_thing",
				Object: "Lead",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Status", Value: stringValue("Closed")},
				},
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	// Provide a $Record so executeRecordUpdate's implicit-$Record path
	// reaches the persistence.Update path.
	result, err := exec.PreviewFlowOutputs(context.Background(), "PreviewFlow",
		map[string]interface{}{
			"$Record": models.SObject{"Id": "fake-target-1", "attributes": map[string]interface{}{"type": "Lead"}},
		},
		newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewFlowOutputs returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatal("IsSuccess=true; PreviewFlowOutputs must reject RecordUpdate")
	}
	found := false
	for _, e := range result.Errors {
		if strings.Contains(e.Message, "recordUpdate is not allowed") {
			found = true
			break
		}
	}
	if !found {
		t.Fatalf("expected RecordUpdate rejection in errors; got %+v", result.Errors)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

func TestPreviewFlowOutputs_RecordDelete_Rejected(t *testing.T) {
	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "delete_thing"},
		},
		RecordDeletes: []models.FlowRecordDelete{
			{
				Name:           "delete_thing",
				Object:         "Lead",
				InputReference: "$Record",
			},
		},
	}
	exec, rp := previewExecutor(t, flowMeta)

	ctx := context.WithValue(context.Background(), flowRuntimeContextKey{}, NewFlowRuntimeContext(context.Background(), nil))

	result, err := exec.PreviewFlowOutputs(ctx, "PreviewFlow", map[string]interface{}{
		"$Record": models.SObject{"Id": "fake-id-001"},
	}, newPreviewUser())
	if err != nil {
		t.Fatalf("PreviewFlowOutputs returned error: %v", err)
	}
	if result == nil {
		t.Fatal("result was nil")
	}
	if result.IsSuccess {
		t.Fatal("IsSuccess=true; PreviewFlowOutputs must reject RecordDelete")
	}
	found := false
	for _, e := range result.Errors {
		if strings.Contains(e.Message, "recordDelete is not allowed") {
			found = true
			break
		}
	}
	if !found {
		t.Fatalf("expected RecordDelete rejection in errors; got %+v", result.Errors)
	}
	if rp.insertCalls+rp.updateCalls+rp.deleteCalls != 0 {
		t.Errorf("inner persistence saw %d Insert / %d Update / %d Delete; want 0/0/0",
			rp.insertCalls, rp.updateCalls, rp.deleteCalls)
	}
}

// ---------------------------------------------------------------------------
// Codex round-4 regression — preview synthetic IDs must be unguessable.
//
// The earlier deterministic format ("preview-<object>-<n>") let a caller
// hand-craft a literal like "preview-Account-1" on a child lookup field,
// and the no-DML overlay's registry shortcut in
// records.validateNonPolymorphicLookups (and peers) trusted (object,id)
// membership alone — so the hand-crafted literal would silently pass
// FK validation even though no synthetic Account had been created in the
// same run. Switching to crypto/rand-backed IDs is the architectural fix:
// only IDs minted by previewRecordId can match the registry.
// ---------------------------------------------------------------------------

func TestPreviewRecordId_UnguessableAndNotDeterministic(t *testing.T) {
	a, err := previewRecordId()
	if err != nil {
		t.Fatalf("previewRecordId returned error: %v", err)
	}
	b, err := previewRecordId()
	if err != nil {
		t.Fatalf("previewRecordId returned error: %v", err)
	}
	if !strings.HasPrefix(a, "preview-") {
		t.Fatalf("previewRecordId = %q; want preview- prefix", a)
	}
	// 16 bytes -> 32 hex chars after the "preview-" literal.
	const expectedLen = len("preview-") + 32
	if len(a) != expectedLen {
		t.Fatalf("len(previewRecordId) = %d; want %d (preview- + 32 hex chars)", len(a), expectedLen)
	}
	if a == b {
		t.Fatalf("previewRecordId returned same ID on two calls: %q (must be random)", a)
	}
	// Anti-bypass: hand-guessable "preview-<object>-<n>" shapes must never
	// be produced by the minter. If this assertion ever flips green for a
	// well-known shape, the registry shortcut becomes forgeable again.
	deterministicShapes := []string{
		"preview-Account-1",
		"preview-Account-2",
		"preview-ParentObject-1",
		"preview-Opportunity-1",
	}
	for _, guess := range deterministicShapes {
		if a == guess || b == guess {
			t.Fatalf("previewRecordId produced a guessable literal %q; round-4 provenance fix regressed", guess)
		}
	}
}

func TestPreviewPersistence_Insert_MintsUnguessableId(t *testing.T) {
	rp := &recordingPersistence{}
	captured := []FlowRecordCreatePreview{}
	pp := &previewPersistence{inner: rp, captured: &captured}

	ctx := constants.WithPreviewSyntheticRecordRegistry(context.Background())
	ctx = constants.WithPreviewSiblingMaxDefaultedFieldRegistry(ctx)
	ctx = withRecordCreateInsertMarker(ctx)

	out, err := pp.Insert(ctx, "Account", models.SObject{"Name": "Acme"}, newPreviewUser())
	if err != nil {
		t.Fatalf("Insert returned error: %v", err)
	}
	previewID, _ := out[constants.FieldID].(string)
	if previewID == "" {
		t.Fatalf("Insert did not stamp a preview Id on the returned row: %#v", out)
	}
	if previewID == "preview-Account-1" {
		t.Fatalf("Insert produced the legacy guessable Id %q; the round-4 unguessable-ID fix regressed", previewID)
	}
	if !strings.HasPrefix(previewID, "preview-") || len(previewID) != len("preview-")+32 {
		t.Fatalf("Insert produced Id %q; want preview-<32 hex chars>", previewID)
	}

	// Captured intent should carry the same unguessable Id.
	if len(captured) != 1 || captured[0].PreviewID != previewID {
		t.Fatalf("captured = %#v; want one entry with PreviewID=%q", captured, previewID)
	}

	// A hand-crafted legacy-format literal must NOT resolve in the registry.
	if _, ok := constants.PreviewSyntheticRecordForObject(ctx, "Account", "preview-Account-1"); ok {
		t.Fatal("PreviewSyntheticRecordForObject resolved the guessable literal 'preview-Account-1' from the registry; provenance gate is forgeable")
	}
}

func TestPreviewPersistence_Insert_CapturesSiblingMaxDefaultProvenance(t *testing.T) {
	rp := &recordingPersistence{}
	rp.previewInsertFn = func(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
		constants.RegisterPreviewSiblingMaxDefaultedFields(ctx, []string{"SortOrder__c"})
		out := clonePreviewSObject(record)
		out["SortOrder__c"] = int64(7)
		return out, nil
	}
	captured := []FlowRecordCreatePreview{}
	pp := &previewPersistence{inner: rp, captured: &captured}

	ctx := constants.WithPreviewSyntheticRecordRegistry(context.Background())
	ctx = constants.WithPreviewSiblingMaxDefaultedFieldRegistry(ctx)
	ctx = withRecordCreateInsertMarker(ctx)

	if _, err := pp.Insert(ctx, "LineItem__c", models.SObject{"ParentId": "parent-1"}, newPreviewUser()); err != nil {
		t.Fatalf("Insert returned error: %v", err)
	}
	if len(captured) != 1 {
		t.Fatalf("captured = %#v, want one preview", captured)
	}
	if got := captured[0].SiblingMaxDefaultedFields; len(got) != 1 || got[0] != "SortOrder__c" {
		t.Fatalf("SiblingMaxDefaultedFields = %#v, want [SortOrder__c]", got)
	}
}

func TestPreviewPersistence_Insert_DoesNotReuseFailedSiblingMaxProvenance(t *testing.T) {
	rp := &recordingPersistence{}
	first := true
	rp.previewInsertFn = func(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
		if first {
			first = false
			constants.RegisterPreviewSiblingMaxDefaultedFields(ctx, []string{"SortOrder__c"})
			return nil, errors.New("preview failed after sibling-max default")
		}
		return clonePreviewSObject(record), nil
	}
	captured := []FlowRecordCreatePreview{}
	pp := &previewPersistence{inner: rp, captured: &captured}

	ctx := constants.WithPreviewSyntheticRecordRegistry(context.Background())
	ctx = constants.WithPreviewSiblingMaxDefaultedFieldRegistry(ctx)
	ctx = withRecordCreateInsertMarker(ctx)

	if _, err := pp.Insert(ctx, "LineItem__c", models.SObject{"ParentId": "parent-1"}, newPreviewUser()); err == nil {
		t.Fatal("first Insert error = nil, want preview failure")
	}
	if _, err := pp.Insert(ctx, "LineItem__c", models.SObject{"ParentId": "parent-1"}, newPreviewUser()); err != nil {
		t.Fatalf("second Insert returned error: %v", err)
	}
	if len(captured) != 1 {
		t.Fatalf("captured = %#v, want only the successful preview", captured)
	}
	if len(captured[0].SiblingMaxDefaultedFields) != 0 {
		t.Fatalf("SiblingMaxDefaultedFields = %#v, want empty after failed provenance", captured[0].SiblingMaxDefaultedFields)
	}
}

// ---------------------------------------------------------------------------
// authorizeProtectedStandardFlow — ConnectorInbound path
//
// Verifies that a pack-owned standard flow is authorized when ctx carries
// ConnectorInboundExecutionAuthorizesFlow=true for the EXACT root flow name,
// AND that the returned ctx does NOT have HasStandardActionProtectedFlowExecution
// stamped (minimal privilege: standard-action-only actions must remain locked).
//
// Round-2 regression test: a DIFFERENT flow name (simulating a protected
// descendant/subflow) must be DENIED under the same ctx — the marker must not
// authorize arbitrary protected standard flows, only the routed root flow.
// ---------------------------------------------------------------------------

func TestAuthorizeProtectedStandardFlow_ConnectorInbound_AuthorizedWithoutStandardActionMarker(t *testing.T) {
	const rootFlowName = "Std_Inbound_LeadEnrich"
	const namespace = "hubspot-crm"
	const childFlowName = "Std_Inbound_LeadEnrich_SubFlow"

	exec, _ := previewExecutor(t, models.FlowMetadata{})
	// Build a pack-owned standard flow version (same shape used by other tests
	// in this file that exercise the protected-standard path).
	// DefinitionOwnerPack must match the routing namespace for authorization.
	fv := &models.FlowVersion{
		Id:                  "fv-inbound-1",
		DefinitionId:        "fd-inbound-1",
		Source:              flowSourceStandard,
		Status:              "Active",
		DefinitionOwnerPack: namespace,
	}

	// Context stamped for root flow + namespace (no standard-action stamps).
	ctx := constants.WithConnectorInboundFlowExecution(context.Background(), rootFlowName, namespace)

	outCtx, result := exec.authorizeProtectedStandardFlow(ctx, rootFlowName, fv)

	// Must be authorized — result must be nil (no denial).
	if result != nil {
		t.Fatalf("authorizeProtectedStandardFlow denied connector-inbound ctx for root flow: %+v", result)
	}

	// SECURITY: the returned ctx must NOT carry HasStandardActionProtectedFlowExecution.
	// If it did, standard-action-only actions (previewRecordUpdate, stampConvertedAnchors)
	// would be unlocked for any connector-routed flow — an authorization bypass.
	if constants.HasStandardActionProtectedFlowExecution(outCtx) {
		t.Errorf("HasStandardActionProtectedFlowExecution = true on returned ctx; want false (standard-action-only actions must stay locked)")
	}

	// The connector-inbound marker for the root flow + namespace must still be present.
	if !constants.ConnectorInboundExecutionAuthorizesFlow(outCtx, rootFlowName, namespace) {
		t.Errorf("ConnectorInboundExecutionAuthorizesFlow(%q, %q) = false on returned ctx; want true", rootFlowName, namespace)
	}

	// SECURITY (round-2 regression): a DIFFERENT protected standard flow
	// (child/subflow invoked under the same ctx) must be DENIED. The marker
	// is root-flow-scoped; it must not grant blanket authorization to all
	// pack-owned protected standard descendants.
	_, childResult := exec.authorizeProtectedStandardFlow(ctx, childFlowName, fv)
	if childResult == nil {
		t.Errorf("authorizeProtectedStandardFlow(%q) returned nil (authorized) for a different flow name under connector-inbound ctx; want a denial result — marker must not authorize arbitrary protected descendants", childFlowName)
	}
}

func TestAuthorizeProtectedStandardFlow_AsyncAfterCommitAuthorizesExactRootVersionOnly(t *testing.T) {
	const (
		rootFlowName = "Std_Async_Record_Continuation"
		rootVersion  = "fv-async-root"
	)
	exec, _ := previewExecutor(t, models.FlowMetadata{})
	root := &models.FlowVersion{
		Id:                  rootVersion,
		DefinitionId:        "fd-async-root",
		Source:              flowSourceStandard,
		Status:              "Active",
		DefinitionOwnerPack: "standard-pack",
	}
	ctx := WithAfterCommitScheduledFlowExecution(context.Background(), rootVersion)

	outCtx, result := exec.authorizeProtectedStandardFlow(ctx, rootFlowName, root)
	if result != nil {
		t.Fatalf("exact AsyncAfterCommit root was denied: %+v", result)
	}
	if constants.HasStandardActionProtectedFlowExecution(outCtx) {
		t.Fatal("AsyncAfterCommit root authorization stamped broader standard-action authority")
	}

	child := &models.FlowVersion{
		Id:                  "fv-async-child",
		DefinitionId:        "fd-async-child",
		Source:              flowSourceStandard,
		Status:              "Active",
		DefinitionOwnerPack: "standard-pack",
	}
	if _, childResult := exec.authorizeProtectedStandardFlow(ctx, "Std_Async_Child", child); childResult == nil {
		t.Fatal("different protected child version was authorized by the root marker")
	}
	if _, unmarkedResult := exec.authorizeProtectedStandardFlow(context.Background(), rootFlowName, root); unmarkedResult == nil {
		t.Fatal("protected root without an AsyncAfterCommit marker was authorized")
	}
}

func TestAuthorizeProtectedStandardFlow_ExactCompanionSubflowDelegation(t *testing.T) {
	const (
		ownerPack       = "hubspot-companion"
		parentVersionID = "fv-companion-parent"
		childFlowName   = "HubSpot_Workflow_Action_Lead_Create_Finalize"
	)
	exec, _ := previewExecutor(t, models.FlowMetadata{})
	parentIdentity := FlowExecutionIdentity{
		APIName:       "HubSpot_Workflow_Action_CampaignMember_Upsert",
		DefinitionID:  "fd-companion-parent",
		VersionID:     parentVersionID,
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	}
	ctx := withFlowRuntimeContext(context.Background(), NewFlowRuntimeContext(context.Background(), nil).WithCurrentFlow(parentIdentity))
	ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
	ctx = withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)

	childVersion := &models.FlowVersion{
		Id:                  "fv-companion-child",
		DefinitionId:        "fd-companion-child",
		Source:              flowSourceStandard,
		Status:              "Active",
		DefinitionOwnerPack: ownerPack,
	}
	if _, result := exec.authorizeProtectedStandardFlow(ctx, childFlowName, childVersion); result != nil {
		t.Fatalf("exact delegated child was denied: %+v", result)
	}
	if _, result := exec.authorizeProtectedStandardFlow(ctx, "Different_Child", childVersion); result == nil {
		t.Fatal("different child name was authorized")
	}
	crossOwner := *childVersion
	crossOwner.DefinitionOwnerPack = "different-pack"
	if _, result := exec.authorizeProtectedStandardFlow(ctx, childFlowName, &crossOwner); result == nil {
		t.Fatal("cross-owner child was authorized")
	}
	if constants.HasStandardActionProtectedFlowExecution(ctx) {
		t.Fatal("delegation stamped unrelated standard-action privilege")
	}
}

// TestAuthorizeProtectedStandardFlow_ConnectorInbound_CrossPack_Rejected verifies
// that round-3 same-pack isolation is enforced: a protected standard flow whose
// DefinitionOwnerPack does NOT match the routing connector namespace is DENIED,
// even when the flow name matches exactly. This closes the cross-pack bypass hole
// where pack A declares a route to pack B's protected flow and runs it under
// SystemContext, bypassing B's guard pipeline.
func TestAuthorizeProtectedStandardFlow_ConnectorInbound_CrossPack_Rejected(t *testing.T) {
	const rootFlowName = "Std_CRM_LeadProcess"
	const routingNamespace = "hubspot-crm"
	const flowOwnerPack = "salesforce-crm" // different pack owns this flow

	exec, _ := previewExecutor(t, models.FlowMetadata{})
	// Flow is a standard pack-owned flow, but owned by a DIFFERENT pack than
	// the routing connector namespace.
	fv := &models.FlowVersion{
		Id:                  "fv-cross-pack-1",
		DefinitionId:        "fd-cross-pack-1",
		Source:              flowSourceStandard,
		Status:              "Active",
		DefinitionOwnerPack: flowOwnerPack,
	}

	// Context stamped for the routing connector (hubspot-crm).
	ctx := constants.WithConnectorInboundFlowExecution(context.Background(), rootFlowName, routingNamespace)

	// SECURITY (round-3): cross-pack route must be REJECTED.
	// flowVersion.DefinitionOwnerPack ("salesforce-crm") != routingNamespace ("hubspot-crm").
	_, result := exec.authorizeProtectedStandardFlow(ctx, rootFlowName, fv)
	if result == nil {
		t.Errorf("authorizeProtectedStandardFlow(%q) returned nil (authorized) for a cross-pack route; want a denial result — pack A must not execute pack B's protected standard flow", rootFlowName)
	}

	// Confirm the same-pack check via the helper directly.
	if constants.ConnectorInboundExecutionAuthorizesFlow(ctx, rootFlowName, flowOwnerPack) {
		t.Errorf("ConnectorInboundExecutionAuthorizesFlow(%q, %q) = true when routing namespace is %q; want false", rootFlowName, flowOwnerPack, routingNamespace)
	}
}

// TestAuthorizeProtectedStandardFlow_ConnectorInbound_SamePack_Authorized verifies
// that a same-pack flow IS authorized (positive case for round-3 check).
func TestAuthorizeProtectedStandardFlow_ConnectorInbound_SamePack_Authorized(t *testing.T) {
	const rootFlowName = "Std_CRM_LeadProcess"
	const namespace = "salesforce-crm"

	exec, _ := previewExecutor(t, models.FlowMetadata{})
	fv := &models.FlowVersion{
		Id:                  "fv-same-pack-1",
		DefinitionId:        "fd-same-pack-1",
		Source:              flowSourceStandard,
		Status:              "Active",
		DefinitionOwnerPack: namespace, // same as routing namespace
	}

	ctx := constants.WithConnectorInboundFlowExecution(context.Background(), rootFlowName, namespace)

	_, result := exec.authorizeProtectedStandardFlow(ctx, rootFlowName, fv)
	if result != nil {
		t.Errorf("authorizeProtectedStandardFlow(%q) denied same-pack connector-inbound route: %+v", rootFlowName, result)
	}
}