package runtimeworker

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

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/backend/internal/managedruntime"
	"github.com/nexuscrm/backend/internal/runtimeflowexecution"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	contractconnector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"golang.org/x/sync/singleflight"
)

const (
	immutableArtifactLoadTimeout = 30 * time.Second
	openedSnapshotCacheLimit     = 8
	verifiedViewCacheLimit       = 16
)

func runtimeFlowMetadataViewFromContext(
	ctx context.Context,
) (runtimeflowexecution.MetadataView, bool) {
	view, ok := metadata.RuntimeFlowWorkViewFromContext(ctx)
	return view, ok && view != nil
}

// metadataBindingError distinguishes corrupt or out-of-scope durable authority
// from temporarily unavailable content-addressed artifacts. Both fail closed;
// only the latter remains retryable.
type metadataBindingError struct {
	cause     error
	permanent bool
}

func (e *metadataBindingError) Error() string {
	if e == nil || e.cause == nil {
		return "runtime worker metadata binding failed"
	}
	return e.cause.Error()
}

func (e *metadataBindingError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.cause
}

func permanentMetadataBindingError(err error) error {
	if err == nil {
		return nil
	}
	return &metadataBindingError{cause: err, permanent: true}
}

func retryableMetadataBindingError(err error) error {
	if err == nil {
		return nil
	}
	return &metadataBindingError{cause: err}
}

func isPermanentMetadataBindingError(err error) bool {
	var bindingErr *metadataBindingError
	return errors.As(err, &bindingErr) && bindingErr.permanent
}

// mutationArtifactBinder reopens only the exact immutable content named by a
// durable row. It has no latest/current selector, Catalog repository, TTL, or
// fallback path.
type mutationArtifactBinder struct {
	store                              catalogsnapshot.Store
	scope                              persistence.RuntimeWorkScope
	compatibility                      catalogsnapshot.Compatibility
	prepareAssignmentPolicyEligibility func(
		context.Context,
		*catalogsnapshot.RuntimeMetadataView,
	) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error)

	cacheMu                     sync.RWMutex
	snapshots                   map[catalogsnapshot.Digest]*catalogsnapshot.CatalogSnapshot
	verifiedViews               map[verifiedRuntimeViewKey]*catalogsnapshot.RuntimeMetadataView
	assignmentPolicyEligibility map[catalogsnapshot.Digest]assignmentPolicyEligibilityBinding
	snapshotLoads               singleflight.Group
	viewLoads                   singleflight.Group
	siblingLoads                singleflight.Group
	assignmentPolicyLoads       singleflight.Group
}

type verifiedRuntimeViewKey struct {
	snapshot catalogsnapshot.Digest
	view     catalogsnapshot.Digest
}

type assignmentPolicyEligibilityBinding struct {
	stager managedruntime.AssignmentPolicyEligibilityAmbientStager
}

func newMutationArtifactBinder(
	store catalogsnapshot.Store,
	scope persistence.RuntimeWorkScope,
	compatibility catalogsnapshot.Compatibility,
	prepareAssignmentPolicyEligibility func(
		context.Context,
		*catalogsnapshot.RuntimeMetadataView,
	) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error),
) (*mutationArtifactBinder, error) {
	if store == nil || scope.Validate() != nil || compatibility.SchemaGeneration == 0 ||
		compatibility.EngineGeneration == 0 || prepareAssignmentPolicyEligibility == nil {
		return nil, errors.New("runtime worker metadata binder: store, scope, compatibility, and lifecycle preparer are required")
	}
	return &mutationArtifactBinder{
		store: store, scope: scope, compatibility: compatibility,
		prepareAssignmentPolicyEligibility: prepareAssignmentPolicyEligibility,
		snapshots:                          make(map[catalogsnapshot.Digest]*catalogsnapshot.CatalogSnapshot),
		verifiedViews:                      make(map[verifiedRuntimeViewKey]*catalogsnapshot.RuntimeMetadataView),
		assignmentPolicyEligibility:        make(map[catalogsnapshot.Digest]assignmentPolicyEligibilityBinding),
	}, nil
}

func (b *mutationArtifactBinder) open(ctx context.Context, binding catalogsnapshot.RuntimeWorkBinding) (*catalogsnapshot.RuntimeMetadataView, error) {
	return b.openProfile(ctx, binding, catalogsnapshot.ProjectionRecordMutationV1)
}

func (b *mutationArtifactBinder) openProfile(
	ctx context.Context,
	binding catalogsnapshot.RuntimeWorkBinding,
	profile catalogsnapshot.ProjectionProfile,
) (*catalogsnapshot.RuntimeMetadataView, error) {
	return b.openProfileForCapability(ctx, binding, profile, b.scope.RuntimeCapabilityID)
}

