package connectors

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math"
	"os"
	"runtime/debug"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/syncstate"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
)

const defaultConnectorInboundWorkerDrainTimeout = 30 * time.Second
const defaultConnectorInboundBacklogReconcileInterval = 5 * time.Second

const (
	defaultConnectorInboundWorkersPerPack = 8
	safeConnectorInboundWorkersPerPack    = 8
	maxConnectorInboundWorkersPerPack     = 128
	maxConnectorInboundMaxProcessingRows  = 512
	envConnectorInboundWorkersPerPack     = "NEXUSCRM_CONNECTOR_INBOUND_WORKERS_PER_PACK"
	envConnectorInboundMaxProcessingRows  = "NEXUSCRM_CONNECTOR_INBOUND_MAX_PROCESSING_ROWS"
	envConnectorInboundQueueProfile       = "NEXUSCRM_CONNECTOR_INBOUND_QUEUE_PROFILE"
	envConnectorInboundStageTiming        = "NEXUSCRM_CONNECTOR_INBOUND_STAGE_TIMING"
	// envConnectorFlowDeadlineMS tunes the per-event deadline (milliseconds) that
	// bounds each post-claim EXTERNAL call — flow execution AND each heartbeated
	// post-claim BatchRead. Unset / invalid => defaultConnectorFlowDeadline (a
	// protective bound above normal connector runtimes); set 0 (or negative) to
	// DISABLE; set > 0 for a custom bound. The deadline bounds a HUNG call that
	// the lease heartbeat would
	// otherwise keep alive forever, pinning its lane (a heartbeated row is no longer
	// recoverable by stale-reclaim). It is independent of StaleProcessingSeconds:
	// the heartbeat (not the deadline) prevents stale reclaim of an alive call, so
	// the deadline need not sit below 300 s and can be generous.
	envConnectorFlowDeadlineMS = "NEXUSCRM_CONNECTOR_FLOW_DEADLINE_MS"
)

const (
	connectorInboundQueueProfileReserved     = "reserved"
	connectorInboundQueueProfileDefaultHeavy = "default-heavy"
	connectorInboundWorkItemFirstPeekGrace   = 2 * time.Minute
)

const (
	connectorInboundWorkItemClaimSchemaColumns   = `WorkId, EventId, InstalledPackId, WorkType, QueueClass, State, RunAfter, Attempt, OwnershipEpoch, FailureCount, FailureClass, FailureSummary, LastFailureAt, ReplayedAt, ReplayOfWorkId, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey, TargetMutationLaneKey, Generation, LaneOrdinal, ProjectionKey, ProjectionActiveKey, DependencyKind, DependencyObjectAPIName, DependencyRecordId, DependencyKeyHash, DependencyConnectorOwnerKey, DependencyConnectorNamespace, DependencyConnectorSourceObject, DependencyDestinationSlotId, DependencyConnectorRemoteId, DependencyBlockedAt, DependencyProbeAfter, TerminalAt, CreatedDate, LastModifiedDate`
	connectorInboundWorkItemAttemptSchemaColumns = `WorkId, Attempt, EventId, FlowExecutionId, OutcomeState, PresentedErrorCode, CauseCode, CauseCategory, CauseElementType, CauseElementName, Retriable, CreatedDate, LastModifiedDate`
	connectorInboundEventOwnerSchemaColumns      = `ExecutionOwner`
)

// Self-healing restart backoff for a worker goroutine that exits unexpectedly
// (an escaped panic, or — defensively — a non-cancel return). A worker stopped
// by Stop()/drain is never restarted; only a still-live unexpected exit is.
const (
	connectorInboundRestartBackoffMin = 1 * time.Second
	connectorInboundRestartBackoffMax = 30 * time.Second
	// A run that outlived the longest backoff window clearly was not in a tight
	// crash-loop, so its next restart resets to the minimum backoff.
	connectorInboundRestartHealthyRunThreshold = connectorInboundRestartBackoffMax
)

// connectorInboundWorkerRestarts counts in-place worker restarts performed by
// the supervisor backstop. It is a lightweight in-package signal only; metrics
// wiring is a separate concern (CIN-OPS-2).
var connectorInboundWorkerRestarts atomic.Int64

var (
	errConnectorInboundWorkItemSchemaGap     = errors.New("connector inbound supervisor: mandatory WorkItem claim schema is unavailable")
	errConnectorInboundWorkItemCapabilityGap = errors.New("connector inbound supervisor: complete WorkItem execution capability is unavailable")
	errConnectorInboundWorkItemCutoverGap    = errors.New("connector inbound supervisor: WorkItem-only execution cutover is not ready")
	errConnectorInboundIdentityProbePanic    = errors.New("connector inbound supervisor: identity convergence resolver panicked")
)

type connectorInboundProfileLoader func(context.Context, *sql.DB, string, string, bool) ([]*connector_callout.ConnectorCalloutProfile, error)
type connectorInboundBatchClientFactory func(string, *connector_callout.Engine, []*connector_callout.ConnectorCalloutProfile) (connector_inbound.BatchReadClient, error)
type connectorInboundWorkerFactory func(context.Context, connector_inbound.WorkerConfig, connector_inbound.WorkItemService, connector_inbound.SyncStateService, connector_inbound.BatchReadClient, connector_inbound.FlowInvoker, connector_inbound_routing.RouteResolver) (*connector_inbound.Worker, error)
type connectorInboundWorkerRunner func(workerCtx, drainCtx context.Context, claimFirst bool, w *connector_inbound.Worker) error
type connectorInboundPackLister func(context.Context, *sql.DB) ([]packstate.ConnectorInboundWorkerPackRow, error)
type connectorInboundBatchAuthTypeResolver func(context.Context, packstate.ConnectorInboundWorkerPackRow, string) (string, error)
type connectorInboundDrainOnlyEligibilityResolver func(context.Context, string, string) (bool, error)
type connectorInboundProjectionFingerprintResolver func(context.Context, *sql.DB, string, string, bool) (string, error)

// RuntimeServingPack is one immutable connector-pack generation selected by
// the independently deployed Integration host. The supervisor resolves only
// these exact identities; it never enumerates the global installed-pack or
// callout-profile inventory for the Runtime handoff role.
type RuntimeServingPack struct {
	Namespace       string
	InstalledPackID string
	PayloadDigest   string
	// InboundRoutes is the complete Catalog-selected route inventory for this
	// exact pack generation. A partial inventory is not serving authority.
	InboundRoutes   []RuntimeServingInboundRoute
	CalloutProfiles []RuntimeServingCalloutProfile
}

// RuntimeServingInboundRoute is one exact Catalog-selected route coordinate.
// InvocationMode remains owned by the retained route definition and is read
// through RuntimeServingSelection.RouteResolver during startup admission.
type RuntimeServingInboundRoute struct {
	EventKind    string
	RouteVersion uint64
}

// RuntimeServingCalloutProfile identifies one exact profile and the credential
// identities authorized by that profile's Catalog binding. Credential values
// remain owned by the Integration service and never enter this selection.
type RuntimeServingCalloutProfile struct {
	Name           string
	CredentialRefs []string
}

// RuntimeServingSelection is frozen before process start. Stopped upgrades
// replace the process with a new selection; this is deliberately not a live
// metadata watcher or callback surface.
type RuntimeServingSelection struct {
	Packs         []RuntimeServingPack
	RouteResolver connector_inbound_routing.RouteResolver
}

type connectorInboundWorkItemGenerationResolver interface {
	ResolveClaimableWorkItemGeneration(context.Context, connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error)
}

type workItemProcessingPressureProvider struct {
	db  *sql.DB
	now func() time.Time
}

func (p workItemProcessingPressureProvider) ProcessingPressure(ctx context.Context, installedPackID string, queueClasses []string) (connector_events.ClaimPressureSnapshot, error) {
	now := time.Now
	if p.now != nil {
		now = p.now
	}
	current := now().UTC()
	snapshot := connector_events.ClaimPressureSnapshot{
		InstalledPackID: installedPackID,
		QueueClasses:    append([]string(nil), queueClasses...),
	}
	if p.db == nil {
		return snapshot, nil
	}
	queueClassSQL, queueClassArgs := workItemPressureQueueClassFilter("w", queueClasses)
	args := []any{
		current,
		installedPackID,
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}
	args = append(args, queueClassArgs...)
	err := p.db.QueryRowContext(ctx, `
SELECT
  COALESCE(SUM(CASE WHEN NOT (w.DependencyKind IS NOT NULL AND w.LeaseOwner IS NULL AND w.LeaseToken IS NULL AND w.LeaseUntil IS NULL) THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN (w.LeaseUntil IS NULL OR w.LeaseUntil <= ?)
                         AND NOT (w.DependencyKind IS NOT NULL AND w.LeaseOwner IS NULL AND w.LeaseToken IS NULL AND w.LeaseUntil IS NULL)
                    THEN 1 ELSE 0 END), 0)
  FROM _System_ConnectorWorkItem w
 WHERE w.InstalledPackId = ?
   AND w.WorkType IN (?, ?)
   AND w.State = 'processing'`+queueClassSQL, args...).Scan(&snapshot.ProcessingCount, &snapshot.StaleProcessingCount)
	if err != nil {
		return connector_events.ClaimPressureSnapshot{}, err
	}
	return snapshot, nil
}

func workItemPressureQueueClassFilter(alias string, queueClasses []string) (string, []any) {
	seen := map[string]struct{}{}
	args := make([]any, 0, len(queueClasses))
	for _, rawClass := range queueClasses {
		queueClass, err := connector_inbound_routing.NormalizeQueueClass(rawClass)
		if err != nil {
			continue
		}
		if _, ok := seen[queueClass]; ok {
			continue
		}
		seen[queueClass] = struct{}{}
		args = append(args, queueClass)
	}
	switch len(args) {
	case 0:
		return "", nil
	case 1:
		return fmt.Sprintf(" AND %s.QueueClass = ?", alias), args
	default:
		return fmt.Sprintf(" AND %s.QueueClass IN (%s)", alias, strings.TrimRight(strings.Repeat("?,", len(args)), ",")), args
	}
}

type connectorInboundWorkItemClaimActivation struct {
	service    connector_inbound.WorkItemService
	generation int64
}

// IdentityConvergenceRemoteSignal is the producer-owned finalized authority
// tuple. The waiting Flow's local candidate and Event source object are
// intentionally absent and are revalidated after wake.
type IdentityConvergenceRemoteSignal = connector_workqueue.IdentityConvergenceRemoteSignal

// Supervisor owns one handle per eligible connector-pack incarnation. Each
// handle may run multiple connector_inbound.Workers so local throughput can be
// raised above the historical single-worker default without changing pack
// lifecycle or queue-lease ownership semantics.
// It can be reconciled after live pack apply/remove operations so worker state
// follows the current installed-pack metadata, not just boot-time snapshots.
type Supervisor struct {
	db        *sql.DB
	engine    *connector_callout.Engine
	events    connector_inbound.EventsService
	syncstate connector_inbound.SyncStateService
	resolver  connector_inbound_routing.RouteResolver
	flow      connector_inbound.FlowInvoker
	// requireWorkItemOnlyCutover is host-owned admission, not part of the
	// published WorkItemService interface. Every production supervisor installs
	// the controlplane receipt verifier in NewSupervisor.
	requireWorkItemOnlyCutover func(context.Context) error

	hostname string
	pid      int

	listPacks                connectorInboundPackLister
	loadProfiles             connectorInboundProfileLoader
	resolveBatchContract     connectorInboundBatchContractResolver
	resolveBatchAuthType     connectorInboundBatchAuthTypeResolver
	resolveDrainEligible     connectorInboundDrainOnlyEligibilityResolver
	resolveProjectionFP      connectorInboundProjectionFingerprintResolver
	newBatchClient           connectorInboundBatchClientFactory
	newWorker                connectorInboundWorkerFactory
	runtimeServingConfigured bool
	runtimeServingRoutes     []runtimeServingRouteCoordinate
	identityConvergence      connector_inbound.IdentityConvergenceResolver
	disableIdentityRecovery  bool
	dependencyPageLimit      int

	lifecycleMu           sync.Mutex
	dependencyReconcileMu sync.Mutex
	dependencyProbeOnce   sync.Once
	dependencyProbeSlot   chan struct{}
	mu                    sync.Mutex
	starting              bool
	startupCancel         context.CancelFunc
	startupDone           chan struct{}
	started               bool
	lifecycleEpoch        uint64
	workers               map[string]connectorInboundWorkerHandle
	blocked               map[string]string
	wg                    sync.WaitGroup
	reconcileCancel       context.CancelFunc
	reconcileDone         <-chan struct{}

	workerDrainTimeout       time.Duration
	backlogReconcileInterval time.Duration
	dependencyStartupTimeout time.Duration
	dependencyProbeTimeout   time.Duration
	dependencyMaxWait        time.Duration
	reconcileStopTimeout     time.Duration
	// restartBackoffMin overrides the minimum self-heal restart backoff; zero
	// uses connectorInboundRestartBackoffMin. Tests set it small for speed.
	restartBackoffMin time.Duration
	// runWorker, when non-nil, replaces the direct worker Run dispatch. Production
	// leaves it nil (the real RunUntilDrained/RunClaimFirstUntilDrained run); tests
	// inject a stub to exercise the restart backstop independently of whether a
	// panic escapes the worker-frame recovers.
	runWorker connectorInboundWorkerRunner
}

