package serverruntime

import (
	"context"
	"database/sql"
	"encoding/hex"
	"fmt"
	"os"
	"reflect"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/backend/internal/bootstrap"
	kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
	connector_callout "github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-connectors/interfaces/rest/connector_webhooks"
	contractconnector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	contractconstants "github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"golang.org/x/sync/singleflight"
)

const (
	connectorPushImmutableCacheMaxEntries = 4096
	connectorPushImmutableLoadTimeout     = 30 * time.Second
)

type connectorPushWorkItemBinding struct {
	providerPackInstallID  string
	companionPackInstallID string
	workItemID             string
	eventID                string
	eventCreatedAt         time.Time
}

type connectorPushGenerationStateKey struct {
	owner         string
	packInstallID string
}

type connectorPushDependencyClosureKey struct {
	providerPackInstallID  string
	companionPackInstallID string
	workItemID             string
	eventID                string
	eventCreatedAt         time.Time
}

type connectorPushProfileCacheKey struct {
	generation connectorPushGenerationStateKey
	name       string
}

// connectorPushGenerationResolver is process-scoped. Provider and companion
// states are addressed by the immutable PackInstallIds on WorkItem/Event
// lineage. The current dependency ledger records only dependency names, not
// their target PackInstallIds, so each WorkItem resolves its own transitive
// closure and admits only generations activated before that Event was
// enqueued. That closure is then frozen for the WorkItem's process lifetime;
// another WorkItem never reuses it. Calls without inbound lineage deliberately
// retain the existing live-active lookup semantics.
type connectorPushGenerationResolver struct {
	db *sql.DB

	mu                   sync.RWMutex
	closureMaintenanceMu sync.Mutex
	bindings             map[string]connectorPushWorkItemBinding
	states               map[connectorPushGenerationStateKey]*connectorPushInstalledPackState
	profiles             map[connectorPushProfileCacheKey]*automation.ConnectorPushProfile
	closures             map[connectorPushDependencyClosureKey]map[string]*connectorPushInstalledPackState

	bindingLoads singleflight.Group
	stateLoads   singleflight.Group
	profileLoads singleflight.Group
	closureLoads singleflight.Group
}

func newConnectorPushGenerationResolver(db *sql.DB) *connectorPushGenerationResolver {
	return &connectorPushGenerationResolver{
		db:       db,
		bindings: make(map[string]connectorPushWorkItemBinding),
		states:   make(map[connectorPushGenerationStateKey]*connectorPushInstalledPackState),
		profiles: make(map[connectorPushProfileCacheKey]*automation.ConnectorPushProfile),
		closures: make(map[connectorPushDependencyClosureKey]map[string]*connectorPushInstalledPackState),
	}
}

type connectorPushProfileResolver struct {
	db          *sql.DB
	generations *connectorPushGenerationResolver
}

func (r connectorPushProfileResolver) ResolveConnectorPushProfile(ctx context.Context, requesterOwnerPack, ownerPack, profileName string) (*automation.ConnectorPushProfile, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	requesterOwnerPack = strings.TrimSpace(requesterOwnerPack)
	profileName = strings.TrimSpace(profileName)
	generations := r.generations
	if generations == nil {
		generations = newConnectorPushGenerationResolver(r.db)
	}
	state, pinned, err := generations.resolve(ctx, ownerPack)
	if err != nil {
		return nil, err
	}
	if pinned {
		// Cross-pack consent and both exact incarnations were admitted when the
		// Event was enqueued. Re-validate the requester's frozen generation, but
		// never switch an in-flight Event to today's active pointers.
		if !strings.EqualFold(requesterOwnerPack, ownerPack) {
			requesterState, requesterPinned, resolveErr := generations.resolve(ctx, requesterOwnerPack)
			if resolveErr != nil {
				return nil, resolveErr
			}
			if !requesterPinned {
				return nil, fmt.Errorf("connector push profile dependency: requester %q is not bound to the inbound event generation", requesterOwnerPack)
			}
			if err := validateConnectorInboundPinnedPackState(requesterState); err != nil {
				return nil, fmt.Errorf("connector push profile dependency: %w", err)
			}
			if err := authorizeConnectorPushPinnedProfileDependency(ctx, r.db, requesterState, ownerPack); err != nil {
				return nil, err
			}
		}
	} else if err := authorizeConnectorPushProfileDependency(ctx, r.db, requesterOwnerPack, ownerPack); err != nil {
		return nil, err
	}
	validateState := validateConnectorInboundInstalledPackState
	if pinned {
		validateState = validateConnectorInboundPinnedPackState
	}
	if err := validateState(state); err != nil ||
		!strings.EqualFold(strings.TrimSpace(state.category), string(packs.ConnectorPack)) {
		return nil, connector_callout.ErrProfileUnavailable
	}
	if !pinned {
		return loadConnectorPushProfile(ctx, r.db, ownerPack, state.packInstallID, profileName)
	}
	return generations.resolveProfile(ctx, state, profileName)
}

func loadConnectorPushProfile(ctx context.Context, db *sql.DB, ownerPack, packInstallID, profileName string) (*automation.ConnectorPushProfile, error) {
	profile, err := connector_callout.LoadProfileForPackInstall(ctx, db, ownerPack, packInstallID, profileName)
	if err != nil {
		return nil, err
	}
	if profile == nil {
		return nil, nil
	}
	return &automation.ConnectorPushProfile{
		OwnerPack:            profile.OwnerPack,
		ProfileName:          profile.ProfileName,
		CreatedPackInstallID: profile.InstalledPackID,
		CredentialRef:        profile.CredentialRef,
		EndpointClass:        profile.EndpointClass,
		RetryPolicy: automation.ConnectorPushRetryPolicy{
			MaxAttempts:          profile.RetryPolicy.MaxAttempts,
			BaseDelay:            profile.RetryPolicy.BaseDelay,
			Multiplier:           profile.RetryPolicy.Multiplier,
			MaxDelay:             profile.RetryPolicy.MaxDelay,
			RetryableStatusCodes: append([]int(nil), profile.RetryPolicy.RetryableStatusCodes...),
		},
	}, nil
}

func cloneConnectorPushProfile(profile *automation.ConnectorPushProfile) *automation.ConnectorPushProfile {
	if profile == nil {
		return nil
	}
	out := *profile
	out.RetryPolicy.RetryableStatusCodes = append([]int(nil), profile.RetryPolicy.RetryableStatusCodes...)
	return &out
}

func authorizeConnectorPushProfileDependency(ctx context.Context, db *sql.DB, requesterOwnerPack, profileOwnerPack string) error {
	if err := connector_webhooks.AuthorizeCrossPackProfileDependency(ctx, db, requesterOwnerPack, profileOwnerPack); err != nil {
		return fmt.Errorf("connector push profile dependency: %w", err)
	}
	return nil
}

func authorizeConnectorPushPinnedProfileDependency(
	ctx context.Context,
	db *sql.DB,
	requesterState *connectorPushInstalledPackState,
	profileOwnerPack string,
) error {
	if requesterState == nil {
		return fmt.Errorf("connector push profile dependency: requester generation is required")
	}
	requesterOwnerPack := strings.TrimSpace(requesterState.owner)
	profileOwnerPack = strings.TrimSpace(profileOwnerPack)
	if requesterOwnerPack == "" || profileOwnerPack == "" {
		return fmt.Errorf("connector push profile dependency: requester and profile owner are required")
	}
	if strings.EqualFold(requesterOwnerPack, profileOwnerPack) {
		return nil
	}
	dependencies, err := loadConnectorPushInstalledDependencyNames(ctx, db, requesterState)
	if err != nil {
		return fmt.Errorf("connector push profile dependency: %w", err)
	}
	for _, dependencyOwner := range dependencies {
		if strings.EqualFold(strings.TrimSpace(dependencyOwner), profileOwnerPack) {
			return nil
		}
	}
	return fmt.Errorf(
		"connector push profile dependency: installed requester generation %q/%q does not declare profile owner %q",
		requesterOwnerPack,
		strings.TrimSpace(requesterState.packInstallID),
		profileOwnerPack,
	)
}

type connectorPushContractResolver struct {
	db          *sql.DB
	generations *connectorPushGenerationResolver
	txExtractor interface {
		ExtractTx(context.Context) interface{}
	}
	componentKindRegistrationSet  packs.ComponentKindRegistrationSet
	resolveInstalledPack          func(packs.InstalledPackResolverInput, packs.InstalledPackResolverOptions) (packs.Pack, error)
	contractDeclarations          func(packs.Pack) ([]bootstrap.ConnectorPushDurableMetadataContract, error)
	objectSchemaResolver          func(context.Context, *sql.DB) func(string) *models.ObjectMetadata
	trustedLocalPackRoot          string
	resolveRecoveryHintPack       func(context.Context, *connectorPushInstalledPackState) (packs.Pack, func(), error)
	resolveLockedSourceGeneration func(
		context.Context,
		*connectorPushInstalledPackState,
		map[string]string,
	) (connectorPushLockedSourceGeneration, error)
	contractCacheMu sync.RWMutex
	contractCache   map[connectorPushContractCacheKey]*bootstrap.ConnectorPushDurableMetadataContract
	contractLoads   singleflight.Group

	immutableSourceCacheMu          sync.RWMutex
	immutableRecoveryHints          map[connectorPushRecoveryHintCacheKey]connectorPushRecoveryHint
	immutableLockedSourceStates     map[connectorPushLockedSourceGenerationCacheKey]connectorPushLockedSourceGeneration
	immutableRecoveryHintLoads      singleflight.Group
	immutableLockedSourceStateLoads singleflight.Group
}

type connectorPushContractCacheKey struct {
	generation connectorPushGenerationStateKey
	flowName   string
	actionName string
}

type connectorPushLockedSourceGeneration struct {
	contracts    []bootstrap.ConnectorPushDurableMetadataContract
	mappings     []contractconnector.ConnectorFieldMappingInput
	dependencies []string
	objectFields map[string][]contractconnector.ConnectorPushObjectFieldMetadata
}

type connectorPushInstalledContractRetryScopeContextKey struct{}

type connectorPushInstalledContractRetryScope struct {
	resolver                *connectorPushContractResolver
	mu                      sync.RWMutex
	recoveryHints           map[connectorPushRecoveryHintCacheKey]connectorPushRecoveryHint
	lockedSourceGenerations map[connectorPushLockedSourceGenerationCacheKey]connectorPushLockedSourceGeneration
}

type connectorPushRecoveryHintCacheKey struct {
	trustedLocalPackRoot string
	state                connectorPushInstalledPackState
	includeContracts     bool
}

type connectorPushRecoveryHint struct {
	contracts    []bootstrap.ConnectorPushDurableMetadataContract
	dependencies []string
}

type connectorPushLockedSourceGenerationCacheKey struct {
	trustedLocalPackRoot string
	state                connectorPushInstalledPackState
	revisions            string
}

func (r *connectorPushContractResolver) WithConnectorInboundRemoteIdentityInstalledContractRetryScope(ctx context.Context) context.Context {
	if connectorPushInstalledContractRetryScopeFromContext(ctx, r) != nil {
		return ctx
	}
	return context.WithValue(ctx, connectorPushInstalledContractRetryScopeContextKey{}, &connectorPushInstalledContractRetryScope{
		resolver: r,
	})
}

func connectorPushInstalledContractRetryScopeFromContext(
	ctx context.Context,
	resolver *connectorPushContractResolver,
) *connectorPushInstalledContractRetryScope {
	if ctx == nil || resolver == nil {
		return nil
	}
	scope, _ := ctx.Value(connectorPushInstalledContractRetryScopeContextKey{}).(*connectorPushInstalledContractRetryScope)
	if scope == nil || scope.resolver != resolver {
		return nil
	}
	return scope
}