func (b *mutationArtifactBinder) openProfileForCapability(
	ctx context.Context,
	binding catalogsnapshot.RuntimeWorkBinding,
	profile catalogsnapshot.ProjectionProfile,
	expectedCapability string,
) (*catalogsnapshot.RuntimeMetadataView, error) {
	if b == nil || b.store == nil {
		return nil, retryableMetadataBindingError(
			errors.New("runtime worker metadata binder: binder is unavailable"),
		)
	}
	if profile == "" {
		return nil, permanentMetadataBindingError(
			errors.New("runtime worker metadata binder: expected profile is required"),
		)
	}
	if binding.OrgID != b.scope.OrgID || binding.AppInstallationID != b.scope.AppInstallationID ||
		binding.RuntimeCapabilityID != expectedCapability {
		return nil, permanentMetadataBindingError(
			errors.New("runtime worker metadata binder: durable work is outside the configured scope"),
		)
	}
	snapshot, err := b.openSnapshot(ctx, binding.CatalogSnapshotDigest)
	if err != nil {
		return nil, err
	}
	view, err := b.openVerifiedRuntimeView(ctx, snapshot, binding.RuntimeMetadataViewDigest)
	if err != nil {
		return nil, err
	}
	if err := binding.VerifyView(view, profile); err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	if err := view.VerifyCompatibility(b.compatibility); err != nil {
		return nil, retryableMetadataBindingError(
			fmt.Errorf("runtime worker metadata binder: verify compatibility: %w", err),
		)
	}
	return view, nil
}

// openSnapshot and openVerifiedRuntimeView retain only immutable,
// content-addressed values. A durable binding still supplies and verifies the
// exact digests, scope, profile, lineage, and compatibility on every attempt;
// the cache removes repeated file reads, canonical JSON decoding, and exact
// projection regeneration. It is therefore an execution cache, not a second
// metadata selector or source of truth.
func (b *mutationArtifactBinder) openSnapshot(
	ctx context.Context,
	digest catalogsnapshot.Digest,
) (*catalogsnapshot.CatalogSnapshot, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if snapshot, ok := b.cachedSnapshot(digest); ok {
		return snapshot, nil
	}
	result := b.snapshotLoads.DoChan(string(digest), func() (any, error) {
		if snapshot, ok := b.cachedSnapshot(digest); ok {
			return snapshot, nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), immutableArtifactLoadTimeout)
		defer cancel()
		artifact, err := b.store.Get(loadCtx, catalogsnapshot.ArtifactCatalogSnapshot, digest)
		if err != nil {
			return nil, retryableMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: load Catalog snapshot: %w", err),
			)
		}
		snapshot, err := catalogsnapshot.OpenCatalogSnapshot(artifact)
		if err != nil {
			return nil, permanentMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: open Catalog snapshot: %w", err),
			)
		}
		b.cacheSnapshot(digest, snapshot)
		return snapshot, nil
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		snapshot, ok := loaded.Val.(*catalogsnapshot.CatalogSnapshot)
		if !ok || snapshot == nil {
			return nil, retryableMetadataBindingError(errors.New("runtime worker metadata binder: snapshot cache returned an invalid value"))
		}
		return snapshot, nil
	}
}

func (b *mutationArtifactBinder) openVerifiedRuntimeView(
	ctx context.Context,
	snapshot *catalogsnapshot.CatalogSnapshot,
	digest catalogsnapshot.Digest,
) (*catalogsnapshot.RuntimeMetadataView, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if snapshot == nil {
		return nil, permanentMetadataBindingError(errors.New("runtime worker metadata binder: Catalog snapshot is required"))
	}
	key := verifiedRuntimeViewKey{snapshot: snapshot.Digest(), view: digest}
	if view, ok := b.cachedVerifiedView(key); ok {
		return view, nil
	}
	loadKey := string(key.snapshot) + "\x00" + string(key.view)
	result := b.viewLoads.DoChan(loadKey, func() (any, error) {
		if view, ok := b.cachedVerifiedView(key); ok {
			return view, nil
		}
		loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), immutableArtifactLoadTimeout)
		defer cancel()
		artifact, err := b.store.Get(loadCtx, catalogsnapshot.ArtifactRuntimeMetadataView, digest)
		if err != nil {
			return nil, retryableMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: load Runtime metadata view: %w", err),
			)
		}
		view, err := catalogsnapshot.OpenRuntimeMetadataView(artifact)
		if err != nil {
			return nil, permanentMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: open Runtime metadata view: %w", err),
			)
		}
		if err := view.VerifySource(snapshot); err != nil {
			return nil, permanentMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: verify Catalog source: %w", err),
			)
		}
		b.cacheVerifiedView(key, view)
		return view, nil
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		view, ok := loaded.Val.(*catalogsnapshot.RuntimeMetadataView)
		if !ok || view == nil {
			return nil, retryableMetadataBindingError(errors.New("runtime worker metadata binder: view cache returned an invalid value"))
		}
		return view, nil
	}
}

func (b *mutationArtifactBinder) cachedSnapshot(digest catalogsnapshot.Digest) (*catalogsnapshot.CatalogSnapshot, bool) {
	b.cacheMu.RLock()
	defer b.cacheMu.RUnlock()
	snapshot, ok := b.snapshots[digest]
	return snapshot, ok
}

func (b *mutationArtifactBinder) cacheSnapshot(digest catalogsnapshot.Digest, snapshot *catalogsnapshot.CatalogSnapshot) {
	b.cacheMu.Lock()
	defer b.cacheMu.Unlock()
	if len(b.snapshots) >= openedSnapshotCacheLimit {
		clear(b.snapshots)
	}
	b.snapshots[digest] = snapshot
}

func (b *mutationArtifactBinder) cachedVerifiedView(key verifiedRuntimeViewKey) (*catalogsnapshot.RuntimeMetadataView, bool) {
	b.cacheMu.RLock()
	defer b.cacheMu.RUnlock()
	view, ok := b.verifiedViews[key]
	return view, ok
}