type connectorInboundWorkerHandle struct {
	requestDrain   context.CancelFunc
	forceCancel    context.CancelFunc
	done           <-chan struct{}
	claimScheduler *connector_inbound.CentralClaimScheduler
	fingerprint    string
	workerCount    int
}

func NewSupervisor(db *sql.DB, engine *connector_callout.Engine, flow FlowExecutor, systemUsers SystemUserLoader) (*Supervisor, error) {
	if flow == nil {
		return nil, fmt.Errorf("connector inbound supervisor: flow executor is required")
	}
	if systemUsers == nil {
		return nil, fmt.Errorf("connector inbound supervisor: system user loader is required")
	}
	return newSupervisor(
		db,
		engine,
		newFlowAdapter(flow, systemUsers),
		connector_inbound.NewWorkItemWorker,
	)
}

// NewRuntimeSupervisor assembles the independent Integration execution role.
// It reuses the existing Supervisor lifecycle and WorkItem owner, but its only
// business-operation handoff is the neutral Runtime command boundary. No
// process-local Flow executor is installed as a fallback.
func NewRuntimeSupervisor(
	db *sql.DB,
	engine *connector_callout.Engine,
	runtimeCommands connector_inbound.RuntimeCommandHandoff,
) (*Supervisor, error) {
	if runtimeCommands == nil {
		return nil, fmt.Errorf(
			"connector inbound supervisor: Runtime command handoff is required",
		)
	}
	supervisor, err := newSupervisor(
		db,
		engine,
		nil,
		func(
			ctx context.Context,
			cfg connector_inbound.WorkerConfig,
			workItems connector_inbound.WorkItemService,
			syncstate connector_inbound.SyncStateService,
			batchRead connector_inbound.BatchReadClient,
			_ connector_inbound.FlowInvoker,
			resolver connector_inbound_routing.RouteResolver,
		) (*connector_inbound.Worker, error) {
			return connector_inbound.NewRuntimeCommandHandoffWorker(
				ctx,
				cfg,
				workItems,
				syncstate,
				batchRead,
				runtimeCommands,
				resolver,
			)
		},
	)
	if err != nil {
		return nil, err
	}
	// Runtime command dependencies are owned by the Runtime command row and
	// Runtime Worker. The independent Integration role must not also reconcile
	// the legacy connector WorkItem dependency columns.
	supervisor.disableIdentityRecovery = true
	return supervisor, nil
}

func newSupervisor(
	db *sql.DB,
	engine *connector_callout.Engine,
	flow connector_inbound.FlowInvoker,
	newWorker connectorInboundWorkerFactory,
) (*Supervisor, error) {
	if db == nil {
		return nil, fmt.Errorf("connector inbound supervisor: db is required")
	}
	if engine == nil {
		return nil, fmt.Errorf("connector inbound supervisor: engine is required")
	}
	if newWorker == nil {
		return nil, fmt.Errorf("connector inbound supervisor: worker factory is required")
	}
	hostname, err := os.Hostname()
	if err != nil || strings.TrimSpace(hostname) == "" {
		hostname = "local"
	}

	return &Supervisor{
		db:        db,
		engine:    engine,
		events:    connector_events.NewService(db, nil, nil),
		syncstate: connector_syncstate.NewService(connector_syncstate.NewRepository(db), connector_syncstate.NewControlplanePackResolver(db)),
		resolver:  connector_inbound_routing.NewDBRouteResolver(db),
		flow:      flow,
		requireWorkItemOnlyCutover: func(ctx context.Context) error {
			return connector_events.RequireWorkItemOnlyExecutionCutover(ctx, db)
		},
		hostname:             hostname,
		pid:                  os.Getpid(),
		listPacks:            packstate.ListConnectorPacksForInboundWorkers,
		loadProfiles:         loadProfilesForInboundWorkerStartup,
		resolveBatchContract: resolveBatchContractForInboundWorkerStartup,
		resolveProjectionFP:  resolveInboundMappingProjectionFingerprint,
		newBatchClient:       connector_inbound.NewCalloutBatchReadAdapter,
		newWorker:            newWorker,
		dependencyMaxWait:    connectorInboundIdentityConvergenceMaxWait,
		workers:              make(map[string]connectorInboundWorkerHandle),
		blocked:              make(map[string]string),
	}, nil
}

// SetRuntimeServingSelection freezes the independently deployed Integration
// supervisor to exact pack/profile identities and a revision-bound route
// resolver. It must run before StartContext.
func (s *Supervisor) SetRuntimeServingSelection(selection RuntimeServingSelection) error {
	if s == nil || selection.RouteResolver == nil {
		return errors.New("connector inbound supervisor: Runtime serving selection and route resolver are required")
	}
	packs, err := normalizeRuntimeServingPacks(selection.Packs)
	if err != nil {
		return err
	}
	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()
	s.mu.Lock()
	started := s.started || s.starting
	s.mu.Unlock()
	if started {
		return errors.New("connector inbound supervisor: Runtime serving selection must be installed before start")
	}
	if s.runtimeServingConfigured {
		return errors.New("connector inbound supervisor: Runtime serving selection is already installed")
	}
	s.listPacks = runtimeServingPackLister(packs)
	s.loadProfiles = runtimeServingProfileLoader(packs)
	s.resolver = selection.RouteResolver
	s.runtimeServingRoutes = runtimeServingRouteCoordinates(packs)
	s.runtimeServingConfigured = true
	return nil
}

type runtimeServingRouteCoordinate struct {
	namespace       string
	installedPackID string
	eventKind       string
	routeVersion    int64
}

func normalizeRuntimeServingPacks(values []RuntimeServingPack) ([]RuntimeServingPack, error) {
	result := make([]RuntimeServingPack, 0, len(values))
	seenPacks := make(map[string]struct{}, len(values))
	for _, raw := range values {
		value := RuntimeServingPack{
			Namespace:       strings.TrimSpace(raw.Namespace),
			InstalledPackID: strings.TrimSpace(raw.InstalledPackID),
			PayloadDigest:   strings.TrimSpace(raw.PayloadDigest),
		}
		if value.Namespace == "" || value.InstalledPackID == "" || value.PayloadDigest == "" {
			return nil, errors.New("connector inbound supervisor: Runtime serving pack identity is incomplete")
		}
		key := workerKey(value.Namespace, value.InstalledPackID)
		if _, duplicate := seenPacks[key]; duplicate {
			return nil, fmt.Errorf("connector inbound supervisor: duplicate Runtime serving pack %s (%s)", value.Namespace, value.InstalledPackID)
		}
		seenPacks[key] = struct{}{}
		seenRoutes := make(map[string]struct{}, len(raw.InboundRoutes))
		for _, rawRoute := range raw.InboundRoutes {
			route := RuntimeServingInboundRoute{
				EventKind:    strings.ToLower(strings.TrimSpace(rawRoute.EventKind)),
				RouteVersion: rawRoute.RouteVersion,
			}
			if route.EventKind == "" || route.RouteVersion == 0 || route.RouteVersion > math.MaxInt64 {
				return nil, fmt.Errorf(
					"connector inbound supervisor: Runtime serving pack %s has an invalid inbound route coordinate",
					value.Namespace,
				)
			}
			if _, duplicate := seenRoutes[route.EventKind]; duplicate {
				return nil, fmt.Errorf(
					"connector inbound supervisor: Runtime serving pack %s has duplicate inbound route %s",
					value.Namespace,
					route.EventKind,
				)
			}
			seenRoutes[route.EventKind] = struct{}{}
			value.InboundRoutes = append(value.InboundRoutes, route)
		}
		if len(value.InboundRoutes) == 0 {
			return nil, fmt.Errorf(
				"connector inbound supervisor: Runtime serving pack %s has no selected inbound routes",
				value.Namespace,
			)
		}
		sort.Slice(value.InboundRoutes, func(i, j int) bool {
			return value.InboundRoutes[i].EventKind < value.InboundRoutes[j].EventKind
		})
		seenProfiles := make(map[string]struct{}, len(raw.CalloutProfiles))
		for _, rawProfile := range raw.CalloutProfiles {
			profile := RuntimeServingCalloutProfile{Name: strings.ToLower(strings.TrimSpace(rawProfile.Name))}
			name := profile.Name
			if name == "" {
				return nil, fmt.Errorf("connector inbound supervisor: Runtime serving pack %s has a blank callout profile", value.Namespace)
			}
			if _, duplicate := seenProfiles[name]; duplicate {
				return nil, fmt.Errorf("connector inbound supervisor: Runtime serving pack %s has duplicate callout profile %s", value.Namespace, name)
			}
			seenProfiles[name] = struct{}{}
			seenCredentials := make(map[string]struct{}, len(rawProfile.CredentialRefs))
			for _, rawCredentialRef := range rawProfile.CredentialRefs {
				credentialRef := strings.TrimSpace(rawCredentialRef)
				if credentialRef == "" {
					return nil, fmt.Errorf("connector inbound supervisor: Runtime serving profile %s/%s has a blank credential reference", value.Namespace, name)
				}
				if _, duplicate := seenCredentials[credentialRef]; duplicate {
					return nil, fmt.Errorf("connector inbound supervisor: Runtime serving profile %s/%s has duplicate credential reference %s", value.Namespace, name, credentialRef)
				}
				seenCredentials[credentialRef] = struct{}{}
				profile.CredentialRefs = append(profile.CredentialRefs, credentialRef)
			}
			sort.Strings(profile.CredentialRefs)
			value.CalloutProfiles = append(value.CalloutProfiles, profile)
		}
		sort.Slice(value.CalloutProfiles, func(i, j int) bool {
			return value.CalloutProfiles[i].Name < value.CalloutProfiles[j].Name
		})
		result = append(result, value)
	}
	sort.Slice(result, func(i, j int) bool {
		return workerKey(result[i].Namespace, result[i].InstalledPackID) <
			workerKey(result[j].Namespace, result[j].InstalledPackID)
	})
	return result, nil
}

func runtimeServingRouteCoordinates(packs []RuntimeServingPack) []runtimeServingRouteCoordinate {
	var result []runtimeServingRouteCoordinate
	for _, pack := range packs {
		for _, route := range pack.InboundRoutes {
			result = append(result, runtimeServingRouteCoordinate{
				namespace:       pack.Namespace,
				installedPackID: pack.InstalledPackID,
				eventKind:       route.EventKind,
				routeVersion:    int64(route.RouteVersion),
			})
		}
	}
	return result
}

func (s *Supervisor) verifyRuntimeServingRoutes(ctx context.Context) error {
	if s == nil || s.resolver == nil || len(s.runtimeServingRoutes) == 0 {
		return errors.New("connector inbound supervisor: Runtime serving route authority is unavailable")
	}
	for _, route := range s.runtimeServingRoutes {
		spec, err := s.resolver.ResolveExecutionSpecByVersion(
			ctx,
			route.installedPackID,
			route.eventKind,
			route.routeVersion,
		)
		if err != nil {
			return fmt.Errorf(
				"connector inbound supervisor: verify Runtime serving route %s/%s (%s) version %d: %w",
				route.namespace,
				route.eventKind,
				route.installedPackID,
				route.routeVersion,
				err,
			)
		}
		spec, err = connector_inbound_routing.NormalizeRouteExecutionSpec(spec)
		if err != nil {
			return fmt.Errorf(
				"connector inbound supervisor: verify Runtime serving route %s/%s (%s) version %d: %w",
				route.namespace,
				route.eventKind,
				route.installedPackID,
				route.routeVersion,
				err,
			)
		}
		if !connector_inbound.RuntimeCommandHandoffSupportsInvocationMode(spec.InvocationMode) {
			return fmt.Errorf(
				"connector inbound supervisor: Runtime serving route %s/%s (%s) version %d selects unsupported invocation mode %q",
				route.namespace,
				route.eventKind,
				route.installedPackID,
				route.routeVersion,
				spec.InvocationMode,
			)
		}
	}
	return nil
}

