package automation

// executor_record_ops_test.go — unit tests for cross-object DML and READ
// introduced in issue #571.
//
// All tests in this file use in-memory stubs (no live DB).
// Integration tests are in flow_trigger_listener_integration_test.go.

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

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/api"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/nexuscrm/shared/pkg/stagetiming"
)

func TestCallerOwnedAtomicFlowNeverDetachesRecordUpdate(t *testing.T) {
	if !shouldDetachAfterSaveSelfUpdate(context.Background()) {
		t.Fatal("ordinary after-save execution lost its self-lock detach guard")
	}
	owned := withAtomicFlowPessimisticRetryOwner(context.Background())
	if shouldDetachAfterSaveSelfUpdate(owned) {
		t.Fatal("caller-owned atomic Flow detached its record update transaction")
	}
}

func TestFilteredRecordUpdateFanoutBoundRejectsBeforeAnyEffect(t *testing.T) {
	records := make(
		[]models.SObject,
		MaxAtomicFilteredRecordOperationRows+1,
	)
	for i := range records {
		records[i] = models.SObject{
			constants.FieldID: fmt.Sprintf("record-%03d", i),
		}
	}
	query := &recordLookupQueryStub{
		result: &models.QueryResult{Records: records},
	}
	capture := &captureUpdate{}
	executor := &FlowExecutor{
		query:       query,
		persistence: &stubPersistenceCapture{capture: capture},
	}
	update := &models.FlowRecordUpdate{
		Name: "BoundedFilteredUpdate", Object: "Widget",
		FilterLogic: "and",
		Filters: []models.FlowRecordFilter{{
			Field: "ExternalKey", Operator: "EqualTo",
			Value: map[string]interface{}{"stringValue": "shared"},
		}},
	}
	_, err := executor.executeRecordUpdate(
		context.Background(),
		&models.FlowInterview{ContextData: map[string]interface{}{}},
		nil,
		update,
		nil,
	)
	if err == nil || !strings.Contains(err.Error(), "governor limit") {
		t.Fatalf("executeRecordUpdate() error = %v, want fan-out rejection", err)
	}
	if len(query.specCalls) != 1 ||
		query.specCalls[0].limit != MaxAtomicFilteredRecordOperationRows+1 {
		t.Fatalf(
			"QueryWithSpec calls = %#v, want one Id-only limit+1 preflight",
			query.specCalls,
		)
	}
	if got := query.specCalls[0].selectFields; len(got) != 1 || got[0] != constants.FieldID {
		t.Fatalf("candidate fields = %v, want [Id]", got)
	}
	if len(capture.calls) != 0 {
		t.Fatalf(
			"filtered update applied %d effects before fan-out rejection",
			len(capture.calls),
		)
	}
}

func TestFilteredRecordDeleteFanoutBoundRejectsBeforeAnyEffect(t *testing.T) {
	records := make(
		[]models.SObject,
		MaxAtomicFilteredRecordOperationRows+1,
	)
	for i := range records {
		records[i] = models.SObject{
			constants.FieldID: fmt.Sprintf("record-%03d", i),
		}
	}
	query := &recordLookupQueryStub{
		result: &models.QueryResult{Records: records},
	}
	persistence := &stubPersistenceCapture{}
	executor := &FlowExecutor{
		query:       query,
		persistence: persistence,
	}
	deleteElement := &models.FlowRecordDelete{
		Name: "BoundedFilteredDelete", Object: "Widget",
		FilterLogic: "and",
		Filters: []models.FlowRecordFilter{{
			Field: "ExternalKey", Operator: "EqualTo",
			Value: map[string]interface{}{"stringValue": "shared"},
		}},
	}
	_, err := executor.executeRecordDelete(
		context.Background(),
		&models.FlowInterview{ContextData: map[string]interface{}{}},
		deleteElement,
		nil,
	)
	if err == nil || !strings.Contains(err.Error(), "governor limit") {
		t.Fatalf("executeRecordDelete() error = %v, want fan-out rejection", err)
	}
	if len(query.specCalls) != 1 ||
		query.specCalls[0].limit != MaxAtomicFilteredRecordOperationRows+1 {
		t.Fatalf(
			"QueryWithSpec calls = %#v, want one Id-only limit+1 preflight",
			query.specCalls,
		)
	}
	if got := query.specCalls[0].selectFields; len(got) != 1 || got[0] != constants.FieldID {
		t.Fatalf("candidate fields = %v, want [Id]", got)
	}
	if len(persistence.deleteCalls) != 0 {
		t.Fatalf(
			"filtered delete applied %d effects before fan-out rejection",
			len(persistence.deleteCalls),
		)
	}
}

func TestPreparedFlowDirectCollectionUpdateRejectsBeforeAnyEffect(t *testing.T) {
	records := make([]models.SObject, MaxAtomicFilteredRecordOperationRows+1)
	for i := range records {
		records[i] = models.SObject{constants.FieldID: fmt.Sprintf("record-%03d", i)}
	}
	capture := &captureUpdate{}
	executor := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: capture},
	}
	ctx, err := withPreparedFlowDirectRecordOperationLimit(
		context.Background(),
		MaxAtomicFilteredRecordOperationRows,
	)
	if err != nil {
		t.Fatal(err)
	}
	_, err = executor.executeRecordUpdate(
		ctx,
		&models.FlowInterview{ContextData: map[string]interface{}{"records": records}},
		nil,
		&models.FlowRecordUpdate{
			Name: "DirectUpdate", Object: "Widget", InputReference: "records",
		},
		nil,
	)
	if err == nil || !strings.Contains(err.Error(), "prepared Flow governor limit") {
		t.Fatalf("executeRecordUpdate() error = %v, want prepared-flow fan-out rejection", err)
	}
	if len(capture.calls) != 0 {
		t.Fatalf("direct update applied %d effects before rejection", len(capture.calls))
	}
}

func TestPreparedFlowDirectCollectionDeleteRejectsBeforeAnyEffect(t *testing.T) {
	records := make([]interface{}, MaxAtomicFilteredRecordOperationRows+1)
	for i := range records {
		records[i] = models.SObject{constants.FieldID: fmt.Sprintf("record-%03d", i)}
	}
	persistence := &stubPersistenceCapture{}
	executor := &FlowExecutor{persistence: persistence}
	ctx, err := withPreparedFlowDirectRecordOperationLimit(
		context.Background(),
		MaxAtomicFilteredRecordOperationRows,
	)
	if err != nil {
		t.Fatal(err)
	}
	_, err = executor.executeRecordDelete(
		ctx,
		&models.FlowInterview{ContextData: map[string]interface{}{"records": records}},
		&models.FlowRecordDelete{
			Name: "DirectDelete", Object: "Widget", InputReference: "records",
		},
		nil,
	)
	if err == nil || !strings.Contains(err.Error(), "prepared Flow governor limit") {
		t.Fatalf("executeRecordDelete() error = %v, want prepared-flow fan-out rejection", err)
	}
	if len(persistence.deleteCalls) != 0 {
		t.Fatalf("direct delete applied %d effects before rejection", len(persistence.deleteCalls))
	}
}

// ---------------------------------------------------------------------------
// Stub implementations
// ---------------------------------------------------------------------------

// captureUpdate records calls to persistence.Update.
type captureUpdate struct {
	calls []updateCall
	err   error
}

type updateCall struct {
	objectName    string
	id            string
	updates       models.SObject
	pessimisticTx bool
}

type stubPessimisticTxKey struct{}

func stubInPessimisticTx(ctx context.Context) bool {
	inTx, _ := ctx.Value(stubPessimisticTxKey{}).(bool)
	return inTx
}

func (c *captureUpdate) Update(ctx context.Context, objectName string, id string, updates models.SObject, _ *models.UserSession) error {
	c.calls = append(c.calls, updateCall{
		objectName:    objectName,
		id:            id,
		updates:       updates,
		pessimisticTx: stubInPessimisticTx(ctx),
	})
	return c.err
}

// minimalPersistenceWithCapture builds a FlowExecutor wired with a minimal
// persistence stub that captures Update calls.
func minimalPersistenceWithCapture(cap *captureUpdate) *FlowExecutor {
	return &FlowExecutor{persistence: &stubPersistenceCapture{capture: cap}}
}

// stubPersistenceCapture implements PersistenceProvider using a captureUpdate
// for Update; all other methods panic or no-op.
type stubPersistenceCapture struct {
	capture            *captureUpdate
	rawRecords         map[string]models.SObject
	rawErr             error
	rawCalls           []persistenceReadCall
	lockRecords        map[string]models.SObject
	lockErr            error
	lockCalls          []persistenceReadCall
	pessimisticTxCalls int
	// insertResult, when non-nil, is returned by Insert so tests can verify the
	// engine-owned post-upsert snapshot channel (AddPostUpsertRecord).
	insertResult models.SObject
	insertCalls  []models.SObject
	// txActive controls the return value of IsTransactionActive. Set to true in
	// tests that assert a publish inside an atomic-flow boundary.
	txActive    bool
	deleteCalls []persistenceReadCall
}

type externalIDUpsertCall struct {
	objectName string
	fieldName  string
	fieldValue string
	record     models.SObject
}

type externalIDUpsertPersistenceCapture struct {
	*stubPersistenceCapture
	result  models.SObject
	created bool
	err     error
	calls   []externalIDUpsertCall
}

type externalIDGuardedUpsertCall struct {
	externalIDUpsertCall
	guardFields []string
}

type externalIDGuardedUpsertPersistenceCapture struct {
	*externalIDUpsertPersistenceCapture
	guardedResult  models.SObject
	guardedCreated bool
	guardedErr     error
	guardedCalls   []externalIDGuardedUpsertCall
}

func (s *externalIDGuardedUpsertPersistenceCapture) UpsertByExternalIDWithAdoptionGuards(
	_ context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
	record models.SObject,
	guardFields []string,
	_ *models.UserSession,
) (models.SObject, bool, error) {
	snapshot := make(models.SObject, len(record))
	for field, value := range record {
		snapshot[field] = value
	}
	s.guardedCalls = append(s.guardedCalls, externalIDGuardedUpsertCall{
		externalIDUpsertCall: externalIDUpsertCall{
			objectName: objectName,
			fieldName:  fieldName,
			fieldValue: fieldValue,
			record:     snapshot,
		},
		guardFields: append([]string(nil), guardFields...),
	})
	return s.guardedResult, s.guardedCreated, s.guardedErr
}

func (s *externalIDUpsertPersistenceCapture) UpsertByUniqueExternalID(
	_ context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
	record models.SObject,
	_ *models.UserSession,
) (models.SObject, bool, error) {
	snapshot := make(models.SObject, len(record))
	for field, value := range record {
		snapshot[field] = value
	}
	s.calls = append(s.calls, externalIDUpsertCall{
		objectName: objectName,
		fieldName:  fieldName,
		fieldValue: fieldValue,
		record:     snapshot,
	})
	return s.result, s.created, s.err
}

type persistenceReadCall struct {
	objectName    string
	id            string
	pessimisticTx bool
}

type detachedLookupCtxKey struct{}

type recordLookupQueryCall struct {
	objectName   string
	filterExpr   string
	limit        int
	detached     bool
	selectFields []string
}

type recordLookupQueryStub struct {
	mu              sync.Mutex
	calls           []recordLookupQueryCall
	specCalls       []recordLookupQueryCall
	result          *models.QueryResult
	queryFunc       func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error)
	projectCalls    int
	projectDetached []bool
	projectErr      error
	projectFunc     func(objectName string, records []models.SObject) []models.SObject
}

func (q *recordLookupQueryStub) QueryWithFilter(ctx context.Context, objectName string, filterExpr string, _ *models.UserSession, _ string, _ string, limit int) (*models.QueryResult, error) {
	detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
	q.mu.Lock()
	q.calls = append(q.calls, recordLookupQueryCall{
		objectName: objectName,
		filterExpr: filterExpr,
		limit:      limit,
		detached:   detached,
	})
	q.mu.Unlock()
	if q.queryFunc != nil {
		return q.queryFunc(ctx, objectName, filterExpr, limit)
	}
	if q.result != nil {
		return q.result, nil
	}
	return &models.QueryResult{}, nil
}

func (q *recordLookupQueryStub) Query(context.Context, api.QueryRequest, *models.UserSession) (*models.QueryResult, error) {
	return &models.QueryResult{}, nil
}

func (q *recordLookupQueryStub) QueryWithSpec(ctx context.Context, req queryspec.QuerySpec, _ *models.UserSession) (*models.QueryResult, error) {
	detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
	q.mu.Lock()
	q.specCalls = append(q.specCalls, recordLookupQueryCall{
		objectName:   req.ObjectAPIName,
		filterExpr:   req.Filter.Expression,
		limit:        req.Limit,
		detached:     detached,
		selectFields: append([]string(nil), req.SelectFields...),
	})
	q.mu.Unlock()
	if q.queryFunc != nil {
		return q.queryFunc(ctx, req.ObjectAPIName, req.Filter.Expression, req.Limit)
	}
	if q.result != nil {
		return q.result, nil
	}
	return &models.QueryResult{}, nil
}

func (q *recordLookupQueryStub) RunAnalytics(context.Context, models.AnalyticsQuery, *models.UserSession) (interface{}, error) {
	return nil, nil
}