func (b *mutationArtifactBinder) cacheVerifiedView(key verifiedRuntimeViewKey, view *catalogsnapshot.RuntimeMetadataView) {
	b.cacheMu.Lock()
	defer b.cacheMu.Unlock()
	if len(b.verifiedViews) >= verifiedViewCacheLimit {
		clear(b.verifiedViews)
	}
	b.verifiedViews[key] = view
}

// prepareAssignmentPolicyEligibilityBinding composes the managed lifecycle
// host from the same content-addressed mutation view already verified for the
// durable row. The cache key is the immutable view digest; it is an execution
// cache only and cannot select current/latest Catalog state.
func (b *mutationArtifactBinder) prepareAssignmentPolicyEligibilityBinding(
	ctx context.Context,
	view *catalogsnapshot.RuntimeMetadataView,
) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if b == nil || b.prepareAssignmentPolicyEligibility == nil {
		return nil, permanentMetadataBindingError(
			errors.New("runtime worker metadata binder: lifecycle preparer is unavailable"),
		)
	}
	if view == nil {
		return nil, permanentMetadataBindingError(
			errors.New("runtime worker metadata binder: mutation view is required for lifecycle preparation"),
		)
	}
	digest := view.Digest()
	if binding, ok := b.cachedAssignmentPolicyEligibility(digest); ok {
		return binding.stager, nil
	}
	result := b.assignmentPolicyLoads.DoChan(string(digest), func() (any, error) {
		if binding, ok := b.cachedAssignmentPolicyEligibility(digest); ok {
			return binding, nil
		}
		prepareCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), immutableArtifactLoadTimeout)
		defer cancel()
		stager, err := b.prepareAssignmentPolicyEligibility(prepareCtx, view)
		if err != nil {
			return nil, retryableMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: prepare assignment-policy lifecycle: %w", err),
			)
		}
		binding := assignmentPolicyEligibilityBinding{stager: stager}
		b.cacheAssignmentPolicyEligibility(digest, binding)
		return binding, nil
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		binding, ok := loaded.Val.(assignmentPolicyEligibilityBinding)
		if !ok {
			return nil, retryableMetadataBindingError(
				errors.New("runtime worker metadata binder: lifecycle cache returned an invalid value"),
			)
		}
		return binding.stager, nil
	}
}

func (b *mutationArtifactBinder) cachedAssignmentPolicyEligibility(
	digest catalogsnapshot.Digest,
) (assignmentPolicyEligibilityBinding, bool) {
	b.cacheMu.RLock()
	defer b.cacheMu.RUnlock()
	binding, ok := b.assignmentPolicyEligibility[digest]
	return binding, ok
}

func (b *mutationArtifactBinder) cacheAssignmentPolicyEligibility(
	digest catalogsnapshot.Digest,
	binding assignmentPolicyEligibilityBinding,
) {
	b.cacheMu.Lock()
	defer b.cacheMu.Unlock()
	if len(b.assignmentPolicyEligibility) >= verifiedViewCacheLimit {
		clear(b.assignmentPolicyEligibility)
	}
	b.assignmentPolicyEligibility[digest] = binding
}

// openSiblingProfile derives only the content address of a second authorized
// projection from the durable row's exact Catalog snapshot, then requires that
// exact view to exist in the publication store. It never selects current/latest
// metadata and does not turn the Worker into a second publication authority.
func (b *mutationArtifactBinder) openSiblingProfile(
	ctx context.Context,
	source catalogsnapshot.RuntimeWorkBinding,
	profile catalogsnapshot.ProjectionProfile,
	capabilityID string,
) (*catalogsnapshot.RuntimeMetadataView, error) {
	view, err := b.openSiblingProfileFromSource(
		ctx,
		source,
		profile,
		capabilityID,
	)
	if err != nil {
		return nil, err
	}
	sibling, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	if err := (catalogsnapshot.RuntimeWorkBindingSet{
		Primary: sibling, FlowWork: &source,
	}).Validate(); err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	return view, nil
}

func (b *mutationArtifactBinder) openSiblingProfileFromSource(
	ctx context.Context,
	source catalogsnapshot.RuntimeWorkBinding,
	profile catalogsnapshot.ProjectionProfile,
	capabilityID string,
) (*catalogsnapshot.RuntimeMetadataView, error) {
	if b == nil || b.store == nil {
		return nil, retryableMetadataBindingError(
			errors.New("runtime worker metadata binder: binder is unavailable"),
		)
	}
	view, err := b.openVerifiedSibling(ctx, source.CatalogSnapshotDigest, source.AppInstallationID, capabilityID)
	if err != nil {
		return nil, err
	}
	sibling, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	if sibling.OrgID != source.OrgID ||
		sibling.AppInstallationID != source.AppInstallationID ||
		sibling.CatalogRevision != source.CatalogRevision ||
		sibling.SchemaGeneration != source.SchemaGeneration ||
		sibling.EngineGeneration != source.EngineGeneration ||
		sibling.CatalogSnapshotDigest != source.CatalogSnapshotDigest {
		return nil, permanentMetadataBindingError(
			errors.New("runtime worker metadata binder: mixed-revision sibling projection"),
		)
	}
	return b.openProfileForCapability(ctx, sibling, profile, capabilityID)
}