func connectorPushInboundGenerationIsPinned(ctx context.Context) bool {
	trace, ok := contractconstants.ConnectorInboundTraceFromContext(ctx)
	return ok && strings.TrimSpace(trace.WorkItemID) != "" && strings.TrimSpace(trace.EventID) != ""
}

func (r *connectorPushContractResolver) ResolveConnectorInboundRemoteIdentityInstalledContracts(
	ctx context.Context,
	owner string,
) (automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	if r == nil || r.db == nil {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("connector inbound installed contract resolver is not wired")
	}
	owner = strings.TrimSpace(owner)
	if owner == "" {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("connector inbound installed contract owner is required")
	}
	generations := r.generations
	if generations == nil {
		generations = newConnectorPushGenerationResolver(r.db)
	}
	state, pinned, err := generations.resolve(ctx, owner)
	if err != nil {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, err
	}
	if state == nil {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("installed pack %q not found", owner)
	}
	return r.resolveConnectorInboundInstalledContractSet(ctx, state, pinned)
}

func (r *connectorPushContractResolver) ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(
	ctx context.Context,
	owner string,
) (automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	sets, err := r.ResolveConnectorInboundRemoteIdentityInstalledContractSetsInAmbientTx(ctx, []string{owner})
	if err != nil {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, err
	}
	owner = strings.TrimSpace(owner)
	set, ok := sets[owner]
	if !ok {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("locked installed contract set for %q is missing", owner)
	}
	return set, nil
}

func (r *connectorPushContractResolver) ResolveConnectorInboundRemoteIdentityInstalledContractSetsInAmbientTx(
	ctx context.Context,
	owners []string,
) (map[string]automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	if r == nil || r.db == nil || r.txExtractor == nil {
		return nil, fmt.Errorf("connector inbound installed contract resolver is not wired for ambient transactions")
	}
	tx, ok := r.txExtractor.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return nil, fmt.Errorf("connector inbound installed contract resolution requires a live ambient transaction")
	}
	owners = uniqueSortedConnectorPushStrings(owners)
	if len(owners) == 0 {
		return nil, fmt.Errorf("connector inbound installed contract owner is required")
	}

	// Source discovery is deliberately non-authoritative. It computes an exact,
	// immutable source snapshot before any row lock; the stable FOR UPDATE pass
	// below fences the union of every requested manifest closure in one global
	// owner order before any snapshot is admitted as recovery authority.
	hintedClosures := make(map[string][]string, len(owners))
	hintedStates := make(map[string]*connectorPushInstalledPackState)
	for _, owner := range owners {
		closure, states, err := r.discoverConnectorInboundInstalledSourceClosure(ctx, owner)
		if err != nil {
			return nil, err
		}
		hintedClosures[owner] = closure
		for dependencyOwner, state := range states {
			if existing := hintedStates[dependencyOwner]; existing != nil && *existing != *state {
				return nil, fmt.Errorf("installed source tuple %q changed during source discovery", dependencyOwner)
			}
			hintedStates[dependencyOwner] = state
		}
	}
	hintedOwners := make([]string, 0, len(hintedStates))
	for dependencyOwner := range hintedStates {
		hintedOwners = append(hintedOwners, dependencyOwner)
	}
	sort.Strings(hintedOwners)
	revisions, err := connectorPushInstalledSourceRevisions(hintedStates)
	if err != nil {
		return nil, err
	}
	resolveLocked := r.resolveLockedSourceGeneration
	if resolveLocked == nil {
		resolveLocked = r.resolveConnectorPushLockedSourceGeneration
	}
	resolved := make(map[string]connectorPushLockedSourceGeneration, len(hintedStates))
	for _, dependencyOwner := range hintedOwners {
		generation, err := r.resolveCachedConnectorPushLockedSourceGeneration(
			ctx,
			hintedStates[dependencyOwner],
			revisions,
			resolveLocked,
		)
		if err != nil {
			return nil, fmt.Errorf("resolve locked installed source pack %q: %w", dependencyOwner, err)
		}
		resolved[dependencyOwner] = generation
	}

	// Exact source revisions are immutable. Reconstruct the full source snapshot
	// before taking row locks, then fence every live installed tuple byte-for-byte
	// in stable order. An inbound WorkItem already carries exact provider and
	// companion PackInstallIds; its source closure was resolved from those exact
	// generations above, so locking the shared active-pointer rows here would only
	// serialize otherwise independent WorkItems and can make the whole connector
	// lane time out under load.
	locked := make(map[string]*connectorPushInstalledPackState, len(hintedOwners))
	pinnedWorkItemGeneration := connectorPushInboundGenerationIsPinned(ctx)
	if pinnedWorkItemGeneration {
		for _, dependencyOwner := range hintedOwners {
			state := hintedStates[dependencyOwner]
			if state == nil {
				return nil, fmt.Errorf("pinned installed source tuple %q disappeared", dependencyOwner)
			}
			copyOfState := *state
			locked[dependencyOwner] = &copyOfState
		}
	} else {
		for _, dependencyOwner := range hintedOwners {
			state, err := lockConnectorPushInstalledPackState(ctx, tx, dependencyOwner)
			if err != nil {
				return nil, err
			}
			if err := validateConnectorInboundInstalledPackState(state); err != nil {
				return nil, err
			}
			if hinted := hintedStates[dependencyOwner]; hinted == nil || *hinted != *state {
				return nil, fmt.Errorf("installed source tuple %q changed while acquiring generation locks", dependencyOwner)
			}
			locked[dependencyOwner] = state
		}
	}

	results := make(map[string]automation.ConnectorInboundRemoteIdentityInstalledContractSet, len(owners))
	for _, owner := range owners {
		ownerGeneration, ok := resolved[owner]
		if !ok {
			return nil, fmt.Errorf("locked installed dependency set is missing owner %q", owner)
		}
		lockedOwners, err := connectorInboundLockedSourceClosure(owner, resolved)
		if err != nil {
			return nil, err
		}
		if strings.Join(hintedClosures[owner], "\x00") != strings.Join(lockedOwners, "\x00") {
			return nil, fmt.Errorf("installed source dependency closure for %q changed while acquiring generation locks", owner)
		}
		lockedOwnerSet := make(map[string]struct{}, len(lockedOwners))
		closureResolved := make(map[string]connectorPushLockedSourceGeneration, len(lockedOwners))
		for _, dependencyOwner := range lockedOwners {
			lockedOwnerSet[dependencyOwner] = struct{}{}
			closureResolved[dependencyOwner] = resolved[dependencyOwner]
		}
		for _, profileOwner := range connectorInboundInstalledDependencyOwners(owner, ownerGeneration.contracts) {
			if _, declared := lockedOwnerSet[profileOwner]; !declared {
				return nil, fmt.Errorf("connector profile owner %q is not declared in the immutable source dependency closure", profileOwner)
			}
		}
		objectFields, err := mergeConnectorPushLockedObjectFields(closureResolved)
		if err != nil {
			return nil, err
		}
		if err := kernelconnector.ValidateConnectorPushDurableMetadataContractsWithOptions(
			ownerGeneration.contracts,
			ownerGeneration.mappings,
			contractconnector.ConnectorPushContractValidationOptions{
				ObjectFields:                   objectFields,
				RequireSourceLookupFieldSchema: true,
			},
		); err != nil {
			return nil, fmt.Errorf("validate locked source connector contracts for %q: %w", owner, err)
		}

		generationParts := make([]string, 0, len(lockedOwners))
		installIDs := make(map[string]string, len(lockedOwners))
		for _, dependencyOwner := range lockedOwners {
			state := locked[dependencyOwner]
			if state == nil {
				return nil, fmt.Errorf("locked installed dependency %q disappeared", dependencyOwner)
			}
			installIDs[dependencyOwner] = strings.TrimSpace(state.packInstallID)
			generationParts = append(generationParts, strings.Join([]string{
				dependencyOwner,
				strings.TrimSpace(state.packInstallID),
				strings.TrimSpace(state.appliedHash),
				strings.TrimSpace(state.sourceRevision),
			}, "@"))
		}
		result := automation.ConnectorInboundRemoteIdentityInstalledContractSet{
			Generation:           strings.Join(generationParts, "|"),
			Contracts:            make([]bootstrap.ConnectorPushDurableMetadataContract, 0, len(ownerGeneration.contracts)),
			Mappings:             append([]contractconnector.ConnectorFieldMappingInput(nil), ownerGeneration.mappings...),
			LockedPackInstallIDs: installIDs,
		}
		for i := range ownerGeneration.contracts {
			result.Contracts = append(result.Contracts, *cloneConnectorPushContract(&ownerGeneration.contracts[i]))
		}
		results[owner] = result
	}
	return results, nil
}

func (r *connectorPushContractResolver) resolveConnectorInboundInstalledContractSet(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	pinned bool,
) (automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	owner := strings.TrimSpace(state.owner)
	validateState := validateConnectorInboundInstalledPackState
	if pinned {
		validateState = validateConnectorInboundPinnedPackState
	}
	if err := validateState(state); err != nil {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, err
	}
	contracts, err := r.resolveContractsFromPack(ctx, state)
	if err != nil {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, err
	}
	if len(contracts) == 0 {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("installed pack %s has no connector push contracts", owner)
	}
	result := automation.ConnectorInboundRemoteIdentityInstalledContractSet{
		Generation: strings.Join([]string{
			owner,
			strings.TrimSpace(state.packInstallID),
			strings.TrimSpace(state.appliedHash),
			strings.TrimSpace(state.sourceRevision),
		}, "@"),
		Contracts: make([]bootstrap.ConnectorPushDurableMetadataContract, 0, len(contracts)),
		LockedPackInstallIDs: map[string]string{
			owner: strings.TrimSpace(state.packInstallID),
		},
	}
	for i := range contracts {
		result.Contracts = append(result.Contracts, *cloneConnectorPushContract(&contracts[i]))
	}
	return result, nil
}

func newConnectorPushContractResolver(db *sql.DB, registrationSet packs.ComponentKindRegistrationSet) *connectorPushContractResolver {
	return &connectorPushContractResolver{
		db:                           db,
		generations:                  newConnectorPushGenerationResolver(db),
		componentKindRegistrationSet: registrationSet,
		resolveInstalledPack:         packs.ResolveInstalledPack,
		contractDeclarations:         connectorPushContractDeclarationsFromPack,
		objectSchemaResolver:         connectorPushRuntimeObjectSchemaResolver,
		trustedLocalPackRoot:         strings.TrimSpace(os.Getenv(connectorPushTrustedLocalPackRootEnv)),
		contractCache:                make(map[connectorPushContractCacheKey]*bootstrap.ConnectorPushDurableMetadataContract),
		immutableRecoveryHints:       make(map[connectorPushRecoveryHintCacheKey]connectorPushRecoveryHint),
		immutableLockedSourceStates:  make(map[connectorPushLockedSourceGenerationCacheKey]connectorPushLockedSourceGeneration),
	}
}

type connectorPushInstalledPackState struct {
	owner           string
	packInstallID   string
	category        string
	status          string
	manifestVersion string
	sourceKind      string
	sourceURI       string
	sourceRevision  string
	appliedHash     string
	desiredHash     string
	isDeleted       int
}