func (q *recordLookupQueryStub) SearchSingleObject(context.Context, string, string, []string, []string, int, int, string, string, string, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (q *recordLookupQueryStub) SearchSuggestions(context.Context, string, string, []string, []string, int, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (q *recordLookupQueryStub) GlobalSearch(context.Context, string, *models.UserSession) ([]models.SearchResult, error) {
	return nil, nil
}

func (q *recordLookupQueryStub) Calculate(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	return record, nil
}

func (q *recordLookupQueryStub) ExecuteRawSQL(context.Context, string, []interface{}, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (q *recordLookupQueryStub) ProjectPreviewSyntheticRecords(ctx context.Context, objectName string, records []models.SObject, _ *models.UserSession) ([]models.SObject, error) {
	q.projectCalls++
	detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
	q.projectDetached = append(q.projectDetached, detached)
	if q.projectErr != nil {
		return nil, q.projectErr
	}
	if q.projectFunc != nil {
		return q.projectFunc(objectName, records), nil
	}
	projected := make([]models.SObject, 0, len(records))
	for _, record := range records {
		projected = append(projected, clonePreviewLookupRow(record))
	}
	return projected, nil
}

type detachingLookupPersistence struct {
	*stubPersistenceCapture
}

type guardedPersistenceCapture struct {
	*stubPersistenceCapture
	lockedCurrent models.SObject
	guardedCalls  int
}

func (p *guardedPersistenceCapture) UpdateGuarded(
	ctx context.Context,
	objectName string,
	id string,
	updates models.SObject,
	user *models.UserSession,
	guard func(context.Context, models.SObject) (bool, error),
) (bool, error) {
	p.guardedCalls++
	matched, err := guard(ctx, p.lockedCurrent)
	if err != nil || !matched {
		return matched, err
	}
	if err := p.Update(ctx, objectName, id, updates, user); err != nil {
		return false, err
	}
	return true, nil
}

func (p *detachingLookupPersistence) DetachTx(ctx context.Context) context.Context {
	return context.WithValue(ctx, detachedLookupCtxKey{}, true)
}

func (p *detachingLookupPersistence) IsTransactionActive(ctx context.Context) bool {
	if detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool); detached {
		return false
	}
	if p == nil || p.stubPersistenceCapture == nil {
		return false
	}
	return p.stubPersistenceCapture.IsTransactionActive(ctx)
}

func (s *stubPersistenceCapture) Update(ctx context.Context, objectName string, id string, updates models.SObject, user *models.UserSession) error {
	if s.capture == nil {
		return nil
	}
	return s.capture.Update(ctx, objectName, id, updates, user)
}
func (s *stubPersistenceCapture) Insert(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	snap := make(models.SObject, len(record))
	for k, v := range record {
		snap[k] = v
	}
	s.insertCalls = append(s.insertCalls, snap)
	if s.insertResult != nil {
		return s.insertResult, nil
	}
	// Default: echo back the input record (simulates persistence stamping Id etc.).
	return record, nil
}
func (s *stubPersistenceCapture) PreviewBeforeInsert(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	return record, nil
}
func (s *stubPersistenceCapture) PrelockUpdateParents(ctx context.Context, objectName string, id string, updates models.SObject, user *models.UserSession) error {
	if s.capture == nil {
		return nil
	}
	return s.capture.Update(ctx, objectName, id, updates, user)
}
func (s *stubPersistenceCapture) PreviewBeforeUpdate(_ context.Context, _ string, _ string, _ models.SObject, _ models.SObject, _ *models.UserSession) (*models.PreparedRecordUpdate, error) {
	return nil, nil
}
func (s *stubPersistenceCapture) CommitPreparedUpdate(_ context.Context, _ string, _ string, _ *models.PreparedRecordUpdate, _ models.SObject, _ *models.UserSession) error {
	return nil
}
func (s *stubPersistenceCapture) Delete(ctx context.Context, objectName string, id string, _ *models.UserSession) error {
	s.deleteCalls = append(s.deleteCalls, persistenceReadCall{
		objectName:    objectName,
		id:            id,
		pessimisticTx: stubInPessimisticTx(ctx),
	})
	return nil
}
func (s *stubPersistenceCapture) GetRaw(ctx context.Context, objectName string, id string) (models.SObject, error) {
	s.rawCalls = append(s.rawCalls, persistenceReadCall{
		objectName:    objectName,
		id:            id,
		pessimisticTx: stubInPessimisticTx(ctx),
	})
	if s.rawErr != nil {
		return nil, s.rawErr
	}
	if s.rawRecords != nil {
		if record, ok := s.rawRecords[objectName+"/"+id]; ok {
			return record, nil
		}
	}
	return nil, nil
}
func (s *stubPersistenceCapture) GetLock(ctx context.Context, objectName string, id string) (models.SObject, error) {
	s.lockCalls = append(s.lockCalls, persistenceReadCall{
		objectName:    objectName,
		id:            id,
		pessimisticTx: stubInPessimisticTx(ctx),
	})
	if s.lockErr != nil {
		return nil, s.lockErr
	}
	if s.lockRecords != nil {
		if record, ok := s.lockRecords[objectName+"/"+id]; ok {
			return record, nil
		}
	}
	if s.rawRecords != nil {
		if record, ok := s.rawRecords[objectName+"/"+id]; ok {
			return record, nil
		}
	}
	return nil, nil
}
func (s *stubPersistenceCapture) GetRecordType(_ context.Context, _ string) (string, error) {
	return "", nil
}
func (s *stubPersistenceCapture) RunInTransaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}
func (s *stubPersistenceCapture) IsTransactionActive(context.Context) bool { return s.txActive }
func (s *stubPersistenceCapture) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	s.pessimisticTxCalls++
	return fn(context.WithValue(ctx, stubPessimisticTxKey{}, true))
}
func (s *stubPersistenceCapture) DetachTx(ctx context.Context) context.Context { return ctx }
func (s *stubPersistenceCapture) RecomputeActiveFormulaProjection(_ context.Context, _ *models.UserSession, _, _ string, _ []string) (models.SObject, error) {
	return nil, nil
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

// parentRecord builds a minimal SObject that represents a related record
// (e.g. Account) reachable via a lookup relationship. It has attributes.type
// stamped so resolveDottedRecordReference can extract the object name.
func parentRecord(objectType, id string) models.SObject {
	return models.SObject{
		constants.FieldID: id,
		"attributes":      map[string]interface{}{"type": objectType},
	}
}

func TestRecordUpdate_InputAssignmentStringValueFromHydratedMetadataWritesLiteral(t *testing.T) {
	var flowMeta models.FlowMetadata
	if err := json.Unmarshal([]byte(`{
		"recordUpdates": [{
			"name": "Update_Target",
			"object": "XRuntimeTarget__c",
			"inputReference": "$Record",
			"inputAssignments": [
				{"field": "Status__c", "value": {"stringValue": "Candidate"}}
			],
			"connector": {"targetReference": "Done"}
		}]
	}`), &flowMeta); err != nil {
		t.Fatalf("unmarshal flow metadata: %v", err)
	}
	if len(flowMeta.RecordUpdates) != 1 {
		t.Fatalf("recordUpdates len = %d, want 1", len(flowMeta.RecordUpdates))
	}

	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)
	record := models.SObject{
		constants.FieldID: "target-001",
		"attributes":      map[string]interface{}{"type": "XRuntimeTarget__c"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record":   record,
			"Candidate": true,
		},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, &flowMeta.RecordUpdates[0], nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Done" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Done")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
	}
	if got := cap.calls[0].updates["Status__c"]; got != "Candidate" {
		t.Fatalf("Status__c update = %#v (%T), want %q", got, got, "Candidate")
	}
}

// ---------------------------------------------------------------------------
// Fact 1 — cross-object DML via dotted inputRef (issue #571)
// ---------------------------------------------------------------------------

// TestRecordUpdate_PlainInputRef_Unchanged_NoDot verifies that a non-dotted
// inputRef (e.g. "$Record") is not routed to the cross-object resolver.
// With a non-zero Id in a non-before-save context, $Record goes through
// persistence.Update normally (AfterSave path).
func TestRecordUpdate_PlainInputRef_Unchanged_NoDot(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		constants.FieldID: "rec-001",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": record,
		},
	}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Self",
		InputReference: "$Record",
		Object:         "Account",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "hello"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	// Non-dotted $Record with a non-zero Id in AfterSave context goes through
	// persistence.Update. The dotted-ref resolver must NOT have intercepted it.
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1 for AfterSave $Record path", len(cap.calls))
	}
	if cap.calls[0].objectName != "Account" {
		t.Errorf("Update objectName = %q, want Account", cap.calls[0].objectName)
	}
}

// TestRecordUpdate_CrossObjectInputRef_SingleHop_AllProcessTypes verifies that
// a dotted inputRef ("$Record.Parent__r") resolves to the related record and
// calls persistence.Update with the correct object name and record ID.
// The test is process-type agnostic — the code must not branch on process type.
func TestRecordUpdate_CrossObjectInputRef_SingleHop_AllProcessTypes(t *testing.T) {
	for _, processType := range []string{"Workflow", "AutoLaunchedFlow", "Flow"} {
		t.Run(processType, func(t *testing.T) {
			cap := &captureUpdate{}
			exec := minimalPersistenceWithCapture(cap)

			parent := parentRecord("Account", "acc-001")
			// $Record.Parent__r is an inline already-loaded related record.
			child := models.SObject{
				constants.FieldID: "child-001",
				"Parent__c":       "acc-001",
				"Parent__r":       parent, // inline; no DB fetch needed
				"attributes":      map[string]interface{}{"type": "Child__c"},
			}
			interview := &models.FlowInterview{
				ContextData: map[string]interface{}{
					"$Record": child,
				},
			}

			update := &models.FlowRecordUpdate{
				Name:           "Update_Parent",
				InputReference: "$Record.Parent__r",
				Object:         "", // object derived from attributes.type
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Description", Value: map[string]interface{}{"stringValue": "set by flow"}},
				},
				Connector: &models.FlowConnector{TargetReference: "Next"},
			}

			nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
			if err != nil {
				t.Fatalf("executeRecordUpdate error = %v", err)
			}
			if nextElem != "Next" {
				t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
			}
			if len(cap.calls) != 1 {
				t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
			}
			call := cap.calls[0]
			if call.objectName != "Account" {
				t.Errorf("Update objectName = %q, want Account", call.objectName)
			}
			if call.id != "acc-001" {
				t.Errorf("Update id = %q, want acc-001", call.id)
			}
			if call.updates["Description"] != "set by flow" {
				t.Errorf("Update Description = %v, want %q", call.updates["Description"], "set by flow")
			}
		})
	}
}

func TestRecordUpdate_FaultConnectorCannotSwallowHistorySequenceReplaySignal(t *testing.T) {
	replayErr := &historySequenceReplaySignalForTest{}
	cap := &captureUpdate{err: fmt.Errorf("record update failed: %w", replayErr)}
	exec := minimalPersistenceWithCapture(cap)
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"$Record": models.SObject{
			constants.FieldID: "child-001",
			"Parent__c":       "acc-001",
			"Parent__r":       parentRecord("Account", "acc-001"),
			"attributes":      map[string]interface{}{"type": "Child__c"},
		},
	}}
	update := &models.FlowRecordUpdate{
		Name:           "Update_Parent",
		InputReference: "$Record.Parent__r",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "set by flow"}},
		},
		FaultConnector: &models.FlowConnector{TargetReference: "Handle_Fault"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if nextElem != "" || !errors.Is(err, replayErr) {
		t.Fatalf("executeRecordUpdate replay result = (%q, %v), want (empty, original replay error)", nextElem, err)
	}
	if _, ok := interview.ContextData["$Flow.FaultMessage"]; ok {
		t.Fatal("history-sequence replay signal was routed to the Flow fault connector")
	}
}

// TestRecordUpdate_CrossObjectInputRef_OneHopNullFK_NoDml verifies that when
// the dotted ref resolves to a null FK (e.g. $Record.Parent__r where Parent__c
// is nil), no DML is performed and the flow progresses normally.
func TestRecordUpdate_CrossObjectInputRef_OneHopNullFK_NoDml(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	// Child record has no parent (Parent__r is nil / absent).
	child := models.SObject{
		constants.FieldID: "child-001",
		"Parent__c":       nil,
		"attributes":      map[string]interface{}{"type": "Child__c"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": child,
		},
	}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Parent",
		InputReference: "$Record.Parent__r",
		Object:         "Account",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "set by flow"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v (null FK should be a skip, not error)", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 (null FK → skip)", len(cap.calls))
	}
}

// TestRecordUpdate_DollarRecord_BeforeSave_InPlace_Unchanged verifies that
// $Record with no ID (before-save context) still takes the in-place path
// and is not misrouted to the dotted-ref resolver.
func TestRecordUpdate_DollarRecord_BeforeSave_InPlace_Unchanged(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		// No Id — simulates a before-save record (not yet persisted).
		"attributes": map[string]interface{}{"type": "Account"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": record,
		},
	}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Self",
		InputReference: "$Record",
		Object:         "Account",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Name", Value: map[string]interface{}{"stringValue": "NewName"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	// In-place path: persistence.Update must NOT be called.
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 for in-place $Record path", len(cap.calls))
	}
	// The in-place update must have modified the record in context.
	updated := interview.ContextData["$Record"].(models.SObject)
	if updated["Name"] != "NewName" {
		t.Fatalf("in-place Name = %v, want NewName", updated["Name"])
	}
}

// TestRecordUpdate_AuditAnnotation_DistinguishesSkippedFromSuccess verifies
// that when resolveDottedRecordReference returns a skip reason, the flow
// progresses normally (no error) and the audit status is "skipped".
// We test this indirectly by confirming no error and no DML.
func TestRecordUpdate_AuditAnnotation_DistinguishesSkippedFromSuccess(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	// inputRef points to a dotted path that doesn't exist in context → skip.
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Missing",
		InputReference: "$Record.SomeLookup__r",
		Object:         "SomeObject",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Name", Value: map[string]interface{}{"stringValue": "val"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v (unresolved dotted ref should skip, not error)", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 (skip path)", len(cap.calls))
	}
}

// TestResolveDottedRecordReference_NonDotted verifies that a non-dotted ref
// returns the "non-dotted-ref-routed-to-resolver" skip reason.
func TestResolveDottedRecordReference_NonDotted(t *testing.T) {
	exec := minimalExecutor()
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}

	_, _, _, reason, ok, err := exec.resolveDottedRecordReference(context.Background(), interview, nil, "$Record")
	if err != nil {
		t.Fatalf("resolveDottedRecordReference($Record): %v", err)
	}
	if ok {
		t.Fatal("resolveDottedRecordReference($Record) ok = true, want false")
	}
	if reason != "non-dotted-ref-routed-to-resolver" {
		t.Fatalf("reason = %q, want non-dotted-ref-routed-to-resolver", reason)
	}
}

// TestResolveDottedRecordReference_Inline_Success verifies that a dotted ref
// whose intermediate is inline (already in $Record) is resolved correctly.
func TestResolveDottedRecordReference_Inline_Success(t *testing.T) {
	exec := minimalExecutor()

	parent := parentRecord("Account", "acc-inline-001")
	child := models.SObject{
		constants.FieldID: "child-001",
		"Parent__r":       parent,
		"attributes":      map[string]interface{}{"type": "Child__c"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": child,
		},
	}

	record, objName, id, reason, ok, err := exec.resolveDottedRecordReference(context.Background(), interview, nil, "$Record.Parent__r")
	if err != nil {
		t.Fatalf("resolveDottedRecordReference($Record.Parent__r): %v", err)
	}
	if !ok {
		t.Fatalf("resolveDottedRecordReference ok = false, reason = %q", reason)
	}
	if objName != "Account" {
		t.Errorf("objectName = %q, want Account", objName)
	}
	if id != "acc-inline-001" {
		t.Errorf("recordId = %q, want acc-inline-001", id)
	}
	if record == nil {
		t.Fatal("record = nil, want non-nil")
	}
}

// TestRecordUpdate_NonDottedRefRoutedToResolver_DefensiveBugIndicator verifies
// that a flow variable that contains a dot but is a literal identifier (not
// a context path) is handled without panic. This is a regression-guard for
// the "non-dotted-ref-routed-to-resolver" defensive path.
func TestRecordUpdate_NonDottedRefRoutedToResolver_DefensiveBugIndicator(t *testing.T) {
	exec := minimalExecutor()
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}

	_, _, _, reason, ok, err := exec.resolveDottedRecordReference(context.Background(), interview, nil, "")
	if err != nil {
		t.Fatalf("resolveDottedRecordReference(empty): %v", err)
	}
	if ok {
		t.Fatal("resolveDottedRecordReference(\"\") ok = true, want false")
	}
	// Empty ref returns (nil, "", "", "", false) — no reason, no panic.
	if reason != "" {
		t.Fatalf("reason = %q, want empty for empty ref", reason)
	}
}

// ---------------------------------------------------------------------------
// Fact 2 surface — executeAssignment cross-object read via resolveValueCtxWithPresence
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------
// Fact 1 — cross-object DML: two-hop null intermediate hop (Gap 2)
// ---------------------------------------------------------------------------

// TestRecordUpdate_CrossObjectInputRef_NullIntermediateHop_NoDml verifies that
// when a 2-hop dotted ref's second hop resolves to a definitive null
// (firstNullHopIndex=1 = intermediate-fk-null), no DML is performed and the
// flow progresses normally.
func TestRecordUpdate_CrossObjectInputRef_NullIntermediateHop_NoDml(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	// Build a 2-hop: $Record → Parent__r (has an SObject) → Grandparent__r (nil FK).
	// The second hop is a nil FK, so firstNullHopIndex=1 → "intermediate-fk-null".
	parent := map[string]interface{}{
		constants.FieldID: "par-001",
		"Grandparent__c":  nil, // FK cleared — definitive null, present
		"attributes":      map[string]interface{}{"type": "Parent__c"},
	}
	child := models.SObject{
		constants.FieldID: "child-002",
		"Parent__c":       "par-001",
		"Parent__r":       parent,
		"attributes":      map[string]interface{}{"type": "Child__c"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": child,
		},
	}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Grandparent",
		InputReference: "$Record.Parent__r.Grandparent__r",
		Object:         "Grandparent__c",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "set by flow"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v (intermediate null FK should skip, not error)", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 (intermediate null FK → skip)", len(cap.calls))
	}
}