// A snapshot has exactly one projection per installation/capability. Resolve
// that tuple from the existing bounded, source-verified cache before compiling
// its content address. No separate selector/index may outlive those entries.
func (b *mutationArtifactBinder) cachedVerifiedSibling(
	snapshot catalogsnapshot.Digest,
	installationID string,
	capabilityID string,
) (*catalogsnapshot.RuntimeMetadataView, bool) {
	b.cacheMu.RLock()
	defer b.cacheMu.RUnlock()
	for key, view := range b.verifiedViews {
		if key.snapshot == snapshot && view.SourceSnapshotDigest() == snapshot &&
			view.AppInstallationID() == installationID && view.CapabilityID() == capabilityID {
			return view, true
		}
	}
	return nil, false
}

func (b *mutationArtifactBinder) openVerifiedSibling(
	ctx context.Context,
	snapshotDigest catalogsnapshot.Digest,
	installationID string,
	capabilityID string,
) (*catalogsnapshot.RuntimeMetadataView, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if view, ok := b.cachedVerifiedSibling(snapshotDigest, installationID, capabilityID); ok {
		return view, nil
	}
	// Length-prefix the installation identity so separators within identities
	// cannot merge distinct immutable tuples.
	loadKey := fmt.Sprintf("%s/%d:%s/%s", snapshotDigest, len(installationID), installationID, capabilityID)
	result := b.siblingLoads.DoChan(loadKey, func() (any, error) {
		if view, ok := b.cachedVerifiedSibling(snapshotDigest, installationID, capabilityID); ok {
			return view, nil
		}
		// Shared orchestration outlives individual callers. Snapshot and view
		// loaders own independent budgets; do not add a combined deadline.
		loadCtx := context.WithoutCancel(ctx)
		snapshot, err := b.openSnapshot(loadCtx, snapshotDigest)
		if err != nil {
			return nil, err
		}
		projected, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
			AppInstallationID: installationID,
			CapabilityID:      capabilityID,
		})
		if err != nil {
			return nil, permanentMetadataBindingError(
				fmt.Errorf("runtime worker metadata binder: derive authorized sibling projection: %w", err),
			)
		}
		// Projection discovers only the address. Execution still requires the
		// published artifact and the existing exact-source verification gate.
		return b.openVerifiedRuntimeView(loadCtx, snapshot, projected.Digest())
	})
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case loaded := <-result:
		if loaded.Err != nil {
			return nil, loaded.Err
		}
		view, ok := loaded.Val.(*catalogsnapshot.RuntimeMetadataView)
		if !ok || view == nil {
			return nil, retryableMetadataBindingError(errors.New("runtime worker metadata binder: sibling load returned an invalid value"))
		}
		return view, nil
	}
}

func (b *mutationArtifactBinder) bindScheduledFlow(
	ctx context.Context,
	encoded string,
	definitionID string,
	versionID string,
	entryElement string,
) (automation.PreparedScheduledFlowExecution, error) {
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(encoded)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, permanentMetadataBindingError(err)
	}
	if binding.RuntimeCapabilityID != flowsemantic.CapabilityName {
		return automation.PreparedScheduledFlowExecution{}, permanentMetadataBindingError(
			errors.New("runtime scheduled Flow: immutable view has the wrong Runtime capability"),
		)
	}
	view, err := b.openProfile(ctx, binding, catalogsnapshot.ProjectionRecordMutationV1)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, err
	}
	mutationView, err := b.openSiblingProfile(
		ctx,
		binding,
		catalogsnapshot.ProjectionRecordMutationV1,
		mutationsemantic.CapabilityName,
	)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, err
	}
	mutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(mutationView)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, permanentMetadataBindingError(err)
	}
	mutationScope := persistence.RuntimeWorkScope{
		OrgID:               mutationBinding.OrgID,
		AppInstallationID:   mutationBinding.AppInstallationID,
		RuntimeCapabilityID: mutationBinding.RuntimeCapabilityID,
	}
	if err := mutationScope.Validate(); err != nil {
		return automation.PreparedScheduledFlowExecution{}, permanentMetadataBindingError(err)
	}
	assignmentPolicyEligibility, err := b.prepareAssignmentPolicyEligibilityBinding(ctx, mutationView)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, err
	}
	plan, err := automation.PrepareScheduledFlowExecutionPlanForEntry(
		ctx,
		view,
		definitionID,
		versionID,
		entryElement,
	)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, permanentMetadataBindingError(err)
	}
	prepared, err := automation.NewPreparedScheduledFlowExecution(
		plan,
		func(executionCtx context.Context) (context.Context, error) {
			return installScheduledFlowExecution(
				executionCtx,
				plan,
				mutationView,
				view,
				mutationScope,
				assignmentPolicyEligibility,
			)
		},
	)
	if err != nil {
		return automation.PreparedScheduledFlowExecution{}, permanentMetadataBindingError(err)
	}
	return prepared, nil
}

func installScheduledFlowExecution(
	ctx context.Context,
	plan automation.ScheduledFlowExecutionPlan,
	mutationView *catalogsnapshot.RuntimeMetadataView,
	flowView *catalogsnapshot.RuntimeMetadataView,
	mutationScope persistence.RuntimeWorkScope,
	assignmentPolicyEligibility managedruntime.AssignmentPolicyEligibilityAmbientStager,
) (context.Context, error) {
	bound, err := installExecution(
		ctx,
		mutationView,
		flowView,
		mutationScope,
		assignmentPolicyEligibility,
	)
	if err != nil || !plan.AsyncAfterCommit {
		return bound, err
	}
	if plan.Root == nil || strings.TrimSpace(plan.Root.Id) == "" {
		return nil, errors.New("runtime scheduled Flow: AsyncAfterCommit plan root is missing")
	}
	return automation.WithAfterCommitScheduledFlowExecution(bound, plan.Root.Id), nil
}