func (r *connectorPushGenerationResolver) resolve(ctx context.Context, owner string) (*connectorPushInstalledPackState, bool, error) {
	owner = strings.TrimSpace(owner)
	if r == nil || r.db == nil || owner == "" {
		return nil, false, fmt.Errorf("connector push generation resolver is not wired")
	}
	trace, ok := contractconstants.ConnectorInboundTraceFromContext(ctx)
	if !ok || strings.TrimSpace(trace.WorkItemID) == "" {
		state, err := lookupConnectorPushInstalledPackState(ctx, r.db, owner)
		return state, false, err
	}
	binding, err := r.resolveWorkItemBinding(ctx, trace.WorkItemID, trace.EventID)
	if err != nil {
		return nil, true, err
	}
	candidates := []string{binding.companionPackInstallID, binding.providerPackInstallID}
	if strings.EqualFold(owner, strings.TrimSpace(trace.Namespace)) {
		candidates[0], candidates[1] = candidates[1], candidates[0]
	}
	seen := make(map[string]struct{}, len(candidates))
	for _, packInstallID := range candidates {
		packInstallID = strings.TrimSpace(packInstallID)
		if packInstallID == "" {
			continue
		}
		if _, exists := seen[packInstallID]; exists {
			continue
		}
		seen[packInstallID] = struct{}{}
		state, found, resolveErr := r.resolveExactState(ctx, owner, packInstallID)
		if resolveErr != nil {
			return nil, true, resolveErr
		}
		if found {
			return state, true, nil
		}
	}
	state, found, resolveErr := r.resolveWorkItemDependencyState(ctx, binding, owner)
	if resolveErr != nil {
		return nil, true, resolveErr
	}
	if found {
		return state, true, nil
	}
	return nil, true, fmt.Errorf(
		"connector push owner %q is not bound to inbound work item %q event %q",
		owner,
		strings.TrimSpace(trace.WorkItemID),
		strings.TrimSpace(trace.EventID),
	)
}

func (r *connectorPushGenerationResolver) resolveWorkItemBinding(ctx context.Context, workItemID, eventID string) (connectorPushWorkItemBinding, error) {
	workItemID = strings.TrimSpace(workItemID)
	eventID = strings.TrimSpace(eventID)
	key := workItemID + "\x00" + eventID
	r.mu.RLock()
	cached, ok := r.bindings[key]
	r.mu.RUnlock()
	if ok {
		return cached, nil
	}
	result := r.bindingLoads.DoChan(key, func() (interface{}, error) {
		r.mu.RLock()
		cached, ok := r.bindings[key]
		r.mu.RUnlock()
		if ok {
			return cached, nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		var binding connectorPushWorkItemBinding
		err := r.db.QueryRowContext(loadCtx, `
SELECT e.InstalledPackId, COALESCE(e.CompanionPackInstallId, ''), e.CreatedDate
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
 WHERE w.WorkId = ?
   AND w.EventId = ?
   AND w.InstalledPackId = e.InstalledPackId
   AND w.WorkType IN ('connector_event', 'workflow_action_apply')
   AND w.State = 'processing'
 LIMIT 1`, workItemID, eventID).Scan(
			&binding.providerPackInstallID,
			&binding.companionPackInstallID,
			&binding.eventCreatedAt,
		)
		if err == sql.ErrNoRows {
			return connectorPushWorkItemBinding{}, fmt.Errorf("connector inbound work item %q event %q has no live immutable generation binding", workItemID, eventID)
		}
		if err != nil {
			return connectorPushWorkItemBinding{}, fmt.Errorf("lookup connector inbound work item %q generation binding: %w", workItemID, err)
		}
		binding.providerPackInstallID = strings.TrimSpace(binding.providerPackInstallID)
		binding.companionPackInstallID = strings.TrimSpace(binding.companionPackInstallID)
		binding.workItemID = workItemID
		binding.eventID = eventID
		if binding.providerPackInstallID == "" {
			return connectorPushWorkItemBinding{}, fmt.Errorf("connector inbound work item %q has a blank provider PackInstallId", workItemID)
		}
		if binding.eventCreatedAt.IsZero() {
			return connectorPushWorkItemBinding{}, fmt.Errorf("connector inbound work item %q event %q has no enqueue timestamp", workItemID, eventID)
		}
		r.mu.Lock()
		if len(r.bindings) >= connectorPushImmutableCacheMaxEntries {
			clear(r.bindings)
		}
		r.bindings[key] = binding
		r.mu.Unlock()
		return binding, nil
	})
	select {
	case <-ctx.Done():
		return connectorPushWorkItemBinding{}, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return connectorPushWorkItemBinding{}, loaded.Err
		}
		return loaded.Val.(connectorPushWorkItemBinding), nil
	}
}

func (r *connectorPushGenerationResolver) resolveWorkItemDependencyState(
	ctx context.Context,
	binding connectorPushWorkItemBinding,
	owner string,
) (*connectorPushInstalledPackState, bool, error) {
	owner = strings.TrimSpace(owner)
	key := connectorPushDependencyClosureKey{
		providerPackInstallID:  strings.TrimSpace(binding.providerPackInstallID),
		companionPackInstallID: strings.TrimSpace(binding.companionPackInstallID),
		workItemID:             strings.TrimSpace(binding.workItemID),
		eventID:                strings.TrimSpace(binding.eventID),
		eventCreatedAt:         binding.eventCreatedAt,
	}
	if owner == "" || key.providerPackInstallID == "" || key.workItemID == "" || key.eventID == "" || key.eventCreatedAt.IsZero() {
		return nil, false, fmt.Errorf("connector inbound work item dependency binding is incomplete")
	}
	closure, err := r.resolveWorkItemDependencyClosure(ctx, key)
	if err != nil {
		return nil, false, err
	}
	state := closure[strings.ToLower(owner)]
	if state == nil {
		return nil, false, nil
	}
	copyOfState := *state
	return &copyOfState, true, nil
}