// TestRecordUpdate_CrossObjectInputRef_TwoHop_Workflow verifies that a two-hop
// dotted inputRef ("$Record.Parent__r.Grandparent__r") resolves to the terminal
// grandparent record and calls persistence.Update on it. This exercises the
// intermediate-hop resolution success path in resolveDottedRecordReference
// (complement of TestRecordUpdate_CrossObjectInputRef_NullIntermediateHop_NoDml).
func TestRecordUpdate_CrossObjectInputRef_TwoHop_Workflow(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	grandparent := parentRecord("Grandparent__c", "gp-001")
	parent := map[string]interface{}{
		constants.FieldID: "par-001",
		"Grandparent__c":  "gp-001",
		"Grandparent__r":  grandparent,
		"attributes":      map[string]interface{}{"type": "Parent__c"},
	}
	child := models.SObject{
		constants.FieldID: "child-001",
		"Parent__c":       "par-001",
		"Parent__r":       parent,
		"attributes":      map[string]interface{}{"type": "Child__c"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": child,
		},
	}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Grandparent_TwoHop",
		InputReference: "$Record.Parent__r.Grandparent__r",
		Object:         "",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "set by two-hop workflow"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
	}
	call := cap.calls[0]
	if call.objectName != "Grandparent__c" {
		t.Errorf("Update objectName = %q, want Grandparent__c", call.objectName)
	}
	if call.id != "gp-001" {
		t.Errorf("Update id = %q, want gp-001", call.id)
	}
	if call.updates["Description"] != "set by two-hop workflow" {
		t.Errorf("Update Description = %v, want %q", call.updates["Description"], "set by two-hop workflow")
	}
}

// ---------------------------------------------------------------------------
// Fact 1 — FieldMapping path: dotted target reference (Gap 3)
// ---------------------------------------------------------------------------

// stubFieldMappingApplier is a test-only FieldMappingApplier that writes
// a fixed set of fields into the target map.
type stubFieldMappingApplier struct {
	mappedFields map[string]interface{}
}

func (s *stubFieldMappingApplier) ApplyMappings(_ context.Context, _, _, _ string, _, target map[string]interface{}) error {
	for k, v := range s.mappedFields {
		target[k] = v
	}
	return nil
}

// executorWithFieldMapping builds a FlowExecutor with a capture persistence and
// a stub FieldMappingApplier. The stub writes mappedFields into the target SObject.
func executorWithFieldMapping(cap *captureUpdate, mappedFields map[string]interface{}) *FlowExecutor {
	return &FlowExecutor{
		persistence:   &stubPersistenceCapture{capture: cap},
		fieldMappings: &stubFieldMappingApplier{mappedFields: mappedFields},
	}
}

func TestRecordUpdate_FieldMappingStageTimingDoesNotPolluteUpdates(t *testing.T) {
	exec := executorWithFieldMapping(&captureUpdate{}, map[string]interface{}{
		"MappedField": "mapped-value",
	})
	collector := stagetiming.NewCollector()
	ctx := stagetiming.WithCollector(context.Background(), collector)
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"sourceRecord": models.SObject{"SourceField": "source-value"},
	}}
	update := &models.FlowRecordUpdate{
		Name:   "MapGenericRecord",
		Object: "GenericObject",
		FieldMapping: &models.FlowFieldMappingRef{
			SourceObject:    "SourceObject",
			SourceReference: "sourceRecord",
			TargetObject:    "GenericObject",
			ActionType:      "generic-map",
		},
	}
	updates := models.SObject{}

	if err := exec.applyRecordUpdateFieldMapping(ctx, interview, update, updates, nil); err != nil {
		t.Fatalf("applyRecordUpdateFieldMapping: %v", err)
	}
	if got := updates["MappedField"]; got != "mapped-value" {
		t.Fatalf("MappedField = %v, want mapped-value", got)
	}
	for _, key := range []string{"mapping_ms", "record_mutation_ms", "source_hash"} {
		if _, exists := updates[key]; exists {
			t.Fatalf("updates unexpectedly contain timing key %q: %#v", key, updates)
		}
	}
	snapshot := collector.Snapshot()
	if snapshot.Count(stagetiming.StageMapping) != 1 {
		t.Fatalf("mapping timing count = %d, want 1", snapshot.Count(stagetiming.StageMapping))
	}
}

func TestApplyRecordUpdatePersistenceStageTimingDoesNotPolluteUpdates(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)
	collector := stagetiming.NewCollector()
	ctx := stagetiming.WithCollector(context.Background(), collector)
	updates := models.SObject{"Status": "Open"}

	if err := exec.applyRecordUpdatePersistence(ctx, "GenericObject", "rec-1", updates, nil); err != nil {
		t.Fatalf("applyRecordUpdatePersistence: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(cap.calls))
	}
	if got := cap.calls[0].updates["Status"]; got != "Open" {
		t.Fatalf("Update Status = %v, want Open", got)
	}
	for _, key := range []string{"record_mutation_ms", "mapping_ms", "source_hash"} {
		if _, exists := cap.calls[0].updates[key]; exists {
			t.Fatalf("updates unexpectedly contain timing key %q: %#v", key, cap.calls[0].updates)
		}
	}
	snapshot := collector.Snapshot()
	if snapshot.Count(stagetiming.StageRecordMutation) != 1 {
		t.Fatalf("record mutation timing count = %d, want 1", snapshot.Count(stagetiming.StageRecordMutation))
	}
}

// TestRecordUpdate_FieldMappingTarget_DottedReference verifies that when a
// FieldMapping uses a dotted TargetReference (e.g. "$Record.Parent__r") with
// mergeMode=fillBlankOnly, the executor routes through flowFieldMappingTargetWithExecutor
// (cross-object resolver) and applies only-blank fills against the resolved record.
func TestRecordUpdate_FieldMappingTarget_DottedReference(t *testing.T) {
	// The parent record already has "Description" set, so the mapped value
	// must NOT be written (fillBlankOnly skips non-blank).
	// "Notes__c" is blank in parent → should be filled.
	parent := models.SObject{
		constants.FieldID: "par-fld-001",
		"Description":     "existing",
		"Notes__c":        "", // blank → should get mapped value
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	child := models.SObject{
		constants.FieldID: "ch-fld-001",
		"Parent__c":       "par-fld-001",
		"Parent__r":       parent,
		"attributes":      map[string]interface{}{"type": "Child__c"},
	}
	source := models.SObject{
		constants.FieldID: "src-001",
		"attributes":      map[string]interface{}{"type": "Source__c"},
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record":       child,
			"$SourceRecord": source,
		},
	}

	cap := &captureUpdate{}
	exec := executorWithFieldMapping(cap, map[string]interface{}{
		"Description": "mapped-description", // target already has this → skip
		"Notes__c":    "mapped-notes",       // target blank → fill
	})

	update := &models.FlowRecordUpdate{
		Name:           "Update_Parent_FillBlank",
		InputReference: "$Record.Parent__r",
		Object:         "Account",
		FieldMapping: &models.FlowFieldMappingRef{
			SourceObject:    "Source__c",
			SourceReference: "$SourceRecord",
			TargetReference: "$Record.Parent__r",
			TargetObject:    "Account",
			ActionType:      "convert",
			MergeMode:       "fillBlankOnly",
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	// The InputReference "$Record.Parent__r" is also dotted so the final DML
	// targets the parent Account.
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
	}
	call := cap.calls[0]
	if call.id != "par-fld-001" {
		t.Errorf("Update id = %q, want par-fld-001", call.id)
	}
	// "Description" was non-blank in target → fillBlankOnly must NOT overwrite it.
	if _, wrote := call.updates["Description"]; wrote {
		t.Errorf("fillBlankOnly wrote Description but target already had a value")
	}
	// "Notes__c" was blank in target → fillBlankOnly must write it.
	if got := call.updates["Notes__c"]; got != "mapped-notes" {
		t.Errorf("Notes__c = %v, want mapped-notes", got)
	}
}

// TestRecordUpdate_FieldMappingTarget_FallsBackToInputReference verifies that
// when TargetReference is empty and InputReference is a dotted path, the
// fillBlankOnly mode falls back to resolving InputReference via the executor
// method (same cross-object path).
func TestRecordUpdate_FieldMappingTarget_FallsBackToInputReference(t *testing.T) {
	parent := models.SObject{
		constants.FieldID: "par-fb-001",
		"Description":     "", // blank → should be filled
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	child := models.SObject{
		constants.FieldID: "ch-fb-001",
		"Parent__c":       "par-fb-001",
		"Parent__r":       parent,
		"attributes":      map[string]interface{}{"type": "Child__c"},
	}
	source := models.SObject{
		constants.FieldID: "src-fb-001",
		"attributes":      map[string]interface{}{"type": "Source__c"},
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record":       child,
			"$SourceRecord": source,
		},
	}

	cap := &captureUpdate{}
	exec := executorWithFieldMapping(cap, map[string]interface{}{
		"Description": "fallback-description",
	})

	update := &models.FlowRecordUpdate{
		Name:           "Update_Parent_FB",
		InputReference: "$Record.Parent__r",
		Object:         "Account",
		FieldMapping: &models.FlowFieldMappingRef{
			SourceObject:    "Source__c",
			SourceReference: "$SourceRecord",
			// TargetReference intentionally empty → falls back to InputReference
			TargetReference: "",
			TargetObject:    "Account",
			ActionType:      "convert",
			MergeMode:       "fillBlankOnly",
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
	}
	call := cap.calls[0]
	if call.id != "par-fb-001" {
		t.Errorf("Update id = %q, want par-fb-001", call.id)
	}
	// "Description" was blank in target → fillBlankOnly must write it.
	if got := call.updates["Description"]; got != "fallback-description" {
		t.Errorf("Description = %v, want fallback-description", got)
	}
}

// ---------------------------------------------------------------------------
// Issue #1119 — per-node recordUpdates filter gating
// ---------------------------------------------------------------------------

// TestRecordUpdate_PerNodeFilters_MatchingFilter_DMLFires verifies that a
// recordUpdates node whose Filters block matches the in-flight $Record fires
// its DML (acceptance criterion 1: Industry='Banking' → Description='A').
//
// The node has no explicit Object or InputReference — the implicit-$Record
// pattern used by "Update triggering record" elements in Flow Builder.
func TestRecordUpdate_PerNodeFilters_MatchingFilter_DMLFires(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		constants.FieldID: "acct-001",
		"Industry":        "Banking",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"$Record": record},
	}

	// Object is intentionally empty — implicit $Record targeting, which is the
	// pattern that exposed the bug (non-empty Object triggers the cross-object
	// SQL path which does evaluate the filter via DB query).
	update := &models.FlowRecordUpdate{
		Name: "NodeA",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Banking"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "A"}},
		},
		Connector: &models.FlowConnector{TargetReference: "NodeB"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "NodeB" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "NodeB")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1 (filter matches)", len(cap.calls))
	}
	if got := cap.calls[0].updates["Description"]; got != "A" {
		t.Fatalf("Description update = %v, want %q", got, "A")
	}
}

// TestRecordUpdate_PerNodeFilters_NonMatchingFilter_DMLSkipped verifies that a
// recordUpdates node whose Filters block does NOT match the in-flight $Record
// skips DML but still follows its connector (acceptance criterion 2: the
// Insurance branch does not fire when Industry='Banking').
//
// The node has no explicit Object or InputReference — the implicit-$Record
// pattern used by "Update triggering record" elements in Flow Builder.
func TestRecordUpdate_PerNodeFilters_NonMatchingFilter_DMLSkipped(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		constants.FieldID: "acct-001",
		"Industry":        "Banking",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"$Record": record},
	}

	// Object is intentionally empty — implicit $Record targeting.
	update := &models.FlowRecordUpdate{
		Name: "NodeB",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Insurance"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "B"}},
		},
		Connector: &models.FlowConnector{TargetReference: "NodeC"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "NodeC" {
		t.Fatalf("nextElem = %q, want %q (connector must be followed even when filter skips DML)", nextElem, "NodeC")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 (filter does not match)", len(cap.calls))
	}
}

// TestRecordUpdate_PerNodeFilters_BeforeSave_NonMatchingFilter_NoInPlace
// verifies that a BeforeSave recordUpdates node whose Filters block does NOT
// match skips the in-place $Record mutation (acceptance criterion: only the
// matching branch sets the field; non-matching branches are no-ops).
//
// The node has no explicit Object or InputReference — the implicit-$Record
// pattern used by "Update triggering record" elements in Flow Builder.
func TestRecordUpdate_PerNodeFilters_BeforeSave_NonMatchingFilter_NoInPlace(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		// No Id — simulates an in-flight BeforeSave record.
		"Industry":   "Banking",
		"attributes": map[string]interface{}{"type": "Account"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"$Record": record},
	}

	// Object is intentionally empty — implicit $Record targeting.
	update := &models.FlowRecordUpdate{
		Name: "NodeC",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Other"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "C"}},
		},
		Connector: &models.FlowConnector{TargetReference: "End"},
	}

	bsCtx := WithBeforeSaveFlow(context.Background())
	// Set the trigger object type so resolveBeforeSaveRecordUpdateEffectForStartObject
	// can prove this is an implicit $Record update (not an ambiguous cross-object one).
	flowRuntimeContextFromCtx(bsCtx).SetObjectType("Account")
	nextElem, err := exec.executeRecordUpdate(bsCtx, interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "End" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "End")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 (BeforeSave in-place path must also respect filter)", len(cap.calls))
	}
	// The in-place $Record must NOT have Description set.
	updated := interview.ContextData["$Record"].(models.SObject)
	if desc, ok := updated["Description"]; ok && desc != nil {
		t.Fatalf("Description = %v after non-matching filter, want nil/absent (filter should block in-place mutation)", desc)
	}
}

// TestRecordUpdate_PerNodeFilters_IsChanged_NoOldRecord_DMLFires verifies that
// when no $Record__Old is in the interview context (CREATE scenario), the
// IsChanged operator correctly returns true so the per-node filter passes and
// DML fires. This is the new-record semantics inherited from
// checkEntryConditionsCtx: a field is considered "changed" on a brand-new
// record.
func TestRecordUpdate_PerNodeFilters_IsChanged_NoOldRecord_DMLFires(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		constants.FieldID: "acct-002",
		"Industry":        "Banking",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	// No $Record__Old in context — simulates a CREATE-triggered flow.
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"$Record": record},
	}

	update := &models.FlowRecordUpdate{
		Name: "NodeIsChangedCreate",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "IsChanged", Value: map[string]interface{}{"booleanValue": true}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "created"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Next"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Next" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Next")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1 (IsChanged=true on new record)", len(cap.calls))
	}
}