type scheduledFlowViewBinderAdapter struct {
	binder *mutationArtifactBinder
}

func (b scheduledFlowViewBinderAdapter) PrepareScheduledFlow(
	ctx context.Context,
	encoded string,
	definitionID string,
	versionID string,
	entryElement string,
) (automation.PreparedScheduledFlowExecution, error) {
	if b.binder == nil {
		return automation.PreparedScheduledFlowExecution{}, retryableMetadataBindingError(
			errors.New("runtime scheduled Flow: metadata binder is unavailable"),
		)
	}
	return b.binder.bindScheduledFlow(ctx, encoded, definitionID, versionID, entryElement)
}

func (scheduledFlowViewBinderAdapter) IsPermanentScheduledFlowAuthorityFailure(err error) bool {
	return isPermanentMetadataBindingError(err)
}

var _ automation.ScheduledFlowExecutionAuthority = scheduledFlowViewBinderAdapter{}
var _ automation.ScheduledFlowAuthorityFailureClassifier = scheduledFlowViewBinderAdapter{}

func (b *mutationArtifactBinder) openCommand(
	ctx context.Context,
	encoded string,
) (*catalogsnapshot.RuntimeMetadataView, persistence.RuntimeWorkScope, error) {
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(encoded)
	if err != nil {
		return nil, persistence.RuntimeWorkScope{}, permanentMetadataBindingError(err)
	}
	view, err := b.openProfile(ctx, binding, catalogsnapshot.ProjectionCommandAcceptanceV1)
	if err != nil {
		return nil, persistence.RuntimeWorkScope{}, err
	}
	scope, err := runtimeWorkScopeFromBinding(binding)
	if err != nil {
		return nil, persistence.RuntimeWorkScope{}, permanentMetadataBindingError(err)
	}
	return view, scope, nil
}

// prepareExecution reopens one durable row's immutable inputs before the
// effect transaction. The routing scope and lifecycle host are derived from
// the same verified mutation view, not independently reconstructed from worker
// configuration or current Catalog state.
func (b *mutationArtifactBinder) prepareExecution(
	ctx context.Context,
	binding catalogsnapshot.RuntimeWorkBinding,
) (
	*catalogsnapshot.RuntimeMetadataView,
	persistence.RuntimeWorkScope,
	managedruntime.AssignmentPolicyEligibilityAmbientStager,
	error,
) {
	view, err := b.open(ctx, binding)
	if err != nil {
		return nil, persistence.RuntimeWorkScope{}, nil, err
	}
	scope, err := runtimeWorkScopeFromBinding(binding)
	if err != nil {
		return nil, persistence.RuntimeWorkScope{}, nil, fmt.Errorf("runtime worker metadata binder: derive durable work scope: %w", err)
	}
	assignmentPolicyEligibility, err := b.prepareAssignmentPolicyEligibilityBinding(ctx, view)
	if err != nil {
		return nil, persistence.RuntimeWorkScope{}, nil, err
	}
	return view, scope, assignmentPolicyEligibility, nil
}

func runtimeWorkScopeFromBinding(binding catalogsnapshot.RuntimeWorkBinding) (persistence.RuntimeWorkScope, error) {
	scope := persistence.RuntimeWorkScope{
		OrgID:               binding.OrgID,
		AppInstallationID:   binding.AppInstallationID,
		RuntimeCapabilityID: binding.RuntimeCapabilityID,
	}
	if err := scope.Validate(); err != nil {
		return persistence.RuntimeWorkScope{}, fmt.Errorf("runtime worker metadata binder: derive durable work scope: %w", err)
	}
	return scope, nil
}

// installExecution restores the complete previously verified execution
// envelope inside the effect transaction. No worker path may install only the
// metadata view: records effects and their nested durable outputs need the
// routing projection from the same durable binding as well.
func installExecution(
	ctx context.Context,
	view, flowWorkView *catalogsnapshot.RuntimeMetadataView,
	scope persistence.RuntimeWorkScope,
	assignmentPolicyEligibility managedruntime.AssignmentPolicyEligibilityAmbientStager,
) (context.Context, error) {
	if view == nil {
		return nil, errors.New("runtime worker metadata binder: Runtime metadata view is required")
	}
	ctx = metadata.WithRuntimeMetadataView(ctx, view)
	if flowWorkView != nil {
		ctx = metadata.WithRuntimeFlowWorkView(ctx, flowWorkView)
	}
	ctx, err := persistence.WithRuntimeWorkScope(ctx, scope)
	if err != nil {
		return nil, fmt.Errorf("runtime worker metadata binder: install durable work scope: %w", err)
	}
	ctx = managedruntime.WithAssignmentPolicyEligibilityAmbientStager(
		ctx,
		assignmentPolicyEligibility,
	)
	return ctx, nil
}