func (r *connectorPushGenerationResolver) resolveWorkItemDependencyClosure(
	ctx context.Context,
	key connectorPushDependencyClosureKey,
) (map[string]*connectorPushInstalledPackState, error) {
	r.mu.RLock()
	cached, ok := r.closures[key]
	r.mu.RUnlock()
	if ok {
		return cloneConnectorPushInstalledPackStates(cached), nil
	}
	loadKey := strings.Join([]string{
		key.workItemID,
		key.eventID,
		key.providerPackInstallID,
		key.companionPackInstallID,
		key.eventCreatedAt.UTC().Format(time.RFC3339Nano),
	}, "\x00")
	result := r.closureLoads.DoChan(loadKey, func() (interface{}, error) {
		r.mu.RLock()
		cached, ok := r.closures[key]
		r.mu.RUnlock()
		if ok {
			return cloneConnectorPushInstalledPackStates(cached), nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		closure, err := loadConnectorPushWorkItemDependencyClosure(loadCtx, r.db, key)
		if err != nil {
			return nil, err
		}
		if err := r.storeWorkItemDependencyClosure(loadCtx, key, closure); err != nil {
			return nil, err
		}
		return cloneConnectorPushInstalledPackStates(closure), nil
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		closure, _ := loaded.Val.(map[string]*connectorPushInstalledPackState)
		return cloneConnectorPushInstalledPackStates(closure), nil
	}
}

func (r *connectorPushGenerationResolver) storeWorkItemDependencyClosure(
	ctx context.Context,
	key connectorPushDependencyClosureKey,
	closure map[string]*connectorPushInstalledPackState,
) error {
	r.mu.Lock()
	if r.closures == nil {
		r.closures = make(map[connectorPushDependencyClosureKey]map[string]*connectorPushInstalledPackState)
	}
	if _, exists := r.closures[key]; exists || len(r.closures) < connectorPushImmutableCacheMaxEntries {
		r.closures[key] = cloneConnectorPushInstalledPackStates(closure)
		r.mu.Unlock()
		return nil
	}
	r.mu.Unlock()

	// Exact transitive IDs are not persisted, so a non-terminal WorkItem's closure
	// must never be evicted. Serialize the rare capacity pass and prune only IDs
	// positively proven to be in one of the durable terminal states. Retry and
	// blocked states, unknown states, and missing rows all remain retained. If no
	// terminal closure can be reclaimed, reject new admission instead of weakening
	// lineage or growing without bound. WorkItem terminal state is monotonic; the
	// workqueue replay contract creates a new WorkId rather than reopening one.
	r.closureMaintenanceMu.Lock()
	defer r.closureMaintenanceMu.Unlock()

	r.mu.Lock()
	if _, exists := r.closures[key]; exists || len(r.closures) < connectorPushImmutableCacheMaxEntries {
		r.closures[key] = cloneConnectorPushInstalledPackStates(closure)
		r.mu.Unlock()
		return nil
	}
	cachedKeys := make([]connectorPushDependencyClosureKey, 0, len(r.closures))
	for cachedKey := range r.closures {
		cachedKeys = append(cachedKeys, cachedKey)
	}
	r.mu.Unlock()

	terminal, err := loadConnectorPushTerminalWorkItemIDs(ctx, r.db, cachedKeys)
	if err != nil {
		return err
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	for cachedKey := range r.closures {
		if _, done := terminal[strings.TrimSpace(cachedKey.workItemID)]; done {
			delete(r.closures, cachedKey)
		}
	}
	if len(r.closures) >= connectorPushImmutableCacheMaxEntries {
		return fmt.Errorf(
			"connector inbound active WorkItem dependency closures reached bounded capacity %d",
			connectorPushImmutableCacheMaxEntries,
		)
	}
	r.closures[key] = cloneConnectorPushInstalledPackStates(closure)
	return nil
}

func loadConnectorPushTerminalWorkItemIDs(
	ctx context.Context,
	db *sql.DB,
	keys []connectorPushDependencyClosureKey,
) (map[string]struct{}, error) {
	if db == nil {
		return nil, fmt.Errorf("connector push generation resolver database is not wired")
	}
	workItemSet := make(map[string]struct{}, len(keys))
	for _, key := range keys {
		workItemID := strings.TrimSpace(key.workItemID)
		if workItemID == "" {
			return nil, fmt.Errorf("cached connector inbound dependency closure has a blank WorkItem ID")
		}
		workItemSet[workItemID] = struct{}{}
	}
	workItemIDs := make([]string, 0, len(workItemSet))
	for workItemID := range workItemSet {
		workItemIDs = append(workItemIDs, workItemID)
	}
	sort.Strings(workItemIDs)
	if len(workItemIDs) == 0 {
		return map[string]struct{}{}, nil
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(workItemIDs)), ",")
	args := make([]interface{}, 0, len(workItemIDs)+3)
	args = append(
		args,
		connector_workqueue.WorkStateProcessed,
		connector_workqueue.WorkStateSkipped,
		connector_workqueue.WorkStateDead,
	)
	for _, workItemID := range workItemIDs {
		args = append(args, workItemID)
	}
	rows, err := db.QueryContext(ctx, `
SELECT WorkId
  FROM _System_ConnectorWorkItem
 WHERE State IN (?, ?, ?)
   AND WorkId IN (`+placeholders+`)`, args...)
	if err != nil {
		return nil, fmt.Errorf("prune terminal connector inbound dependency closures: %w", err)
	}
	defer rows.Close()
	terminal := make(map[string]struct{})
	for rows.Next() {
		var workItemID string
		if err := rows.Scan(&workItemID); err != nil {
			return nil, fmt.Errorf("scan terminal connector inbound WorkItem: %w", err)
		}
		workItemID = strings.TrimSpace(workItemID)
		if workItemID != "" {
			terminal[workItemID] = struct{}{}
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate terminal connector inbound WorkItems: %w", err)
	}
	return terminal, nil
}

func loadConnectorPushWorkItemDependencyClosure(
	ctx context.Context,
	db *sql.DB,
	key connectorPushDependencyClosureKey,
) (map[string]*connectorPushInstalledPackState, error) {
	if db == nil {
		return nil, fmt.Errorf("connector push generation resolver database is not wired")
	}
	installIDs := []string{key.providerPackInstallID}
	if key.companionPackInstallID != "" && key.companionPackInstallID != key.providerPackInstallID {
		installIDs = append(installIDs, key.companionPackInstallID)
	}
	states, err := loadConnectorPushBoundRootStates(ctx, db, installIDs)
	if err != nil {
		return nil, err
	}
	foundInstallIDs := make(map[string]struct{}, len(states))
	queue := make([]string, 0, len(states))
	for owner, state := range states {
		if err := validateConnectorInboundPinnedPackState(state); err != nil {
			return nil, fmt.Errorf("bound installed dependency %q: %w", owner, err)
		}
		foundInstallIDs[strings.TrimSpace(state.packInstallID)] = struct{}{}
		queue = append(queue, owner)
	}
	for _, installID := range installIDs {
		if _, ok := foundInstallIDs[installID]; !ok {
			return nil, fmt.Errorf("connector inbound work item bound PackInstallId %q was not found", installID)
		}
	}
	sort.Strings(queue)
	for len(queue) > 0 {
		owner := queue[0]
		queue = queue[1:]
		state := states[owner]
		dependencies, err := loadConnectorPushInstalledDependencyNames(ctx, db, state)
		if err != nil {
			return nil, err
		}
		for _, dependencyOwner := range dependencies {
			keyOwner := strings.ToLower(strings.TrimSpace(dependencyOwner))
			if keyOwner == "" {
				return nil, fmt.Errorf("installed dependency of %q has a blank owner", owner)
			}
			if _, exists := states[keyOwner]; exists {
				continue
			}
			if len(states) >= connectorInboundInstalledSourceMaxPacks {
				return nil, fmt.Errorf("installed source dependency closure exceeds maximum pack count %d", connectorInboundInstalledSourceMaxPacks)
			}
			dependencyState, err := lookupConnectorPushPinnedDependencyState(
				ctx,
				db,
				dependencyOwner,
				key.eventCreatedAt,
			)
			if err != nil {
				return nil, err
			}
			if dependencyState == nil {
				return nil, fmt.Errorf("installed source dependency %q of %q not found", dependencyOwner, owner)
			}
			if err := validateConnectorInboundPinnedPackState(dependencyState); err != nil {
				return nil, err
			}
			states[keyOwner] = dependencyState
			queue = append(queue, keyOwner)
			sort.Strings(queue)
		}
	}
	return states, nil
}

func lookupConnectorPushPinnedDependencyState(
	ctx context.Context,
	db *sql.DB,
	owner string,
	eventCreatedAt time.Time,
) (*connectorPushInstalledPackState, error) {
	if eventCreatedAt.IsZero() {
		return nil, fmt.Errorf("connector inbound event enqueue timestamp is required")
	}
	state, err := lookupConnectorPushInstalledPackState(ctx, db, owner)
	if err != nil || state == nil {
		return state, err
	}
	var lineageAt sql.NullTime
	err = db.QueryRowContext(ctx, `
SELECT a.ActivatedDate
  FROM _System_InstalledPack p
  JOIN _System_PackActivation a
    ON a.PackName = p.Name AND COALESCE(a.IsDeleted, 0) = 0
  JOIN _System_PackGeneration g
    ON g.PackInstallId = a.ActivePackInstallId
   AND g.PackName = p.Name
   AND COALESCE(g.IsDeleted, 0) = 0
 WHERE p.Name = ?
   AND a.ActivePackInstallId = ?
   AND g.PackInstallId = ?
 LIMIT 1`, owner, state.packInstallID, state.packInstallID).Scan(&lineageAt)
	if err == sql.ErrNoRows {
		return nil, fmt.Errorf("installed source dependency %q generation changed while resolving inbound event lineage", owner)
	}
	if err != nil {
		return nil, fmt.Errorf("lookup installed source dependency %q activation lineage: %w", owner, err)
	}
	if !lineageAt.Valid {
		return nil, fmt.Errorf(
			"installed source dependency %q generation %q has no activation lineage",
			owner,
			state.packInstallID,
		)
	}
	if !lineageAt.Time.Before(eventCreatedAt) {
		return nil, fmt.Errorf(
			"installed source dependency %q generation %q was activated or modified at %s, not before connector event enqueue %s",
			owner,
			state.packInstallID,
			lineageAt.Time.UTC().Format(time.RFC3339Nano),
			eventCreatedAt.UTC().Format(time.RFC3339Nano),
		)
	}
	return state, nil
}

func loadConnectorPushBoundRootStates(
	ctx context.Context,
	db *sql.DB,
	installIDs []string,
) (map[string]*connectorPushInstalledPackState, error) {
	if len(installIDs) == 0 || len(installIDs) > 2 {
		return nil, fmt.Errorf("connector inbound work item root generation binding is invalid")
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(installIDs)), ",")
	args := make([]interface{}, 0, len(installIDs)*2)
	for _, installID := range installIDs {
		args = append(args, installID)
	}
	for _, installID := range installIDs {
		args = append(args, installID)
	}
	rows, err := db.QueryContext(ctx, `
SELECT PackName, PackInstallId, Category, Status, COALESCE(IsDeleted, 0),
       COALESCE(ManifestFormatVersion, ''), COALESCE(SourceKind, ''),
       COALESCE(SourceURI, ''), COALESCE(SourceRevision, ''),
       COALESCE(AppliedPayloadHash, ''), COALESCE(DesiredPayloadHash, '')
  FROM _System_PackGeneration
 WHERE PackInstallId IN (`+placeholders+`)
   AND COALESCE(IsDeleted, 0) = 0
UNION ALL
SELECT p.Name, p.PackInstallId, p.Category, p.Status, COALESCE(p.IsDeleted, 0),
       COALESCE(p.ManifestFormatVersion, ''), COALESCE(p.SourceKind, ''),
       COALESCE(p.SourceURI, ''), COALESCE(p.SourceRevision, ''),
       COALESCE(p.AppliedPayloadHash, ''), COALESCE(p.DesiredPayloadHash, '')
  FROM _System_InstalledPack p
 WHERE p.PackInstallId IN (`+placeholders+`)
   AND COALESCE(p.IsDeleted, 0) = 0
   AND NOT EXISTS (
       SELECT 1 FROM _System_PackGeneration g
        WHERE g.PackInstallId = p.PackInstallId
          AND COALESCE(g.IsDeleted, 0) = 0
   )`, args...)
	if err != nil {
		return nil, fmt.Errorf("load connector inbound work item root generations: %w", err)
	}
	defer rows.Close()
	states := make(map[string]*connectorPushInstalledPackState, len(installIDs))
	for rows.Next() {
		state := &connectorPushInstalledPackState{}
		if err := rows.Scan(
			&state.owner,
			&state.packInstallID,
			&state.category,
			&state.status,
			&state.isDeleted,
			&state.manifestVersion,
			&state.sourceKind,
			&state.sourceURI,
			&state.sourceRevision,
			&state.appliedHash,
			&state.desiredHash,
		); err != nil {
			return nil, err
		}
		state.owner = strings.TrimSpace(state.owner)
		ownerKey := strings.ToLower(state.owner)
		if ownerKey == "" {
			return nil, fmt.Errorf("connector inbound work item root generation owner is blank")
		}
		if existing := states[ownerKey]; existing != nil && *existing != *state {
			return nil, fmt.Errorf("connector inbound work item root generation owner %q is ambiguous", state.owner)
		}
		states[ownerKey] = state
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return states, nil
}

func loadConnectorPushInstalledDependencyNames(
	ctx context.Context,
	db *sql.DB,
	state *connectorPushInstalledPackState,
) ([]string, error) {
	if state == nil || strings.TrimSpace(state.owner) == "" || strings.TrimSpace(state.packInstallID) == "" {
		return nil, fmt.Errorf("installed dependency state is incomplete")
	}
	rows, err := db.QueryContext(ctx, `
SELECT DependsOnPackName
  FROM _System_InstalledPackDependency
 WHERE DependentPackName = ?
   AND DependentPackInstallId = ?
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY DependsOnPackName`, state.owner, state.packInstallID)
	if err != nil {
		return nil, fmt.Errorf("load installed dependencies for %q/%q: %w", state.owner, state.packInstallID, err)
	}
	defer rows.Close()
	var dependencies []string
	for rows.Next() {
		var dependency string
		if err := rows.Scan(&dependency); err != nil {
			return nil, err
		}
		if dependency = strings.TrimSpace(dependency); dependency != "" {
			dependencies = append(dependencies, dependency)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return uniqueSortedConnectorPushStrings(dependencies), nil
}

func cloneConnectorPushInstalledPackStates(
	states map[string]*connectorPushInstalledPackState,
) map[string]*connectorPushInstalledPackState {
	if states == nil {
		return nil
	}
	result := make(map[string]*connectorPushInstalledPackState, len(states))
	for owner, state := range states {
		if state == nil {
			result[owner] = nil
			continue
		}
		copyOfState := *state
		result[owner] = &copyOfState
	}
	return result
}

func (r *connectorPushGenerationResolver) resolveExactState(ctx context.Context, owner, packInstallID string) (*connectorPushInstalledPackState, bool, error) {
	key := connectorPushGenerationStateKey{owner: strings.TrimSpace(owner), packInstallID: strings.TrimSpace(packInstallID)}
	r.mu.RLock()
	cached, ok := r.states[key]
	r.mu.RUnlock()
	if ok {
		if cached == nil {
			return nil, false, nil
		}
		state := *cached
		return &state, true, nil
	}
	loadKey := key.owner + "\x00" + key.packInstallID
	result := r.stateLoads.DoChan(loadKey, func() (interface{}, error) {
		r.mu.RLock()
		cached, ok := r.states[key]
		r.mu.RUnlock()
		if ok {
			if cached == nil {
				return (*connectorPushInstalledPackState)(nil), nil
			}
			state := *cached
			return &state, nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		state, found, err := lookupConnectorPushExactGenerationState(loadCtx, r.db, key.owner, key.packInstallID)
		if err != nil {
			return state, err
		}
		r.mu.Lock()
		if len(r.states) >= connectorPushImmutableCacheMaxEntries {
			clear(r.states)
		}
		if found {
			copyOfState := *state
			r.states[key] = &copyOfState
		} else {
			// An exact (owner, PackInstallId) pair is immutable. Cache negative
			// candidate probes too: dependency owners are intentionally tested
			// against the provider and companion roots before the frozen closure.
			r.states[key] = nil
		}
		r.mu.Unlock()
		return state, nil
	})
	select {
	case <-ctx.Done():
		return nil, false, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, false, loaded.Err
		}
		state, _ := loaded.Val.(*connectorPushInstalledPackState)
		return state, state != nil, nil
	}
}

func lookupConnectorPushExactGenerationState(ctx context.Context, db *sql.DB, owner, packInstallID string) (*connectorPushInstalledPackState, bool, error) {
	state := &connectorPushInstalledPackState{owner: owner}
	err := db.QueryRowContext(ctx, `
SELECT PackInstallId, Category, Status, COALESCE(IsDeleted, 0),
       COALESCE(ManifestFormatVersion, ''), COALESCE(SourceKind, ''),
       COALESCE(SourceURI, ''), COALESCE(SourceRevision, ''),
       COALESCE(AppliedPayloadHash, ''), COALESCE(DesiredPayloadHash, '')
  FROM _System_PackGeneration
 WHERE PackName = ?
   AND PackInstallId = ?
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`, owner, packInstallID).Scan(
		&state.packInstallID,
		&state.category,
		&state.status,
		&state.isDeleted,
		&state.manifestVersion,
		&state.sourceKind,
		&state.sourceURI,
		&state.sourceRevision,
		&state.appliedHash,
		&state.desiredHash,
	)
	if err == sql.ErrNoRows {
		// Legacy installations predate PackGeneration. Keep the lookup exact by
		// incarnation; never fall back to the active pointer for an inbound Event.
		err = db.QueryRowContext(ctx, `
SELECT PackInstallId, Category, Status, COALESCE(IsDeleted, 0),
       COALESCE(ManifestFormatVersion, ''), COALESCE(SourceKind, ''),
       COALESCE(SourceURI, ''), COALESCE(SourceRevision, ''),
       COALESCE(AppliedPayloadHash, ''), COALESCE(DesiredPayloadHash, '')
  FROM _System_InstalledPack
 WHERE Name = ?
   AND PackInstallId = ?
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`, owner, packInstallID).Scan(
			&state.packInstallID,
			&state.category,
			&state.status,
			&state.isDeleted,
			&state.manifestVersion,
			&state.sourceKind,
			&state.sourceURI,
			&state.sourceRevision,
			&state.appliedHash,
			&state.desiredHash,
		)
	}
	if err == sql.ErrNoRows {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, fmt.Errorf("lookup immutable pack generation %q/%q: %w", owner, packInstallID, err)
	}
	return state, true, nil
}

func (r *connectorPushGenerationResolver) resolveProfile(ctx context.Context, state *connectorPushInstalledPackState, profileName string) (*automation.ConnectorPushProfile, error) {
	key := connectorPushProfileCacheKey{
		generation: connectorPushGenerationStateKey{owner: strings.TrimSpace(state.owner), packInstallID: strings.TrimSpace(state.packInstallID)},
		name:       strings.TrimSpace(profileName),
	}
	r.mu.RLock()
	cached, ok := r.profiles[key]
	r.mu.RUnlock()
	if ok {
		return cloneConnectorPushProfile(cached), nil
	}
	loadKey := key.generation.owner + "\x00" + key.generation.packInstallID + "\x00" + key.name
	result := r.profileLoads.DoChan(loadKey, func() (interface{}, error) {
		r.mu.RLock()
		cached, ok := r.profiles[key]
		r.mu.RUnlock()
		if ok {
			return cloneConnectorPushProfile(cached), nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		profile, err := loadConnectorPushProfile(loadCtx, r.db, key.generation.owner, key.generation.packInstallID, key.name)
		if err != nil {
			return nil, err
		}
		r.mu.Lock()
		if len(r.profiles) >= connectorPushImmutableCacheMaxEntries {
			clear(r.profiles)
		}
		r.profiles[key] = cloneConnectorPushProfile(profile)
		r.mu.Unlock()
		return cloneConnectorPushProfile(profile), nil
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		profile, _ := loaded.Val.(*automation.ConnectorPushProfile)
		return cloneConnectorPushProfile(profile), nil
	}
}

func (r *connectorPushContractResolver) ResolveConnectorPushContract(ctx context.Context, q automation.ConnectorPushContractQuery) (*bootstrap.ConnectorPushDurableMetadataContract, error) {
	if r == nil || r.db == nil {
		return nil, fmt.Errorf("connector_push_contract resolver database is not wired")
	}
	owner := strings.TrimSpace(q.ConnectorOwnerKey)
	flowName := strings.TrimSpace(q.FlowDeveloperName)
	actionName := strings.TrimSpace(q.FlowActionName)

	if r.generations == nil {
		r.generations = newConnectorPushGenerationResolver(r.db)
	}
	state, pinned, err := r.generations.resolve(ctx, owner)
	if err != nil {
		return nil, err
	}
	if state == nil {
		return nil, nil
	}
	status := strings.TrimSpace(state.status)
	statusAllowed := strings.EqualFold(status, "active") ||
		(pinned && strings.EqualFold(status, "draining"))
	if state.isDeleted != 0 || !statusAllowed {
		return nil, nil
	}
	if strings.TrimSpace(state.appliedHash) == "" || strings.TrimSpace(state.desiredHash) == "" || strings.TrimSpace(state.appliedHash) != strings.TrimSpace(state.desiredHash) {
		return nil, fmt.Errorf("installed pack %s hash state is not active and converged", owner)
	}
	if pinned {
		return r.resolvePinnedConnectorPushContract(ctx, state, flowName, actionName)
	}
	contract, err := r.resolveContractFromPack(ctx, state, flowName, actionName)
	if err != nil {
		return nil, err
	}
	return cloneConnectorPushContract(contract), nil
}

func (r *connectorPushContractResolver) resolvePinnedConnectorPushContract(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	flowName string,
	actionName string,
) (*bootstrap.ConnectorPushDurableMetadataContract, error) {
	key := connectorPushContractCacheKey{
		generation: connectorPushGenerationStateKey{owner: strings.TrimSpace(state.owner), packInstallID: strings.TrimSpace(state.packInstallID)},
		flowName:   strings.TrimSpace(flowName),
		actionName: strings.TrimSpace(actionName),
	}
	r.contractCacheMu.RLock()
	cached, ok := r.contractCache[key]
	r.contractCacheMu.RUnlock()
	if ok {
		return cloneConnectorPushContract(cached), nil
	}
	loadKey := key.generation.owner + "\x00" + key.generation.packInstallID + "\x00" + key.flowName + "\x00" + key.actionName
	result := r.contractLoads.DoChan(loadKey, func() (interface{}, error) {
		r.contractCacheMu.RLock()
		cached, ok := r.contractCache[key]
		r.contractCacheMu.RUnlock()
		if ok {
			return cloneConnectorPushContract(cached), nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		contract, err := r.resolveContractFromPack(loadCtx, state, key.flowName, key.actionName)
		if err != nil {
			return nil, err
		}
		r.contractCacheMu.Lock()
		if len(r.contractCache) >= connectorPushImmutableCacheMaxEntries {
			clear(r.contractCache)
		}
		r.contractCache[key] = cloneConnectorPushContract(contract)
		r.contractCacheMu.Unlock()
		return cloneConnectorPushContract(contract), nil
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		contract, _ := loaded.Val.(*bootstrap.ConnectorPushDurableMetadataContract)
		return cloneConnectorPushContract(contract), nil
	}
}

func (r *connectorPushContractResolver) lookupInstalledPackState(ctx context.Context, owner string) (*connectorPushInstalledPackState, error) {
	return lookupConnectorPushInstalledPackState(ctx, r.db, owner)
}

func lookupConnectorPushInstalledPackState(ctx context.Context, db *sql.DB, owner string) (*connectorPushInstalledPackState, error) {
	state := &connectorPushInstalledPackState{owner: owner}
	err := db.QueryRowContext(ctx, `
	SELECT CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN p.PackInstallId ELSE COALESCE(g.PackInstallId, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN p.Category ELSE COALESCE(g.Category, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN p.Status ELSE COALESCE(g.Status, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.IsDeleted, 0) ELSE COALESCE(g.IsDeleted, 1) END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.ManifestFormatVersion, '') ELSE COALESCE(g.ManifestFormatVersion, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.SourceKind, '') ELSE COALESCE(g.SourceKind, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.SourceURI, '') ELSE COALESCE(g.SourceURI, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.SourceRevision, '') ELSE COALESCE(g.SourceRevision, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.AppliedPayloadHash, '') ELSE COALESCE(g.AppliedPayloadHash, '') END,
	       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.DesiredPayloadHash, '') ELSE COALESCE(g.DesiredPayloadHash, '') END
  FROM _System_InstalledPack p
  LEFT JOIN _System_PackActivation a
    ON a.PackName = p.Name AND COALESCE(a.IsDeleted, 0) = 0
  LEFT JOIN _System_PackGeneration g
    ON g.PackInstallId = a.ActivePackInstallId
   AND g.PackName = p.Name
   AND COALESCE(g.IsDeleted, 0) = 0
 WHERE p.Name = ?
	 LIMIT 1`, owner).Scan(&state.packInstallID, &state.category, &state.status, &state.isDeleted, &state.manifestVersion, &state.sourceKind, &state.sourceURI, &state.sourceRevision, &state.appliedHash, &state.desiredHash)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("lookup installed pack %q: %w", owner, err)
	}
	return state, nil
}

func lockConnectorPushInstalledPackState(ctx context.Context, tx *sql.Tx, owner string) (*connectorPushInstalledPackState, error) {
	rows, err := tx.QueryContext(ctx, `
SELECT CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN p.PackInstallId ELSE COALESCE(g.PackInstallId, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN p.Category ELSE COALESCE(g.Category, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN p.Status ELSE COALESCE(g.Status, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.IsDeleted, 0) ELSE COALESCE(g.IsDeleted, 1) END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.ManifestFormatVersion, '') ELSE COALESCE(g.ManifestFormatVersion, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.SourceKind, '') ELSE COALESCE(g.SourceKind, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.SourceURI, '') ELSE COALESCE(g.SourceURI, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.SourceRevision, '') ELSE COALESCE(g.SourceRevision, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.AppliedPayloadHash, '') ELSE COALESCE(g.AppliedPayloadHash, '') END,
       CASE WHEN COALESCE(a.ActivePackInstallId, '') = '' THEN COALESCE(p.DesiredPayloadHash, '') ELSE COALESCE(g.DesiredPayloadHash, '') END
  FROM _System_InstalledPack p
  LEFT JOIN _System_PackActivation a
    ON a.PackName = p.Name AND COALESCE(a.IsDeleted, 0) = 0
  LEFT JOIN _System_PackGeneration g
    ON g.PackInstallId = a.ActivePackInstallId
   AND g.PackName = p.Name
   AND COALESCE(g.IsDeleted, 0) = 0
 WHERE p.Name = ?
 FOR UPDATE`, owner)
	if err != nil {
		return nil, fmt.Errorf("lock installed pack %q: %w", owner, err)
	}
	defer rows.Close()
	var state *connectorPushInstalledPackState
	for rows.Next() {
		if state != nil {
			return nil, fmt.Errorf("installed pack %q has multiple active generations", owner)
		}
		candidate := &connectorPushInstalledPackState{owner: owner}
		if err := rows.Scan(&candidate.packInstallID, &candidate.category, &candidate.status, &candidate.isDeleted, &candidate.manifestVersion, &candidate.sourceKind, &candidate.sourceURI, &candidate.sourceRevision, &candidate.appliedHash, &candidate.desiredHash); err != nil {
			return nil, err
		}
		state = candidate
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if state == nil {
		return nil, fmt.Errorf("installed pack %q not found", owner)
	}
	return state, nil
}

func validateConnectorInboundInstalledPackState(state *connectorPushInstalledPackState) error {
	return validateConnectorInboundInstalledPackStateStatus(state, false)
}

func validateConnectorInboundPinnedPackState(state *connectorPushInstalledPackState) error {
	return validateConnectorInboundInstalledPackStateStatus(state, true)
}

func validateConnectorInboundInstalledPackStateStatus(state *connectorPushInstalledPackState, allowDraining bool) error {
	if state == nil || strings.TrimSpace(state.owner) == "" || strings.TrimSpace(state.packInstallID) == "" {
		return fmt.Errorf("installed pack state is incomplete")
	}
	status := strings.TrimSpace(state.status)
	statusAllowed := strings.EqualFold(status, "active") ||
		(allowDraining && strings.EqualFold(status, "draining"))
	if state.isDeleted != 0 || !statusAllowed ||
		strings.TrimSpace(state.appliedHash) == "" || strings.TrimSpace(state.desiredHash) == "" ||
		strings.TrimSpace(state.appliedHash) != strings.TrimSpace(state.desiredHash) {
		if allowDraining {
			return fmt.Errorf("installed pack %s is not active/draining and converged", state.owner)
		}
		return fmt.Errorf("installed pack %s is not active and converged", state.owner)
	}
	return nil
}

func connectorInboundInstalledDependencyOwners(owner string, contracts []bootstrap.ConnectorPushDurableMetadataContract) []string {
	seen := map[string]struct{}{strings.TrimSpace(owner): {}}
	for i := range contracts {
		profileOwner := strings.TrimSpace(contracts[i].ProfileOwnerPack)
		if profileOwner == "" {
			profileOwner = strings.TrimSpace(contracts[i].ConnectorOwnerKey)
		}
		if profileOwner != "" {
			seen[profileOwner] = struct{}{}
		}
	}
	owners := make([]string, 0, len(seen))
	for dependencyOwner := range seen {
		if dependencyOwner != "" {
			owners = append(owners, dependencyOwner)
		}
	}
	sort.Strings(owners)
	return owners
}

func connectorPushInstalledSourceRevisions(states map[string]*connectorPushInstalledPackState) (map[string]string, error) {
	revisions := make(map[string]string, len(states)*2)
	owners := make([]string, 0, len(states))
	for owner := range states {
		owners = append(owners, owner)
	}
	sort.Strings(owners)
	for _, owner := range owners {
		state := states[owner]
		if state == nil {
			return nil, fmt.Errorf("installed source state for %q is missing", owner)
		}
		revision := strings.TrimSpace(state.sourceRevision)
		for _, sourceName := range []string{strings.TrimSpace(state.sourceURI), strings.TrimSpace(owner)} {
			if sourceName == "" {
				continue
			}
			sourceName = strings.ToLower(sourceName)
			if existing := strings.TrimSpace(revisions[sourceName]); existing != "" && existing != revision {
				return nil, fmt.Errorf("installed source name %q resolves to multiple revisions", sourceName)
			}
			revisions[sourceName] = revision
		}
	}
	return revisions, nil
}

const (
	connectorInboundInstalledSourceMaxPacks = 128
	connectorInboundInstalledSourceMaxDepth = 64
)

func (r *connectorPushContractResolver) discoverConnectorInboundInstalledSourceClosure(ctx context.Context, owner string) ([]string, map[string]*connectorPushInstalledPackState, error) {
	owner = strings.TrimSpace(owner)
	if owner == "" {
		return nil, nil, fmt.Errorf("connector inbound installed contract owner is required")
	}
	visitState := map[string]uint8{}
	states := map[string]*connectorPushInstalledPackState{}
	stack := make([]string, 0, 8)
	var rootContracts []bootstrap.ConnectorPushDurableMetadataContract
	var visit func(string, int) error
	visit = func(current string, depth int) error {
		current = strings.TrimSpace(current)
		if current == "" {
			return fmt.Errorf("installed source dependency name is blank")
		}
		if depth > connectorInboundInstalledSourceMaxDepth {
			return fmt.Errorf("installed source dependency closure exceeds maximum depth %d", connectorInboundInstalledSourceMaxDepth)
		}
		switch visitState[current] {
		case 1:
			return fmt.Errorf("installed source dependency cycle detected: %s", strings.Join(append(append([]string(nil), stack...), current), " -> "))
		case 2:
			return nil
		}
		if len(states) >= connectorInboundInstalledSourceMaxPacks {
			return fmt.Errorf("installed source dependency closure exceeds maximum pack count %d", connectorInboundInstalledSourceMaxPacks)
		}
		visitState[current] = 1
		stack = append(stack, current)
		defer func() { stack = stack[:len(stack)-1] }()

		generations := r.generations
		if generations == nil {
			return fmt.Errorf("connector push generation resolver is not wired")
		}
		state, pinned, err := generations.resolve(ctx, current)
		if err != nil {
			return err
		}
		if state == nil {
			return fmt.Errorf("installed source dependency %q not found", current)
		}
		if err := validateConnectorInboundInstalledPackStateForHint(state, r.trustedLocalPackRoot, pinned); err != nil {
			return err
		}
		hint, err := r.resolveCachedConnectorPushRecoveryHint(ctx, state, current == owner)
		if err != nil {
			return fmt.Errorf("discover installed source dependency %q: %w", current, err)
		}
		states[current] = state
		dependencies := append([]string(nil), hint.dependencies...)
		if current == owner {
			rootContracts = cloneConnectorPushContracts(hint.contracts)
		}
		for _, dependency := range dependencies {
			if err := visit(dependency, depth+1); err != nil {
				return err
			}
		}
		visitState[current] = 2
		return nil
	}
	if err := visit(owner, 0); err != nil {
		return nil, nil, err
	}
	for _, profileOwner := range connectorInboundInstalledDependencyOwners(owner, rootContracts) {
		if _, declared := states[profileOwner]; !declared {
			return nil, nil, fmt.Errorf("connector profile owner %q is not declared in the immutable source dependency closure", profileOwner)
		}
	}
	owners := make([]string, 0, len(states))
	for dependencyOwner := range states {
		owners = append(owners, dependencyOwner)
	}
	sort.Strings(owners)
	return owners, states, nil
}

func validateConnectorInboundInstalledPackStateForHint(state *connectorPushInstalledPackState, trustedLocalPackRoot string, allowDraining bool) error {
	if state == nil {
		return fmt.Errorf("installed pack hint state is incomplete")
	}
	if err := validateConnectorInboundInstalledPackStateStatus(state, allowDraining); err != nil {
		return err
	}
	if strings.TrimSpace(state.owner) == "" || strings.TrimSpace(state.packInstallID) == "" ||
		strings.TrimSpace(state.category) == "" || strings.TrimSpace(state.manifestVersion) == "" ||
		strings.TrimSpace(state.sourceURI) == "" || strings.TrimSpace(state.sourceRevision) == "" {
		return fmt.Errorf("installed pack %s hint state is incomplete", state.owner)
	}
	if !IsImmutableConnectorPushSourceRevision(state.sourceRevision) {
		return fmt.Errorf("installed pack %s hint requires an immutable 40-hex source revision", state.owner)
	}
	switch strings.TrimSpace(state.sourceKind) {
	case packs.PackSourceKindOfficialRepo:
		return nil
	case packs.PackSourceKindLocalPath:
		if strings.TrimSpace(trustedLocalPackRoot) == "" {
			return fmt.Errorf("installed pack %s local_path hint requires %s", state.owner, connectorPushTrustedLocalPackRootEnv)
		}
		return nil
	default:
		return fmt.Errorf("installed pack %s hint requires an immutable official_repo or release-scoped local_path source", state.owner)
	}
}

func (r *connectorPushContractResolver) resolveConnectorPushRecoveryHintPack(
	ctx context.Context,
	state *connectorPushInstalledPackState,
) (packs.Pack, func(), error) {
	if r.resolveRecoveryHintPack != nil {
		return r.resolveRecoveryHintPack(ctx, state)
	}
	if strings.TrimSpace(state.sourceKind) != packs.PackSourceKindLocalPath {
		resolvedPack, err := r.resolveInstalledPackFromState(ctx, state)
		return resolvedPack, func() {}, err
	}
	resolvedPack, cleanup, err := r.resolveConnectorPushRecoverySourcePack(ctx, state, nil)
	if err != nil {
		return nil, cleanup, err
	}
	return resolvedPack, cleanup, nil
}

func (r *connectorPushContractResolver) resolveConnectorPushLockedSourceGeneration(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	revisions map[string]string,
) (connectorPushLockedSourceGeneration, error) {
	if err := validateConnectorInboundPinnedPackState(state); err != nil {
		return connectorPushLockedSourceGeneration{}, err
	}
	resolvedPack, cleanup, err := r.resolveConnectorPushRecoverySourcePack(ctx, state, revisions)
	if err != nil {
		return connectorPushLockedSourceGeneration{}, err
	}
	defer cleanup()

	result := connectorPushLockedSourceGeneration{
		contracts:    resolvedPack.ConnectorPushContractDeclarations(),
		objectFields: map[string][]contractconnector.ConnectorPushObjectFieldMetadata{},
	}
	for _, dependency := range resolvedPack.PackDependencies() {
		if name := strings.TrimSpace(dependency.Name); name != "" {
			result.dependencies = append(result.dependencies, name)
		}
	}
	result.dependencies = uniqueSortedConnectorPushStrings(result.dependencies)
	for _, component := range resolvedPack.Components() {
		if component.Kind() != packs.ComponentKindConnectorFieldMapping {
			continue
		}
		payload, err := component.ExpectedPayload()
		if err != nil {
			return connectorPushLockedSourceGeneration{}, err
		}
		mappings, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload(payload, strings.TrimSpace(state.owner))
		if err != nil {
			return connectorPushLockedSourceGeneration{}, err
		}
		result.mappings = append(result.mappings, mappings...)
	}
	specs, err := packs.ParseObjectSpecsPayloads(resolvedPack.RuntimeObjectSpecPayloads(), strings.TrimSpace(state.owner))
	if err != nil {
		return connectorPushLockedSourceGeneration{}, err
	}
	for _, spec := range specs {
		objectName := strings.ToLower(strings.TrimSpace(spec.APIName))
		if objectName == "" {
			continue
		}
		for _, field := range spec.Fields {
			fieldName := strings.TrimSpace(field.APIName)
			if fieldName == "" {
				continue
			}
			result.objectFields[objectName] = append(result.objectFields[objectName], contractconnector.ConnectorPushObjectFieldMetadata{
				APIName:     fieldName,
				Type:        strings.TrimSpace(field.Type),
				ReferenceTo: append([]string(nil), field.ReferenceTo...),
			})
		}
	}
	return result, nil
}

func (r *connectorPushContractResolver) resolveConnectorPushRecoverySourcePack(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	revisions map[string]string,
) (*packs.SourceBackedPack, func(), error) {
	cleanup := func() {}
	if state == nil || !IsImmutableConnectorPushSourceRevision(state.sourceRevision) {
		return nil, cleanup, fmt.Errorf("strict recovery requires an immutable 40-hex source revision")
	}
	sourceKind := strings.TrimSpace(state.sourceKind)
	sourceURI := strings.TrimSpace(state.sourceURI)
	switch sourceKind {
	case packs.PackSourceKindOfficialRepo:
	case packs.PackSourceKindLocalPath:
		snapshotURI, snapshotCleanup, err := materializeConnectorPushTrustedLocalPackSnapshot(
			ctx,
			r.trustedLocalPackRoot,
			sourceURI,
			strings.TrimSpace(state.sourceRevision),
			strings.TrimSpace(state.owner),
			strings.TrimSpace(state.category),
			strings.TrimSpace(state.manifestVersion),
		)
		if err != nil {
			return nil, cleanup, err
		}
		sourceURI = snapshotURI
		cleanup = snapshotCleanup
	default:
		return nil, cleanup, fmt.Errorf("strict recovery requires an immutable official_repo or release-scoped local_path source")
	}

	opts := packs.DefaultPackSourceOptions()
	objectSchemaResolver := r.objectSchemaResolver
	if objectSchemaResolver == nil {
		objectSchemaResolver = connectorPushRuntimeObjectSchemaResolver
	}
	// Source-backed construction may need dependency-owned object shapes to
	// parse connector contracts. This live resolver is only a provisional parse
	// input: recovery authority is granted later only after the complete locked
	// source closure supplies and validates the same fields immutably.
	opts.ObjectSchemaResolver = objectSchemaResolver(ctx, r.db)
	opts.OfficialRepoSourceRevisions = make(map[string]string, len(revisions))
	for name, revision := range revisions {
		name = strings.TrimSpace(name)
		revision = strings.TrimSpace(revision)
		if name != "" && revision != "" {
			opts.OfficialRepoSourceRevisions[name] = revision
			opts.OfficialRepoSourceRevisions[strings.ToLower(name)] = revision
		}
	}
	resolvedPack, err := packs.ResolveInstalledSourceBackedPackForInstalledRowAtSourceRevisionWithPackSourceOptionsAndComponentKindRegistrationSet(
		strings.TrimSpace(state.owner),
		packs.Category(strings.TrimSpace(state.category)),
		strings.TrimSpace(state.manifestVersion),
		sourceKind,
		sourceURI,
		strings.TrimSpace(state.sourceRevision),
		opts,
		r.componentKindRegistrationSet,
	)
	if err != nil {
		cleanup()
		return nil, func() {}, err
	}
	if _, err := packs.ValidateStoredPackPayloadHash(resolvedPack, strings.TrimSpace(state.appliedHash)); err != nil {
		cleanup()
		return nil, func() {}, fmt.Errorf("locked installed payload hash: %w", err)
	}
	return resolvedPack, cleanup, nil
}

func (r *connectorPushContractResolver) resolveCachedConnectorPushRecoveryHint(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	includeContracts bool,
) (connectorPushRecoveryHint, error) {
	if state == nil {
		return connectorPushRecoveryHint{}, fmt.Errorf("installed source state is required")
	}
	key := connectorPushRecoveryHintCacheKey{
		trustedLocalPackRoot: strings.TrimSpace(r.trustedLocalPackRoot),
		state:                *state,
		includeContracts:     includeContracts,
	}
	if connectorPushInboundGenerationIsPinned(ctx) {
		return r.resolveImmutableConnectorPushRecoveryHint(ctx, key, state, includeContracts)
	}
	scope := connectorPushInstalledContractRetryScopeFromContext(ctx, r)
	if scope != nil {
		scope.mu.RLock()
		cached, ok := scope.recoveryHints[key]
		scope.mu.RUnlock()
		if ok {
			return cloneConnectorPushRecoveryHint(cached), nil
		}
	}
	hint, err := r.resolveConnectorPushRecoveryHintUncached(ctx, state, includeContracts)
	if err != nil {
		return connectorPushRecoveryHint{}, err
	}
	hint = cloneConnectorPushRecoveryHint(hint)
	if scope == nil {
		return hint, nil
	}
	scope.mu.Lock()
	if scope.recoveryHints == nil {
		scope.recoveryHints = make(map[connectorPushRecoveryHintCacheKey]connectorPushRecoveryHint)
	}
	if existing, exists := scope.recoveryHints[key]; exists {
		hint = cloneConnectorPushRecoveryHint(existing)
	} else {
		scope.recoveryHints[key] = cloneConnectorPushRecoveryHint(hint)
	}
	scope.mu.Unlock()
	return hint, nil
}

func (r *connectorPushContractResolver) resolveConnectorPushRecoveryHintUncached(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	includeContracts bool,
) (connectorPushRecoveryHint, error) {
	resolvedPack, cleanup, err := r.resolveConnectorPushRecoveryHintPack(ctx, state)
	if err != nil {
		return connectorPushRecoveryHint{}, err
	}
	defer cleanup()
	hint := connectorPushRecoveryHint{}
	if sourcePack, ok := resolvedPack.(packs.PackDependencyProvider); ok {
		for _, dependency := range sourcePack.PackDependencies() {
			hint.dependencies = append(hint.dependencies, dependency.Name)
		}
	}
	hint.dependencies = uniqueSortedConnectorPushStrings(hint.dependencies)
	if includeContracts {
		declarations := r.contractDeclarations
		if declarations == nil {
			declarations = connectorPushContractDeclarationsFromPack
		}
		hint.contracts, err = declarations(resolvedPack)
		if err != nil {
			return connectorPushRecoveryHint{}, err
		}
	}
	return cloneConnectorPushRecoveryHint(hint), nil
}

func (r *connectorPushContractResolver) resolveImmutableConnectorPushRecoveryHint(
	ctx context.Context,
	key connectorPushRecoveryHintCacheKey,
	state *connectorPushInstalledPackState,
	includeContracts bool,
) (connectorPushRecoveryHint, error) {
	r.immutableSourceCacheMu.RLock()
	cached, ok := r.immutableRecoveryHints[key]
	r.immutableSourceCacheMu.RUnlock()
	if ok {
		return cloneConnectorPushRecoveryHint(cached), nil
	}
	loadKey := fmt.Sprintf("%#v", key)
	result := r.immutableRecoveryHintLoads.DoChan(loadKey, func() (interface{}, error) {
		r.immutableSourceCacheMu.RLock()
		cached, ok := r.immutableRecoveryHints[key]
		r.immutableSourceCacheMu.RUnlock()
		if ok {
			return cloneConnectorPushRecoveryHint(cached), nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		hint, err := r.resolveConnectorPushRecoveryHintUncached(loadCtx, state, includeContracts)
		if err != nil {
			return connectorPushRecoveryHint{}, err
		}
		r.immutableSourceCacheMu.Lock()
		if r.immutableRecoveryHints == nil {
			r.immutableRecoveryHints = make(map[connectorPushRecoveryHintCacheKey]connectorPushRecoveryHint)
		}
		if len(r.immutableRecoveryHints) >= connectorPushImmutableCacheMaxEntries {
			clear(r.immutableRecoveryHints)
		}
		r.immutableRecoveryHints[key] = cloneConnectorPushRecoveryHint(hint)
		r.immutableSourceCacheMu.Unlock()
		return cloneConnectorPushRecoveryHint(hint), nil
	})
	select {
	case <-ctx.Done():
		return connectorPushRecoveryHint{}, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return connectorPushRecoveryHint{}, loaded.Err
		}
		hint, _ := loaded.Val.(connectorPushRecoveryHint)
		return cloneConnectorPushRecoveryHint(hint), nil
	}
}

func (r *connectorPushContractResolver) resolveCachedConnectorPushLockedSourceGeneration(
	ctx context.Context,
	state *connectorPushInstalledPackState,
	revisions map[string]string,
	resolve func(context.Context, *connectorPushInstalledPackState, map[string]string) (connectorPushLockedSourceGeneration, error),
) (connectorPushLockedSourceGeneration, error) {
	if state == nil {
		return connectorPushLockedSourceGeneration{}, fmt.Errorf("installed source state is required")
	}
	key := connectorPushLockedSourceGenerationCacheKey{
		trustedLocalPackRoot: strings.TrimSpace(r.trustedLocalPackRoot),
		state:                *state,
		revisions:            connectorPushSourceRevisionSetCacheKey(revisions),
	}
	if connectorPushInboundGenerationIsPinned(ctx) {
		return r.resolveImmutableConnectorPushLockedSourceGeneration(ctx, key, state, revisions, resolve)
	}
	scope := connectorPushInstalledContractRetryScopeFromContext(ctx, r)
	if scope != nil {
		scope.mu.RLock()
		cached, ok := scope.lockedSourceGenerations[key]
		scope.mu.RUnlock()
		if ok {
			return cloneConnectorPushLockedSourceGeneration(cached), nil
		}
	}

	generation, err := resolve(ctx, state, revisions)
	if err != nil {
		return connectorPushLockedSourceGeneration{}, err
	}
	generation = cloneConnectorPushLockedSourceGeneration(generation)
	if scope == nil {
		return generation, nil
	}
	scope.mu.Lock()
	if scope.lockedSourceGenerations == nil {
		scope.lockedSourceGenerations = make(map[connectorPushLockedSourceGenerationCacheKey]connectorPushLockedSourceGeneration)
	}
	if existing, exists := scope.lockedSourceGenerations[key]; exists {
		generation = cloneConnectorPushLockedSourceGeneration(existing)
	} else {
		scope.lockedSourceGenerations[key] = cloneConnectorPushLockedSourceGeneration(generation)
	}
	scope.mu.Unlock()
	return generation, nil
}

func (r *connectorPushContractResolver) resolveImmutableConnectorPushLockedSourceGeneration(
	ctx context.Context,
	key connectorPushLockedSourceGenerationCacheKey,
	state *connectorPushInstalledPackState,
	revisions map[string]string,
	resolve func(context.Context, *connectorPushInstalledPackState, map[string]string) (connectorPushLockedSourceGeneration, error),
) (connectorPushLockedSourceGeneration, error) {
	r.immutableSourceCacheMu.RLock()
	cached, ok := r.immutableLockedSourceStates[key]
	r.immutableSourceCacheMu.RUnlock()
	if ok {
		return cloneConnectorPushLockedSourceGeneration(cached), nil
	}
	loadKey := fmt.Sprintf("%#v", key)
	result := r.immutableLockedSourceStateLoads.DoChan(loadKey, func() (interface{}, error) {
		r.immutableSourceCacheMu.RLock()
		cached, ok := r.immutableLockedSourceStates[key]
		r.immutableSourceCacheMu.RUnlock()
		if ok {
			return cloneConnectorPushLockedSourceGeneration(cached), nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), connectorPushImmutableLoadTimeout)
		defer cancel()
		generation, err := resolve(loadCtx, state, revisions)
		if err != nil {
			return connectorPushLockedSourceGeneration{}, err
		}
		generation = cloneConnectorPushLockedSourceGeneration(generation)
		r.immutableSourceCacheMu.Lock()
		if r.immutableLockedSourceStates == nil {
			r.immutableLockedSourceStates = make(map[connectorPushLockedSourceGenerationCacheKey]connectorPushLockedSourceGeneration)
		}
		if len(r.immutableLockedSourceStates) >= connectorPushImmutableCacheMaxEntries {
			clear(r.immutableLockedSourceStates)
		}
		r.immutableLockedSourceStates[key] = cloneConnectorPushLockedSourceGeneration(generation)
		r.immutableSourceCacheMu.Unlock()
		return cloneConnectorPushLockedSourceGeneration(generation), nil
	})
	select {
	case <-ctx.Done():
		return connectorPushLockedSourceGeneration{}, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return connectorPushLockedSourceGeneration{}, loaded.Err
		}
		generation, _ := loaded.Val.(connectorPushLockedSourceGeneration)
		return cloneConnectorPushLockedSourceGeneration(generation), nil
	}
}

func connectorPushSourceRevisionSetCacheKey(revisions map[string]string) string {
	keys := make([]string, 0, len(revisions))
	for name := range revisions {
		keys = append(keys, name)
	}
	sort.Strings(keys)
	var result strings.Builder
	for _, name := range keys {
		revision := revisions[name]
		_, _ = fmt.Fprintf(&result, "%d:%s%d:%s", len(name), name, len(revision), revision)
	}
	return result.String()
}

func cloneConnectorPushRecoveryHint(hint connectorPushRecoveryHint) connectorPushRecoveryHint {
	return connectorPushRecoveryHint{
		contracts:    cloneConnectorPushContracts(hint.contracts),
		dependencies: append([]string(nil), hint.dependencies...),
	}
}

func cloneConnectorPushLockedSourceGeneration(generation connectorPushLockedSourceGeneration) connectorPushLockedSourceGeneration {
	result := connectorPushLockedSourceGeneration{
		contracts:    cloneConnectorPushContracts(generation.contracts),
		mappings:     append([]contractconnector.ConnectorFieldMappingInput(nil), generation.mappings...),
		dependencies: append([]string(nil), generation.dependencies...),
	}
	if generation.objectFields != nil {
		result.objectFields = make(map[string][]contractconnector.ConnectorPushObjectFieldMetadata, len(generation.objectFields))
		for objectName, fields := range generation.objectFields {
			clonedFields := append([]contractconnector.ConnectorPushObjectFieldMetadata(nil), fields...)
			for i := range clonedFields {
				clonedFields[i].ReferenceTo = append([]string(nil), clonedFields[i].ReferenceTo...)
			}
			result.objectFields[objectName] = clonedFields
		}
	}
	return result
}

func cloneConnectorPushContracts(contracts []bootstrap.ConnectorPushDurableMetadataContract) []bootstrap.ConnectorPushDurableMetadataContract {
	result := make([]bootstrap.ConnectorPushDurableMetadataContract, 0, len(contracts))
	for i := range contracts {
		result = append(result, *cloneConnectorPushContract(&contracts[i]))
	}
	return result
}

func connectorInboundLockedSourceClosure(
	owner string,
	resolved map[string]connectorPushLockedSourceGeneration,
) ([]string, error) {
	owner = strings.TrimSpace(owner)
	seen := map[string]uint8{}
	var visit func(string, int) error
	visit = func(current string, depth int) error {
		current = strings.TrimSpace(current)
		if current == "" {
			return fmt.Errorf("locked source dependency name is blank")
		}
		if depth > connectorInboundInstalledSourceMaxDepth {
			return fmt.Errorf("locked source dependency closure exceeds maximum depth %d", connectorInboundInstalledSourceMaxDepth)
		}
		switch seen[current] {
		case 1:
			return fmt.Errorf("locked source dependency cycle detected at %q", current)
		case 2:
			return nil
		}
		generation, ok := resolved[current]
		if !ok {
			return fmt.Errorf("locked source dependency %q is missing", current)
		}
		if len(seen) >= connectorInboundInstalledSourceMaxPacks {
			return fmt.Errorf("locked source dependency closure exceeds maximum pack count %d", connectorInboundInstalledSourceMaxPacks)
		}
		seen[current] = 1
		for _, dependency := range uniqueSortedConnectorPushStrings(generation.dependencies) {
			if err := visit(dependency, depth+1); err != nil {
				return err
			}
		}
		seen[current] = 2
		return nil
	}
	if err := visit(owner, 0); err != nil {
		return nil, err
	}
	owners := make([]string, 0, len(seen))
	for dependencyOwner, state := range seen {
		if state != 2 {
			continue
		}
		if dependencyOwner != "" {
			owners = append(owners, dependencyOwner)
		}
	}
	sort.Strings(owners)
	return owners, nil
}

func mergeConnectorPushLockedObjectFields(
	resolved map[string]connectorPushLockedSourceGeneration,
) (map[string][]contractconnector.ConnectorPushObjectFieldMetadata, error) {
	byObject := map[string]map[string]contractconnector.ConnectorPushObjectFieldMetadata{}
	owners := make([]string, 0, len(resolved))
	for owner := range resolved {
		owners = append(owners, owner)
	}
	sort.Strings(owners)
	for _, owner := range owners {
		for objectName, fields := range resolved[owner].objectFields {
			objectName = strings.ToLower(strings.TrimSpace(objectName))
			if objectName == "" {
				continue
			}
			if byObject[objectName] == nil {
				byObject[objectName] = map[string]contractconnector.ConnectorPushObjectFieldMetadata{}
			}
			for _, field := range fields {
				key := strings.ToLower(strings.TrimSpace(field.APIName))
				if key == "" {
					continue
				}
				if existing, ok := byObject[objectName][key]; ok && !reflect.DeepEqual(existing, field) {
					return nil, fmt.Errorf("locked source object schema conflicts for %s.%s", objectName, field.APIName)
				}
				byObject[objectName][key] = field
			}
		}
	}
	result := make(map[string][]contractconnector.ConnectorPushObjectFieldMetadata, len(byObject))
	for objectName, fields := range byObject {
		keys := make([]string, 0, len(fields))
		for key := range fields {
			keys = append(keys, key)
		}
		sort.Strings(keys)
		for _, key := range keys {
			result[objectName] = append(result[objectName], fields[key])
		}
	}
	return result, nil
}

func uniqueSortedConnectorPushStrings(values []string) []string {
	seen := map[string]struct{}{}
	for _, value := range values {
		if value = strings.TrimSpace(value); value != "" {
			seen[value] = struct{}{}
		}
	}
	result := make([]string, 0, len(seen))
	for value := range seen {
		result = append(result, value)
	}
	sort.Strings(result)
	return result
}

// IsImmutableConnectorPushSourceRevision reports whether revision is an exact
// commit identity suitable for strict source-backed recovery. Setup preflight
// and the runtime resolver share this validator so compiled source locks and
// installed source tuples cannot drift to different immutability rules.
func IsImmutableConnectorPushSourceRevision(revision string) bool {
	revision = strings.TrimSpace(revision)
	if len(revision) != 40 {
		return false
	}
	_, err := hex.DecodeString(revision)
	return err == nil
}

func (r *connectorPushContractResolver) resolveContractFromPack(ctx context.Context, state *connectorPushInstalledPackState, flowName, actionName string) (*bootstrap.ConnectorPushDurableMetadataContract, error) {
	contracts, err := r.resolveContractsFromPack(ctx, state)
	if err != nil {
		return nil, err
	}
	owner := strings.TrimSpace(state.owner)
	flowName = strings.TrimSpace(flowName)
	actionName = strings.TrimSpace(actionName)
	var matched *bootstrap.ConnectorPushDurableMetadataContract
	for _, contract := range contracts {
		if strings.TrimSpace(contract.ConnectorOwnerKey) != owner {
			continue
		}
		if strings.TrimSpace(contract.FlowDeveloperName) != flowName {
			continue
		}
		if strings.TrimSpace(contract.FlowActionName) != actionName {
			continue
		}
		contractCopy := contract
		if matched != nil {
			return nil, fmt.Errorf("connector_push_contract %s/%s/%s is ambiguous", owner, flowName, actionName)
		}
		matched = &contractCopy
	}
	return matched, nil
}

func (r *connectorPushContractResolver) resolveContractsFromPack(ctx context.Context, state *connectorPushInstalledPackState) ([]bootstrap.ConnectorPushDurableMetadataContract, error) {
	declarations := r.contractDeclarations
	if declarations == nil {
		declarations = connectorPushContractDeclarationsFromPack
	}
	resolvedPack, err := r.resolveInstalledPackFromState(ctx, state)
	if err != nil {
		return nil, err
	}
	contracts, err := declarations(resolvedPack)
	if err != nil {
		return nil, err
	}
	return contracts, nil
}

func (r *connectorPushContractResolver) resolveInstalledPackFromState(ctx context.Context, state *connectorPushInstalledPackState) (packs.Pack, error) {
	resolve := r.resolveInstalledPack
	if resolve == nil {
		resolve = packs.ResolveInstalledPack
	}
	objectSchemaResolver := r.objectSchemaResolver
	if objectSchemaResolver == nil {
		objectSchemaResolver = connectorPushRuntimeObjectSchemaResolver
	}
	owner := strings.TrimSpace(state.owner)
	resolvedPack, err := resolve(
		packs.InstalledPackResolverInput{
			Name:                  owner,
			Category:              packs.Category(strings.TrimSpace(state.category)),
			AppliedPayloadHash:    state.appliedHash,
			ManifestFormatVersion: state.manifestVersion,
			SourceKind:            state.sourceKind,
			SourceURI:             state.sourceURI,
			SourceRevision:        state.sourceRevision,
		},
		packs.InstalledPackResolverOptions{
			AllowBundledFallback:         false,
			ObjectSchemaResolver:         objectSchemaResolver(ctx, r.db),
			ComponentKindRegistrationSet: &r.componentKindRegistrationSet,
		},
	)
	if err != nil {
		return nil, err
	}
	return resolvedPack, nil
}

// connector_push_contracts may reference objects supplied by a dependency pack.
// Resolve them from the installed metadata so source-backed contract recovery can
// validate that bounded dependency surface rather than treating it as absent.
func connectorPushRuntimeObjectSchemaResolver(ctx context.Context, db *sql.DB) func(string) *models.ObjectMetadata {
	if db == nil {
		return nil
	}
	metadataRepo := persistence.NewMetadataRepository(db)
	schemaRepo := persistence.NewSchemaRepository(db, nil)
	metadataSvc := metadata.NewMetadataService(metadataRepo, metadata.NewSchemaManager(schemaRepo), nil)
	return func(apiName string) *models.ObjectMetadata {
		return metadataSvc.GetSchema(ctx, apiName)
	}
}

func connectorPushContractDeclarationsFromPack(resolvedPack packs.Pack) ([]bootstrap.ConnectorPushDurableMetadataContract, error) {
	pack, ok := resolvedPack.(*packs.SourceBackedPack)
	if !ok {
		packName := ""
		if resolvedPack != nil {
			packName = resolvedPack.Name()
		}
		return nil, fmt.Errorf("installed pack %s did not resolve to source-backed pack", packName)
	}
	return pack.ConnectorPushContractDeclarations(), nil
}

func cloneConnectorPushContract(contract *bootstrap.ConnectorPushDurableMetadataContract) *bootstrap.ConnectorPushDurableMetadataContract {
	if contract == nil {
		return nil
	}
	out := *contract
	out.PreviousActionSlotIDs = append([]string(nil), contract.PreviousActionSlotIDs...)
	out.DiagnosticErrorFieldPaths = append([]string(nil), contract.DiagnosticErrorFieldPaths...)
	if contract.CreateConflictResolution != nil {
		resolution := *contract.CreateConflictResolution
		resolution.StatusCodes = append([]int(nil), contract.CreateConflictResolution.StatusCodes...)
		if contract.CreateConflictResolution.DiagnosticMatch != nil {
			resolution.DiagnosticMatch = make(map[string]string, len(contract.CreateConflictResolution.DiagnosticMatch))
			for key, value := range contract.CreateConflictResolution.DiagnosticMatch {
				resolution.DiagnosticMatch[key] = value
			}
		}
		out.CreateConflictResolution = &resolution
	}
	out.Associations = make([]bootstrap.ConnectorPushAssociationContract, len(contract.Associations))
	for i, assoc := range contract.Associations {
		out.Associations[i] = cloneConnectorPushAssociation(assoc)
	}
	return &out
}

func cloneConnectorPushAssociation(assoc bootstrap.ConnectorPushAssociationContract) bootstrap.ConnectorPushAssociationContract {
	out := assoc
	if assoc.Junction != nil {
		junction := *assoc.Junction
		junction.Filters = make([]bootstrap.ConnectorPushAssociationJunctionFilter, len(assoc.Junction.Filters))
		for i, filter := range assoc.Junction.Filters {
			junction.Filters[i] = filter
			junction.Filters[i].Equals = cloneConnectorPushTemplateValue(filter.Equals)
		}
		out.Junction = &junction
	}
	out.TargetPreviousActionSlotIDs = append([]string(nil), assoc.TargetPreviousActionSlotIDs...)
	out.PayloadTemplate = cloneConnectorPushTemplateValue(assoc.PayloadTemplate)
	out.InlinePhases = append([]string(nil), assoc.InlinePhases...)
	out.CalloutBodyTemplate = cloneConnectorPushTemplateValue(assoc.CalloutBodyTemplate)
	out.CalloutPhases = append([]string(nil), assoc.CalloutPhases...)
	return out
}

func cloneConnectorPushTemplateValue(value interface{}) interface{} {
	switch typed := value.(type) {
	case map[string]interface{}:
		out := make(map[string]interface{}, len(typed))
		for key, nested := range typed {
			out[key] = cloneConnectorPushTemplateValue(nested)
		}
		return out
	case []interface{}:
		out := make([]interface{}, len(typed))
		for i, nested := range typed {
			out[i] = cloneConnectorPushTemplateValue(nested)
		}
		return out
	case []string:
		return append([]string(nil), typed...)
	case []int:
		return append([]int(nil), typed...)
	default:
		return typed
	}
}