// TestRecordUpdate_PerNodeFilters_IsChanged_UnchangedField_DMLSkipped verifies
// that when $Record__Old IS in the interview context with the same field value,
// the IsChanged operator correctly returns false, the per-node filter blocks
// DML, and the connector is still followed.
func TestRecordUpdate_PerNodeFilters_IsChanged_UnchangedField_DMLSkipped(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		constants.FieldID: "acct-003",
		"Industry":        "Banking",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	// $Record__Old has the SAME Industry value — Industry was not changed.
	oldRecord := models.SObject{
		constants.FieldID: "acct-003",
		"Industry":        "Banking",
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record":      record,
			"$Record__Old": oldRecord,
		},
	}

	update := &models.FlowRecordUpdate{
		Name: "NodeIsChangedUpdate",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "IsChanged", Value: map[string]interface{}{"booleanValue": true}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "should-not-write"}},
		},
		Connector: &models.FlowConnector{TargetReference: "End"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "End" {
		t.Fatalf("nextElem = %q, want %q (connector must be followed even when IsChanged=false)", nextElem, "End")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("persistence.Update called %d times, want 0 (Industry unchanged, IsChanged filter should block DML)", len(cap.calls))
	}
}

// TestRecordUpdate_PerNodeFilters_ChainedAfterSave_ContextSynced verifies that
// after a successful implicit-$Record AfterSave persistence.Update call, the
// written fields are merged back into interview.ContextData["$Record"] so that
// a subsequent chained node's per-node filter can read the committed state.
//
// Scenario:
//   - Node A: filter Industry=Banking → writes Description=A (persistence.Update called)
//   - After Node A: interview context must have Description="A"
//   - Node B: filter Description=A (reads from context) → writes Rating=Hot
//
// Without the post-persistence sync (issue #1119), Node B's filter reads the
// original nil Description, skips DML, and only 1 persistence.Update call is
// recorded. With the fix both nodes fire → 2 calls.
func TestRecordUpdate_PerNodeFilters_ChainedAfterSave_ContextSynced(t *testing.T) {
	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	record := models.SObject{
		constants.FieldID: "acct-chain-001",
		"Industry":        "Banking",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"$Record": record},
	}

	// Node A: implicit-$Record, AfterSave (record has an ID, context.Background()
	// is not a BeforeSave context). Filter Industry=Banking matches → writes Description=A.
	nodeA := &models.FlowRecordUpdate{
		Name: "NodeA",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Banking"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "A"}},
		},
		Connector: &models.FlowConnector{TargetReference: "NodeB"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, nodeA, nil)
	if err != nil {
		t.Fatalf("Node A executeRecordUpdate error = %v", err)
	}
	if nextElem != "NodeB" {
		t.Fatalf("Node A nextElem = %q, want %q", nextElem, "NodeB")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("after Node A: persistence.Update called %d times, want 1", len(cap.calls))
	}
	// Context sync: Description must be visible to Node B's filter.
	ctxRec, ok := interview.ContextData["$Record"].(models.SObject)
	if !ok {
		t.Fatal("interview.ContextData[\"$Record\"] is not a models.SObject after Node A")
	}
	if got := ctxRec["Description"]; got != "A" {
		t.Fatalf("after Node A context sync: Description = %v, want %q", got, "A")
	}

	// Node B: filter Description=A — reads the synced context; fires DML if synced.
	nodeB := &models.FlowRecordUpdate{
		Name: "NodeB",
		Filters: []models.FlowRecordFilter{
			{Field: "Description", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "A"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Rating", Value: map[string]interface{}{"stringValue": "Hot"}},
		},
		Connector: &models.FlowConnector{TargetReference: "End"},
	}

	nextElem, err = exec.executeRecordUpdate(context.Background(), interview, nil, nodeB, nil)
	if err != nil {
		t.Fatalf("Node B executeRecordUpdate error = %v", err)
	}
	if nextElem != "End" {
		t.Fatalf("Node B nextElem = %q, want %q", nextElem, "End")
	}
	if len(cap.calls) != 2 {
		t.Fatalf("after Node B: persistence.Update called %d times, want 2 (Node B filter matched synced context)", len(cap.calls))
	}
	if got := cap.calls[1].updates["Rating"]; got != "Hot" {
		t.Fatalf("Node B Rating update = %v, want %q", got, "Hot")
	}
}

// TestRecordUpdate_FilteredQuerySelfUpdate_ContextSynced verifies that after a
// successful filtered-query persistence.Update that targets the triggering
// record, the written fields are merged back into interview.ContextData["$Record"]
// so that a subsequent chained node's per-node filter reads the committed state.
//
// Scenario:
//   - Node A: Object="Account", InputReference="", Filters=[Industry EqualTo Banking]
//     → filtered-query path; query stub returns trigger record; writes
//     Description="A"; $Record must be synced with Description="A".
//   - Node B: Object="", InputReference="" (→ implicit $Record), Filters=[Description EqualTo A]
//     → per-node filter gate reads synced Description; writes Rating="Hot".
func TestRecordUpdate_FilteredQuerySelfUpdate_ContextSynced(t *testing.T) {
	cap := &captureUpdate{}
	triggerRecord := models.SObject{
		constants.FieldID: "acct-fq-001",
		"Industry":        "Banking",
		"attributes":      map[string]interface{}{"type": "Account"},
	}
	queryStub := &limitCapturingQuery{
		result: &models.QueryResult{
			Records: []models.SObject{triggerRecord},
		},
	}
	exec := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: cap},
		query:       queryStub,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"$Record": triggerRecord},
	}

	// Node A: filtered-query path. Object="Account" makes targetObject non-empty,
	// so executeRecordUpdate routes through the filtered-query block (not the
	// implicit-$Record single-record path). The query stub returns the trigger
	// record; applyRecordUpdatePersistence is called with id == triggerRecordID.
	// After the fix, $Record.Description must be synced to "A".
	nodeA := &models.FlowRecordUpdate{
		Name:   "NodeA",
		Object: "Account",
		Filters: []models.FlowRecordFilter{
			{Field: "Industry", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Banking"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Description", Value: map[string]interface{}{"stringValue": "A"}},
		},
		Connector: &models.FlowConnector{TargetReference: "NodeB"},
	}

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, nodeA, nil)
	if err != nil {
		t.Fatalf("Node A executeRecordUpdate error = %v", err)
	}
	if nextElem != "NodeB" {
		t.Fatalf("Node A nextElem = %q, want %q", nextElem, "NodeB")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("after Node A: persistence.Update called %d times, want 1", len(cap.calls))
	}
	ctxRec, ok := interview.ContextData["$Record"].(models.SObject)
	if !ok {
		t.Fatal("interview.ContextData[\"$Record\"] is not a models.SObject after Node A")
	}
	if got := ctxRec["Description"]; got != "A" {
		t.Fatalf("after Node A filtered-query context sync: Description = %v, want %q", got, "A")
	}

	// Node B: implicit-$Record path (Object="", InputReference="").
	// The per-node filter gate reads the synced Description; fires DML if synced.
	nodeB := &models.FlowRecordUpdate{
		Name: "NodeB",
		Filters: []models.FlowRecordFilter{
			{Field: "Description", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "A"}},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Rating", Value: map[string]interface{}{"stringValue": "Hot"}},
		},
		Connector: &models.FlowConnector{TargetReference: "End"},
	}

	nextElem, err = exec.executeRecordUpdate(context.Background(), interview, nil, nodeB, nil)
	if err != nil {
		t.Fatalf("Node B executeRecordUpdate error = %v", err)
	}
	if nextElem != "End" {
		t.Fatalf("Node B nextElem = %q, want %q", nextElem, "End")
	}
	if len(cap.calls) != 2 {
		t.Fatalf("after Node B: persistence.Update called %d times, want 2 (Node B filter matched synced context)", len(cap.calls))
	}
	if got := cap.calls[1].updates["Rating"]; got != "Hot" {
		t.Fatalf("Node B Rating update = %v, want %q", got, "Hot")
	}
}

// TestRecordUpdate_PerNodeFilters_ChainedMutuallyExclusive_ExplicitRecordRef
// verifies the acceptance-criteria gap left by the original issue #1119 fix:
// three chained recordUpdates nodes authored with an explicit
// InputReference="$Record" must behave identically to the existing implicit
// $Record tests. Exactly one mutually-exclusive branch may write, all-false
// chains must still traverse their connectors, and BeforeSave must accept the
// explicit "$Record" reference for in-place mutation.
func TestRecordUpdate_PerNodeFilters_ChainedMutuallyExclusive_ExplicitRecordRef(t *testing.T) {
	buildNode := func(name, industry, description, next string) *models.FlowRecordUpdate {
		return &models.FlowRecordUpdate{
			Name:           name,
			InputReference: "$Record",
			Filters: []models.FlowRecordFilter{
				{Field: "Industry", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": industry}},
			},
			InputAssignments: []models.FlowInputAssignment{
				{Field: "Description", Value: map[string]interface{}{"stringValue": description}},
			},
			Connector: &models.FlowConnector{TargetReference: next},
		}
	}

	type modeCase struct {
		name       string
		beforeSave bool
	}
	type triggerCase struct {
		name            string
		industry        string
		wantDesc        string
		matchStep       int
		wantCallsByStep []int
	}

	modes := []modeCase{
		{name: "AfterSave", beforeSave: false},
		{name: "BeforeSave", beforeSave: true},
	}
	triggers := []triggerCase{
		{name: "Banking", industry: "Banking", wantDesc: "A", matchStep: 0, wantCallsByStep: []int{1, 1, 1}},
		{name: "Insurance", industry: "Insurance", wantDesc: "B", matchStep: 1, wantCallsByStep: []int{0, 1, 1}},
		{name: "Other", industry: "Other", wantDesc: "C", matchStep: 2, wantCallsByStep: []int{0, 0, 1}},
		{name: "NoMatch", industry: "Healthcare", wantDesc: "", matchStep: -1, wantCallsByStep: []int{0, 0, 0}},
	}
	wantNexts := []string{"NodeB", "NodeC", "End"}

	for _, mode := range modes {
		t.Run(mode.name, func(t *testing.T) {
			for _, tc := range triggers {
				t.Run(tc.name, func(t *testing.T) {
					cap := &captureUpdate{}
					exec := minimalPersistenceWithCapture(cap)

					record := models.SObject{
						"Industry":   tc.industry,
						"attributes": map[string]interface{}{"type": "Account"},
					}
					if !mode.beforeSave {
						record[constants.FieldID] = "acct-explicit-record-ref-001"
					}

					interview := &models.FlowInterview{
						ContextData: map[string]interface{}{"$Record": record},
					}

					ctx := context.Background()
					if mode.beforeSave {
						ctx = WithBeforeSaveFlow(ctx)
						flowRuntimeContextFromCtx(ctx).SetObjectType("Account")
					}

					nodes := []*models.FlowRecordUpdate{
						buildNode("NodeA", "Banking", "A", "NodeB"),
						buildNode("NodeB", "Insurance", "B", "NodeC"),
						buildNode("NodeC", "Other", "C", "End"),
					}

					for i, node := range nodes {
						nextElem, err := exec.executeRecordUpdate(ctx, interview, nil, node, nil)
						if err != nil {
							t.Fatalf("%s executeRecordUpdate error = %v", node.Name, err)
						}
						if nextElem != wantNexts[i] {
							t.Fatalf("%s nextElem = %q, want %q", node.Name, nextElem, wantNexts[i])
						}

						if mode.beforeSave {
							if len(cap.calls) != 0 {
								t.Fatalf("after %s: persistence.Update called %d times, want 0 for BeforeSave in-place updates", node.Name, len(cap.calls))
							}
						} else if len(cap.calls) != tc.wantCallsByStep[i] {
							t.Fatalf("after %s: persistence.Update called %d times, want %d", node.Name, len(cap.calls), tc.wantCallsByStep[i])
						}

						ctxRecord, ok := interview.ContextData["$Record"].(models.SObject)
						if !ok {
							t.Fatalf("after %s: interview.ContextData[\"$Record\"] is not a models.SObject", node.Name)
						}
						wantDescAfterStep := ""
						if tc.matchStep >= 0 && i >= tc.matchStep {
							wantDescAfterStep = tc.wantDesc
						}
						gotDesc, hasDesc := ctxRecord["Description"]
						if wantDescAfterStep == "" {
							if hasDesc && gotDesc != nil {
								t.Fatalf("after %s: Description = %v, want nil/absent", node.Name, gotDesc)
							}
							continue
						}
						if gotDesc != wantDescAfterStep {
							t.Fatalf("after %s: Description = %v, want %q", node.Name, gotDesc, wantDescAfterStep)
						}
					}

					if mode.beforeSave {
						if len(cap.calls) != 0 {
							t.Fatalf("BeforeSave persistence.Update called %d times, want 0", len(cap.calls))
						}
					} else {
						wantCalls := tc.wantCallsByStep[len(tc.wantCallsByStep)-1]
						if len(cap.calls) != wantCalls {
							t.Fatalf("AfterSave persistence.Update called %d times, want %d", len(cap.calls), wantCalls)
						}
						if tc.wantDesc == "" {
							if len(cap.calls) != 0 {
								t.Fatalf("AfterSave no-match wrote %d times, want 0", len(cap.calls))
							}
						} else if got := cap.calls[0].updates["Description"]; got != tc.wantDesc {
							t.Fatalf("AfterSave Description update = %v, want %q", got, tc.wantDesc)
						}
					}
				})
			}
		})
	}
}

// TestWorkflowAssign_CrossObjectRead_ThreeLevel_RegressionGuard verifies that
// resolveValueCtxWithPresence can walk a 3-level dotted path and return the
// leaf value. This exercises the Surface 2A path used by executeAssignment.
func TestWorkflowAssign_CrossObjectRead_ThreeLevel_RegressionGuard(t *testing.T) {
	exec := minimalExecutor()

	grandparent := map[string]interface{}{
		"attributes": map[string]interface{}{"type": "Grandparent__c"},
		"Name":       "leaf-value",
	}
	parent := map[string]interface{}{
		"attributes":     map[string]interface{}{"type": "Parent__c"},
		"Grandparent__r": grandparent,
	}
	root := models.SObject{
		constants.FieldID: "root-001",
		"Parent__r":       parent,
		"attributes":      map[string]interface{}{"type": "Root__c"},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": root,
		},
	}

	// resolveValueCtxWithPresence wraps resolveNestedReferenceCtxWithPresence
	// for elementReference inputs.
	value := map[string]interface{}{
		"elementReference": "$Record.Parent__r.Grandparent__r.Name",
	}
	got, present := exec.resolveValueCtxWithPresence(context.Background(), interview, nil, value)
	if !present {
		t.Fatal("resolveValueCtxWithPresence present = false, want true")
	}
	if got != "leaf-value" {
		t.Fatalf("resolveValueCtxWithPresence = %v, want %q", got, "leaf-value")
	}
}

// ---------------------------------------------------------------------------
// Engine-owned post-upsert snapshot channel (issue #2080)
// ---------------------------------------------------------------------------