func (b *mutationArtifactBinder) bindEncoded(ctx context.Context, encoded string) (eventfeature.DurableExecutionInstaller, error) {
	bindingSet, err := catalogsnapshot.OpenRuntimeWorkBindingSet(encoded)
	if err != nil {
		return nil, err
	}
	view, scope, assignmentPolicyEligibility, err := b.prepareExecution(ctx, bindingSet.Primary)
	if err != nil {
		return nil, err
	}
	var flowWorkView *catalogsnapshot.RuntimeMetadataView
	if bindingSet.FlowWork != nil {
		flowWorkView, err = b.openProfileForCapability(
			ctx,
			*bindingSet.FlowWork,
			catalogsnapshot.ProjectionRecordMutationV1,
			flowsemantic.CapabilityName,
		)
		if err != nil {
			return nil, err
		}
	}
	return func(executionCtx context.Context) (context.Context, error) {
		return installExecution(
			executionCtx,
			view,
			flowWorkView,
			scope,
			assignmentPolicyEligibility,
		)
	}, nil
}

func (b *mutationArtifactBinder) bindMutation(ctx context.Context, binding catalogsnapshot.RuntimeWorkBinding) (context.Context, error) {
	view, scope, assignmentPolicyEligibility, err := b.prepareExecution(ctx, binding)
	if err != nil {
		return nil, err
	}
	return installExecution(ctx, view, nil, scope, assignmentPolicyEligibility)
}

type commandExecutionAuthority struct {
	authorizationView           *catalogsnapshot.RuntimeMetadataView
	executionView               *catalogsnapshot.RuntimeMetadataView
	flowWorkView                *catalogsnapshot.RuntimeMetadataView
	scope                       persistence.RuntimeWorkScope
	assignmentPolicyEligibility managedruntime.AssignmentPolicyEligibilityAmbientStager
}

func (a commandExecutionAuthority) AllowsCommand(kind string, version uint32) bool {
	return a.authorizationView != nil &&
		a.authorizationView.AllowsCommand(kind, version)
}

func (a commandExecutionAuthority) CommandExecutionContext(
	kind string,
	version uint32,
	authorizedWorkloadID string,
	definitionRef string,
	definitionVersion string,
) (string, bool) {
	if a.authorizationView == nil {
		return "", false
	}
	return a.authorizationView.CommandExecutionContext(
		kind,
		version,
		authorizedWorkloadID,
		definitionRef,
		definitionVersion,
	)
}

func (a commandExecutionAuthority) Bind(ctx context.Context) (context.Context, error) {
	bound, err := installExecution(
		ctx,
		a.executionView,
		a.flowWorkView,
		a.scope,
		a.assignmentPolicyEligibility,
	)
	return bound, err
}

type commandViewBinderAdapter struct {
	binder *mutationArtifactBinder
}

func (b commandViewBinderAdapter) OpenCommand(
	ctx context.Context,
	encoded string,
	kind string,
	version uint32,
) (platformCommandExecutionAuthority, error) {
	if b.binder == nil {
		return nil, errors.New("runtime platform command: metadata binder is unavailable")
	}
	view, scope, err := b.binder.openCommand(ctx, encoded)
	if err != nil {
		return nil, err
	}
	authority := commandExecutionAuthority{
		authorizationView: view,
		executionView:     view,
		scope:             scope,
	}
	if kind != RuntimeFlowCommandKind ||
		(version != RuntimeFlowCommandPayloadFormatVersion &&
			version != RuntimeFlowCommandPayloadFormatVersionV2 &&
			version != RuntimeFlowCommandPayloadFormatVersionV3) {
		return authority, nil
	}
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(encoded)
	if err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	flowView, err := b.binder.openSiblingProfileFromSource(
		ctx,
		binding,
		catalogsnapshot.ProjectionRecordMutationV1,
		flowsemantic.CapabilityName,
	)
	if err != nil {
		return nil, err
	}
	mutationView, err := b.binder.openSiblingProfileFromSource(
		ctx,
		binding,
		catalogsnapshot.ProjectionRecordMutationV1,
		mutationsemantic.CapabilityName,
	)
	if err != nil {
		return nil, err
	}
	mutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(
		mutationView,
	)
	if err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	mutationScope, err := runtimeWorkScopeFromBinding(mutationBinding)
	if err != nil {
		return nil, permanentMetadataBindingError(err)
	}
	assignmentPolicyEligibility, err := b.binder.prepareAssignmentPolicyEligibilityBinding(ctx, mutationView)
	if err != nil {
		return nil, err
	}
	authority.executionView = mutationView
	authority.flowWorkView = flowView
	authority.scope = mutationScope
	authority.assignmentPolicyEligibility = assignmentPolicyEligibility
	return authority, nil
}

// runtimeConnectorInboundFinalizationAuthorityResolver is an adapter over the
// immutable RuntimeMetadataView already bound to this command attempt. It owns
// no cache, repository lookup, credential value or provider effect.
type runtimeConnectorInboundFinalizationAuthorityResolver struct{}