func runtimeServingPackLister(packs []RuntimeServingPack) connectorInboundPackLister {
	return func(ctx context.Context, db *sql.DB) ([]packstate.ConnectorInboundWorkerPackRow, error) {
		rows := make([]packstate.ConnectorInboundWorkerPackRow, 0, len(packs))
		for _, selected := range packs {
			row, found, err := packstate.ResolveActiveConnectorPackByInstall(
				ctx,
				db,
				selected.Namespace,
				selected.InstalledPackID,
			)
			if err != nil {
				return nil, err
			}
			if !found {
				return nil, fmt.Errorf(
					"connector inbound supervisor: selected Runtime serving pack %s (%s) is not an active ready generation",
					selected.Namespace,
					selected.InstalledPackID,
				)
			}
			if row.PayloadHash != selected.PayloadDigest {
				return nil, fmt.Errorf(
					"connector inbound supervisor: selected Runtime serving pack %s (%s) payload digest does not match",
					selected.Namespace,
					selected.InstalledPackID,
				)
			}
			rows = append(rows, packstate.ConnectorInboundWorkerPackRow{ConnectorPackRow: *row})
		}
		return rows, nil
	}
}

func runtimeServingProfileLoader(packs []RuntimeServingPack) connectorInboundProfileLoader {
	byPack := make(map[string]RuntimeServingPack, len(packs))
	for _, pack := range packs {
		byPack[workerKey(pack.Namespace, pack.InstalledPackID)] = pack
	}
	return func(ctx context.Context, db *sql.DB, ownerPack, packInstallID string, drainOnly bool) ([]*connector_callout.ConnectorCalloutProfile, error) {
		if drainOnly {
			return nil, errors.New("connector inbound supervisor: Runtime serving selection cannot start a drain-only generation")
		}
		selected, found := byPack[workerKey(ownerPack, packInstallID)]
		if !found {
			return nil, errors.New("connector inbound supervisor: pack is outside Runtime serving selection")
		}
		profiles := make([]*connector_callout.ConnectorCalloutProfile, 0, len(selected.CalloutProfiles))
		for _, selectedProfile := range selected.CalloutProfiles {
			profile, err := connector_callout.LoadProfileForPackInstall(ctx, db, ownerPack, packInstallID, selectedProfile.Name)
			if err != nil {
				return nil, fmt.Errorf("load exact Runtime serving profile %s/%s: %w", ownerPack, selectedProfile.Name, err)
			}
			authorizedCredentials := make(map[string]struct{}, len(selectedProfile.CredentialRefs))
			for _, credentialRef := range selectedProfile.CredentialRefs {
				authorizedCredentials[credentialRef] = struct{}{}
			}
			for _, credentialRef := range runtimeServingProfileCredentialRefs(profile) {
				if _, authorized := authorizedCredentials[credentialRef]; !authorized {
					return nil, fmt.Errorf(
						"connector inbound supervisor: Runtime serving profile %s/%s credential is not authorized by its selection",
						ownerPack,
						selectedProfile.Name,
					)
				}
			}
			profiles = append(profiles, profile)
		}
		return profiles, nil
	}
}

func runtimeServingProfileCredentialRefs(profile *connector_callout.ConnectorCalloutProfile) []string {
	if profile == nil {
		return nil
	}
	result := make([]string, 0, 2)
	if credentialRef := strings.TrimSpace(profile.CredentialRef); credentialRef != "" {
		result = append(result, credentialRef)
	}
	if batchCredentialRef := strings.TrimSpace(profile.BatchSpec.CredentialRef); batchCredentialRef != "" &&
		(len(result) == 0 || batchCredentialRef != result[0]) {
		result = append(result, batchCredentialRef)
	}
	return result
}

func (s *Supervisor) SetBatchAuthTypeResolver(resolve connectorInboundBatchAuthTypeResolver) {
	s.resolveBatchAuthType = resolve
}

func (s *Supervisor) SetDrainOnlyEligibilityResolver(resolve connectorInboundDrainOnlyEligibilityResolver) {
	s.resolveDrainEligible = resolve
}

// SetIdentityConvergenceResolver installs the host-owned authority probe used
// after a durable dependency block and during restart/periodic recovery.
func (s *Supervisor) SetIdentityConvergenceResolver(resolve connector_inbound.IdentityConvergenceResolver) {
	s.identityConvergence = resolve
}

// WakeIdentityConvergenceRecord is the post-commit record-event bridge. The
// persisted dependency key makes duplicate create/update notifications safe.
func (s *Supervisor) WakeIdentityConvergenceRecord(ctx context.Context, objectAPIName, recordID string) (int64, error) {
	repository := connector_workqueue.NewRepository(s.db, nil, nil)
	if owner, ok := s.events.(interface {
		WorkItemRepository() *connector_workqueue.Repository
	}); ok && owner.WorkItemRepository() != nil {
		repository = owner.WorkItemRepository()
	}
	now := time.Now().UTC()
	woken, err := repository.WakeRecordDependency(ctx, objectAPIName, recordID, now)
	if err != nil {
		if scheduleErr := repository.ScheduleRecordIdentityConvergenceProbeRetry(ctx, objectAPIName, recordID, now); scheduleErr != nil {
			return woken, errors.Join(
				fmt.Errorf("connector inbound supervisor: wake local record dependency: %w", err),
				fmt.Errorf("connector inbound supervisor: persist local record wake retry: %w", scheduleErr),
			)
		}
		return woken, err
	}
	if woken == 0 {
		return woken, err
	}
	s.wakeAllLiveWorkItemClaimSchedulers()
	return woken, nil
}

// WakeIdentityConvergenceRemoteIdentity is the post-commit finalized-link
// bridge. It wakes by the producer-owned durable authority tuple only; every
// matching Flow must re-read its complete Event scope and durable authority,
// then fail closed on conflict or ambiguity.
func (s *Supervisor) WakeIdentityConvergenceRemoteIdentity(ctx context.Context, signal IdentityConvergenceRemoteSignal) (int64, error) {
	repository := connector_workqueue.NewRepository(s.db, nil, nil)
	if owner, ok := s.events.(interface {
		WorkItemRepository() *connector_workqueue.Repository
	}); ok && owner.WorkItemRepository() != nil {
		repository = owner.WorkItemRepository()
	}
	woken, err := newIdentityConvergenceRemoteWaker(repository).WakeIdentityConvergenceRemoteIdentity(ctx, signal)
	if err != nil {
		return woken, err
	}
	if woken == 0 {
		return woken, nil
	}
	s.wakeAllLiveWorkItemClaimSchedulers()
	return woken, nil
}

func (s *Supervisor) wakeAllLiveWorkItemClaimSchedulers() {
	s.mu.Lock()
	handles := make([]connectorInboundWorkerHandle, 0, len(s.workers))
	for _, handle := range s.workers {
		handles = append(handles, handle)
	}
	s.mu.Unlock()
	for _, handle := range handles {
		if !connectorInboundWorkerHandleDone(handle) && handle.claimScheduler != nil {
			handle.claimScheduler.WakeIfRunning()
		}
	}
}

func (s *Supervisor) Start() error {
	return s.StartContext(connectorInboundSupervisorContext())
}