// TestRecordCreate_PopulatesPostUpsertRecord verifies that a successful
// executeRecordCreate publishes the persistence-returned record to the
// engine-owned registry (AddPostUpsertRecord) so later flow elements can read
// the source record id and stamped remote id without trusting $Record.
func TestRecordCreate_PopulatesPostUpsertRecord(t *testing.T) {
	const objectName = "Lead"
	const remoteIDField = "HubSpot_Id__c"

	// Stub Insert to return a record with the engine-stamped Id and remoteIDField.
	insertedRecord := models.SObject{
		constants.FieldID: "00Q-engine-id",
		remoteIDField:     "hs-remote-42",
		"FirstName":       "Test",
	}
	stub := &stubPersistenceCapture{
		insertResult: insertedRecord,
		txActive:     true, // gate requires IsTransactionActive=true for atomic-flow publish
	}
	exec := &FlowExecutor{persistence: stub}

	create := &models.FlowRecordCreate{
		Name:   "Create_Lead",
		Object: objectName,
		InputAssignments: []models.FlowInputAssignment{
			{Field: "FirstName", Value: map[string]interface{}{"stringValue": "Test"}},
		},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	// Build a ctx that carries a FlowRuntimeContext with the atomic-flow boundary
	// set, matching what runPossiblyAtomic stamps for real inbound-flow executions.
	// The gate (publishPostUpsertRecord) requires both isWithinAtomicFlowBoundary
	// and IsTransactionActive to be true before publishing.
	baseCtx := context.Background()
	frc := NewFlowRuntimeContext(baseCtx, nil)
	ctx := withFlowRuntimeContext(baseCtx, frc)
	ctx = withAtomicFlowBoundary(ctx)

	_, err := exec.executeRecordCreate(ctx, interview, nil, create, nil)
	if err != nil {
		t.Fatalf("executeRecordCreate error = %v", err)
	}

	// The engine-owned registry must carry the persistence-returned record.
	recs := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(objectName)
	if len(recs) != 1 {
		t.Fatalf("PostUpsertRecords(Lead) = %d entries after executeRecordCreate; expected 1", len(recs))
	}
	snap := recs[0]
	if id, _ := snap[constants.FieldID].(string); id != "00Q-engine-id" {
		t.Errorf("PostUpsertRecords[0] Id = %q, want 00Q-engine-id", id)
	}
	if remote, _ := snap[remoteIDField].(string); remote != "hs-remote-42" {
		t.Errorf("PostUpsertRecords[0] %s = %q, want hs-remote-42", remoteIDField, remote)
	}
}

func TestRecordCreate_AtomicExternalIDUpsertAdoptsExistingOwner(t *testing.T) {
	const (
		objectName = "Lead"
		fieldName  = "Salesforce_Identity_Key__c"
		fieldValue = "00D-org:Lead:00Q-source"
	)
	base := &stubPersistenceCapture{txActive: true}
	stub := &externalIDUpsertPersistenceCapture{
		stubPersistenceCapture: base,
		result: models.SObject{
			constants.FieldID: "00Q-local-winner",
			fieldName:         fieldValue,
		},
		created: false,
	}
	exec := &FlowExecutor{persistence: stub}
	create := &models.FlowRecordCreate{
		Name:                    "CreateOrAdoptLead",
		Object:                  objectName,
		UpsertByExternalIDField: fieldName,
		InputAssignments: []models.FlowInputAssignment{
			{Field: fieldName, Value: map[string]interface{}{"stringValue": fieldValue}},
			{Field: "LastName", Value: map[string]interface{}{"stringValue": "Inbound"}},
		},
		AssignRecordIdToReference: "leadId",
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	baseCtx := context.Background()
	frc := NewFlowRuntimeContext(baseCtx, nil)
	ctx := withAtomicFlowBoundary(withFlowRuntimeContext(baseCtx, frc))

	_, err := exec.executeRecordCreate(ctx, interview, nil, create, nil)
	if err != nil {
		t.Fatalf("executeRecordCreate error = %v", err)
	}
	if len(base.insertCalls) != 0 {
		t.Fatalf("Insert calls = %d, want 0 for external-ID upsert", len(base.insertCalls))
	}
	if len(stub.calls) != 1 {
		t.Fatalf("UpsertByExternalID calls = %d, want 1", len(stub.calls))
	}
	call := stub.calls[0]
	if call.objectName != objectName || call.fieldName != fieldName || call.fieldValue != fieldValue {
		t.Fatalf("UpsertByExternalID call = %#v", call)
	}
	if got := interview.ContextData["leadId"]; got != "00Q-local-winner" {
		t.Fatalf("leadId = %v, want adopted owner id", got)
	}
	records := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(objectName)
	if len(records) != 1 || records[0][constants.FieldID] != "00Q-local-winner" {
		t.Fatalf("post-upsert records = %#v, want adopted owner", records)
	}
}

func TestRecordCreate_AtomicExternalIDUpsertDefersMissingIdentityToPersistence(t *testing.T) {
	base := &stubPersistenceCapture{txActive: true}
	stub := &externalIDUpsertPersistenceCapture{
		stubPersistenceCapture: base,
		result: models.SObject{
			constants.FieldID:            "00Q-defaulted",
			"Salesforce_Identity_Key__c": "schema-default-or-before-save",
		},
		created: true,
	}
	exec := &FlowExecutor{persistence: stub}
	create := &models.FlowRecordCreate{
		Name:                    "CreateOrAdoptLead",
		Object:                  "Lead",
		UpsertByExternalIDField: "Salesforce_Identity_Key__c",
	}

	baseCtx := context.Background()
	ctx := withAtomicFlowBoundary(withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil)))
	_, err := exec.executeRecordCreate(
		ctx,
		&models.FlowInterview{ContextData: map[string]interface{}{}},
		nil,
		create,
		nil,
	)
	if err != nil {
		t.Fatalf("executeRecordCreate error = %v, want persistence-owned identity resolution", err)
	}
	if len(stub.calls) != 1 || stub.calls[0].fieldValue != "" || len(base.insertCalls) != 0 {
		t.Fatalf("deferred identity calls = %#v, direct inserts=%d; want one blank-value atomic upsert", stub.calls, len(base.insertCalls))
	}
}

func TestRecordCreate_AtomicExternalIDUpsertRejectsAmbiguousCaseAliases(t *testing.T) {
	base := &stubPersistenceCapture{}
	stub := &externalIDUpsertPersistenceCapture{stubPersistenceCapture: base}
	exec := &FlowExecutor{persistence: stub}
	create := &models.FlowRecordCreate{
		Name:                    "CreateOrAdoptLead",
		Object:                  "Lead",
		UpsertByExternalIDField: "Salesforce_Identity_Key__c",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Salesforce_Identity_Key__c", Value: map[string]interface{}{"stringValue": "salesforce:org:Lead:one"}},
			{Field: "salesforce_identity_key__c", Value: map[string]interface{}{"stringValue": "salesforce:org:Lead:other"}},
		},
	}

	_, err := exec.executeRecordCreate(
		context.Background(),
		&models.FlowInterview{ContextData: map[string]interface{}{}},
		nil,
		create,
		nil,
	)
	if err == nil || !strings.Contains(err.Error(), "ambiguous case-insensitive values") {
		t.Fatalf("executeRecordCreate error = %v, want ambiguous identity rejection", err)
	}
	if len(stub.calls) != 0 || len(base.insertCalls) != 0 {
		t.Fatalf("persistence mutated after ambiguous identity: upsert=%d insert=%d", len(stub.calls), len(base.insertCalls))
	}
}

func TestRecordCreate_GuardedExternalIDAdoptionRoutesOnlyOwnershipConflict(t *testing.T) {
	const (
		externalField = "Salesforce_Identity_Key__c"
		guardField    = "HubSpot_Id__c"
	)
	for _, test := range []struct {
		name     string
		err      error
		wantNext string
	}{
		{
			name:     "occupied owner uses dedicated quarantine branch",
			err:      fmt.Errorf("occupied: %w", ports.ErrExternalIDUpsertAdoptionGuardConflict),
			wantNext: "QuarantineClaim",
		},
		{
			name:     "ordinary failure remains fail closed",
			err:      errors.New("database unavailable"),
			wantNext: "GenericFault",
		},
		{
			name: "rollback ambiguity cannot route adoption branch",
			err: appErrors.NewInternalError(
				"pessimistic transaction failed with rollback ambiguity",
				fmt.Errorf("occupied: %w", ports.ErrExternalIDUpsertAdoptionGuardConflict),
			),
			wantNext: "GenericFault",
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			base := &stubPersistenceCapture{}
			plain := &externalIDUpsertPersistenceCapture{stubPersistenceCapture: base}
			stub := &externalIDGuardedUpsertPersistenceCapture{
				externalIDUpsertPersistenceCapture: plain,
				guardedErr:                         test.err,
			}
			exec := &FlowExecutor{persistence: stub}
			create := &models.FlowRecordCreate{
				Name:                      "CreateOrAdoptLead",
				Object:                    "Lead",
				UpsertByExternalIDField:   externalField,
				UpsertAdoptionGuardFields: []string{guardField},
				AdoptionConflictConnector: &models.FlowConnector{TargetReference: "QuarantineClaim"},
				FaultConnector:            &models.FlowConnector{TargetReference: "GenericFault"},
				InputAssignments: []models.FlowInputAssignment{
					{Field: externalField, Value: map[string]interface{}{"stringValue": "00D:Lead:00Q"}},
					{Field: guardField, Value: map[string]interface{}{"stringValue": "hs-42"}},
				},
			}
			interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
			next, err := exec.executeRecordCreate(context.Background(), interview, nil, create, nil)
			if err != nil || next != test.wantNext {
				t.Fatalf("executeRecordCreate() = (%q, %v), want (%q, nil)", next, err, test.wantNext)
			}
			if len(stub.guardedCalls) != 1 || len(stub.guardedCalls[0].guardFields) != 1 || stub.guardedCalls[0].guardFields[0] != guardField {
				t.Fatalf("guarded calls = %#v, want exact guard", stub.guardedCalls)
			}
			if len(plain.calls) != 0 || len(base.insertCalls) != 0 {
				t.Fatalf("guarded path degraded to another mutation: plain=%d insert=%d", len(plain.calls), len(base.insertCalls))
			}
		})
	}
}

func TestRecordCreate_AtomicExternalIDUpsertFaultConnectorCannotSwallowRetryMarker(t *testing.T) {
	const externalField = "Salesforce_Identity_Key__c"
	base := &stubPersistenceCapture{}
	stub := &externalIDUpsertPersistenceCapture{
		stubPersistenceCapture: base,
		err:                    fmt.Errorf("concurrent winner: %w", ports.ErrUniqueFieldUpsertRace),
	}
	exec := &FlowExecutor{persistence: stub}
	create := &models.FlowRecordCreate{
		Name:                    "CreateOrAdoptLead",
		Object:                  "Lead",
		UpsertByExternalIDField: externalField,
		FaultConnector:          &models.FlowConnector{TargetReference: "GenericFault"},
		InputAssignments: []models.FlowInputAssignment{{
			Field: externalField,
			Value: map[string]interface{}{"stringValue": "00D:Lead:00Q"},
		}},
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}

	next, err := exec.executeRecordCreate(context.Background(), interview, nil, create, nil)
	if next != "" || !ports.IsUniqueFieldUpsertRace(err) {
		t.Fatalf("executeRecordCreate() = (%q, %v), want empty target and retry marker", next, err)
	}
	if _, routed := interview.ContextData["$Flow.FaultMessage"]; routed {
		t.Fatal("unique upsert retry marker was routed through the ordinary FaultConnector")
	}
	if len(stub.calls) != 1 || len(base.insertCalls) != 0 {
		t.Fatalf("upsert/direct insert calls = %d/%d, want 1/0", len(stub.calls), len(base.insertCalls))
	}
}

func TestRecordCreate_GuardedExternalIDAdoptionRequiresStrongCapability(t *testing.T) {
	plain := &externalIDUpsertPersistenceCapture{stubPersistenceCapture: &stubPersistenceCapture{}}
	exec := &FlowExecutor{persistence: plain}
	create := &models.FlowRecordCreate{
		Name:                      "CreateOrAdoptLead",
		Object:                    "Lead",
		UpsertByExternalIDField:   "Salesforce_Identity_Key__c",
		UpsertAdoptionGuardFields: []string{"HubSpot_Id__c"},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Salesforce_Identity_Key__c", Value: map[string]interface{}{"stringValue": "00D:Lead:00Q"}},
			{Field: "HubSpot_Id__c", Value: map[string]interface{}{"stringValue": "hs-42"}},
		},
	}
	_, err := exec.executeRecordCreate(
		context.Background(),
		&models.FlowInterview{ContextData: map[string]interface{}{}},
		nil,
		create,
		nil,
	)
	if err == nil || !strings.Contains(err.Error(), "requires guarded atomic external-ID upsert persistence") {
		t.Fatalf("executeRecordCreate() error = %v, want guarded capability rejection", err)
	}
	if len(plain.calls) != 0 || len(plain.insertCalls) != 0 {
		t.Fatalf("unsupported guarded path mutated persistence: plain=%d insert=%d", len(plain.calls), len(plain.insertCalls))
	}
}

// TestRecordUpdate_PopulatesPostUpsertRecord verifies that applyRecordUpdatePersistence
// (the shared update persistence boundary) publishes the engine-confirmed update
// to the post-upsert registry (P1-1: all update paths covered centrally).
func TestRecordUpdate_PopulatesPostUpsertRecord(t *testing.T) {
	const objectName = "Contact"
	const remoteIDField = "External_Id__c"

	cap := &captureUpdate{}
	stub := &stubPersistenceCapture{
		capture:  cap,
		txActive: true, // gate requires IsTransactionActive=true for atomic-flow publish
	}
	exec := &FlowExecutor{persistence: stub}

	update := &models.FlowRecordUpdate{
		Name:           "Update_Contact",
		Object:         objectName,
		InputReference: "contactVar",
		InputAssignments: []models.FlowInputAssignment{
			{Field: remoteIDField, Value: map[string]interface{}{"stringValue": "ext-updated"}},
		},
	}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"contactVar": models.SObject{
				constants.FieldID: "003-existing",
				remoteIDField:     "ext-old",
			},
		},
	}

	// Build a ctx with atomic-flow boundary set — mirrors the real inbound-flow
	// execution path. The gate requires both predicates true to publish.
	baseCtx := context.Background()
	frc := NewFlowRuntimeContext(baseCtx, nil)
	ctx := withFlowRuntimeContext(baseCtx, frc)
	ctx = withAtomicFlowBoundary(ctx)

	_, err := exec.executeRecordUpdate(ctx, interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}

	// applyRecordUpdatePersistence should have published a snapshot.
	recs := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(objectName)
	if len(recs) != 1 {
		t.Fatalf("PostUpsertRecords(Contact) = %d entries after executeRecordUpdate; expected 1", len(recs))
	}
	snap := recs[0]
	if id, _ := snap[constants.FieldID].(string); id != "003-existing" {
		t.Errorf("PostUpsertRecords[0] Id = %q, want 003-existing", id)
	}
	if remote, _ := snap[remoteIDField].(string); remote != "ext-updated" {
		t.Errorf("PostUpsertRecords[0] %s = %q, want ext-updated", remoteIDField, remote)
	}
}