func (runtimeConnectorInboundFinalizationAuthorityResolver) ResolveConnectorInboundRemoteIdentityFinalizationAuthority(
	ctx context.Context,
	query automation.ConnectorPushContractQuery,
) (*automation.ConnectorInboundRemoteIdentityFinalizationAuthority, error) {
	view, ok := metadata.RuntimeMetadataViewFromContext(ctx)
	if !ok || view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return nil, errors.New(
			"runtime connector inbound finalization: record-mutation view is not bound",
		)
	}
	values := view.ConnectorInboundFinalizationAuthorities(query.ConnectorOwnerKey)
	var selected *catalogsnapshot.ConnectorInboundFinalizationAuthority
	for i := range values {
		value := values[i]
		if value.Contract.FlowDeveloperName != query.FlowDeveloperName ||
			value.Contract.FlowActionName != query.FlowActionName {
			continue
		}
		if selected != nil {
			return nil, fmt.Errorf(
				"runtime connector inbound finalization: selector %s/%s/%s is ambiguous",
				query.ConnectorOwnerKey,
				query.FlowDeveloperName,
				query.FlowActionName,
			)
		}
		selected = &value
	}
	if selected == nil {
		return nil, nil
	}
	profile := selected.Profile
	return &automation.ConnectorInboundRemoteIdentityFinalizationAuthority{
		Contract: selected.Contract,
		Profile: automation.ConnectorPushProfile{
			OwnerPack:            profile.OwnerPack,
			ProfileName:          profile.ProfileName,
			CreatedPackInstallID: profile.Generation,
			CredentialRef:        profile.CredentialRef,
			EndpointClass:        profile.EndpointClass,
			RetryPolicy: automation.ConnectorPushRetryPolicy{
				MaxAttempts:          profile.MaxAttempts,
				BaseDelay:            time.Duration(profile.BaseDelayNanos),
				Multiplier:           profile.Multiplier,
				MaxDelay:             time.Duration(profile.MaxDelayNanos),
				RetryableStatusCodes: append([]int(nil), profile.RetryableStatusCodes...),
			},
		},
		OutboundMappings: append(
			[]contractconnector.ConnectorFieldMappingInput(nil),
			selected.OutboundMappings...,
		),
	}, nil
}

// runtimeConnectorInboundRecoveryInstalledResolver projects the exact
// immutable finalization closures already bound to this command into the
// generic recovery engine's installed-binding port. It performs no repository
// lookup and owns no cache.
type runtimeConnectorInboundRecoveryInstalledResolver struct{}

func (runtimeConnectorInboundRecoveryInstalledResolver) ResolveConnectorInboundRemoteIdentityInstalledContracts(
	ctx context.Context,
	connectorOwnerKey string,
) (automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return runtimeConnectorInboundInstalledContractSet(ctx, connectorOwnerKey)
}

func (runtimeConnectorInboundRecoveryInstalledResolver) ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(
	ctx context.Context,
	connectorOwnerKey string,
) (automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return runtimeConnectorInboundInstalledContractSet(ctx, connectorOwnerKey)
}

func (runtimeConnectorInboundRecoveryInstalledResolver) ResolveConnectorInboundRemoteIdentityInstalledContractSetsInAmbientTx(
	ctx context.Context,
	connectorOwnerKeys []string,
) (map[string]automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	sets := make(map[string]automation.ConnectorInboundRemoteIdentityInstalledContractSet, len(connectorOwnerKeys))
	for _, connectorOwnerKey := range connectorOwnerKeys {
		connectorOwnerKey = strings.TrimSpace(connectorOwnerKey)
		if connectorOwnerKey == "" {
			return nil, errors.New("runtime connector inbound installed contracts: connector owner key is blank")
		}
		if _, exists := sets[connectorOwnerKey]; exists {
			continue
		}
		set, err := runtimeConnectorInboundInstalledContractSet(ctx, connectorOwnerKey)
		if err != nil {
			return nil, err
		}
		sets[connectorOwnerKey] = set
	}
	return sets, nil
}

// runtimeConnectorInboundInstalledContractSet projects the complete immutable
// connector authority already bound to the worker attempt. The projection is
// deliberately used for both current and ambient-transaction resolver ports:
// this worker never consults a repository/current installation while moving a
// remote identity.
func runtimeConnectorInboundInstalledContractSet(
	ctx context.Context,
	connectorOwnerKey string,
) (automation.ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	view, ok := metadata.RuntimeMetadataViewFromContext(ctx)
	if !ok || view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, errors.New(
			"runtime connector inbound installed contracts: record-mutation view is not bound",
		)
	}
	connectorOwnerKey = strings.TrimSpace(connectorOwnerKey)
	if connectorOwnerKey == "" {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, errors.New(
			"runtime connector inbound installed contracts: connector owner key is blank",
		)
	}
	authorities := view.ConnectorInboundFinalizationAuthorities(connectorOwnerKey)
	if len(authorities) == 0 {
		return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf(
			"runtime connector inbound installed contracts: no immutable authority for %s", connectorOwnerKey,
		)
	}
	set := automation.ConnectorInboundRemoteIdentityInstalledContractSet{
		Generation:           string(view.Digest()),
		Contracts:            make([]contractconnector.ConnectorPushDurableMetadataContract, 0, len(authorities)),
		LockedPackInstallIDs: make(map[string]string),
	}
	for _, authority := range authorities {
		contract := authority.Contract
		if strings.TrimSpace(contract.ConnectorOwnerKey) != connectorOwnerKey {
			return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, errors.New(
				"runtime connector inbound installed contracts: authority owner does not match selection",
			)
		}
		profileOwnerPack := strings.TrimSpace(contract.ProfileOwnerPack)
		if profileOwnerPack == "" {
			profileOwnerPack = connectorOwnerKey
		}
		if strings.TrimSpace(authority.Profile.OwnerPack) != profileOwnerPack ||
			strings.TrimSpace(authority.Profile.Generation) == "" {
			return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, errors.New(
				"runtime connector inbound installed contracts: authority profile is incomplete or mismatched",
			)
		}
		if current, exists := set.LockedPackInstallIDs[profileOwnerPack]; exists && current != authority.Profile.Generation {
			return automation.ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf(
				"runtime connector inbound installed contracts: profile owner %q has mixed installed generations", profileOwnerPack,
			)
		}
		set.LockedPackInstallIDs[profileOwnerPack] = authority.Profile.Generation
		set.Contracts = append(set.Contracts, contract)
		set.Mappings = append(set.Mappings, authority.OutboundMappings...)
	}
	return set, nil
}