// StartContext starts the independently deployed Integration worker families
// under the caller's startup context. The legacy Start entry point remains for
// in-process compositions, while serving processes use this form so a shutdown
// signal or startup deadline cancels database and authority reads before any
// listener is advertised ready. After successful startup, Stop remains the
// sole lifecycle owner; expiration of a startup deadline must not stop a
// published worker pool.
func (s *Supervisor) StartContext(ctx context.Context) error {
	if ctx == nil {
		return errors.New("connector inbound supervisor: startup context is required")
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	var startEpoch uint64
	for {
		s.lifecycleMu.Lock()
		s.mu.Lock()
		if s.started {
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return nil
		}
		if s.starting {
			startupDone := s.startupDone
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			select {
			case <-startupDone:
				if err := ctx.Err(); err != nil {
					return err
				}
				continue
			case <-ctx.Done():
				return ctx.Err()
			}
		}
		if s.disableIdentityRecovery && !s.runtimeServingConfigured {
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return errors.New("connector inbound supervisor: Runtime handoff requires a serving selection")
		}

		startupCtx, startupCancel := context.WithCancel(ctx)
		s.starting = true
		s.startupCancel = startupCancel
		s.startupDone = make(chan struct{})
		s.lifecycleEpoch++
		startEpoch = s.lifecycleEpoch
		s.mu.Unlock()
		s.lifecycleMu.Unlock()

		var routeErr error
		if s.runtimeServingConfigured {
			// Route resolution is host-owned code. Run it synchronously outside
			// lifecycleMu: Stop can cancel and invalidate this attempt without
			// waiting, while keeping the attempt single-flight until the resolver
			// actually returns. A resolver that ignores cancellation therefore
			// cannot strand the lifecycle lock or accumulate orphan goroutines.
			routeErr = s.verifyRuntimeServingRoutes(startupCtx)
		}
		startupCancel()

		s.lifecycleMu.Lock()
		s.mu.Lock()
		if !s.starting {
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return nil
		}
		startupStillCurrent := s.lifecycleEpoch == startEpoch
		s.starting = false
		s.startupCancel = nil
		startupDone := s.startupDone
		s.startupDone = nil
		close(startupDone)
		if !startupStillCurrent {
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return nil
		}
		if routeErr != nil {
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return routeErr
		}
		if err := ctx.Err(); err != nil {
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return err
		}
		s.started = true
		s.blocked = make(map[string]string)
		s.mu.Unlock()

		// The route probe above is deliberately lock-free, but worker publication
		// is one lifecycle operation. Keep route-result validation, initial
		// reconciliation, and the corresponding unlock in this explicit scope so
		// Stop cannot interleave between publishing started and installing workers.
		if err := s.reconcileStarted(ctx, true); err != nil {
			s.mu.Lock()
			s.started = false
			s.mu.Unlock()
			s.lifecycleMu.Unlock()
			return err
		}
		s.lifecycleMu.Unlock()
		break
	}

	// Authority resolvers are host-provided and may block. Do not execute them
	// while holding the lifecycle mutex: Stop must always be able to mark the
	// supervisor stopped even if a resolver ignores cancellation. The timeout
	// bounds the Start caller; a concurrent Stop makes the post-probe state check
	// prevent the backlog loop from being started afterward.
	var startupErr error
	if !s.disableIdentityRecovery {
		startupCtx, startupCancel := context.WithTimeout(ctx, s.effectiveDependencyStartupTimeout())
		startupErr = s.reconcileIdentityConvergenceDependenciesAtStartup(startupCtx)
		startupCancel()
	}

	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()
	s.mu.Lock()
	started := s.started
	currentEpoch := s.lifecycleEpoch
	s.mu.Unlock()
	if !started || currentEpoch != startEpoch {
		return nil
	}
	if err := ctx.Err(); err != nil {
		s.mu.Lock()
		s.started = false
		s.mu.Unlock()
		s.stopStartedWorkers(s.workerKeys())
		return err
	}
	if startupErr != nil {
		// The durable wait is already preserved and every resolver/wake failure
		// enrolled its one-shot retry. Startup reconciliation is recovery work,
		// not a readiness gate for unrelated inbound workers; the periodic loop
		// continues it from durable state. The WorkItem/worker capability gates
		// above remain fail-closed.
		log.Printf("WARN connector inbound supervisor: deferred startup identity convergence recovery: %v", startupErr)
	}
	s.startBacklogReconciler()
	return nil
}

// Reconcile refreshes the running worker set after live connector-pack
// lifecycle changes. Existing eligible workers are rebuilt deliberately: their
// batch profile, handler-contract, and credential bindings are derived from
// mutable live metadata and must not stay stale across setup apply/reapply.
func (s *Supervisor) Reconcile(ctx context.Context) error {
	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()

	if ctx == nil {
		ctx = connectorInboundSupervisorContext()
	}
	s.mu.Lock()
	if !s.started {
		s.mu.Unlock()
		return nil
	}
	s.blocked = make(map[string]string)
	s.mu.Unlock()
	return s.reconcileStarted(ctx, false)
}

func (s *Supervisor) ReconcileChanged(ctx context.Context) error {
	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()

	if ctx == nil {
		ctx = connectorInboundSupervisorContext()
	}
	s.mu.Lock()
	if !s.started {
		s.mu.Unlock()
		return nil
	}
	s.blocked = make(map[string]string)
	s.mu.Unlock()
	return s.reconcileChanged(ctx)
}

// PruneWorkerIncarnation stops only the exact connector-pack incarnation.
// It never lists desired workers, starts a replacement, or reconciles an
// unrelated namespace. Control-plane failure handling uses this after the
// failed incarnation has been durably removed from eligibility.
func (s *Supervisor) PruneWorkerIncarnation(namespace, installID string) error {
	if s == nil || strings.TrimSpace(namespace) == "" ||
		strings.TrimSpace(namespace) != namespace ||
		strings.TrimSpace(installID) == "" ||
		strings.TrimSpace(installID) != installID ||
		strings.Contains(namespace, "/") || strings.Contains(installID, "/") {
		return errors.New(
			"connector inbound supervisor: exact worker namespace and install identity are required",
		)
	}
	key := workerKey(namespace, installID)
	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()
	s.mu.Lock()
	delete(s.blocked, key)
	s.mu.Unlock()
	s.stopWorkers([]string{key}, false)
	return nil
}

func (s *Supervisor) reconcileStarted(ctx context.Context, startup bool) error {
	return s.reconcile(ctx, startup, true, false)
}

func (s *Supervisor) reconcileChanged(ctx context.Context) error {
	return s.reconcile(ctx, false, false, false)
}

func (s *Supervisor) reconcileBacklog(ctx context.Context) error {
	if ctx == nil {
		ctx = connectorInboundSupervisorContext()
	}
	if err := ctx.Err(); err != nil {
		return nil
	}
	s.mu.Lock()
	started := s.started
	s.mu.Unlock()
	if !started {
		return nil
	}
	// Legacy identity recovery is deliberately outside lifecycleMu. Stop can
	// cancel this reconcile context before waiting for the background loop.
	if !s.disableIdentityRecovery {
		if _, err := s.reconcileDueIdentityConvergenceDependencies(ctx); err != nil {
			return err
		}
	}
	if !s.lifecycleMu.TryLock() {
		return nil
	}
	defer s.lifecycleMu.Unlock()
	if err := ctx.Err(); err != nil {
		return nil
	}
	s.mu.Lock()
	if !s.started {
		s.mu.Unlock()
		return nil
	}
	s.mu.Unlock()
	return s.reconcile(ctx, false, false, true)
}

const connectorInboundIdentityConvergenceReconcileLimit = 1000
const connectorInboundIdentityConvergencePeriodicLimit = 25
const connectorInboundIdentityConvergenceStartupMaxPages = 100
const connectorInboundIdentityConvergenceStartupTimeout = 30 * time.Second
const connectorInboundIdentityConvergenceProbeTimeout = 5 * time.Second
const connectorInboundIdentityConvergenceMaxWait = 30 * time.Minute
const connectorInboundReconcileStopTimeout = 5 * time.Second

func (s *Supervisor) reconcileIdentityConvergenceDependenciesAtStartup(ctx context.Context) error {
	var cursor *connector_workqueue.IdentityConvergenceCursor
	for page := 0; page < connectorInboundIdentityConvergenceStartupMaxPages; page++ {
		_, next, err := s.reconcileIdentityConvergenceDependencyPage(ctx, cursor)
		if err != nil {
			return err
		}
		if next == nil {
			return nil
		}
		cursor = next
	}
	return fmt.Errorf("connector inbound supervisor: identity convergence startup recovery exceeded %d pages", connectorInboundIdentityConvergenceStartupMaxPages)
}

func (s *Supervisor) reconcileIdentityConvergenceDependencyPage(
	ctx context.Context,
	cursor *connector_workqueue.IdentityConvergenceCursor,
) (int64, *connector_workqueue.IdentityConvergenceCursor, error) {
	s.dependencyReconcileMu.Lock()
	defer s.dependencyReconcileMu.Unlock()
	resolver := s.identityConvergence
	repository := connector_workqueue.NewRepository(s.db, nil, nil)
	if owner, ok := s.events.(interface {
		WorkItemRepository() *connector_workqueue.Repository
	}); ok && owner.WorkItemRepository() != nil {
		repository = owner.WorkItemRepository()
	}
	limit := s.dependencyPageLimit
	if limit <= 0 {
		limit = connectorInboundIdentityConvergenceReconcileLimit
	}
	var total int64
	if cursor == nil {
		invalidWorkIDs, err := repository.ListInvalidIdentityConvergenceWaitIDs(ctx, limit)
		if err != nil {
			return 0, nil, fmt.Errorf("connector inbound supervisor: list invalid identity convergence dependencies: %w", err)
		}
		total, err = s.terminalizeInvalidIdentityConvergenceWaits(ctx, repository, invalidWorkIDs, time.Now().UTC())
		if err != nil {
			return total, nil, err
		}
	}
	waits, err := repository.ListIdentityConvergenceWaits(ctx, cursor, limit)
	if err != nil {
		return total, nil, fmt.Errorf("connector inbound supervisor: list identity convergence dependencies: %w", err)
	}
	if len(waits) == 0 {
		return total, nil, nil
	}
	now := time.Now().UTC()
	reconciled, err := s.reconcileIdentityConvergenceWaits(ctx, repository, resolver, waits, now)
	total += reconciled
	if err != nil {
		return total, nil, err
	}
	if len(waits) < limit {
		return total, nil, nil
	}
	last := waits[len(waits)-1]
	return total, &connector_workqueue.IdentityConvergenceCursor{
		DependencyBlockedAt: last.DependencyBlockedAt,
		WorkID:              last.WorkID,
	}, nil
}

func (s *Supervisor) reconcileDueIdentityConvergenceDependencies(ctx context.Context) (int64, error) {
	s.dependencyReconcileMu.Lock()
	defer s.dependencyReconcileMu.Unlock()
	resolver := s.identityConvergence
	repository := connector_workqueue.NewRepository(s.db, nil, nil)
	if owner, ok := s.events.(interface {
		WorkItemRepository() *connector_workqueue.Repository
	}); ok && owner.WorkItemRepository() != nil {
		repository = owner.WorkItemRepository()
	}
	now := time.Now().UTC()
	var waits []connector_workqueue.IdentityConvergenceWait
	var err error
	if maxWait := s.effectiveDependencyMaxWait(); maxWait > 0 {
		waits, err = repository.ListDueOrExpiredIdentityConvergenceWaits(ctx, now, now.Add(-maxWait), connectorInboundIdentityConvergencePeriodicLimit)
	} else {
		waits, err = repository.ListDueIdentityConvergenceWaits(ctx, now, connectorInboundIdentityConvergencePeriodicLimit)
	}
	if err != nil {
		return 0, fmt.Errorf("connector inbound supervisor: list due identity convergence dependencies: %w", err)
	}
	reconciled, err := s.reconcileIdentityConvergenceWaits(ctx, repository, resolver, waits, now)
	return reconciled, err
}

func (s *Supervisor) reconcileIdentityConvergenceWaits(
	ctx context.Context,
	repository *connector_workqueue.Repository,
	resolver connector_inbound.IdentityConvergenceResolver,
	waits []connector_workqueue.IdentityConvergenceWait,
	now time.Time,
) (int64, error) {
	var total int64
	for _, wait := range waits {
		if connector_workqueue.IsIdentityConvergenceWaitInvalid(wait) {
			terminalized, err := s.terminalizeInvalidIdentityConvergenceWaits(ctx, repository, []string{wait.WorkID}, now)
			if err != nil {
				return total, err
			}
			total += terminalized
			continue
		}
		expired := identityConvergenceWaitExpired(wait, now, s.effectiveDependencyMaxWait())
		if resolver == nil {
			if err := scheduleIdentityConvergenceProbeRetry(ctx, repository, wait, now); err != nil {
				return total, fmt.Errorf("connector inbound supervisor: defer identity convergence dependency without resolver work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err)
			}
			continue
		}
		probe := connector_inbound.IdentityConvergenceProbe{
			Namespace:  wait.Namespace,
			ObjectKind: wait.ObjectKind,
			SourceID:   wait.SourceID,
			Dependency: wait.Dependency,
		}
		result, probeErr := s.resolveIdentityConvergenceBounded(ctx, resolver, probe)
		if probeErr != nil {
			// Resolver failures may carry exact record or remote selectors. Keep
			// them inside the resolver boundary; the fixed class plus redacted key
			// hash is sufficient to correlate the persisted wait safely.
			log.Printf("WARN connector inbound supervisor: identity convergence probe failed work_id=%s dependency_key_hash=%s failure_class=resolver_error", wait.WorkID, wait.Dependency.KeyHash)
			if err := scheduleIdentityConvergenceProbeRetry(ctx, repository, wait, now); err != nil {
				return total, fmt.Errorf("connector inbound supervisor: schedule identity convergence probe retry work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err)
			}
			// A child probe deadline is a bounded, persisted retry, not a reason to
			// stop every unrelated inbound worker. Only the caller's own canceled
			// lifecycle context interrupts this page after its retry has been saved.
			if err := ctx.Err(); err != nil {
				return total, fmt.Errorf("connector inbound supervisor: identity convergence reconciliation stopped work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err)
			}
			continue
		}
		if !result.Converged() {
			// Elapsed time is only a bounded-wait policy after the current owner has
			// authoritatively reported that the dependency is still unresolved. A
			// missing or failed resolver remains retryable above; it must not let the
			// clock outrank an owner fact that may already have committed.
			if expired {
				expiredNow, err := repository.ExpireIdentityConvergenceWait(ctx, wait, now)
				if err != nil {
					return total, fmt.Errorf("connector inbound supervisor: expire unresolved identity convergence dependency work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err)
				}
				if expiredNow {
					total++
					log.Printf("WARN connector inbound supervisor: unresolved identity convergence wait expired after fresh owner read work_id=%s dependency_key_hash=%s failure_class=%s", wait.WorkID, wait.Dependency.KeyHash, connector_workqueue.FailureClassIdentityConvergenceExpired)
				}
				continue
			}
			if err := scheduleIdentityConvergenceProbeRetry(ctx, repository, wait, now); err != nil {
				return total, fmt.Errorf("connector inbound supervisor: reschedule unresolved identity convergence probe work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err)
			}
			continue
		}
		var woken int64
		var err error
		if result.ExactRecord {
			woken, err = repository.WakeRecordDependency(ctx, wait.Dependency.ObjectAPIName, wait.Dependency.RecordID, now)
		} else if result.RemoteIdentity {
			woken, err = repository.WakeRemoteIdentity(ctx, IdentityConvergenceRemoteSignal{
				ConnectorOwnerKey:  wait.Dependency.ConnectorOwnerKey,
				ConnectorNamespace: wait.Dependency.ConnectorNamespace,
				DestinationSlotID:  wait.Dependency.DestinationSlotID,
				ConnectorRemoteID:  wait.Dependency.ConnectorRemoteID,
			}, now)
		}
		if err != nil {
			if scheduleErr := scheduleIdentityConvergenceProbeRetry(ctx, repository, wait, now); scheduleErr != nil {
				return total, errors.Join(
					fmt.Errorf("connector inbound supervisor: wake identity convergence dependency work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err),
					fmt.Errorf("connector inbound supervisor: schedule failed wake retry work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, scheduleErr),
				)
			}
			return total, fmt.Errorf("connector inbound supervisor: wake identity convergence dependency work_id=%s dependency_key_hash=%s: %w", wait.WorkID, wait.Dependency.KeyHash, err)
		}
		if woken > 0 {
			total += woken
			s.wakeLiveWorkItemClaimScheduler(workerKey(wait.Namespace, wait.InstalledPackID))
		}
	}
	return total, nil
}

func (s *Supervisor) terminalizeInvalidIdentityConvergenceWaits(
	ctx context.Context,
	repository *connector_workqueue.Repository,
	workIDs []string,
	now time.Time,
) (int64, error) {
	var total int64
	for _, workID := range workIDs {
		terminalized, err := repository.TerminalizeInvalidIdentityConvergenceWait(ctx, workID, now)
		if err != nil {
			return total, fmt.Errorf("connector inbound supervisor: terminalize invalid identity convergence dependency work_id=%s: %w", workID, err)
		}
		if !terminalized {
			continue
		}
		total++
		log.Printf(
			"WARN connector inbound supervisor: invalid identity convergence wait terminalized work_id=%s failure_class=%s",
			workID,
			connector_workqueue.FailureClassIdentityConvergenceInvalid,
		)
	}
	return total, nil
}

func identityConvergenceWaitExpired(wait connector_workqueue.IdentityConvergenceWait, now time.Time, maxWait time.Duration) bool {
	return maxWait > 0 &&
		!wait.DependencyBlockedAt.IsZero() &&
		!now.IsZero() &&
		!wait.DependencyBlockedAt.Add(maxWait).After(now)
}

func scheduleIdentityConvergenceProbeRetry(
	ctx context.Context,
	repository *connector_workqueue.Repository,
	wait connector_workqueue.IdentityConvergenceWait,
	now time.Time,
) error {
	if wait.DependencyProbeAfter != nil {
		return repository.RescheduleIdentityConvergenceProbeRetry(
			ctx,
			wait.WorkID,
			wait.Dependency.KeyHash,
			*wait.DependencyProbeAfter,
			now,
		)
	}
	return repository.ScheduleIdentityConvergenceProbeRetry(ctx, wait.WorkID, wait.Dependency.KeyHash, now)
}

type identityConvergenceProbeOutcome struct {
	result connector_inbound.IdentityConvergenceResult
	err    error
}

func (s *Supervisor) resolveIdentityConvergenceBounded(
	ctx context.Context,
	resolver connector_inbound.IdentityConvergenceResolver,
	probe connector_inbound.IdentityConvergenceProbe,
) (connector_inbound.IdentityConvergenceResult, error) {
	if ctx == nil {
		ctx = connectorInboundSupervisorContext()
	}
	probeCtx, cancel := context.WithTimeout(ctx, s.effectiveDependencyProbeTimeout())
	defer cancel()
	s.dependencyProbeOnce.Do(func() {
		s.dependencyProbeSlot = make(chan struct{}, 1)
	})
	select {
	case s.dependencyProbeSlot <- struct{}{}:
	case <-probeCtx.Done():
		return connector_inbound.IdentityConvergenceResult{}, probeCtx.Err()
	}
	outcome := make(chan identityConvergenceProbeOutcome, 1)
	go func() {
		defer func() {
			<-s.dependencyProbeSlot
			if recover() != nil {
				// The resolver is host supplied. Its panic must not take down the
				// process or strand the single-flight slot; it follows the same
				// durable retry path as any other resolver failure. Do not expose
				// the recovered value because it may contain host-owned data.
				outcome <- identityConvergenceProbeOutcome{err: errConnectorInboundIdentityProbePanic}
			}
		}()
		result, err := resolver.ResolveIdentityConvergence(probeCtx, probe)
		outcome <- identityConvergenceProbeOutcome{result: result, err: err}
	}()
	select {
	case resolved := <-outcome:
		return acceptIdentityConvergenceProbeOutcome(probeCtx, resolved)
	case <-probeCtx.Done():
		return connector_inbound.IdentityConvergenceResult{}, probeCtx.Err()
	}
}

func acceptIdentityConvergenceProbeOutcome(
	probeCtx context.Context,
	resolved identityConvergenceProbeOutcome,
) (connector_inbound.IdentityConvergenceResult, error) {
	// The resolver result and the deadline can become ready in the same select.
	// A result returned after cancellation is not a fresh owner fact even when
	// the resolver reports nil error, so the deadline must win deterministically.
	if err := probeCtx.Err(); err != nil {
		return connector_inbound.IdentityConvergenceResult{}, err
	}
	return resolved.result, resolved.err
}

func (s *Supervisor) reconcile(ctx context.Context, startup bool, force bool, preserveExistingOnReadError bool) error {
	if s.runtimeServingConfigured {
		// A stopped-upgrade Runtime selection is exact authority, not a best-effort
		// cache. Do not preserve a worker when an exact pack/profile/projection read
		// can no longer verify that authority. The separate WorkItem activation probe
		// is queue state, not metadata authority; a transient probe error preserves
		// the already-bound worker and its generation below.
		preserveExistingOnReadError = false
	}
	packRows, err := s.listPacks(ctx, s.db)
	if err != nil {
		if s.runtimeServingConfigured && !startup {
			s.stopWorkers(s.workerKeys(), false)
		}
		if startup {
			s.resetStarted()
		}
		return err
	}

	desired := make(map[string]struct{}, len(packRows))
	for _, row := range packRows {
		desired[workerKey(row.Namespace, row.InstallID)] = struct{}{}
	}
	var removed []string
	for _, key := range s.workerKeys() {
		if _, ok := desired[key]; !ok {
			removed = append(removed, key)
		}
	}
	s.stopWorkers(removed, false)

	startedKeys := make([]string, 0, len(packRows))
	for _, row := range packRows {
		key := workerKey(row.Namespace, row.InstallID)
		workerCount := connectorInboundWorkersPerPackFromEnv()
		maxProcessingRows, explicitProcessingBudget := connectorInboundMaxProcessingRowsFromEnv(workerCount)
		queueProfileMode := connectorInboundQueueProfileModeFromEnv()
		currentFingerprint := s.workerFingerprint(key)
		activation, err := s.resolveWorkItemClaimActivation(ctx, row)
		if err != nil {
			if !startup &&
				currentFingerprint != "" &&
				!errors.Is(err, errConnectorInboundWorkItemSchemaGap) &&
				!errors.Is(err, errConnectorInboundWorkItemCapabilityGap) &&
				!errors.Is(err, errConnectorInboundWorkItemCutoverGap) {
				log.Printf("WARN connector inbound supervisor: preserving existing WorkItem worker for %s (%s) after readiness probe failed: %v", row.Namespace, row.InstallID, err)
				continue
			}
			s.stopWorkers([]string{key}, false)
			if row.DrainOnly {
				s.blockDrainOnlyWorker(row, "verify WorkItem readiness", err)
				continue
			}
			if startup {
				s.stopStartedWorkers(startedKeys)
			}
			return err
		}
		eligible, err := s.drainOnlyWorkerStillEligible(ctx, row)
		if err != nil {
			s.stopWorkers([]string{key}, false)
			s.blockDrainOnlyWorker(row, "verify drain eligibility", err)
			continue
		}
		if !eligible {
			s.stopWorkers([]string{key}, false)
			continue
		}

		baseFingerprint := workerFingerprint(row, "", workerCount, maxProcessingRows, explicitProcessingBudget, queueProfileMode)
		workItemWakeChecked := false

		profiles, err := s.loadProfiles(ctx, s.db, row.Namespace, row.InstallID, row.DrainOnly)
		if err != nil {
			if preserveExistingOnReadError && !startup && currentFingerprint != "" {
				log.Printf("WARN connector inbound supervisor: preserving existing worker for %s (%s) after periodic profile read failed: %v", row.Namespace, row.InstallID, err)
				continue
			}
			s.stopWorkers([]string{key}, false)
			if row.DrainOnly {
				s.blockDrainOnlyWorker(row, "load profiles", err)
				continue
			}
			if startup {
				s.stopStartedWorkers(startedKeys)
			}
			return fmt.Errorf("connector inbound supervisor: load profiles for %s (%s): %w", row.Namespace, row.InstallID, err)
		}
		if !force && currentFingerprint == baseFingerprint && !profilesDeclareBatchProjectionBodyField(profiles) {
			workItemWakeChecked = true
			if s.wakeLiveWorkItemClaimScheduler(key) {
				continue
			}
		}

		projectionFingerprint := ""
		if profilesDeclareBatchProjectionBodyField(profiles) {
			projectionFingerprint, err = s.resolveProjectionFingerprint(ctx, row)
			if err != nil {
				if preserveExistingOnReadError && !startup && currentFingerprint != "" {
					log.Printf("WARN connector inbound supervisor: preserving existing worker for %s (%s) after periodic projection fingerprint read failed: %v", row.Namespace, row.InstallID, err)
					continue
				}
				s.stopWorkers([]string{key}, false)
				if row.DrainOnly {
					s.blockDrainOnlyWorker(row, "resolve projection fingerprint", err)
					continue
				}
				if startup {
					s.stopStartedWorkers(startedKeys)
				}
				return fmt.Errorf("connector inbound supervisor: resolve projection fingerprint for %s (%s): %w", row.Namespace, row.InstallID, err)
			}
		}
		fingerprint := workerFingerprint(row, projectionFingerprint, workerCount, maxProcessingRows, explicitProcessingBudget, queueProfileMode)
		if !force && currentFingerprint == fingerprint {
			if !workItemWakeChecked && s.wakeLiveWorkItemClaimScheduler(key) {
				continue
			}
		}

		// Rebuild even when the key is already running; live pack metadata may
		// have changed without changing namespace/install identity.
		s.stopWorkers([]string{key}, false)

		contract, err := s.resolveBatchContract(ctx, s.db, row.Namespace, row.InstallID, profiles, row.DrainOnly)
		if err != nil {
			if row.DrainOnly {
				s.blockDrainOnlyWorker(row, "resolve startup contract", err)
				continue
			}
			if startup {
				s.stopStartedWorkers(startedKeys)
			}
			return fmt.Errorf("connector inbound supervisor: resolve startup contract for %s (%s): %w", row.Namespace, row.InstallID, err)
		}

		effectiveProfiles := profiles
		if len(contract.requiredObjectKinds) > 0 {
			effectiveProfiles = filterProfilesForRequiredObjectKinds(profiles, contract.requiredObjectKinds)
		}
		duplicateNoopFallback := len(contract.requiredObjectKinds) == 0 && hasDuplicateExecutableBatchCoverage(effectiveProfiles)
		if !duplicateNoopFallback {
			if err := s.validateBatchCredentials(ctx, row, effectiveProfiles); err != nil {
				if row.DrainOnly {
					if drainOnlyWorkerNoLongerEligible(err) {
						continue
					}
					s.blockDrainOnlyWorker(row, "validate batch credentials", err)
					continue
				}
				if startup {
					s.stopStartedWorkers(startedKeys)
				}
				return fmt.Errorf("connector inbound supervisor: validate batch credentials for %s (%s): %w", row.Namespace, row.InstallID, err)
			}
		}

		batchClient, err := s.buildBatchClient(row, effectiveProfiles, contract)
		if err != nil {
			if row.DrainOnly {
				s.blockDrainOnlyWorker(row, "build batch client", err)
				continue
			}
			if startup {
				s.stopStartedWorkers(startedKeys)
			}
			return fmt.Errorf("connector inbound supervisor: build batch client for %s (%s): %w", row.Namespace, row.InstallID, err)
		}
		// Wire the drain-time companion incarnation re-check so cross-pack events
		// verify the frozen companion app-pack incarnation is still active before
		// executing.
		companionInstalled := companionInstallStillInstalledChecker(s.db)

		workerCtx, forceCancel := connectorInboundWorkerContext()
		drainCtx, requestDrain := connectorInboundDrainContext()
		done := make(chan struct{})
		workers := make([]*connector_inbound.Worker, 0, workerCount)
		enableStageTiming := connectorInboundStageTimingFromEnv()
		flowDeadline := connectorFlowDeadlineFromEnv()
		claimScheduler := connector_inbound.NewCentralClaimScheduler()
		workItemService := activation.service
		if workItemService == nil {
			err := fmt.Errorf("connector inbound supervisor: work_item claim path requires one complete WorkItem service")
			s.stopWorkers([]string{key}, false)
			if row.DrainOnly {
				s.blockDrainOnlyWorker(row, "wire work item claim path", err)
				continue
			}
			if startup {
				s.stopStartedWorkers(startedKeys)
			}
			return err
		}
		pressureProvider := workItemProcessingPressureProvider{
			db:  s.db,
			now: time.Now,
		}
		workItemGeneration := activation.generation
		workItemSchedulerOwner := s.workItemSchedulerOwnerID(row)
		workItemSchedulerFencing := s.workItemSchedulerFencingToken(row)
		stats := s.db.Stats()
		log.Printf("event=connector_inbound_supervisor_profile elapsed_ms=0 action=start namespace=%s installed_pack_id=%s workers=%d max_processing_rows=%d explicit_processing_budget=%t queue_profile=%s stage_timing=%t db_pool_max_open=%d db_pool_in_use=%d", row.Namespace, row.InstallID, workerCount, maxProcessingRows, explicitProcessingBudget, queueProfileMode, enableStageTiming, stats.MaxOpenConnections, stats.InUse)
		claimLimiter := connector_inbound.NewSQLClaimLimiter(connector_inbound.SQLClaimLimiterConfig{
			DBStats:                s.db,
			Pressure:               pressureProvider,
			Profile:                enableStageTiming,
			MaxProcessingRows:      maxProcessingRows,
			GlobalProcessingBudget: explicitProcessingBudget,
		})
		for i := 0; i < workerCount; i++ {
			workerID := s.workerID(row.Namespace)
			if workerCount > 1 {
				workerID = fmt.Sprintf("%s#%02d", workerID, i+1)
			}
			profile := connectorInboundWorkItemWorkerQueueProfile(workerCount, i, queueProfileMode)
			batchSize := profile.batchSize
			queueClasses := strings.Join(profile.queueClasses, ",")
			if queueClasses == "" {
				queueClasses = "all"
			}
			preferredQueueClasses := strings.Join(profile.preferredQueueClasses, ",")
			if preferredQueueClasses == "" {
				preferredQueueClasses = "none"
			}
			log.Printf("event=connector_inbound_worker_profile elapsed_ms=0 namespace=%s installed_pack_id=%s worker_id=%s ordinal=%d worker_count=%d max_processing_rows=%d explicit_processing_budget=%t queue_profile=%s stage_timing=%t queue_classes=%s preferred_queue_classes=%s batch_size=%d", row.Namespace, row.InstallID, workerID, i+1, workerCount, maxProcessingRows, explicitProcessingBudget, queueProfileMode, enableStageTiming, queueClasses, preferredQueueClasses, batchSize)
			worker, err := s.newWorker(ctx,
				connector_inbound.WorkerConfig{
					Namespace:         row.Namespace,
					PackInstallID:     row.InstallID,
					WorkerID:          workerID,
					BatchSize:         batchSize,
					QueueClasses:      profile.queueClasses,
					ClaimLimiter:      claimLimiter,
					EnableStageTiming: enableStageTiming,
					FlowDeadline:      flowDeadline,

					WorkItemGeneration:            workItemGeneration,
					WorkItemSchedulerOwnerID:      workItemSchedulerOwner,
					WorkItemSchedulerFencingToken: workItemSchedulerFencing,
				},
				workItemService,
				s.syncstate,
				batchClient,
				s.flow,
				s.resolver,
			)
			if err != nil {
				requestDrain()
				forceCancel()
				if startup {
					s.stopStartedWorkers(startedKeys)
				}
				return fmt.Errorf("connector inbound supervisor: construct worker for %s (%s): %w", row.Namespace, row.InstallID, err)
			}
			worker.SetWorkItemQueueClassPreference(profile.preferredQueueClasses)
			worker.UseCentralClaimScheduler(claimScheduler)
			worker.SetCompanionInstalledChecker(companionInstalled)
			worker.SetOutboundEchoProbe(connector_inbound.SQLOutboundEchoProbe{DB: s.db})
			workers = append(workers, worker)
		}
		s.mu.Lock()
		s.workers[key] = connectorInboundWorkerHandle{
			requestDrain:   requestDrain,
			forceCancel:    forceCancel,
			done:           done,
			claimScheduler: claimScheduler,
			fingerprint:    fingerprint,
			workerCount:    workerCount,
		}
		s.wg.Add(len(workers))
		s.mu.Unlock()
		log.Printf("connector inbound supervisor: constructed worker pool for %s (%s): workers=%d central_scheduler=true generation=%d drain_only=%t",
			row.Namespace, row.InstallID, workerCount, workItemGeneration, row.DrainOnly)

		startedKeys = append(startedKeys, key)
		groupWG := &sync.WaitGroup{}
		groupWG.Add(len(workers))
		for i, worker := range workers {
			ordinal := i + 1
			go func(namespace, installID string, workerOrdinal int, w *connector_inbound.Worker, wg *sync.WaitGroup) {
				defer s.wg.Done()
				defer wg.Done()
				log.Printf("connector inbound supervisor: worker loop entered for %s (%s) #%d: claim_first=true",
					namespace, installID, workerOrdinal)
				s.superviseWorker(workerCtx, drainCtx, namespace, installID, workerOrdinal, true, w)
			}(row.Namespace, row.InstallID, ordinal, worker, groupWG)
		}
		go func(wg *sync.WaitGroup) {
			wg.Wait()
			close(done)
		}(groupWG)
	}

	return nil
}

func (s *Supervisor) resolveWorkItemClaimActivation(ctx context.Context, row packstate.ConnectorInboundWorkerPackRow) (connectorInboundWorkItemClaimActivation, error) {
	service, ok := s.events.(connector_inbound.WorkItemService)
	if !ok || service == nil || !service.HonorsClaimWorkTypes() {
		return connectorInboundWorkItemClaimActivation{}, fmt.Errorf("%w for %s (%s)", errConnectorInboundWorkItemCapabilityGap, row.Namespace, row.InstallID)
	}
	if s.requireWorkItemOnlyCutover == nil {
		return connectorInboundWorkItemClaimActivation{}, fmt.Errorf("%w for %s (%s): supervisor receipt verifier is not configured", errConnectorInboundWorkItemCutoverGap, row.Namespace, row.InstallID)
	}
	if err := s.requireWorkItemOnlyCutover(ctx); err != nil {
		return connectorInboundWorkItemClaimActivation{}, fmt.Errorf("%w for %s (%s): %w", errConnectorInboundWorkItemCutoverGap, row.Namespace, row.InstallID, err)
	}
	ready, err := s.workItemClaimSchemaReady(ctx)
	if err != nil {
		return connectorInboundWorkItemClaimActivation{}, fmt.Errorf("connector inbound supervisor: verify work_item claim schema for %s (%s): %w", row.Namespace, row.InstallID, err)
	}
	if !ready {
		return connectorInboundWorkItemClaimActivation{}, fmt.Errorf("%w for %s (%s)", errConnectorInboundWorkItemSchemaGap, row.Namespace, row.InstallID)
	}
	generation, err := s.resolveWorkItemClaimGeneration(ctx, row, service)
	if err != nil {
		return connectorInboundWorkItemClaimActivation{}, err
	}
	return connectorInboundWorkItemClaimActivation{service: service, generation: generation}, nil
}

func (s *Supervisor) workItemClaimSchemaReady(ctx context.Context) (bool, error) {
	if s.db == nil {
		return false, fmt.Errorf("db is required")
	}
	if err := probeWorkItemClaimSchema(ctx, s.db, `_System_ConnectorWorkItem`, connectorInboundWorkItemClaimSchemaColumns); err != nil {
		if workItemClaimSchemaMissing(err) {
			return false, nil
		}
		return false, err
	}
	if err := probeWorkItemClaimSchema(ctx, s.db, `_System_ConnectorEvent`, connectorInboundEventOwnerSchemaColumns); err != nil {
		if workItemClaimSchemaMissing(err) {
			return false, nil
		}
		return false, err
	}
	if err := probeWorkItemClaimSchema(ctx, s.db, `_System_ConnectorWorkAttempt`, connectorInboundWorkItemAttemptSchemaColumns); err != nil {
		if workItemClaimSchemaMissing(err) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func probeWorkItemClaimSchema(ctx context.Context, db *sql.DB, tableName, columns string) error {
	rows, err := db.QueryContext(ctx, fmt.Sprintf("SELECT %s FROM %s LIMIT 0", columns, tableName))
	if err != nil {
		return err
	}
	return rows.Close()
}

func workItemClaimSchemaMissing(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	missingMarkers := []string{
		"unknown column",
		"unknown table",
		"doesn't exist",
		"does not exist",
		"no such table",
		"no such column",
		"error 1054",
		"error 1146",
	}
	for _, marker := range missingMarkers {
		if strings.Contains(msg, marker) {
			return true
		}
	}
	return false
}

// superviseWorker runs one connector-inbound worker and restarts it in place if
// it exits unexpectedly — an escaped panic, or (defensively) a non-cancel
// return — while neither force-cancel nor drain has been requested. A worker
// stopped by Stop()/drain is left stopped, preserving drain semantics exactly:
// the watcher's wg.Wait() still observes this goroutine return, so `done` closes
// and Stop returns within its drain deadline.
//
// The recover here is a belt-and-suspenders backstop; the primary per-event
// recover lives in the worker frame (CIN-P0-1). This frame holds NO claim token,
// so it MUST NOT finalize the in-flight event — doing so would violate the
// single-arbiter CAS lease invariant. Its job is only: log + count + backoff +
// restart, restoring capacity that would otherwise be dropped silently.
func (s *Supervisor) superviseWorker(workerCtx, drainCtx context.Context, namespace, installID string, ordinal int, claimFirst bool, w *connector_inbound.Worker) {
	backoff := newConnectorInboundRestartBackoff(
		s.effectiveRestartBackoffMin(),
		connectorInboundRestartBackoffMax,
		connectorInboundRestartHealthyRunThreshold,
	)
	for {
		if shuttingDown(workerCtx, drainCtx) {
			return
		}
		start := time.Now()
		runErr, panicValue, panicStack := s.runWorkerOnce(workerCtx, drainCtx, claimFirst, w)

		// Force-cancel or drain ends supervision: never restart on shutdown, even
		// if a panic raced with the cancel. done must still close.
		if shuttingDown(workerCtx, drainCtx) {
			if panicValue != nil {
				log.Printf("WARN connector inbound worker %s (%s) #%d panicked during shutdown: %v\n%s",
					namespace, installID, ordinal, panicValue, panicStack)
			} else if runErr != nil && !errors.Is(runErr, context.Canceled) {
				log.Printf("connector inbound worker %s (%s) #%d stopped: %v", namespace, installID, ordinal, runErr)
			}
			return
		}

		if claimFirst && panicValue == nil && runErr == nil {
			continue
		}

		// Still live but the run exited unexpectedly: restore capacity in place.
		connectorInboundWorkerRestarts.Add(1)
		sleep := backoff.next(time.Since(start))
		if panicValue != nil {
			log.Printf("WARN connector inbound worker %s (%s) #%d panicked; restarting in %s: %v\n%s",
				namespace, installID, ordinal, sleep, panicValue, panicStack)
		} else {
			log.Printf("WARN connector inbound worker %s (%s) #%d exited unexpectedly (%v); restarting in %s",
				namespace, installID, ordinal, runErr, sleep)
		}
		if !sleepUntilRestartOrShutdown(sleep, workerCtx, drainCtx) {
			return
		}
	}
}

// runWorkerOnce invokes the worker's drain loop once, converting an escaped
// panic into a recovered value + stack rather than crashing the process. It
// captures debug.Stack (goroutine frames only — never the event payload).
func (s *Supervisor) runWorkerOnce(workerCtx, drainCtx context.Context, claimFirst bool, w *connector_inbound.Worker) (runErr error, panicValue any, panicStack []byte) {
	defer func() {
		if r := recover(); r != nil {
			panicValue = r
			panicStack = debug.Stack()
		}
	}()
	switch {
	case s.runWorker != nil:
		runErr = s.runWorker(workerCtx, drainCtx, claimFirst, w)
	case claimFirst:
		runErr = w.RunClaimFirstUntilDrained(workerCtx, drainCtx.Done())
	default:
		runErr = w.RunUntilDrained(workerCtx, drainCtx.Done())
	}
	return runErr, panicValue, panicStack
}

// shuttingDown reports whether force-cancel or cooperative drain has been
// requested. Both contexts are cancel-derived, so Err() != nil is the signal.
func shuttingDown(workerCtx, drainCtx context.Context) bool {
	return workerCtx.Err() != nil || drainCtx.Err() != nil
}

// sleepUntilRestartOrShutdown waits for d, returning true if the wait completed
// (proceed to restart) or false if force-cancel/drain intervened (stop). This is
// what keeps Stop() bounded: a pending backoff is abandoned the moment drain or
// force-cancel fires.
func sleepUntilRestartOrShutdown(d time.Duration, workerCtx, drainCtx context.Context) bool {
	if d <= 0 {
		return !shuttingDown(workerCtx, drainCtx)
	}
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-workerCtx.Done():
		return false
	case <-drainCtx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func (s *Supervisor) effectiveRestartBackoffMin() time.Duration {
	if s.restartBackoffMin > 0 {
		return s.restartBackoffMin
	}
	return connectorInboundRestartBackoffMin
}

// connectorInboundRestartBackoff is a bounded-exponential restart schedule. It
// starts at min, doubles per crash up to max, and resets to min after a run that
// outlived healthyRunThreshold (an isolated panic, not a crash-loop).
type connectorInboundRestartBackoff struct {
	minBackoff          time.Duration
	maxBackoff          time.Duration
	healthyRunThreshold time.Duration
	current             time.Duration
}

func newConnectorInboundRestartBackoff(minBackoff, maxBackoff, healthyRunThreshold time.Duration) *connectorInboundRestartBackoff {
	if minBackoff <= 0 {
		minBackoff = connectorInboundRestartBackoffMin
	}
	if maxBackoff < minBackoff {
		maxBackoff = minBackoff
	}
	return &connectorInboundRestartBackoff{
		minBackoff:          minBackoff,
		maxBackoff:          maxBackoff,
		healthyRunThreshold: healthyRunThreshold,
		current:             minBackoff,
	}
}

// next returns how long to wait before the next restart, given how long the
// just-exited run lasted, then advances the schedule (capped at maxBackoff).
func (b *connectorInboundRestartBackoff) next(ranFor time.Duration) time.Duration {
	if b.healthyRunThreshold > 0 && ranFor >= b.healthyRunThreshold {
		b.current = b.minBackoff
	}
	sleep := b.current
	b.current *= 2
	if b.current > b.maxBackoff {
		b.current = b.maxBackoff
	}
	return sleep
}

func (s *Supervisor) workerFingerprint(key string) string {
	s.mu.Lock()
	defer s.mu.Unlock()
	if handle, ok := s.workers[key]; ok {
		select {
		case <-handle.done:
			return ""
		default:
		}
		return handle.fingerprint
	}
	return ""
}

func (s *Supervisor) validateBatchCredentials(ctx context.Context, row packstate.ConnectorInboundWorkerPackRow, profiles []*connector_callout.ConnectorCalloutProfile) error {
	if s.resolveBatchAuthType == nil {
		return nil
	}
	rowCtx := ctx
	if row.DrainOnly {
		rowCtx = namedcredential.WithDrainOnlyInstalledPack(rowCtx, row.Namespace, row.InstallID)
	}
	for _, profile := range profiles {
		if !profileBatchSpecConfigured(profile) || !profileHasUsableBatchObjectCoverage(profile) {
			continue
		}
		credentialRef := strings.TrimSpace(profile.BatchSpec.CredentialRef)
		if _, err := s.resolveBatchAuthType(rowCtx, row, credentialRef); err != nil {
			return fmt.Errorf("resolve auth type for credential %s: %w", credentialRef, err)
		}
	}
	return nil
}

func drainOnlyWorkerNoLongerEligible(err error) bool {
	return errors.Is(err, namedcredential.ErrConnectorPackInactive)
}

func (s *Supervisor) drainOnlyWorkerStillEligible(ctx context.Context, row packstate.ConnectorInboundWorkerPackRow) (bool, error) {
	if !row.DrainOnly || s.resolveDrainEligible == nil {
		return true, nil
	}
	return s.resolveDrainEligible(ctx, row.Namespace, row.InstallID)
}

func (s *Supervisor) resolveProjectionFingerprint(ctx context.Context, row packstate.ConnectorInboundWorkerPackRow) (string, error) {
	if s.resolveProjectionFP == nil {
		return "", nil
	}
	return s.resolveProjectionFP(ctx, s.db, row.Namespace, row.InstallID, row.DrainOnly)
}

func (s *Supervisor) Stop() error {
	s.lifecycleMu.Lock()
	s.mu.Lock()
	if s.starting {
		startupCancel := s.startupCancel
		s.lifecycleEpoch++
		s.mu.Unlock()
		s.lifecycleMu.Unlock()
		startupCancel()
		return nil
	}
	if !s.started {
		s.mu.Unlock()
		s.lifecycleMu.Unlock()
		return nil
	}
	handles := make([]connectorInboundWorkerHandle, 0, len(s.workers))
	for _, handle := range s.workers {
		handles = append(handles, handle)
	}
	reconcileCancel := s.reconcileCancel
	reconcileDone := s.reconcileDone
	s.reconcileCancel = nil
	s.reconcileDone = nil
	s.workers = make(map[string]connectorInboundWorkerHandle)
	s.started = false
	s.mu.Unlock()

	if reconcileCancel != nil {
		reconcileCancel()
	}
	if reconcileDone != nil {
		timer := time.NewTimer(s.effectiveReconcileStopTimeout())
		defer timer.Stop()
		select {
		case <-reconcileDone:
		case <-timer.C:
			log.Printf("WARN connector inbound supervisor: reconcile loop did not stop before bounded shutdown timeout")
		}
	}
	s.stopWorkerHandles(handles)
	s.lifecycleMu.Unlock()
	return nil
}

func (s *Supervisor) stopStartedWorkers(keys []string) {
	s.stopWorkers(keys, true)
}

func (s *Supervisor) stopWorkers(keys []string, markStopped bool) {
	s.mu.Lock()
	handles := make([]connectorInboundWorkerHandle, 0, len(keys))
	for _, key := range keys {
		if handle, ok := s.workers[key]; ok {
			handles = append(handles, handle)
			delete(s.workers, key)
		}
	}
	if markStopped {
		s.started = false
	}
	s.mu.Unlock()

	s.stopWorkerHandles(handles)
}

func (s *Supervisor) wakeLiveWorkItemClaimScheduler(key string) bool {
	s.mu.Lock()
	handle, ok := s.workers[key]
	s.mu.Unlock()
	if !ok {
		return false
	}
	if connectorInboundWorkerHandleDone(handle) {
		return false
	}
	if handle.claimScheduler != nil {
		return handle.claimScheduler.WakeIfRunning()
	}
	return false
}

func connectorInboundWorkerHandleDone(handle connectorInboundWorkerHandle) bool {
	if handle.done == nil {
		return false
	}
	select {
	case <-handle.done:
		return true
	default:
		return false
	}
}

func (s *Supervisor) stopWorkerHandles(handles []connectorInboundWorkerHandle) {
	stopWorkerHandles(handles, s.workerDrainDeadline())
}

func (s *Supervisor) workerDrainDeadline() time.Duration {
	if s.workerDrainTimeout > 0 {
		return s.workerDrainTimeout
	}
	return defaultConnectorInboundWorkerDrainTimeout
}

func (s *Supervisor) startBacklogReconciler() {
	interval := s.effectiveBacklogReconcileInterval()
	if interval <= 0 {
		return
	}
	ctx, cancel := context.WithCancel(connectorInboundSupervisorContext())
	done := make(chan struct{})
	s.mu.Lock()
	s.reconcileCancel = cancel
	s.reconcileDone = done
	s.mu.Unlock()

	go func() {
		defer close(done)
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if err := s.reconcileBacklog(ctx); err != nil {
					log.Printf("WARN connector inbound supervisor: periodic backlog reconcile failed: %v", err)
				}
			}
		}
	}()
}

func (s *Supervisor) effectiveBacklogReconcileInterval() time.Duration {
	if s.backlogReconcileInterval != 0 {
		return s.backlogReconcileInterval
	}
	return defaultConnectorInboundBacklogReconcileInterval
}

func (s *Supervisor) effectiveDependencyStartupTimeout() time.Duration {
	if s.dependencyStartupTimeout > 0 {
		return s.dependencyStartupTimeout
	}
	return connectorInboundIdentityConvergenceStartupTimeout
}

func (s *Supervisor) effectiveDependencyProbeTimeout() time.Duration {
	if s.dependencyProbeTimeout > 0 {
		return s.dependencyProbeTimeout
	}
	return connectorInboundIdentityConvergenceProbeTimeout
}

func (s *Supervisor) effectiveDependencyMaxWait() time.Duration {
	return s.dependencyMaxWait
}

func (s *Supervisor) effectiveReconcileStopTimeout() time.Duration {
	if s.reconcileStopTimeout > 0 {
		return s.reconcileStopTimeout
	}
	return connectorInboundReconcileStopTimeout
}

func connectorInboundWorkerContext() (context.Context, context.CancelFunc) {
	return context.WithCancel(connectorInboundSupervisorContext())
}

func connectorInboundDrainContext() (context.Context, context.CancelFunc) {
	return context.WithCancel(connectorInboundSupervisorContext())
}

func connectorInboundSupervisorContext() context.Context {
	return context.Background()
}

func connectorInboundWorkersPerPackFromEnv() int {
	raw := strings.TrimSpace(os.Getenv(envConnectorInboundWorkersPerPack))
	if raw == "" {
		return defaultConnectorInboundWorkersPerPack
	}
	value, err := strconv.Atoi(raw)
	if err != nil || value < 1 {
		log.Printf("connector inbound supervisor: invalid %s=%q; using default %d", envConnectorInboundWorkersPerPack, raw, defaultConnectorInboundWorkersPerPack)
		return defaultConnectorInboundWorkersPerPack
	}
	if value > maxConnectorInboundWorkersPerPack {
		log.Printf("connector inbound supervisor: %s=%d exceeds max %d; clamping", envConnectorInboundWorkersPerPack, value, maxConnectorInboundWorkersPerPack)
		return maxConnectorInboundWorkersPerPack
	}
	if value > safeConnectorInboundWorkersPerPack {
		log.Printf("connector inbound supervisor: unsafe high %s=%d exceeds conservative default %d; worker count is preserved for compatibility, and processing budget remains controlled by %s", envConnectorInboundWorkersPerPack, value, safeConnectorInboundWorkersPerPack, envConnectorInboundMaxProcessingRows)
	}
	return value
}

func connectorInboundMaxProcessingRowsFromEnv(workerCount int) (int, bool) {
	defaultBudget := connector_inbound.MaxProcessingRowsForWorkers(workerCount)
	raw := strings.TrimSpace(os.Getenv(envConnectorInboundMaxProcessingRows))
	if raw == "" {
		if workerCount > safeConnectorInboundWorkersPerPack {
			log.Printf("connector inbound supervisor: unsafe high %s=%d is using legacy derived processing budget %d; set %s explicitly to decouple worker fan-out from processing budget", envConnectorInboundWorkersPerPack, workerCount, defaultBudget, envConnectorInboundMaxProcessingRows)
		}
		return defaultBudget, false
	}
	value, err := strconv.Atoi(raw)
	if err != nil || value < 1 {
		log.Printf("connector inbound supervisor: invalid %s=%q; using default %d", envConnectorInboundMaxProcessingRows, raw, defaultBudget)
		return defaultBudget, false
	}
	if value > maxConnectorInboundMaxProcessingRows {
		log.Printf("connector inbound supervisor: %s=%d exceeds max %d; clamping", envConnectorInboundMaxProcessingRows, value, maxConnectorInboundMaxProcessingRows)
		return maxConnectorInboundMaxProcessingRows, true
	}
	return value, true
}

func connectorInboundQueueProfileModeFromEnv() string {
	raw := strings.ToLower(strings.TrimSpace(os.Getenv(envConnectorInboundQueueProfile)))
	switch raw {
	case "", connectorInboundQueueProfileReserved:
		return connectorInboundQueueProfileReserved
	case connectorInboundQueueProfileDefaultHeavy:
		return connectorInboundQueueProfileDefaultHeavy
	default:
		log.Printf("connector inbound supervisor: invalid %s=%q; using %s", envConnectorInboundQueueProfile, raw, connectorInboundQueueProfileReserved)
		return connectorInboundQueueProfileReserved
	}
}

type connectorInboundQueueProfile struct {
	queueClasses          []string
	preferredQueueClasses []string
	batchSize             int
}

func connectorInboundWorkerQueueProfile(workerCount, ordinal int, mode string) connectorInboundQueueProfile {
	if workerCount <= 1 {
		return connectorInboundQueueProfile{batchSize: 100}
	}
	latencyWorkers := max(1, workerCount/4)
	if mode == connectorInboundQueueProfileDefaultHeavy {
		if ordinal < latencyWorkers {
			return connectorInboundQueueProfile{
				queueClasses: []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
				batchSize:    5,
			}
		}
		return connectorInboundQueueProfile{
			queueClasses: []string{
				connector_inbound_routing.QueueClassInteractive,
				connector_inbound_routing.QueueClassDefault,
				connector_inbound_routing.QueueClassBulk,
			},
			batchSize: 25,
		}
	}
	bulkWorkers := max(1, workerCount/4)
	if latencyWorkers+bulkWorkers > workerCount {
		bulkWorkers = max(0, workerCount-latencyWorkers)
	}
	switch {
	case ordinal < latencyWorkers:
		return connectorInboundQueueProfile{
			queueClasses: []string{connector_inbound_routing.QueueClassInteractive, connector_inbound_routing.QueueClassDefault},
			batchSize:    5,
		}
	case ordinal < latencyWorkers+bulkWorkers:
		return connectorInboundQueueProfile{
			queueClasses: []string{connector_inbound_routing.QueueClassBulk},
			batchSize:    100,
		}
	default:
		return connectorInboundQueueProfile{
			queueClasses: []string{
				connector_inbound_routing.QueueClassInteractive,
				connector_inbound_routing.QueueClassDefault,
				connector_inbound_routing.QueueClassBulk,
			},
			batchSize: 25,
		}
	}
}

func connectorInboundWorkItemWorkerQueueProfile(workerCount, ordinal int, mode string) connectorInboundQueueProfile {
	profile := connectorInboundWorkerQueueProfile(workerCount, ordinal, mode)
	if mode != connectorInboundQueueProfileDefaultHeavy || workerCount <= 1 || len(profile.queueClasses) != 3 {
		return profile
	}
	if workerCount == 2 {
		profile.preferredQueueClasses = []string{
			connector_inbound_routing.QueueClassInteractive,
			connector_inbound_routing.QueueClassDefault,
		}
		return profile
	}
	// Keep a bulk liveness floor, but leave the middle half of the fleet truly
	// shared. With 16 workers this yields 4 latency-only, 8 affinity-free, and 4
	// bulk-preferred executors instead of pinning all 12 all-class executors.
	bulkPreferredWorkers := max(1, workerCount/4)
	if ordinal >= workerCount-bulkPreferredWorkers {
		profile.preferredQueueClasses = []string{connector_inbound_routing.QueueClassBulk}
	}
	return profile
}

func connectorInboundStageTimingFromEnv() bool {
	raw := strings.TrimSpace(os.Getenv(envConnectorInboundStageTiming))
	if raw == "" {
		return false
	}
	value, err := strconv.ParseBool(raw)
	if err != nil {
		log.Printf("connector inbound supervisor: invalid %s=%q; stage timing disabled", envConnectorInboundStageTiming, raw)
		return false
	}
	return value
}

// defaultConnectorFlowDeadline bounds a hung flow by default without making the
// platform default as aggressive as a per-connector rollout knob. Most healthy
// connector inbound flows should complete far sooner; services that have proved
// a lower cap in production can tune envConnectorFlowDeadlineMS explicitly.
// Set the env var to 0 to disable.
const defaultConnectorFlowDeadline = 10 * time.Minute

// connectorFlowDeadlineFromEnv reads the per-event flow deadline (milliseconds).
// Unset / empty / invalid => defaultConnectorFlowDeadline (a protective bound, not
// off — a hung flow must not pin its lane forever even with no ops config). An
// explicit 0 or negative DISABLES the deadline. The heartbeat — not this deadline
// — prevents stale-reclaim, so the deadline is free to be generous.
func connectorFlowDeadlineFromEnv() time.Duration {
	raw := strings.TrimSpace(os.Getenv(envConnectorFlowDeadlineMS))
	if raw == "" {
		return defaultConnectorFlowDeadline
	}
	ms, err := strconv.Atoi(raw)
	if err != nil {
		log.Printf("connector inbound supervisor: invalid %s=%q; using default flow deadline %s", envConnectorFlowDeadlineMS, raw, defaultConnectorFlowDeadline)
		return defaultConnectorFlowDeadline
	}
	if ms <= 0 {
		return 0 // explicit 0 or negative disables the deadline
	}
	return time.Duration(ms) * time.Millisecond
}

func stopWorkerHandles(handles []connectorInboundWorkerHandle, drainTimeout time.Duration) {
	if len(handles) == 0 {
		return
	}
	allDone := workerHandlesDone(handles)
	for _, handle := range handles {
		if handle.requestDrain != nil {
			handle.requestDrain()
		}
	}
	if waitForWorkerHandles(allDone, drainTimeout) {
		return
	}
	for _, handle := range handles {
		if handle.forceCancel != nil {
			handle.forceCancel()
		}
	}
	<-allDone
}

func workerHandlesDone(handles []connectorInboundWorkerHandle) <-chan struct{} {
	done := make(chan struct{})
	go func() {
		for _, handle := range handles {
			<-handle.done
		}
		close(done)
	}()
	return done
}

func waitForWorkerHandles(done <-chan struct{}, timeout time.Duration) bool {
	if timeout <= 0 {
		<-done
		return true
	}
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	select {
	case <-done:
		return true
	case <-timer.C:
		return false
	}
}

func (s *Supervisor) workerKeys() []string {
	s.mu.Lock()
	defer s.mu.Unlock()

	keys := make([]string, 0, len(s.workers))
	for key := range s.workers {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	return keys
}

func (s *Supervisor) resetStarted() {
	s.mu.Lock()
	s.started = false
	s.mu.Unlock()
}

func (s *Supervisor) buildBatchClient(row packstate.ConnectorInboundWorkerPackRow, profiles []*connector_callout.ConnectorCalloutProfile, contract connectorInboundBatchContract) (connector_inbound.BatchReadClient, error) {
	engine := s.engine
	if engine == nil {
		return nil, fmt.Errorf("connector inbound: batch engine is required for %s (%s)", row.Namespace, row.InstallID)
	}
	var err error
	profiles, err = applyBatchProjectionFields(profiles, contract.projectionFields)
	if err != nil {
		return nil, err
	}
	if len(contract.requiredObjectKinds) == 0 {
		batchClient, err := s.newBatchClient(row.Namespace, engine, profiles)
		if err != nil {
			// Payload-only namespaces with no executable batch profile fall back to
			// no-op enrichment by design; that is the normal optional-enrichment path
			// and must stay silent to avoid log spam on healthy workers.
			if errors.Is(err, connector_inbound.ErrNoEligibleBatchProfiles) {
				return connector_inbound.NewNoopBatchReadClient(), nil
			}
			// Duplicate same-objectKind batch coverage should never reach the runtime:
			// install-time validation fails closed on it (validateConnectorCalloutProfileBatchCoverageOverlaps).
			// If it does reach here (installed-row drift / out-of-band state) we keep the
			// fail-safe degradation to payload-only, but it must be observable rather than
			// silent so the ownership drift can be diagnosed. The wrapped err already
			// carries the conflicting namespace/objectKind.
			if !row.DrainOnly && errors.Is(err, connector_inbound.ErrMultipleEligibleBatchProfiles) {
				log.Printf("connector inbound supervisor: %s (%s) duplicate batch object coverage at runtime; degrading to payload-only (no enrichment): %v", row.Namespace, row.InstallID, err)
				return connector_inbound.NewNoopBatchReadClient(), nil
			}
			return nil, err
		}
		return wrapDrainOnlyBatchClient(row, batchClient), nil
	}

	eligibleObjectKinds := collectEligibleBatchObjectKinds(profiles)
	if missing := sortedMissingObjectKinds(contract.requiredObjectKinds, eligibleObjectKinds); len(missing) > 0 {
		return nil, fmt.Errorf("connector inbound startup contract requires batch coverage for %s", strings.Join(missing, ", "))
	}

	requiredProfiles := filterProfilesForRequiredObjectKinds(profiles, contract.requiredObjectKinds)
	batchClient, err := s.newBatchClient(row.Namespace, engine, requiredProfiles)
	if err != nil {
		return nil, err
	}
	return wrapDrainOnlyBatchClient(row, requiredEnrichmentBatchReadClient{
		inner:               batchClient,
		requiredObjectKinds: contract.requiredObjectKinds,
	}), nil
}

func filterProfilesForRequiredObjectKinds(profiles []*connector_callout.ConnectorCalloutProfile, required map[string]struct{}) []*connector_callout.ConnectorCalloutProfile {
	if len(required) == 0 {
		return profiles
	}
	filtered := make([]*connector_callout.ConnectorCalloutProfile, 0, len(profiles))
	for _, profile := range profiles {
		if profile == nil {
			filtered = append(filtered, profile)
			continue
		}
		filteredMap := make(map[string]string)
		for objectKind, objectPath := range profile.ObjectPathMap {
			if _, ok := required[strings.TrimSpace(objectKind)]; ok {
				filteredMap[objectKind] = objectPath
			}
		}
		if len(filteredMap) == 0 {
			filteredMap = map[string]string{}
		}
		cp := *profile
		cp.ObjectPathMap = filteredMap
		filtered = append(filtered, &cp)
	}
	return filtered
}

func wrapDrainOnlyBatchClient(row packstate.ConnectorInboundWorkerPackRow, client connector_inbound.BatchReadClient) connector_inbound.BatchReadClient {
	if !row.DrainOnly {
		return client
	}
	return drainOnlyBatchReadClient{
		inner:     client,
		namespace: row.Namespace,
		installID: row.InstallID,
	}
}

type drainOnlyBatchReadClient struct {
	inner     connector_inbound.BatchReadClient
	namespace string
	installID string
}

func (c drainOnlyBatchReadClient) BatchRead(ctx context.Context, namespace, objectKind string, sourceIDs []string) (map[string]json.RawMessage, error) {
	ctx = namedcredential.WithDrainOnlyInstalledPack(ctx, c.namespace, c.installID)
	return c.inner.BatchRead(ctx, namespace, objectKind, sourceIDs)
}

func (s *Supervisor) blockDrainOnlyWorker(row packstate.ConnectorInboundWorkerPackRow, phase string, err error) {
	key := workerKey(row.Namespace, row.InstallID)
	reason := fmt.Sprintf("%s: %v", phase, err)

	s.mu.Lock()
	if s.blocked == nil {
		s.blocked = make(map[string]string)
	}
	s.blocked[key] = reason
	s.mu.Unlock()

	log.Printf("connector inbound drain worker %s (%s) blocked: %s", row.Namespace, row.InstallID, reason)
}

func companionInstallStillInstalledChecker(db *sql.DB) func(context.Context, string) (bool, error) {
	return func(ctx context.Context, companionPackInstallID string) (bool, error) {
		companionPackInstallID = strings.TrimSpace(companionPackInstallID)
		if companionPackInstallID == "" {
			return false, nil
		}
		var one int
		err := db.QueryRowContext(ctx, `
SELECT 1
  FROM _System_InstalledPack
 WHERE PackInstallId = ?
   AND Category = ?
   AND Status = ?
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`, companionPackInstallID, string(packstate.AppPack), activeInstalledPackStatus).Scan(&one)
		if err == sql.ErrNoRows {
			return false, nil
		}
		return err == nil, err
	}
}

func (s *Supervisor) workerID(namespace string) string {
	return fmt.Sprintf("connector-inbound-%s-%s-%d", namespace, s.hostname, s.pid)
}

func (s *Supervisor) workItemSchedulerOwnerID(row packstate.ConnectorInboundWorkerPackRow) string {
	return fmt.Sprintf("connector-inbound-%s-%s-%s-%d#workitem-scheduler", strings.TrimSpace(row.Namespace), strings.TrimSpace(row.InstallID), s.hostname, s.pid)
}

func (s *Supervisor) workItemSchedulerFencingToken(row packstate.ConnectorInboundWorkerPackRow) string {
	return fmt.Sprintf("connector-inbound-%s-%s-%s-%d#workitem-scheduler-fence", strings.TrimSpace(row.Namespace), strings.TrimSpace(row.InstallID), s.hostname, s.pid)
}

func (s *Supervisor) resolveWorkItemClaimGeneration(ctx context.Context, row packstate.ConnectorInboundWorkerPackRow, resolver connectorInboundWorkItemGenerationResolver) (int64, error) {
	if resolver == nil {
		return 0, nil
	}
	generation, found, err := resolver.ResolveClaimableWorkItemGeneration(connector_workqueue.WithClaimWorkTypes(ctx, []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	}), connector_workqueue.ResolveClaimableGenerationInput{
		InstalledPackID: row.InstallID,
		Generation:      0,
		Now:             time.Now().UTC(),
	})
	if err != nil {
		return 0, fmt.Errorf("connector inbound supervisor: resolve work_item generation for %s (%s): %w", row.Namespace, row.InstallID, err)
	}
	if !found {
		return 0, nil
	}
	return generation, nil
}

func workerKey(namespace, installID string) string {
	return strings.TrimSpace(namespace) + "/" + strings.TrimSpace(installID)
}

func workerFingerprint(row packstate.ConnectorInboundWorkerPackRow, projectionFingerprint string, workerCount, maxProcessingRows int, explicitProcessingBudget bool, queueProfileMode string) string {
	return strings.Join([]string{
		strings.TrimSpace(row.Namespace),
		strings.TrimSpace(row.InstallID),
		strings.TrimSpace(row.Status),
		strings.TrimSpace(row.PayloadHash),
		fmt.Sprintf("%t", row.DrainOnly),
		strings.TrimSpace(projectionFingerprint),
		strconv.Itoa(workerCount),
		strconv.Itoa(maxProcessingRows),
		fmt.Sprintf("%t", explicitProcessingBudget),
		strings.TrimSpace(queueProfileMode),
	}, "\x00")
}