func TestFilteredRecordUpdate_RecheckFiltersAtWriteUsesLockedCurrentRow(t *testing.T) {
	const (
		objectName  = "Example__c"
		remoteField = "RemoteKey__c"
		activeField = "Inactive__c"
		targetID    = "example-1"
		desired     = "remote-expected"
	)

	tests := []struct {
		name        string
		locked      models.SObject
		wantMatched bool
	}{
		{
			name: "pre-query match but locked mirror is different",
			locked: models.SObject{
				constants.FieldID: targetID,
				activeField:       false,
				remoteField:       "remote-concurrent",
			},
		},
		{
			name: "pre-query match but locked lifecycle is inactive",
			locked: models.SObject{
				constants.FieldID: targetID,
				activeField:       true,
				remoteField:       "",
			},
		},
		{
			name: "locked row remains active with blank mirror",
			locked: models.SObject{
				constants.FieldID: targetID,
				activeField:       false,
				remoteField:       "",
			},
			wantMatched: true,
		},
		{
			name: "locked row remains active with same mirror",
			locked: models.SObject{
				constants.FieldID: targetID,
				activeField:       false,
				remoteField:       desired,
			},
			wantMatched: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cap := &captureUpdate{}
			persistence := &guardedPersistenceCapture{
				stubPersistenceCapture: &stubPersistenceCapture{capture: cap, txActive: true},
				lockedCurrent:          tt.locked,
			}
			query := &recordLookupQueryStub{result: &models.QueryResult{
				Records: []models.SObject{{constants.FieldID: targetID}},
			}}
			exec := &FlowExecutor{persistence: persistence, query: query}
			update := &models.FlowRecordUpdate{
				Name:                  "GuardedUpdate",
				Object:                objectName,
				FilterLogic:           "1 AND 2 AND (3 OR 4)",
				RecheckFiltersAtWrite: true,
				Filters: []models.FlowRecordFilter{
					{Field: constants.FieldID, Operator: "EqualTo", Value: stringValue(targetID)},
					{Field: activeField, Operator: "NotEqualTo", Value: true},
					{Field: remoteField, Operator: "IsBlank"},
					{Field: remoteField, Operator: "EqualTo", Value: stringValue(desired)},
				},
				InputAssignments: []models.FlowInputAssignment{
					{Field: remoteField, Value: stringValue(desired)},
				},
			}
			interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
			baseCtx := context.Background()
			ctx := withAtomicFlowBoundary(withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil)))

			if _, err := exec.executeRecordUpdate(ctx, interview, nil, update, nil); err != nil {
				t.Fatalf("executeRecordUpdate() error = %v", err)
			}
			if persistence.guardedCalls != 1 {
				t.Fatalf("UpdateGuarded calls = %d, want 1", persistence.guardedCalls)
			}
			wantWrites := 0
			if tt.wantMatched {
				wantWrites = 1
			}
			if len(cap.calls) != wantWrites {
				t.Fatalf("Update calls = %d, want %d", len(cap.calls), wantWrites)
			}
			postUpsert := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(objectName)
			if len(postUpsert) != wantWrites {
				t.Fatalf("post-upsert records = %d, want %d", len(postUpsert), wantWrites)
			}
			if mutated := flowRuntimeContextFromCtx(ctx).HasRecordMutation(); mutated != tt.wantMatched {
				t.Fatalf("record mutation marker = %v, want %v", mutated, tt.wantMatched)
			}
		})
	}
}

func TestFilteredRecordUpdate_RecheckFiltersAtWriteRejectsAfterSaveTriggerRecordBeforeAnyWrite(t *testing.T) {
	const (
		objectName = "Example__c"
		triggerID  = "example-trigger"
		otherID    = "example-other"
	)

	cap := &captureUpdate{}
	persistence := &guardedPersistenceCapture{
		stubPersistenceCapture: &stubPersistenceCapture{capture: cap, txActive: true},
		lockedCurrent: models.SObject{
			constants.FieldID: triggerID,
			"Active__c":       true,
		},
	}
	query := &recordLookupQueryStub{result: &models.QueryResult{
		// Put the non-trigger row first to prove the executor rejects the
		// complete result set before it starts processing any candidate.
		Records: []models.SObject{
			{constants.FieldID: otherID},
			{constants.FieldID: triggerID},
		},
	}}
	exec := &FlowExecutor{persistence: persistence, query: query}
	update := &models.FlowRecordUpdate{
		Name:                  "GuardedAfterSaveSelfUpdate",
		Object:                objectName,
		RecheckFiltersAtWrite: true,
		Filters: []models.FlowRecordFilter{
			{Field: "Active__c", Operator: "EqualTo", Value: true},
		},
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Status__c", Value: stringValue("ready")},
		},
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"$Record": models.SObject{constants.FieldID: triggerID},
	}}
	baseCtx := context.Background()
	ctx := withAtomicFlowBoundary(withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil)))

	_, err := exec.executeRecordUpdate(ctx, interview, nil, update, nil)
	if err == nil || !strings.Contains(err.Error(), "guarded after-save RecordUpdate") {
		t.Fatalf("executeRecordUpdate() error = %v, want guarded after-save self-update rejection", err)
	}
	if persistence.guardedCalls != 0 {
		t.Fatalf("UpdateGuarded calls = %d, want 0", persistence.guardedCalls)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("Update calls = %d, want 0", len(cap.calls))
	}
	if got := len(flowRuntimeContextFromCtx(ctx).PostUpsertRecords(objectName)); got != 0 {
		t.Fatalf("post-upsert records = %d, want 0", got)
	}
	if flowRuntimeContextFromCtx(ctx).HasRecordMutation() {
		t.Fatal("record mutation marker = true, want false")
	}
}

func TestGuardedRecordUpdateFiltersFreezeRHSOnceForSelectorAndLockedGuard(t *testing.T) {
	const field = "DynamicValue__c"
	exec := &FlowExecutor{}
	resolveCalls := 0
	resolved, err := exec.resolveCurrentRecordFiltersWithResolver("", []models.FlowRecordFilter{{
		Field: field, Operator: "EqualTo", Value: map[string]interface{}{"elementReference": "$Flow.CurrentDateTime"},
	}}, func(_ int, _ interface{}) interface{} {
		resolveCalls++
		return fmt.Sprintf("frozen-%d", resolveCalls)
	})
	if err != nil {
		t.Fatalf("resolveCurrentRecordFiltersWithResolver() error = %v", err)
	}

	selector, err := exec.buildGuardedFilterFromResolvedConditions(resolved, "", false)
	if err != nil {
		t.Fatalf("buildGuardedFilterFromResolvedConditions() error = %v", err)
	}
	matched, err := exec.lockedRecordMatchesResolvedFilters(context.Background(), models.SObject{field: "frozen-1"}, resolved, "")
	if err != nil {
		t.Fatalf("lockedRecordMatchesResolvedFilters() error = %v", err)
	}
	if resolveCalls != 1 {
		t.Fatalf("RHS resolver calls = %d, want exactly 1", resolveCalls)
	}
	if !strings.Contains(selector, "'frozen-1'") || strings.Contains(selector, "frozen-2") {
		t.Fatalf("selector = %q, want only frozen RHS", selector)
	}
	if !matched {
		t.Fatal("locked guard = false, want same frozen RHS to match")
	}
}

func TestGuardedRecordUpdateProcessBuilderSelectorMatchesLockedNullSemantics(t *testing.T) {
	const (
		inactiveField = "Inactive__c"
		remoteField   = "RemoteKey__c"
		desired       = "remote-expected"
	)
	exec := &FlowExecutor{}
	resolved, err := exec.resolveCurrentRecordFiltersWithResolver("", []models.FlowRecordFilter{
		{Field: inactiveField, Operator: "NotEqualTo", Value: true},
		{Field: remoteField, Operator: "IsBlank"},
		{Field: remoteField, Operator: "EqualTo", Value: desired},
	}, func(_ int, value interface{}) interface{} { return value })
	if err != nil {
		t.Fatalf("resolveCurrentRecordFiltersWithResolver() error = %v", err)
	}
	const logic = "1 AND (2 OR 3)"
	selector, err := exec.buildGuardedFilterFromResolvedConditions(resolved, logic, true)
	if err != nil {
		t.Fatalf("buildGuardedFilterFromResolvedConditions() error = %v", err)
	}
	for _, want := range []string{
		"(Inactive__c != true OR Inactive__c IS NULL)",
		"FLOWISBLANK(RemoteKey__c)",
		"RemoteKey__c = 'remote-expected'",
	} {
		if !strings.Contains(selector, want) {
			t.Fatalf("selector = %q, want %q", selector, want)
		}
	}

	tests := []struct {
		name   string
		locked models.SObject
		want   bool
	}{
		{name: "null inactive and blank mirror", locked: models.SObject{inactiveField: nil, remoteField: ""}, want: true},
		{name: "false inactive and same mirror", locked: models.SObject{inactiveField: false, remoteField: desired}, want: true},
		{name: "inactive row", locked: models.SObject{inactiveField: true, remoteField: ""}},
		{name: "different mirror", locked: models.SObject{inactiveField: false, remoteField: "remote-other"}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			matched, matchErr := exec.lockedRecordMatchesResolvedFilters(context.Background(), tt.locked, resolved, logic)
			if matchErr != nil {
				t.Fatalf("lockedRecordMatchesResolvedFilters() error = %v", matchErr)
			}
			if matched != tt.want {
				t.Fatalf("locked match = %v, want %v", matched, tt.want)
			}
		})
	}
}

func TestFilteredRecordUpdate_RecheckFiltersAtWriteCanonicalizesMultilineLogic(t *testing.T) {
	const (
		targetID    = "example-1"
		activeField = "Active__c"
	)
	capture := &captureUpdate{}
	persistence := &guardedPersistenceCapture{
		stubPersistenceCapture: &stubPersistenceCapture{capture: capture, txActive: true},
		lockedCurrent: models.SObject{
			constants.FieldID: targetID,
			activeField:       false,
		},
	}
	query := &recordLookupQueryStub{result: &models.QueryResult{Records: []models.SObject{{constants.FieldID: targetID}}}}
	exec := &FlowExecutor{persistence: persistence, query: query}
	update := &models.FlowRecordUpdate{
		Name:                  "GuardedMultilineLogic",
		Object:                "Example__c",
		RecheckFiltersAtWrite: true,
		FilterLogic:           "1 AND\n2",
		Filters: []models.FlowRecordFilter{
			{Field: constants.FieldID, Operator: "EqualTo", Value: stringValue(targetID)},
			{Field: activeField, Operator: "EqualTo", Value: true},
		},
		InputAssignments: []models.FlowInputAssignment{{
			Field: "RemoteKey__c", Value: stringValue("remote-expected"),
		}},
	}
	ctx := withAtomicFlowBoundary(withFlowRuntimeContext(context.Background(), NewFlowRuntimeContext(context.Background(), nil)))
	if _, err := exec.executeRecordUpdate(ctx, &models.FlowInterview{ContextData: map[string]interface{}{}}, nil, update, nil); err != nil {
		t.Fatalf("executeRecordUpdate() error = %v", err)
	}
	if len(query.specCalls) != 1 {
		t.Fatalf("QueryWithSpec calls = %d, want 1", len(query.specCalls))
	}
	for _, want := range []string{"Id = 'example-1'", "Active__c = true"} {
		if !strings.Contains(query.specCalls[0].filterExpr, want) {
			t.Fatalf("selector = %q, want condition %q", query.specCalls[0].filterExpr, want)
		}
	}
	if got := query.specCalls[0].selectFields; len(got) != 1 || got[0] != constants.FieldID {
		t.Fatalf("candidate fields = %v, want [Id]", got)
	}
	if persistence.guardedCalls != 1 || len(capture.calls) != 0 {
		t.Fatalf("guarded calls/writes = %d/%d, want 1/0", persistence.guardedCalls, len(capture.calls))
	}
	if got := len(flowRuntimeContextFromCtx(ctx).PostUpsertRecords("Example__c")); got != 0 {
		t.Fatalf("post-upsert records = %d, want 0", got)
	}
}

func TestValidateGuardedRecordUpdateFilterShapeCanonicalizesLineBreaks(t *testing.T) {
	filters := []models.FlowRecordFilter{
		{Field: constants.FieldID, Operator: "EqualTo", Value: stringValue("example-1")},
		{Field: "Active__c", Operator: "EqualTo", Value: true},
	}
	for _, logic := range []string{"1 AND\n2", "1 AND\r2"} {
		canonical, err := validateGuardedRecordUpdateFilterShape(filters, logic)
		if err != nil {
			t.Fatalf("validateGuardedRecordUpdateFilterShape(%q) error = %v", logic, err)
		}
		if canonical != "1 AND 2" {
			t.Fatalf("canonical filterLogic = %q, want 1 AND 2", canonical)
		}
	}
}

func TestFilteredRecordUpdate_RecheckFiltersAtWriteRejectsInvalidFilterContractBeforeQuery(t *testing.T) {
	const targetID = "example-1"
	tests := []struct {
		name        string
		filters     []models.FlowRecordFilter
		filterLogic string
		wantErr     string
	}{
		{
			name:    "unknown operator",
			filters: []models.FlowRecordFilter{{Field: constants.FieldID, Operator: "EqualTypo", Value: stringValue(targetID)}},
			wantErr: "unsupported operator",
		},
		{
			name:    "between outside narrow contract",
			filters: []models.FlowRecordFilter{{Field: constants.FieldID, Operator: "Between", Value: []interface{}{targetID, targetID}}},
			wantErr: "unsupported operator",
		},
		{
			name:    "in outside narrow contract",
			filters: []models.FlowRecordFilter{{Field: constants.FieldID, Operator: "In", Value: []interface{}{targetID}}},
			wantErr: "unsupported operator",
		},
		{
			name:    "not in outside narrow contract",
			filters: []models.FlowRecordFilter{{Field: constants.FieldID, Operator: "NotIn", Value: []interface{}{targetID}}},
			wantErr: "unsupported operator",
		},
		{
			name:    "invalid field identifier",
			filters: []models.FlowRecordFilter{{Field: "Bad Field", Operator: "EqualTo", Value: stringValue(targetID)}},
			wantErr: "invalid field",
		},
		{
			name:    "blank field identifier",
			filters: []models.FlowRecordFilter{{Field: "", Operator: "EqualTo", Value: stringValue(targetID)}},
			wantErr: "invalid field",
		},
		{
			name:    "equal blank resolved value",
			filters: []models.FlowRecordFilter{{Field: "RemoteKey__c", Operator: "EqualTo", Value: stringValue("")}},
			wantErr: "nonblank comparison value",
		},
		{
			name:    "not equal nil resolved value",
			filters: []models.FlowRecordFilter{{Field: "RemoteKey__c", Operator: "NotEqualTo"}},
			wantErr: "nonblank comparison value",
		},
		{
			name:        "out of range filter logic",
			filters:     []models.FlowRecordFilter{{Field: constants.FieldID, Operator: "EqualTo", Value: stringValue(targetID)}},
			filterLogic: "1 AND 2",
			wantErr:     "out of range",
		},
		{
			name: "bad filter logic parentheses",
			filters: []models.FlowRecordFilter{
				{Field: constants.FieldID, Operator: "EqualTo", Value: stringValue(targetID)},
				{Field: "Inactive__c", Operator: "EqualTo", Value: false},
				{Field: "RemoteKey__c", Operator: "IsBlank"},
			},
			filterLogic: "1 AND (2 OR 3",
			wantErr:     "unmatched opening parenthesis",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			capture := &captureUpdate{}
			persistence := &guardedPersistenceCapture{
				stubPersistenceCapture: &stubPersistenceCapture{capture: capture, txActive: true},
				lockedCurrent:          models.SObject{constants.FieldID: targetID},
			}
			query := &recordLookupQueryStub{result: &models.QueryResult{Records: []models.SObject{{constants.FieldID: targetID}}}}
			exec := &FlowExecutor{persistence: persistence, query: query}
			update := &models.FlowRecordUpdate{
				Name:                  "GuardedUpdate",
				Object:                "Example__c",
				RecheckFiltersAtWrite: true,
				Filters:               tt.filters,
				FilterLogic:           tt.filterLogic,
				InputAssignments: []models.FlowInputAssignment{{
					Field: "RemoteKey__c", Value: stringValue("remote-expected"),
				}},
			}
			interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
			_, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
				t.Fatalf("executeRecordUpdate() error = %v, want %q", err, tt.wantErr)
			}
			if len(query.calls) != 0 || len(query.specCalls) != 0 {
				t.Fatalf("query calls = filter:%d spec:%d, want zero", len(query.calls), len(query.specCalls))
			}
			if persistence.guardedCalls != 0 || len(capture.calls) != 0 {
				t.Fatalf("guarded calls = %d, writes = %d; want zero", persistence.guardedCalls, len(capture.calls))
			}
		})
	}
}