func (runtimeConnectorInboundRecoveryInstalledResolver) ResolveConnectorInboundRemoteIdentityRecoveryInstalledBindings(
	ctx context.Context,
	selector automation.ConnectorInboundRemoteIdentityRecoverySelector,
) (automation.ConnectorInboundRemoteIdentityRecoveryInstalledBindings, error) {
	view, ok := metadata.RuntimeMetadataViewFromContext(ctx)
	if !ok || view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return automation.ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, errors.New(
			"runtime connector inbound recovery: record-mutation view is not bound",
		)
	}
	authorities := view.ConnectorInboundFinalizationAuthorities(selector.ConnectorOwnerKey)
	bindings := make([]automation.ConnectorInboundRemoteIdentityRecoveryBinding, 0)
	for _, authority := range authorities {
		contract := authority.Contract
		if contract.DestinationSlotID != selector.DestinationSlotID {
			continue
		}
		// RecoverySourceEligibility is mandatory only for the headless Runtime
		// command profile. Keep that policy at this adapter boundary so the
		// pre-existing product-validated recovery owner retains its published
		// authority set.
		if authority.RecoverySourceEligibility == nil ||
			len(authority.RecoverySourceEligibility.RequiredFalseBooleanFields) == 0 {
			continue
		}
		profileOwner := contract.ProfileOwnerPack
		if profileOwner == "" {
			profileOwner = contract.ConnectorOwnerKey
		}
		snapshot := automation.ConnectorPushDurableSnapshot{
			RequesterOwnerPack: contract.ConnectorOwnerKey,
			ProfileOwnerPack:   profileOwner,
			// ConnectorNamespace is the established durable snapshot lookup key
			// for the immutable connector contract, despite its legacy name. The
			// authenticated inbound route namespace remains on the recovery
			// selector and evidence; persisting it here would make the downstream
			// connector worker reopen the contract under the wrong owner.
			ConnectorNamespace:   contract.ConnectorOwnerKey,
			FlowDeveloperName:    contract.FlowDeveloperName,
			FlowActionName:       contract.FlowActionName,
			TargetObject:         contract.TargetObject,
			ProfileName:          contract.ProfileName,
			CreatedPackInstallID: authority.Profile.Generation,
			CredentialRef:        authority.Profile.CredentialRef,
			EndpointClass:        authority.Profile.EndpointClass,
			RetryPolicy: automation.ConnectorPushRetryPolicy{
				MaxAttempts:          authority.Profile.MaxAttempts,
				BaseDelay:            time.Duration(authority.Profile.BaseDelayNanos),
				Multiplier:           authority.Profile.Multiplier,
				MaxDelay:             time.Duration(authority.Profile.MaxDelayNanos),
				RetryableStatusCodes: append([]int(nil), authority.Profile.RetryableStatusCodes...),
			},
			CreatePathTemplate:      contract.CreatePathTemplate,
			UpdatePathTemplate:      contract.UpdatePathTemplate,
			ResponseIDPath:          contract.ResponseIDPath,
			RemoteIDField:           contract.RemoteIDField,
			RemoteIDProvenanceField: contract.RemoteIDProvenanceField,
			MappingFingerprint:      automation.ConnectorPushFieldMappingFingerprint(authority.OutboundMappings),
			BodyFingerprint:         automation.ConnectorPushContractBodyFingerprint(&contract, authority.OutboundMappings),
			CreateConflictResolutionFingerprint: automation.ConnectorPushCreateConflictResolutionFingerprint(
				contract.CreateConflictResolution,
			),
		}
		snapshot.ProfileContractHash = automation.ConnectorPushProfileContractHash(snapshot)
		eligibility := append(
			[]string(nil),
			authority.RecoverySourceEligibility.RequiredFalseBooleanFields...,
		)
		bindings = append(bindings, automation.ConnectorInboundRemoteIdentityRecoveryBinding{
			EvidenceBinding: automation.ConnectorInboundRemoteIDEvidenceBinding{
				SourceObject: contract.SourceObject, RemoteIDField: contract.RemoteIDField,
			},
			Identity: automation.ConnectorPushDurableIdentity{
				ConnectorOwnerKey: contract.ConnectorOwnerKey,
				ActionSlotID:      contract.ActionSlotID, ActionType: contract.ActionType,
				SourceObject: contract.SourceObject, DestinationSlotID: contract.DestinationSlotID,
			},
			Snapshot: snapshot, RecoverySourceEligibility: eligibility,
		})
	}
	if len(bindings) == 0 {
		return automation.ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, fmt.Errorf(
			"runtime connector inbound recovery: no immutable authority for %s/%s",
			selector.ConnectorOwnerKey,
			selector.DestinationSlotID,
		)
	}
	return automation.ConnectorInboundRemoteIdentityRecoveryInstalledBindings{
		Generation: string(view.Digest()), Bindings: bindings,
	}, nil
}