func TestFilteredRecordUpdate_OrdinaryUnknownOperatorKeepsLegacySelectorBehavior(t *testing.T) {
	const targetID = "example-1"
	capture := &captureUpdate{}
	persistence := &stubPersistenceCapture{capture: capture}
	query := &recordLookupQueryStub{result: &models.QueryResult{Records: []models.SObject{{constants.FieldID: targetID}}}}
	exec := &FlowExecutor{persistence: persistence, query: query}
	update := &models.FlowRecordUpdate{
		Name:    "OrdinaryUpdate",
		Object:  "Example__c",
		Filters: []models.FlowRecordFilter{{Field: constants.FieldID, Operator: "EqualTypo", Value: stringValue(targetID)}},
		InputAssignments: []models.FlowInputAssignment{{
			Field: "Name", Value: stringValue("legacy-write"),
		}},
	}
	if _, err := exec.executeRecordUpdate(context.Background(), &models.FlowInterview{ContextData: map[string]interface{}{}}, nil, update, nil); err != nil {
		t.Fatalf("executeRecordUpdate() error = %v", err)
	}
	if len(query.specCalls) != 1 || query.specCalls[0].filterExpr != "1=1" {
		t.Fatalf("query calls = %#v, want one legacy 1=1 selector", query.specCalls)
	}
	if len(capture.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(capture.calls))
	}
}

func TestFilteredRecordUpdate_RecheckFiltersAtWriteFailsClosedWithoutPersistenceCapability(t *testing.T) {
	const targetID = "example-1"
	persistence := &stubPersistenceCapture{capture: &captureUpdate{}, txActive: true}
	exec := &FlowExecutor{
		persistence: persistence,
		query: &recordLookupQueryStub{result: &models.QueryResult{
			Records: []models.SObject{{constants.FieldID: targetID}},
		}},
	}
	update := &models.FlowRecordUpdate{
		Name:                  "GuardedUpdate",
		Object:                "Example__c",
		RecheckFiltersAtWrite: true,
		Filters: []models.FlowRecordFilter{{
			Field: constants.FieldID, Operator: "EqualTo", Value: stringValue(targetID),
		}},
		InputAssignments: []models.FlowInputAssignment{{
			Field: "RemoteKey__c", Value: stringValue("remote-expected"),
		}},
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	ctx := withAtomicFlowBoundary(withFlowRuntimeContext(context.Background(), NewFlowRuntimeContext(context.Background(), nil)))

	_, err := exec.executeRecordUpdate(ctx, interview, nil, update, nil)
	if err == nil || !strings.Contains(err.Error(), "requires locked-current persistence support") {
		t.Fatalf("executeRecordUpdate() error = %v, want missing guarded persistence capability", err)
	}
	if len(persistence.capture.calls) != 0 {
		t.Fatalf("ordinary Update calls = %d, want zero", len(persistence.capture.calls))
	}
}

// TestPublishPostUpsertRecord_PublishValidityGate is the regression test for
// #2080 Gate-0 v7 (unified publish-validity invariant). It verifies that
// publishPostUpsertRecord enforces both gate predicates:
//
//	(a) A DetachTx'd / non-atomic ctx (IsTransactionActive=false) → registry
//	    stays EMPTY. The register action's GetLock cannot see a write that is
//	    not in the ambient atomic-flow tx, so publishing it would cause stale
//	    evidence.
//
//	(b) An atomic ctx (isWithinAtomicFlowBoundary=true AND
//	    IsTransactionActive=true) → record IS published.
//
// Revert-proof note: removing the IsTransactionActive predicate from
// publishPostUpsertRecord makes sub-test (a) FAIL (detached write leaks into
// registry), confirming that predicate is load-bearing.
func TestPublishPostUpsertRecord_PublishValidityGate(t *testing.T) {
	const object = "Lead"
	record := models.SObject{constants.FieldID: "00Q-test-id", "External_Id__c": "ext-42"}

	t.Run("detached_ctx_skips_publish", func(t *testing.T) {
		// IsTransactionActive=false simulates a DetachTx'd write context: the
		// ambient-tx field has been cleared so the register's GetLock cannot see it.
		stub := &stubPersistenceCapture{txActive: false}
		exec := &FlowExecutor{persistence: stub}

		// ctx carries the atomic boundary marker (FlowRuntimeContext field), which
		// DetachTx preserves — confirming that isWithinAtomicFlowBoundary alone is
		// insufficient and IsTransactionActive is the decisive gate predicate.
		baseCtx := context.Background()
		ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
		ctx = withAtomicFlowBoundary(ctx)

		exec.publishPostUpsertRecord(ctx, object, record)

		recs := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(object)
		if len(recs) != 0 {
			t.Errorf("detached ctx: expected 0 registry entries (gate should block), got %d", len(recs))
		}
	})

	t.Run("atomic_ctx_publishes", func(t *testing.T) {
		// Both predicates true: within atomic boundary AND transaction is active.
		stub := &stubPersistenceCapture{txActive: true}
		exec := &FlowExecutor{persistence: stub}

		baseCtx := context.Background()
		ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
		ctx = withAtomicFlowBoundary(ctx)

		exec.publishPostUpsertRecord(ctx, object, record)

		recs := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(object)
		if len(recs) != 1 {
			t.Fatalf("atomic ctx: expected 1 registry entry, got %d", len(recs))
		}
		if id, _ := recs[0][constants.FieldID].(string); id != "00Q-test-id" {
			t.Errorf("registry[0] Id = %q, want 00Q-test-id", id)
		}
	})

	t.Run("non_atomic_boundary_skips_publish", func(t *testing.T) {
		// isWithinAtomicFlowBoundary=false (no withAtomicFlowBoundary call):
		// the first gate predicate blocks even if IsTransactionActive is true.
		stub := &stubPersistenceCapture{txActive: true}
		exec := &FlowExecutor{persistence: stub}

		baseCtx := context.Background()
		ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
		// Deliberately NOT calling withAtomicFlowBoundary.

		exec.publishPostUpsertRecord(ctx, object, record)

		recs := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(object)
		if len(recs) != 0 {
			t.Errorf("non-atomic ctx: expected 0 registry entries (gate should block), got %d", len(recs))
		}
	})
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_UsesDetachedQueryAndAppendsMatchingPostUpsertRow(t *testing.T) {
	query := &recordLookupQueryStub{}
	query.queryFunc = func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
		detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
		if detached {
			return &models.QueryResult{Done: true}, nil
		}
		if objectName != "Account" || !strings.Contains(filterExpr, "001-created") || limit != 1 {
			t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
		}
		return &models.QueryResult{Done: true, TotalSize: 1, Records: []models.SObject{{
			constants.FieldID: "001-created",
			"Name":            "Created In Tx",
		}}}, nil
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-created",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:                     "LookupCreatedAccount",
		Object:                   "Account",
		OutputReference:          "$lookedAccount",
		GetFirstRecordOnly:       true,
		StoreOutputAutomatically: false,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-created",
		"Name":            "Created In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must read the ambient atomic tx")
	}

	got, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want projected Account row", interview.ContextData["$lookedAccount"])
	}
	if id, _ := got[constants.FieldID].(string); id != "001-created" {
		t.Fatalf("overlay Account Id = %q, want %q", id, "001-created")
	}
	if name, _ := got["Name"].(string); name != "Created In Tx" {
		t.Fatalf("overlay Account Name = %q, want %q", name, "Created In Tx")
	}
}

func TestRecordLookup_BeforeSaveRequestCache_ReusesClonedResult(t *testing.T) {
	query := &recordLookupQueryStub{
		result: &models.QueryResult{
			Done:      true,
			TotalSize: 1,
			Records: []models.SObject{{
				constants.FieldID: "001-cache",
				"Name":            "Cached Account",
			}},
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-cache",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupCachedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	flowRuntimeContextFromCtx(ctx).SetObjectType("Lead")

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("first executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 1 {
		t.Fatalf("after first lookup, QueryWithFilter() calls = %d, want 1", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("first lookup did not use detached ctx")
	}
	first, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || first == nil {
		t.Fatalf("first ContextData[$lookedAccount] = %#v, want Account row", interview.ContextData["$lookedAccount"])
	}
	first["Name"] = "mutated output"

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("second executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 1 {
		t.Fatalf("after second lookup, QueryWithFilter() calls = %d, want 1 cache hit", len(query.calls))
	}
	second, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || second == nil {
		t.Fatalf("second ContextData[$lookedAccount] = %#v, want Account row", interview.ContextData["$lookedAccount"])
	}
	if got := second["Name"]; got != "Cached Account" {
		t.Fatalf("second cached Account Name = %v, want original cached clone", got)
	}
}

func TestRecordLookup_BeforeSaveSingleflightCoalescesConcurrentMisses(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	var queryCalls int32

	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			if n := atomic.AddInt32(&queryCalls, 1); n == 1 {
				close(started)
				select {
				case <-release:
				case <-ctx.Done():
					return nil, ctx.Err()
				}
			}
			return &models.QueryResult{
				Done:      true,
				TotalSize: 1,
				Records: []models.SObject{{
					constants.FieldID: "001-shared",
					"Name":            "Shared Account",
				}},
			}, nil
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := NewFlowExecutor(nil, nil, nil, persistence, query, nil, nil)

	lookup := &models.FlowRecordLookup{
		Name:               "LookupSharedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}
	runLookup := func() error {
		trigger := models.SObject{
			constants.FieldID: "00Q-trigger",
			"AccountId":       "001-shared",
		}
		models.StampObjectType(trigger, "Lead")
		interview := &models.FlowInterview{
			ContextData: map[string]interface{}{
				"$Record": trigger,
			},
		}
		baseCtx := context.Background()
		ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
		ctx = WithBeforeSaveFlow(ctx)
		flowRuntimeContextFromCtx(ctx).SetObjectType("Lead")

		if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
			return err
		}
		record, ok := interview.ContextData["$lookedAccount"].(models.SObject)
		if !ok || record == nil {
			return fmt.Errorf("ContextData[$lookedAccount] = %#v, want Account row", interview.ContextData["$lookedAccount"])
		}
		if got := record["Name"]; got != "Shared Account" {
			return fmt.Errorf("looked Account Name = %v, want Shared Account", got)
		}
		return nil
	}

	errCh := make(chan error, 2)
	go func() { errCh <- runLookup() }()
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("first lookup did not start")
	}
	go func() { errCh <- runLookup() }()
	time.Sleep(25 * time.Millisecond)
	close(release)

	for i := 0; i < 2; i++ {
		if err := <-errCh; err != nil {
			t.Fatalf("lookup %d error = %v", i+1, err)
		}
	}
	if got := atomic.LoadInt32(&queryCalls); got != 1 {
		t.Fatalf("query calls = %d, want 1 coalesced call", got)
	}
	query.mu.Lock()
	recordedCalls := len(query.calls)
	query.mu.Unlock()
	if recordedCalls != 1 {
		t.Fatalf("recorded QueryWithFilter calls = %d, want 1", recordedCalls)
	}
}

func TestRecordLookup_BeforeSaveSameObjectActiveTxSkipsSingleflight(t *testing.T) {
	started := make(chan struct{})
	secondStarted := make(chan struct{})
	release := make(chan struct{})
	var queryCalls int32

	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			switch atomic.AddInt32(&queryCalls, 1) {
			case 1:
				close(started)
				select {
				case <-release:
				case <-ctx.Done():
					return nil, ctx.Err()
				}
			case 2:
				close(secondStarted)
			}
			return &models.QueryResult{
				Done:      true,
				TotalSize: 1,
				Records: []models.SObject{{
					constants.FieldID: "00Q-self",
					"Name":            "Self Lead",
				}},
			}, nil
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	lookup := &models.FlowRecordLookup{
		Name:               "LookupSelfLead",
		Object:             "Lead",
		OutputReference:    "$lookedLead",
		GetFirstRecordOnly: true,
		CoalesceInFlight:   true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Email",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.Email"},
		}},
	}
	runLookup := func() error {
		trigger := models.SObject{
			constants.FieldID: "00Q-trigger",
			"Email":           "self@example.invalid",
		}
		models.StampObjectType(trigger, "Lead")
		interview := &models.FlowInterview{
			ContextData: map[string]interface{}{
				"$Record": trigger,
			},
		}
		baseCtx := context.Background()
		ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
		ctx = WithBeforeSaveFlow(ctx)
		flowRuntimeContextFromCtx(ctx).SetObjectType("Lead")

		if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
			return err
		}
		record, ok := interview.ContextData["$lookedLead"].(models.SObject)
		if !ok || record == nil {
			return fmt.Errorf("ContextData[$lookedLead] = %#v, want Lead row", interview.ContextData["$lookedLead"])
		}
		return nil
	}

	errCh := make(chan error, 2)
	go func() { errCh <- runLookup() }()
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("first lookup did not start")
	}
	go func() { errCh <- runLookup() }()
	select {
	case <-secondStarted:
	case <-time.After(time.Second):
		close(release)
		t.Fatal("second same-object lookup did not run its own query")
	}
	close(release)

	for i := 0; i < 2; i++ {
		if err := <-errCh; err != nil {
			t.Fatalf("lookup %d error = %v", i+1, err)
		}
	}
	if got := atomic.LoadInt32(&queryCalls); got != 2 {
		t.Fatalf("query calls = %d, want 2 independent active-tx lookups", got)
	}
	query.mu.Lock()
	defer query.mu.Unlock()
	for i, call := range query.calls {
		if call.detached {
			t.Fatalf("query call %d detached = true, want same-object before-save ambient tx", i+1)
		}
	}
}

func TestRecordLookup_AfterSaveSameObjectActiveTxCoalesceFallbackUsesAmbientTx(t *testing.T) {
	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
			if detached {
				return &models.QueryResult{Done: true}, nil
			}
			if objectName != "Lead" || !strings.Contains(filterExpr, "same@example.invalid") {
				t.Fatalf("QueryWithFilter(%q, %q), want Lead lookup by trigger Email", objectName, filterExpr)
			}
			return &models.QueryResult{
				Done:      true,
				TotalSize: 1,
				Records: []models.SObject{{
					constants.FieldID: "00Q-trigger",
					"Email":           "same@example.invalid",
				}},
			}, nil
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"Email":           "same@example.invalid",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupSelfLead",
		Object:             "Lead",
		OutputReference:    "$lookedLead",
		GetFirstRecordOnly: true,
		CoalesceInFlight:   true,
		SortField:          "CreatedDate",
		SortOrder:          "DESC",
		Filters: []models.FlowRecordFilter{{
			Field:    "Email",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.Email"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).SetObjectType("Lead")

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	query.mu.Lock()
	calls := append([]recordLookupQueryCall(nil), query.calls...)
	query.mu.Unlock()
	if len(calls) != 1 {
		t.Fatalf("QueryWithFilter calls = %d, want 1 fallback lookup", len(calls))
	}
	if calls[0].detached {
		t.Fatal("after-save same-object CoalesceInFlight fallback detached active tx; want ambient trigger transaction")
	}
	got, ok := interview.ContextData["$lookedLead"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedLead] = %#v, want Lead row from ambient tx", interview.ContextData["$lookedLead"])
	}
	if id, _ := got[constants.FieldID].(string); id != "00Q-trigger" {
		t.Fatalf("looked Lead Id = %q, want triggering row", id)
	}
}

func TestRecordLookup_AfterSaveSameObjectActiveTxBatchableLookupSkipsBatchDetach(t *testing.T) {
	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
			if detached {
				return &models.QueryResult{Done: true}, nil
			}
			if objectName != "Lead" || !strings.Contains(filterExpr, "00Q-trigger") {
				t.Fatalf("QueryWithFilter(%q, %q), want Lead lookup by trigger Id", objectName, filterExpr)
			}
			return &models.QueryResult{
				Done:      true,
				TotalSize: 1,
				Records: []models.SObject{{
					constants.FieldID: "00Q-trigger",
					"Email":           "same@example.invalid",
				}},
			}, nil
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{
		query:               query,
		persistence:         persistence,
		recordLookupBatcher: newFlowRecordLookupBatcher(time.Millisecond, 1),
	}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"Email":           "same@example.invalid",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupSelfLeadByID",
		Object:             "Lead",
		OutputReference:    "$lookedLead",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    constants.FieldID,
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.Id"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).SetObjectType("Lead")

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	query.mu.Lock()
	filterCalls := append([]recordLookupQueryCall(nil), query.calls...)
	specCalls := append([]recordLookupQueryCall(nil), query.specCalls...)
	query.mu.Unlock()
	if len(specCalls) != 0 {
		t.Fatalf("QueryWithSpec calls = %d, want 0 because same-object active tx must skip detached batching", len(specCalls))
	}
	if len(filterCalls) != 1 {
		t.Fatalf("QueryWithFilter calls = %d, want 1 ambient fallback lookup", len(filterCalls))
	}
	if filterCalls[0].detached {
		t.Fatal("after-save same-object batchable lookup detached active tx; want ambient trigger transaction")
	}
	got, ok := interview.ContextData["$lookedLead"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedLead] = %#v, want Lead row from ambient tx", interview.ContextData["$lookedLead"])
	}
	if id, _ := got[constants.FieldID].(string); id != "00Q-trigger" {
		t.Fatalf("looked Lead Id = %q, want triggering row", id)
	}
}

func TestRecordLookup_RequestCacheDisabledOutsideBeforeSave(t *testing.T) {
	query := &recordLookupQueryStub{
		result: &models.QueryResult{
			Done:      true,
			TotalSize: 1,
			Records: []models.SObject{{
				constants.FieldID: "001-cache",
				"Name":            "Cached Account",
			}},
		},
	}
	exec := &FlowExecutor{query: query, persistence: &stubPersistenceCapture{}}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-cache",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupCachedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("first executeRecordLookup() error = %v", err)
	}
	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("second executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want 2 when cache is disabled outside before-save", len(query.calls))
	}
}

func TestRecordLookup_InFlightCoalesceHintCoalescesConcurrentMissesOutsideBeforeSave(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	var queryCalls int32

	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			if n := atomic.AddInt32(&queryCalls, 1); n == 1 {
				close(started)
				select {
				case <-release:
				case <-ctx.Done():
					return nil, ctx.Err()
				}
			}
			return &models.QueryResult{
				Done:      true,
				TotalSize: 1,
				Records: []models.SObject{{
					constants.FieldID: "005-owner",
					"Name":            "Integration User",
				}},
			}, nil
		},
	}
	exec := NewFlowExecutor(nil, nil, nil, &stubPersistenceCapture{}, query, nil, nil)

	lookup := &models.FlowRecordLookup{
		Name:               "FindDefaultOwner",
		Object:             "User",
		OutputReference:    "$defaultOwnerUser",
		GetFirstRecordOnly: true,
		CoalesceInFlight:   true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Email",
			Operator: "EqualTo",
			Value:    "admin@example.invalid",
		}},
	}
	runLookup := func() error {
		interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
		baseCtx := context.Background()
		ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
		if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
			return err
		}
		record, ok := interview.ContextData["$defaultOwnerUser"].(models.SObject)
		if !ok || record == nil {
			return fmt.Errorf("ContextData[$defaultOwnerUser] = %#v, want User row", interview.ContextData["$defaultOwnerUser"])
		}
		if got := record["Name"]; got != "Integration User" {
			return fmt.Errorf("default owner Name = %v, want Integration User", got)
		}
		return nil
	}

	errCh := make(chan error, 2)
	go func() { errCh <- runLookup() }()
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("first lookup did not start")
	}
	go func() { errCh <- runLookup() }()
	time.Sleep(25 * time.Millisecond)
	close(release)

	for i := 0; i < 2; i++ {
		if err := <-errCh; err != nil {
			t.Fatalf("lookup %d error = %v", i+1, err)
		}
	}
	if got := atomic.LoadInt32(&queryCalls); got != 1 {
		t.Fatalf("query calls = %d, want 1 coalesced call", got)
	}
	query.mu.Lock()
	recordedCalls := len(query.calls)
	query.mu.Unlock()
	if recordedCalls != 1 {
		t.Fatalf("recorded QueryWithFilter calls = %d, want 1", recordedCalls)
	}
}

func TestRecordLookup_BeforeSaveRequestCacheClearedByInPlaceRecordUpdate(t *testing.T) {
	query := &recordLookupQueryStub{}
	query.queryFunc = func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
		callNo := len(query.calls)
		return &models.QueryResult{
			Done:      true,
			TotalSize: 1,
			Records: []models.SObject{{
				constants.FieldID: "001-cache",
				"Name":            "Account Snapshot " + string(rune('0'+callNo)),
			}},
		}, nil
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-cache",
		"Company":         "Before",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupCachedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}
	update := &models.FlowRecordUpdate{
		Name:           "MutateTriggerRecord",
		InputReference: "$Record",
		InputAssignments: []models.FlowInputAssignment{{
			Field: "Company",
			Value: "After",
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	flowRuntimeContextFromCtx(ctx).SetObjectType("Lead")

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("first executeRecordLookup() error = %v", err)
	}
	if _, err := exec.executeRecordUpdate(ctx, interview, nil, update, nil); err != nil {
		t.Fatalf("executeRecordUpdate() error = %v", err)
	}
	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("second executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want 2 because in-place RecordUpdate clears cache", len(query.calls))
	}
	second, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || second == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want Account row", interview.ContextData["$lookedAccount"])
	}
	if got := second["Name"]; got != "Account Snapshot 2" {
		t.Fatalf("second Account Name = %v, want fresh post-clear query result", got)
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_DoesNotAppendUnreadablePostUpsertRow(t *testing.T) {
	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
			if detached {
				return &models.QueryResult{Done: true}, nil
			}
			if objectName != "Account" || !strings.Contains(filterExpr, "001-hidden") || limit != 1 {
				t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
			}
			return &models.QueryResult{Done: true}, nil // Simulates record-level read access filtering out the row.
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-hidden",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupHiddenAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-hidden",
		"Name":            "Hidden In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must read the ambient atomic tx")
	}
	if got := interview.ContextData["$lookedAccount"]; got != nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want nil when access projection filters the post-upsert row", got)
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_OverlayUsesUserReadContext(t *testing.T) {
	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
			if detached {
				return &models.QueryResult{Done: true}, nil
			}
			if objectName != "Account" || !strings.Contains(filterExpr, "001-hidden") || limit != 1 {
				t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
			}
			if got := constants.GetSecurityContext(ctx); got != constants.UserContext {
				t.Fatalf("overlay security context = %v, want UserContext", got)
			}
			return &models.QueryResult{Done: true}, nil
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-hidden",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupHiddenAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "record-triggered flow lookup overlay system-only test")
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-hidden",
		"Name":            "Hidden In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, &models.UserSession{ID: "restricted-user"}); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must keep the ambient tx")
	}
	if got := interview.ContextData["$lookedAccount"]; got != nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want nil when bare SystemContext is forced to user-read overlay", got)
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_OverlayPreservesStandardActionInternalReadScope(t *testing.T) {
	query := &recordLookupQueryStub{
		queryFunc: func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
			detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
			if detached {
				return &models.QueryResult{Done: true}, nil
			}
			if objectName != "Account" || !strings.Contains(filterExpr, "001-hidden") || limit != 1 {
				t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
			}
			if got := constants.GetSecurityContext(ctx); got != constants.UserContext {
				t.Fatalf("overlay security context = %v, want UserContext", got)
			}
			if !constants.HasStandardActionExecutionContext(ctx) {
				t.Fatal("overlay ctx lost StandardActionExecutionContext")
			}
			if !constants.HasStandardActionInternalAccessScopes(ctx) || !constants.IsScopedInternalAccess(ctx, "Account", constants.PermRead) {
				t.Fatal("overlay ctx lost Account read StandardActionInternalAccessScope")
			}
			if !constants.HasStandardActionProtectedFlowExecution(ctx) {
				t.Fatal("overlay ctx lost StandardActionProtectedFlowExecution marker")
			}
			return &models.QueryResult{Done: true, TotalSize: 1, Records: []models.SObject{{
				constants.FieldID: "001-hidden",
				"Name":            "Scoped Row",
			}}}, nil
		},
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-hidden",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupHiddenAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := constants.WithStandardActionInternalAccessScopes(context.Background(), "standard-action-1", []constants.StandardActionInternalAccessScope{{
		StandardActionID: "standard-action-1",
		ObjectAPIName:    "Account",
		Operation:        constants.PermRead,
	}})
	baseCtx = constants.WithStandardActionExecutionRootFlow(baseCtx, "standard-action-1", "Std_ConvertLead")
	baseCtx = constants.WithStandardActionProtectedFlowExecution(baseCtx, "Std_ConvertLead")
	baseCtx = constants.WithSecurityContext(baseCtx, constants.SystemContext, "record-triggered flow lookup overlay test")
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-hidden",
		"Name":            "Hidden In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, &models.UserSession{ID: "restricted-user"}); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must keep the ambient tx")
	}
	got, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want scoped Account row", interview.ContextData["$lookedAccount"])
	}
	if id, _ := got[constants.FieldID].(string); id != "001-hidden" {
		t.Fatalf("overlay Account Id = %q, want %q", id, "001-hidden")
	}
	if name, _ := got["Name"].(string); name != "Scoped Row" {
		t.Fatalf("overlay Account Name = %q, want %q", name, "Scoped Row")
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_WithoutMatchingPostUpsertRecordKeepsDetachedQueryResult(t *testing.T) {
	query := &recordLookupQueryStub{result: &models.QueryResult{
		Done:      true,
		TotalSize: 1,
		Records: []models.SObject{{
			constants.FieldID: "001-committed",
			"Name":            "Committed Account",
		}},
	}}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-committed",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-other",
		"Name":            "Other In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must read the ambient atomic tx")
	}

	got, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want committed Account row", interview.ContextData["$lookedAccount"])
	}
	if id, _ := got[constants.FieldID].(string); id != "001-committed" {
		t.Fatalf("Account Id = %q, want %q", id, "001-committed")
	}
	if name, _ := got["Name"].(string); name != "Committed Account" {
		t.Fatalf("Account Name = %q, want %q", name, "Committed Account")
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_UsesAmbientOverlayForUpdatedPostUpsertSnapshot(t *testing.T) {
	query := &recordLookupQueryStub{}
	query.queryFunc = func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
		detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
		if detached {
			return &models.QueryResult{Done: true}, nil
		}
		if objectName != "Account" || !strings.Contains(filterExpr, "001-updated") || limit != 1 {
			t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
		}
		return &models.QueryResult{Done: true, TotalSize: 1, Records: []models.SObject{{
			constants.FieldID: "001-updated",
			"Name":            "Updated In Tx",
		}}}, nil
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-updated",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupUpdatedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-updated",
		"Name":            "Updated In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must read the ambient atomic tx")
	}
	got, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want updated Account row", interview.ContextData["$lookedAccount"])
	}
	if name, _ := got["Name"].(string); name != "Updated In Tx" {
		t.Fatalf("Account Name = %q, want updated overlay row", name)
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_ReplacesDetachedResultWhenOwnedPostUpsertRowHasSameID(t *testing.T) {
	query := &recordLookupQueryStub{}
	query.queryFunc = func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
		detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
		if detached {
			return &models.QueryResult{Done: true, TotalSize: 1, Records: []models.SObject{{
				constants.FieldID: "001-created",
				"Name":            "Committed Snapshot",
			}}}, nil
		}
		if objectName != "Account" || !strings.Contains(filterExpr, "001-created") || limit != 1 {
			t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
		}
		return &models.QueryResult{Done: true, TotalSize: 1, Records: []models.SObject{{
			constants.FieldID: "001-created",
			"Name":            "Updated In Tx",
		}}}, nil
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-created",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupCreatedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-created",
		"Name":            "Created In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must read the ambient atomic tx")
	}

	got, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want committed Account row", interview.ContextData["$lookedAccount"])
	}
	if id, _ := got[constants.FieldID].(string); id != "001-created" {
		t.Fatalf("Account Id = %q, want %q", id, "001-created")
	}
	if name, _ := got["Name"].(string); name != "Updated In Tx" {
		t.Fatalf("Account Name = %q, want ambient overlay row for execution-owned Id", name)
	}
}

func TestRecordLookup_BeforeSaveCrossObjectAtomic_RefillsDetachedWindowWhenOwnedPostUpsertRowNoLongerMatches(t *testing.T) {
	query := &recordLookupQueryStub{}
	query.queryFunc = func(ctx context.Context, objectName string, filterExpr string, limit int) (*models.QueryResult, error) {
		detached, _ := ctx.Value(detachedLookupCtxKey{}).(bool)
		if detached {
			if limit != 2 {
				t.Fatalf("detached QueryWithFilter() limit = %d, want over-fetch window of 2", limit)
			}
			return &models.QueryResult{Done: true, TotalSize: 2, Records: []models.SObject{
				{
					constants.FieldID: "001-moved",
					"Name":            "Committed Snapshot",
				},
				{
					constants.FieldID: "001-fallback",
					"Name":            "Committed Fallback",
				},
			}}, nil
		}
		if objectName != "Account" || !strings.Contains(filterExpr, "001-moved") || limit != 1 {
			t.Fatalf("overlay QueryWithFilter(%q, %q, limit=%d), want Account owned-id lookup", objectName, filterExpr, limit)
		}
		return &models.QueryResult{Done: true}, nil
	}
	persistence := &detachingLookupPersistence{stubPersistenceCapture: &stubPersistenceCapture{txActive: true}}
	exec := &FlowExecutor{query: query, persistence: persistence}

	trigger := models.SObject{
		constants.FieldID: "00Q-trigger",
		"AccountId":       "001-moved",
	}
	models.StampObjectType(trigger, "Lead")
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": trigger,
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "LookupMovedAccount",
		Object:             "Account",
		OutputReference:    "$lookedAccount",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{{
			Field:    "Id",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "$Record.AccountId"},
		}},
	}

	baseCtx := context.Background()
	ctx := withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))
	ctx = WithBeforeSaveFlow(ctx)
	ctx = withAtomicFlowBoundary(ctx)
	flowRuntimeContextFromCtx(ctx).AddPostUpsertRecord("Account", models.SObject{
		constants.FieldID: "001-moved",
		"Name":            "Moved Out Of Scope In Tx",
	})

	if _, err := exec.executeRecordLookup(ctx, interview, lookup, nil); err != nil {
		t.Fatalf("executeRecordLookup() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want detached primary + ambient overlay", len(query.calls))
	}
	if !query.calls[0].detached {
		t.Fatal("QueryWithFilter() did not run on a detached query ctx; detached fresh reads must remain the default path")
	}
	if query.calls[1].detached {
		t.Fatal("overlay QueryWithFilter() ran on detached ctx; same-execution overlay must read the ambient atomic tx")
	}
	got, ok := interview.ContextData["$lookedAccount"].(models.SObject)
	if !ok || got == nil {
		t.Fatalf("ContextData[$lookedAccount] = %#v, want fallback committed Account row", interview.ContextData["$lookedAccount"])
	}
	if id, _ := got[constants.FieldID].(string); id != "001-fallback" {
		t.Fatalf("Account Id = %q, want fallback committed row", id)
	}
	if name, _ := got["Name"].(string); name != "Committed Fallback" {
		t.Fatalf("Account Name = %q, want fallback committed row", name)
	}
}