package connector_inbound

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"reflect"
	"runtime/debug"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	flowcommandv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/stagetiming"
	zlog "github.com/rs/zerolog/log"
)

const (
	defaultPollInterval            = 5 * time.Second
	defaultCentralClaimRefillDelay = 200 * time.Millisecond
	defaultFinalizeTimeout         = 5 * time.Second
	defaultDrainBatchSize          = 100
)

// ErrInvalidWorkerConfig is returned when required WorkerConfig fields are
// missing.
var ErrInvalidWorkerConfig = errors.New("connector_inbound: invalid worker config")

// ErrBatchReadRecordShape signals that a record returned by BatchRead failed
// shape validation (the JSON bytes could not be decoded into map[string]any).
// This is an INT-10 boundary fault — the fault is at the batch-read boundary,
// NOT in the original webhook payload. Callers MUST treat this as retriable
// (failWithSyncState) rather than terminal (failTerminal), because the
// dispatcher already validated the parser payload shape before enqueue, so any
// shape failure at this stage is outside the parser/dispatcher boundary.
var ErrBatchReadRecordShape = errors.New("connector_inbound: batch read returned non-object record")

// WorkerConfig is per-pack-incarnation config. The application services layer
// may construct one or more [Worker] instances for a (Namespace, PackInstallID)
// pair after discovering active incarnations via
// [connector_events.Service.ListClaimablePackIncarnations].
type WorkerConfig struct {
	// Namespace is the connector pack's namespace (e.g. "crm-ns"). Required.
	Namespace string

	// PackInstallID is the value of _System_InstalledPack.PackInstallId for this
	// pack incarnation (fresh UUID minted on every install/rebirth).
	// Passed to [EventsService.ClaimNext] so workers for a reborn pack cannot
	// drain events from an earlier incarnation sharing the same namespace.
	// Obtain via [packs.ResolveConnectorPackByNamespace] and use the returned
	// ConnectorPackRow.InstallID. Do NOT pass the stable Id column — events are
	// queued and claimed by incarnation (PackInstallId), not by stable Id.
	// Required.
	PackInstallID string

	// WorkerID is an opaque identifier for this worker instance (e.g.
	// hostname+pid). Written to the ClaimedBy column for observability. Required.
	WorkerID string

	// PollInterval is the sleep duration between empty-queue polls. Zero defaults
	// to 5 s. Positive values are used as-is; set to a small value in tests.
	PollInterval time.Duration

	// MaxClaims, when > 0, causes [Worker.Run] to return nil after processing
	// exactly MaxClaims events. Zero means unlimited — suitable for production.
	// Set > 0 in tests to bound execution.
	//
	// Note: the counter tracks CLAIMED events (including ones that fail
	// prepare/exec), matching the single-claim semantics — MaxClaims is a
	// claim ceiling, not a success ceiling.
	MaxClaims int

	// FinalizeTimeout bounds MarkProcessed/MarkFailed writes. Finalization uses
	// a fresh context so shutdown cancellation cannot strand an event after Flow
	// side effects have completed. Zero defaults to 5 s.
	FinalizeTimeout time.Duration

	// BatchSize is the maximum number of events claimed per loop iteration.
	// Zero defaults to defaultDrainBatchSize (100). Positive values are used
	// as-is; set to a small value in tests to bound execution per iteration.
	BatchSize int

	// QueueClasses, when non-empty, restricts this worker to route-declared queue
	// classes. Empty means all classes for backwards-compatible callers.
	QueueClasses []string

	// ClaimLimiter, when set, runs before any queue row is claimed. It is a
	// side-effect-free pressure gate over DB pool stats and live connector-event
	// pressure, so overloaded workers back off before mutating queue leases.
	ClaimLimiter ClaimLimiter

	// EnableStageTiming emits per-event connector inbound stage timing logs.
	EnableStageTiming bool

	// LeaseHeartbeatInterval is the period at which the worker renews the lease of
	// each in-flight claimed event (RenewLease bumps LastModifiedDate) so a
	// slow-but-alive flow keeps its row past the StaleProcessingSeconds reclaim
	// cutoff and is never double-executed by stale-reclaim. Zero defaults to
	// StaleProcessingSeconds/3 (~100 s for the 300 s cutoff). The heartbeat is the
	// PRIMARY double-exec fix and is always on (it needs no operator tuning); it is
	// a no-op only when the injected EventsService does not implement RenewLease.
	LeaseHeartbeatInterval time.Duration

	// FlowDeadline is the OPT-IN, DEFAULT-OFF per-event flow execution deadline.
	// Zero disables the deadline — a legitimately slow-but-completing call is never
	// interrupted. When > 0, BOTH the flow span AND each post-claim BatchRead run
	// under context.WithTimeout(FlowDeadline); on timeout the event is released via
	// ReleaseClaimForRetry (NOT MarkFailed — a hung-but-not-dead call must not burn
	// the dead-letter budget), so a wedged read self-heals on backoff exactly like a
	// hung flow. The deadline exists because a heartbeated post-claim call is no
	// longer recoverable by stale-reclaim — without it a wedged call would pin its
	// lane forever. The supervisor injects a generous default
	// (defaultConnectorFlowDeadline, 30min) from NEXUSCRM_CONNECTOR_FLOW_DEADLINE_MS;
	// far above any real runtime, so it only fires for a genuinely wedged call.
	FlowDeadline time.Duration

	// ClaimPath configures the retained in-package legacy test harness. Production
	// construction is exclusively [NewWorkItemWorker], which fixes the claim path
	// to WorkItems and rejects these split fields. Production callers must not run
	// legacy Event claimers and WorkItem claimers for the same pack/generation.
	ClaimPath ClaimPath

	WorkItemClaims      WorkItemClaimService
	WorkItemEventLoader WorkItemEventLoader
	WorkItemEventBridge WorkItemEventBridge
	WorkItemGeneration  int64
	// WorkItemSchedulerOwnerID and WorkItemSchedulerFencingToken retain the
	// stable instance-scoped identity forwarded to published/custom claim and
	// bridge extensions. In-tree workers do not acquire SchedulerLease; the
	// repository uses WorkItem row CAS and LeaseToken as execution authority.
	WorkItemSchedulerOwnerID      string
	WorkItemSchedulerFencingToken string
	WorkItemLeaseDuration         time.Duration
}

type IdentityConvergenceProbe struct {
	Namespace  string
	ObjectKind string
	SourceID   string
	Dependency models.FlowExecutionDependency
}

// IdentityConvergenceResult names which authority became observable. Exact
// record commits wake every waiter for the same selector. Remote-identity
// convergence wakes only the immutable connector source locator; it must not
// grant authority to a different source that happens to await the same record.
type IdentityConvergenceResult struct {
	ExactRecord    bool
	RemoteIdentity bool
}

func (r IdentityConvergenceResult) Converged() bool {
	return r.ExactRecord || r.RemoteIdentity
}

type IdentityConvergenceResolver interface {
	ResolveIdentityConvergence(ctx context.Context, probe IdentityConvergenceProbe) (IdentityConvergenceResult, error)
}

type connectorInboundExecutionTiming struct {
	startedAt       time.Time
	buildInputsMS   int64
	flowCallMS      int64
	finalizeMS      int64
	heartbeatStopMS int64
	flowPath        string
	batchGroupSize  int
}

type connectorInboundWorkItemTraceResolver interface {
	WorkItemIDForEvent(eventID string) (string, bool)
}

type connectorInboundWorkItemAttemptTraceResolver interface {
	WorkItemAttemptForEvent(eventID string) (int, bool)
}

type connectorInboundExecutionTimingKey struct{}

func withConnectorInboundExecutionTiming(ctx context.Context, timing *connectorInboundExecutionTiming) context.Context {
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	if timing == nil {
		return ctx
	}
	return context.WithValue(ctx, connectorInboundExecutionTimingKey{}, timing)
}

func connectorInboundExecutionTimingFromContext(ctx context.Context) *connectorInboundExecutionTiming {
	if ctx == nil {
		return nil
	}
	timing, _ := ctx.Value(connectorInboundExecutionTimingKey{}).(*connectorInboundExecutionTiming)
	return timing
}

func connectorInboundTimingMillis(started time.Time) int64 {
	if started.IsZero() {
		return 0
	}
	return int64(time.Since(started) / time.Millisecond)
}

// EventsService is the worker's narrow view of [connector_events.Service].
// Defined here (not imported as concrete) so [worker_test.go] can substitute
// a simple struct mock without sqlmock, and so the worker package avoids
// importing the full connector_events surface.
type EventsService interface {
	ClaimNext(ctx context.Context, installedPackID, workerID string) (*connector_events.ConnectorEvent, bool, error)
	PeekBatch(ctx context.Context, installedPackID string, n int) ([]*connector_events.ConnectorEvent, error)
	ClaimByID(ctx context.Context, eventID, workerID, installedPackID string, deliveryID sql.NullString, deliverySequence int) (string, time.Time, bool, error)
	MarkProcessed(ctx context.Context, eventID, claimToken string) error
	MarkSkipped(ctx context.Context, eventID, claimToken string) error
	MarkFailed(ctx context.Context, eventID, claimToken, errorMsg string) error
}

type identityConvergenceBlockingEventsService interface {
	SupportsIdentityConvergenceBlocking() bool
	BlockOnIdentityConvergence(ctx context.Context, eventID, claimToken string, probe IdentityConvergenceProbe) error
}

type peekedCandidateForgetter interface {
	ForgetPeekedCandidate(eventID string)
}

type ClaimPath string

const (
	ClaimPathLegacyEvent ClaimPath = "legacy_event"
	ClaimPathWorkItem    ClaimPath = "work_item"
)

type WorkItemClaimService interface {
	// Deprecated: row-CAS claim services are not called through this hook.
	AcquireSchedulerLease(ctx context.Context, in connector_workqueue.SchedulerLeaseInput) (bool, error)
	// Deprecated: row-CAS claim services are not called through this hook.
	RenewSchedulerLease(ctx context.Context, in connector_workqueue.SchedulerLeaseInput) (bool, error)
	// ClaimNextWorkItem must honor connector_workqueue.ClaimWorkTypes(ctx).
	// Returning a WorkItem outside a non-empty executor scope violates the
	// claim contract and can starve work the executor is able to run.
	ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error)
	CompleteWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error)
	RetryWorkItem(ctx context.Context, workID, leaseToken string, runAfter, now time.Time) (bool, error)
	RenewWorkItemLease(ctx context.Context, workID, leaseToken string, leaseUntil, now time.Time) (bool, error)
}

// WorkItemClaimInputService is the single admitted boundary for WorkItem
// selection and immutable Event input. HonorsClaimWorkTypes is an explicit
// promise because ignoring the request-local executor scope can hide runnable
// work behind a WorkType this worker cannot execute.
//
// Mutable failure, retry, lease, terminal, and coalescing transitions remain on
// the existing claim/bridge surfaces until their focused cutover phase.
type WorkItemClaimInputService interface {
	HonorsClaimWorkTypes() bool
	ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error)
	PeekClaimableWorkItems(ctx context.Context, in connector_workqueue.ClaimWorkItemInput, limit int) ([]*connector_workqueue.WorkItem, error)
	ClaimWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error)
	ClaimStaleWorkItemByEvent(ctx context.Context, in connector_workqueue.ClaimWorkItemByEventInput) (*connector_workqueue.WorkItem, bool, error)
	ResolveClaimableWorkItemGeneration(ctx context.Context, in connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error)
	LoadEventForWorkItem(ctx context.Context, eventID string) (*connector_events.ConnectorEvent, error)
	LoadEventsForWorkItems(ctx context.Context, eventIDs []string) (map[string]*connector_events.ConnectorEvent, error)
	LoadSourceProjectionPayloads(ctx context.Context, workID string) ([]*connector_events.ConnectorEvent, error)
}

// WorkItemService is the mandatory production execution boundary. One service
// owns selection, immutable Event input, failure budgeting, lease and terminal
// transitions, and atomic coalescing. It deliberately excludes SchedulerLease
// and Event mutation hooks: WorkItem row CAS is the live executor authority.
type WorkItemService interface {
	WorkItemClaimInputService
	FailWorkItem(ctx context.Context, in connector_workqueue.FailWorkItemInput) (*connector_workqueue.FailWorkItemResult, bool, error)
	CompleteWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error)
	RetryWorkItem(ctx context.Context, workID, leaseToken string, runAfter, now time.Time) (bool, error)
	RenewWorkItemLease(ctx context.Context, workID, leaseToken string, leaseUntil, now time.Time) (bool, error)
	ClaimNextCoalescedExecutorWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error)
	ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error)
}

// WorkItemCutoverVerifier is required by the latest WorkItem constructor. A
// caller must pass the activation context that owns cancellation, so receipt
// verification cannot outlive worker startup or a reconcile shutdown.
type WorkItemCutoverVerifier interface {
	RequireWorkItemOnlyExecutionCutover(context.Context) error
	RequireSourceProjectionProtocol(context.Context, string) error
}

// WorkItemRowCASClaimAuthority is an optional capability for claim services
// whose WorkItem row transition is already the sole distributed authority.
// Workers skip legacy SchedulerLease hooks only when this capability is
// explicitly present and enabled. Existing custom services retain the former
// lease-hook lifecycle until they adopt this interface.
type WorkItemRowCASClaimAuthority interface {
	UsesWorkItemRowCASClaimAuthority() bool
}

// WorkItemExecutorStateAuthority is an optional capability for claim services
// whose WorkItem row owns processing, retry, failure, lease, and terminal
// state. The in-tree repository enables it; custom services retain the legacy
// Event-mirror adapter until they explicitly adopt this contract.
type WorkItemExecutorStateAuthority interface {
	UsesWorkItemExecutorStateAuthority() bool
}

type WorkItemClaimGenerationResolver interface {
	// ResolveClaimableWorkItemGeneration must use the same request-local
	// connector_workqueue.ClaimWorkTypes scope as the subsequent claim.
	ResolveClaimableWorkItemGeneration(ctx context.Context, in connector_workqueue.ResolveClaimableGenerationInput) (int64, bool, error)
}

type WorkItemEventLoader interface {
	LoadEventForWorkItem(ctx context.Context, eventID string) (*connector_events.ConnectorEvent, error)
	LoadSourceProjectionPayloads(ctx context.Context, workID string) ([]*connector_events.ConnectorEvent, error)
}

type WorkItemEventBridge interface {
	MarkEventProcessingForWorkItem(ctx context.Context, eventID, workerID, leaseToken string, claimedAt, now time.Time) error
	ClaimNextCoalescedWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*connector_events.WorkItemCoalescedSiblingClaim, bool, error)
	MarkEventTerminalForWorkItem(ctx context.Context, eventID, leaseToken, terminalStatus, errorMsg string, now time.Time) error
	ReleaseEventForWorkItemRetry(ctx context.Context, eventID, leaseToken, errorMsg string, runAfter, now time.Time, retryCount int) error
	RenewEventLeaseForWorkItem(ctx context.Context, eventID, leaseToken string, now time.Time) error
}

// scheduler runs the connector drain loop for one pack incarnation.
//
// The default scheduler preserves the historical worker behavior: each worker
// asks for a claim permit, claims at most one execution unit, runs it through the
// injected executor, releases the permit, then backs off on empty/error polls.
// Later central-claim-loop work can replace this scheduler without changing the
// event execution adapter.
type scheduler interface {
	run(ctx context.Context, drain <-chan struct{}, executor executor) error
}

type wakeableScheduler interface {
	Wake()
}

// schedulerConfig is the worker state a scheduler needs to preserve existing
// drain-loop behavior without reaching into Worker internals.
type schedulerConfig struct {
	MaxClaims             int
	BatchSize             int
	PollInterval          time.Duration
	ClaimPath             ClaimPath
	QueueClasses          []string
	PreferredQueueClasses []string
}

// claimExecutionRequest is a scheduler-owned claim attempt.
type claimExecutionRequest struct {
	MaxClaims        int
	StaleReclaimOnly bool
	Event            *connector_events.ConnectorEvent
	Candidate        schedulerClaimCandidate
	HasCandidate     bool
}

// claimExecutionResult reports how the executor resolved one claim attempt.
type claimExecutionResult struct {
	Claimed        bool
	ClaimedRows    int
	ClaimAttempted bool
	ClaimLatency   time.Duration
	ClaimEligible  time.Time
	ClaimErr       error
	ProcessingErr  error
}

// executor is the scheduler's adapter to existing worker behavior.
//
// The current Worker implementation still performs the actual ClaimNext,
// PeekBatch, ClaimByID, and event execution. The interface makes that boundary
// explicit so a later scheduler can centralize claim orchestration while reusing
// the execution path for already-claimed work.
type executor interface {
	schedulerConfig() schedulerConfig
	requestClaimPermit(ctx context.Context, drain <-chan struct{}, batchSize int) (ClaimLimiterDecision, bool, error)
	requestClaimPermitForScheduler(ctx context.Context, batchSize int) (ClaimLimiterDecision, bool, time.Duration, error)
	claimNextForScheduler(ctx context.Context, staleOnly bool) (*connector_events.ConnectorEvent, bool, error)
	executeClaim(ctx context.Context, drain <-chan struct{}, request claimExecutionRequest) (claimExecutionResult, error)
	releaseClaimPermitFromScheduler(decision ClaimLimiterDecision, claimedRows int)
}

type schedulerClaimCandidate struct {
	event              *connector_events.ConnectorEvent
	workID             string
	workType           string
	queueClass         string
	generation         int64
	generationResolved bool
	staleOnly          bool
	eligibleAt         time.Time
}

type schedulerBatchCandidateExecutor interface {
	peekClaimCandidatesForScheduler(ctx context.Context, limit int, staleOnly bool) ([]schedulerClaimCandidate, error)
	claimByIDForScheduler(ctx context.Context, candidate schedulerClaimCandidate) (*connector_events.ConnectorEvent, bool, error)
}

type schedulerCandidatePermitSizer interface {
	claimPermitBatchSizeForScheduler(ctx context.Context, candidate schedulerClaimCandidate) int
}

type compatibilityScheduler struct{}

type CentralClaimSchedulerStats struct {
	StartedAt                     time.Time
	RegisteredWorkers             int
	IdleWorkers                   int
	BusyWorkers                   int
	ClaimAttempts                 int64
	ClaimSuccess                  int64
	ClaimEmpty                    int64
	ClaimErrors                   int64
	ClaimCASLost                  int64
	PeekCandidateCount            int64
	PeekErrors                    int64
	PermitDenied                  int64
	PermitErrors                  int64
	ClaimLatencySamples           int64
	ClaimLatencyTotal             time.Duration
	EligibleToClaimLatencySamples int64
	EligibleToClaimLatencyTotal   time.Duration
	EligibleToClaimLatencyMax     time.Duration
}

// CentralClaimScheduler coordinates one claim loop for a pack incarnation while
// still letting multiple worker executors process already-claimed events. It is
// intentionally per-pack: callers should construct one scheduler for the set of
// workers created for the same Namespace/PackInstallID and inject that same
// scheduler into every worker in that set.
type CentralClaimScheduler struct {
	mu    sync.Mutex
	regs  []*centralClaimRegistration
	next  int
	stats CentralClaimSchedulerStats

	running          bool
	done             chan struct{}
	wakeCh           chan struct{}
	resultCh         chan centralClaimResult
	err              error
	dispatchLogCount int
}

type centralClaimRegistration struct {
	executor             executor
	work                 chan centralClaimWork
	resultCh             chan centralClaimResult
	busy                 bool
	closed               bool
	processed            int
	nextPoll             time.Time
	inflightCandidateKey string
}

type centralClaimWork struct {
	permit            ClaimLimiterDecision
	additionalPermits []ClaimLimiterDecision
	request           claimExecutionRequest
}

type centralClaimResult struct {
	reg        *centralClaimRegistration
	result     claimExecutionResult
	panicValue any
}

var errCentralClaimSchedulerPanic = errors.New("central claim scheduler panic")

// NewCentralClaimScheduler returns an empty scheduler. It starts its coordinator
// lazily on the first RunClaimFirstUntilDrained call that registers an executor.
func NewCentralClaimScheduler() *CentralClaimScheduler {
	return &CentralClaimScheduler{}
}

func (s *CentralClaimScheduler) Stats() CentralClaimSchedulerStats {
	if s == nil {
		return CentralClaimSchedulerStats{}
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	stats := s.stats
	stats.RegisteredWorkers = len(s.regs)
	stats.IdleWorkers = 0
	stats.BusyWorkers = 0
	for _, reg := range s.regs {
		if reg == nil || reg.closed {
			continue
		}
		if reg.busy {
			stats.BusyWorkers++
		} else {
			stats.IdleWorkers++
		}
	}
	return stats
}

// Wake asks a running scheduler to re-check claimable work without waiting for
// the next poll tick. It is intentionally best-effort: callers use it as a
// backstop when external materialization makes new lane heads visible.
func (s *CentralClaimScheduler) Wake() {
	_ = s.WakeIfRunning()
}

func (s *CentralClaimScheduler) WakeIfRunning() bool {
	if s == nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.running {
		return false
	}
	s.clearDeferredRegistrationsLocked()
	hasLiveRegistration := false
	for _, reg := range s.regs {
		if reg != nil && !reg.closed {
			hasLiveRegistration = true
			break
		}
	}
	if !hasLiveRegistration {
		return false
	}
	s.signalWakeLocked()
	return true
}

type classAwareEventsService interface {
	ClaimNextForClasses(ctx context.Context, installedPackID, workerID string, queueClasses []string) (*connector_events.ConnectorEvent, bool, error)
	PeekBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*connector_events.ConnectorEvent, error)
}

// staleReclaimingEventsService is the optional capability the worker uses under a
// limp permit (ClaimLimiterDecision.StaleReclaimOnly): claim/peek ONLY a
// stale-processing row so a degraded tick recovers a stranded lease without
// admitting a fresh pending/failed row past the cap. The production
// connector_events.Service implements it; if the injected service does not, the
// worker reclaims nothing this tick (it never falls back to a general claim,
// which could over-admit).
type staleReclaimingEventsService interface {
	ReclaimStaleNext(ctx context.Context, installedPackID, workerID string) (*connector_events.ConnectorEvent, bool, error)
	ReclaimStaleNextForClasses(ctx context.Context, installedPackID, workerID string, queueClasses []string) (*connector_events.ConnectorEvent, bool, error)
	PeekStaleBatch(ctx context.Context, installedPackID string, n int) ([]*connector_events.ConnectorEvent, error)
	PeekStaleBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*connector_events.ConnectorEvent, error)
}

type sourceLaneCoalescingEventsService interface {
	ClaimNextCoalescedSibling(ctx context.Context, currentEventID, currentClaimToken, workerID string) (*connector_events.ConnectorEvent, bool, error)
}

type sourceScopedLaneCoalescingEventsService interface {
	ClaimNextCoalescedSiblingSourceScoped(ctx context.Context, currentEventID, currentClaimToken, workerID string) (*connector_events.ConnectorEvent, bool, error)
}

// settledProjectionPayloadEventsService restores the source-projection
// generation for a durable WorkItem. The WorkItem keeps only its selected
// EventId, while preserve-payload Flows must still receive every raw Event
// collapsed since the prior terminal source projection, across deliveries.
type settledProjectionPayloadEventsService interface {
	LoadSettledProjectionPayloads(ctx context.Context, eventID string) ([]*connector_events.ConnectorEvent, error)
}

type claimReleasingEventsService interface {
	ReleaseClaim(ctx context.Context, eventID, claimToken string) error
	// ReleaseClaimForRetry returns the leased row for a backed-off re-drive
	// without incrementing RetryCount. Used for a retriable flow fault
	// (IsRetriable on a failed result) so a concurrent-loser is re-driven to
	// self-heal rather than dead-lettered on the genuine-failure budget.
	ReleaseClaimForRetry(ctx context.Context, eventID, claimToken, reason string) error
}

// SyncStateService is the worker's narrow view of connector_syncstate.Service.
// Defined here for the same mockability reason as [EventsService].
type SyncStateService interface {
	RecordError(ctx context.Context, namespace, object, direction, errorMsg string) error
}

// Worker claims events from the connector_events queue, resolves the target
// AutoLaunched Flow via the injected [RouteResolver], batch-fetches detail
// records via [BatchReadClient], invokes the Flow, and acks or fails the event.
//
// Construct production workers with [NewWorkItemWorker]. The zero value is not
// usable.
type Worker struct {
	cfg WorkerConfig
	// workItemPreferredQueueClasses is a non-durable scheduler hint set through
	// SetWorkItemQueueClassPreference. It deliberately stays outside WorkerConfig
	// so published callers using unkeyed WorkerConfig literals remain compatible.
	workItemPreferredQueueClasses []string
	events                        EventsService
	syncstate                     SyncStateService
	batchRead                     BatchReadClient
	flow                          FlowInvoker
	runtimeCommands               RuntimeCommandHandoff
	resolver                      connector_inbound_routing.RouteResolver
	// outboundEchoProbe optionally annotates inbound payloads with durable
	// outbound push evidence before Flow execution. It is provider-neutral:
	// implementations inspect recent finalized connector-push attempts for the
	// same remote object/property.
	outboundEchoProbe OutboundEchoProbe
	// checkCompanionInstalled verifies that the frozen companion app-pack
	// incarnation (PackInstallId) is still active at drain time. Returns
	// (true, nil) only for the current active row, (false, nil) when
	// absent/inactive/deleted, and (false, err) on DB error.
	// If not set (nil), cross-pack companion events FAIL CLOSED (failTerminal)
	// rather than executing without the incarnation re-check.
	// Same-pack/legacy events (CompanionFlowVersionID=="") never call this.
	checkCompanionInstalled  func(ctx context.Context, companionPackInstallID string) (bool, error)
	scheduler                scheduler
	schedulerPermitModeCache sync.Map
}

// NewWorkItemWorker constructs the production WorkItem executor. The supplied
// service is complete by contract and is the only queue dependency used by the
// live adapter; split WorkItem and legacy Event mutation dependencies are
// rejected. The latest constructor also requires a receipt verifier; custom
// services from the superseded v0.1.x contract are intentionally unsupported.
func NewWorkItemWorker(
	ctx context.Context,
	cfg WorkerConfig,
	workItems WorkItemService,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
) (*Worker, error) {
	return newWorkItemWorker(ctx, cfg, workItems, syncstate, batchRead, flow, resolver, true)
}

// NewRuntimeCommandHandoffWorker constructs the independent Integration
// execution mode. It reuses the same durable WorkItem, route, enrichment,
// heartbeat, retry, and terminal owners, but transfers the prepared operation
// through Runtime.AcceptCommand instead of invoking an in-process Flow engine.
func NewRuntimeCommandHandoffWorker(
	ctx context.Context,
	cfg WorkerConfig,
	workItems WorkItemService,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	runtimeCommands RuntimeCommandHandoff,
	resolver connector_inbound_routing.RouteResolver,
) (*Worker, error) {
	if ctx == nil {
		return nil, fmt.Errorf("%w: activation context is nil", ErrInvalidWorkerConfig)
	}
	if workItemClaimInputServiceIsNil(workItems) {
		return nil, fmt.Errorf("%w: workItems is nil", ErrInvalidWorkerConfig)
	}
	if !workItems.HonorsClaimWorkTypes() {
		return nil, fmt.Errorf("%w: WorkItem service must honor ClaimWorkTypes", ErrInvalidWorkerConfig)
	}
	cutover, ok := workItems.(WorkItemCutoverVerifier)
	if !ok {
		return nil, fmt.Errorf("%w: WorkItem service must verify the WorkItem-only cutover receipt", ErrInvalidWorkerConfig)
	}
	if err := cutover.RequireWorkItemOnlyExecutionCutover(ctx); err != nil {
		return nil, fmt.Errorf("%w: verify WorkItem-only cutover receipt: %w", ErrInvalidWorkerConfig, err)
	}
	if err := cutover.RequireSourceProjectionProtocol(ctx, cfg.PackInstallID); err != nil {
		return nil, fmt.Errorf("%w: verify source projection protocol: %w", ErrInvalidWorkerConfig, err)
	}
	if runtimeCommands == nil {
		return nil, fmt.Errorf("%w: Runtime command handoff is nil", ErrInvalidWorkerConfig)
	}
	// Every neutral Runtime command request requires a caller deadline. Keep the
	// legacy in-process Flow worker's opt-in/default-off behavior unchanged, but
	// fail this cross-process mode closed before it can claim work when its
	// existing per-event deadline is disabled.
	if cfg.FlowDeadline <= 0 {
		return nil, fmt.Errorf(
			"%w: Runtime command handoff requires a positive FlowDeadline",
			ErrInvalidWorkerConfig,
		)
	}
	if err := runtimeCommands.VerifyReady(ctx); err != nil {
		return nil, fmt.Errorf(
			"%w: verify Runtime command handoff readiness: %w",
			ErrInvalidWorkerConfig,
			err,
		)
	}
	var err error
	cfg, err = normalizeWorkerConfigWithoutFlow(cfg, syncstate, batchRead, resolver)
	if err != nil {
		return nil, err
	}
	if err := validateMandatoryWorkItemConfig(cfg); err != nil {
		return nil, err
	}
	cfg.ClaimPath = ClaimPathWorkItem
	events := newMandatoryWorkItemEventsService(workItems, cfg)
	worker := buildWorker(cfg, events, syncstate, batchRead, nil, resolver)
	worker.runtimeCommands = runtimeCommands
	return worker, nil
}

func newWorkItemWorker(
	ctx context.Context,
	cfg WorkerConfig,
	workItems WorkItemService,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
	requireCutoverReceipt bool,
) (*Worker, error) {
	if ctx == nil {
		return nil, fmt.Errorf("%w: activation context is nil", ErrInvalidWorkerConfig)
	}
	if workItemClaimInputServiceIsNil(workItems) {
		return nil, fmt.Errorf("%w: workItems is nil", ErrInvalidWorkerConfig)
	}
	if !workItems.HonorsClaimWorkTypes() {
		return nil, fmt.Errorf("%w: WorkItem service must honor ClaimWorkTypes", ErrInvalidWorkerConfig)
	}
	if requireCutoverReceipt {
		cutover, ok := workItems.(WorkItemCutoverVerifier)
		if !ok {
			return nil, fmt.Errorf("%w: WorkItem service must verify the WorkItem-only cutover receipt", ErrInvalidWorkerConfig)
		}
		if err := cutover.RequireWorkItemOnlyExecutionCutover(ctx); err != nil {
			return nil, fmt.Errorf("%w: verify WorkItem-only cutover receipt: %w", ErrInvalidWorkerConfig, err)
		}
		if err := cutover.RequireSourceProjectionProtocol(ctx, cfg.PackInstallID); err != nil {
			return nil, fmt.Errorf("%w: verify source projection protocol: %w", ErrInvalidWorkerConfig, err)
		}
	}
	var err error
	cfg, err = normalizeWorkerConfig(cfg, syncstate, batchRead, flow, resolver)
	if err != nil {
		return nil, err
	}
	if err := validateMandatoryWorkItemConfig(cfg); err != nil {
		return nil, err
	}
	cfg.ClaimPath = ClaimPathWorkItem
	events := newMandatoryWorkItemEventsService(workItems, cfg)
	return buildWorker(cfg, events, syncstate, batchRead, flow, resolver), nil
}

func normalizeWorkerConfig(
	cfg WorkerConfig,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
) (WorkerConfig, error) {
	if flow == nil {
		return WorkerConfig{}, fmt.Errorf("%w: flow is nil", ErrInvalidWorkerConfig)
	}
	return normalizeWorkerConfigWithoutFlow(cfg, syncstate, batchRead, resolver)
}

func normalizeWorkerConfigWithoutFlow(
	cfg WorkerConfig,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	resolver connector_inbound_routing.RouteResolver,
) (WorkerConfig, error) {
	if cfg.Namespace == "" {
		return WorkerConfig{}, fmt.Errorf("%w: Namespace is empty", ErrInvalidWorkerConfig)
	}
	if cfg.PackInstallID == "" {
		return WorkerConfig{}, fmt.Errorf("%w: PackInstallID is empty", ErrInvalidWorkerConfig)
	}
	if cfg.WorkerID == "" {
		return WorkerConfig{}, fmt.Errorf("%w: WorkerID is empty", ErrInvalidWorkerConfig)
	}
	if syncstate == nil {
		return WorkerConfig{}, fmt.Errorf("%w: syncstate is nil", ErrInvalidWorkerConfig)
	}
	if resolver == nil {
		return WorkerConfig{}, fmt.Errorf("%w: resolver is nil", ErrInvalidWorkerConfig)
	}

	if batchRead == nil {
		return WorkerConfig{}, fmt.Errorf("%w: batchRead is required (production wiring must inject a real BatchReadClient)", ErrInvalidWorkerConfig)
	}

	if cfg.PollInterval == 0 {
		cfg.PollInterval = defaultPollInterval
	}
	if cfg.BatchSize <= 0 {
		cfg.BatchSize = defaultDrainBatchSize
	}
	if cfg.FinalizeTimeout < 0 {
		return WorkerConfig{}, fmt.Errorf("%w: FinalizeTimeout is negative", ErrInvalidWorkerConfig)
	}
	if cfg.FinalizeTimeout == 0 {
		cfg.FinalizeTimeout = defaultFinalizeTimeout
	}
	return cfg, nil
}

func buildWorker(
	cfg WorkerConfig,
	events EventsService,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
) *Worker {
	return &Worker{
		cfg:       cfg,
		events:    events,
		syncstate: syncstate,
		batchRead: batchRead,
		flow:      flow,
		resolver:  resolver,
	}
}

// SetCompanionInstalledChecker injects the drain-time companion incarnation
// re-check function. The function must query _System_InstalledPack to verify
// that the frozen PackInstallId is still active. When absent, inactive,
// deleted, or DB-error, processOneWithDrain calls failTerminal.
// If not set (nil), cross-pack companion events FAIL CLOSED (failTerminal)
// rather than executing without the incarnation re-check.
// Same-pack/legacy events (CompanionFlowVersionID=="") never call this.
//
// Production wiring: call this from the application services layer after
// worker construction, passing a closure over a live *sql.DB.
func (w *Worker) SetCompanionInstalledChecker(fn func(ctx context.Context, companionPackInstallID string) (bool, error)) {
	w.checkCompanionInstalled = fn
}

// SetOutboundEchoProbe injects optional outbound-push evidence used by flows
// that need to distinguish provider-authored deltas from delayed echoes of a
// Nexus-originated connector push. Nil disables enrichment.
func (w *Worker) SetOutboundEchoProbe(probe OutboundEchoProbe) {
	w.outboundEchoProbe = probe
}

// UseCentralClaimScheduler makes this worker consume work from a scheduler-owned
// claim path instead of independently scanning the connector event queue.
func (w *Worker) UseCentralClaimScheduler(s *CentralClaimScheduler) {
	w.scheduler = s
}

// SetWorkItemQueueClassPreference sets this WorkItem worker's optional
// first-choice queue-class tier. QueueClasses continues to define eligibility;
// preference is a non-durable, work-conserving hint and is ignored for stale
// reclaim. Call it during worker construction, before Run.
func (w *Worker) SetWorkItemQueueClassPreference(queueClasses []string) {
	w.workItemPreferredQueueClasses = append([]string(nil), queueClasses...)
	if service, ok := w.events.(*workItemEventsService); ok {
		service.setPreferredQueueClasses(queueClasses)
	}
}

// UseDistributedWorkItemClaims preserves the published opt-in for custom claim
// services that already use WorkItem row CAS and lane-head ordering without
// implementing WorkItemRowCASClaimAuthority. Repository-backed workers enable
// the same behavior through that capability automatically.
//
// Deprecated: new claim services should implement
// WorkItemRowCASClaimAuthority.
func (w *Worker) UseDistributedWorkItemClaims() {
	if service, ok := w.events.(*workItemEventsService); ok {
		service.distributedClaims = true
	}
}

// DistributedWorkItemClaimsEnabled reports whether this worker's claim service
// uses WorkItem row CAS as its distributed authority, either through the
// capability contract or the retained compatibility opt-in.
func (w *Worker) DistributedWorkItemClaimsEnabled() bool {
	service, ok := w.events.(*workItemEventsService)
	return ok && service.distributedClaimsEnabled()
}

// WorkItemQueueClassPreference returns this worker's configured first-choice
// queue-class tier. The returned slice may be modified by the caller.
func (w *Worker) WorkItemQueueClassPreference() []string {
	return append([]string(nil), w.workItemPreferredQueueClasses...)
}

// NewWorkItemWorkerForTest is the test-only nil-BatchRead convenience for the
// mandatory WorkItem constructor. It keeps the same receipt contract as
// [NewWorkItemWorker]; production wiring must use the latter.
func NewWorkItemWorkerForTest(
	ctx context.Context,
	cfg WorkerConfig,
	workItems WorkItemService,
	syncstate SyncStateService,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
) (*Worker, error) {
	if batchRead == nil {
		batchRead = noopBatchReadClient{}
	}
	return newWorkItemWorker(ctx, cfg, workItems, syncstate, batchRead, flow, resolver, true)
}

// Run polls the event queue until ctx is cancelled (or cfg.MaxClaims > 0 events
// have been processed). Returns ctx.Err() on cancellation.
//
// Processing errors from individual events are best-effort logged via the
// returned error from [processOne] — Run does not abort on per-event failures.
func (w *Worker) Run(ctx context.Context) error {
	return w.RunUntilDrained(ctx, nil)
}

// RunUntilDrained is Run plus a cooperative drain signal. Closing drain stops
// new claims immediately, including an in-flight PeekBatch via a context
// cancellation, but does not cancel an event that has already been claimed.
// The caller may still cancel ctx after its drain deadline to force a
// non-terminal retry path for work that does not finish in time.
//
// Each loop iteration: PEEK up to cfg.BatchSize claimable rows (no lease held),
// issue ONE BatchRead per objectKind for prefetch, then PER-ROW: ClaimByID
// (CAS arbiter), prepareEvent, executeEvent. This gives ~N× throughput over a
// single-event loop while preserving single-event claim/lease/finalize semantics.
func (w *Worker) RunUntilDrained(ctx context.Context, drain <-chan struct{}) error {
	if normalizeClaimPath(w.cfg.ClaimPath) == ClaimPathWorkItem {
		return w.RunClaimFirstUntilDrained(ctx, drain)
	}
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	var processed int
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if drainRequested(drain) {
			return nil
		}
		if w.cfg.MaxClaims > 0 && processed >= w.cfg.MaxClaims {
			return nil
		}

		permit, allowed, err := w.waitForClaimPermit(ctx, drain, w.cfg.BatchSize)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			continue
		}
		if !allowed {
			continue
		}

		claimed := w.drainPeekedBatch(ctx, drain, permit, processed)
		processed += claimed // count CLAIMED rows (CAS winners), NOT len(peeked)
		if claimed == 0 {
			// Empty peek, or all peeked rows lost to other workers — avoid busy-spin.
			timer := time.NewTimer(w.cfg.PollInterval)
			select {
			case <-ctx.Done():
				stopTimer(timer)
				return ctx.Err()
			case <-drain:
				stopTimer(timer)
				return nil
			case <-timer.C:
			}
		}
	}
}

// drainPeekedBatch runs one peek→claim→process→finalize iteration under the
// already-acquired claim permit. It is wrapped in two backstops so a panic
// anywhere in the body (the flow call is already wrapped by safeExecuteFlow;
// this covers prefetch/decode/prepare/coalesce/batchRead/flush) cannot crash the
// in-process worker goroutine (which would take down the whole backend) and
// cannot leave a claimed row stranded in 'processing':
//
//   - a DEFERRED permit release captures the FINAL claimed count, so a recovered
//     panic still frees the reservation (claimReservation has no expiry/sweep; a
//     leaked permit accretes toward MaxProcessingRows and silently throttles the
//     pack to a permanent stall);
//   - a recover backstop finalizes the WHOLE in-flight unit — the accumulating
//     batch group AND/OR a loose claimed row — via finalizeAfterPanic.
//
// Returns the number of CAS-won claims this iteration (the count survives a
// recovered panic via the named return).
func (w *Worker) drainPeekedBatch(ctx context.Context, drain <-chan struct{}, permit ClaimLimiterDecision, processed int) (claimed int) {
	var pendingBatch *pendingBatchGroup
	var inflight *connector_events.ConnectorEvent
	defer func() { w.releaseClaimPermit(permit, claimed) }()
	defer func() {
		if r := recover(); r != nil {
			w.finalizeAfterPanic(ctx, pendingBatch, inflight, r)
		}
	}()

	// Compute peek size for this iteration, respecting MaxClaims ceiling.
	n := w.cfg.BatchSize
	if permit.MaxClaims > 0 && permit.MaxClaims < n {
		n = permit.MaxClaims
	}
	if w.cfg.MaxClaims > 0 {
		remaining := w.cfg.MaxClaims - processed
		if remaining < n {
			n = remaining
		}
	}

	// PEEK — no claim, no lease. Cancel the SELECT if drain fires.
	peekCtx, cancelPeek := context.WithCancel(ctx)
	stopDrainCancel := cancelContextOnDrain(cancelPeek, drain)
	peeked, _ := w.peekBatch(peekCtx, n, permit.StaleReclaimOnly) // peek err → empty → backoff
	stopDrainCancel()
	cancelPeek()

	if len(peeked) == 0 {
		return 0 // empty queue — caller backs off (deferred release frees the permit)
	}

	// PREFETCH (no lease): best-effort decode sourceID only, ONE BatchRead per
	// objectKind. Errors are caught at the kind level; if prefetch for a kind
	// errors, cache[kind] stays absent and per-row single-read fallback fires.
	// cache is REBUILT every loop iteration.
	cache := make(map[string]map[string]json.RawMessage)
	sourceIDsByKind := make(map[string][]string)
	seenByKind := make(map[string]map[string]bool)
	for _, ev := range peeked {
		// SKIP prefetch for rows that must pass a gate BEFORE any enrichment read:
		//  - cross-pack rows (CompanionFlowVersionID != "") — the companion
		//    fail-closed contract requires ZERO batch reads before the companion
		//    incarnation check passes; they read per-row AFTER that gate below.
		//  - RouteVersion-invalid rows — terminally invalid (failTerminal in
		//    prepareEvent), so reading enrichment first is wasted on a doomed row.
		//  - route rows with flowExecutionFilter — the filter must evaluate
		//    against the durable event payload before any enrichment read.
		if ev.CompanionFlowVersionID != "" || !ev.RouteVersion.Valid || ev.RouteVersion.Int64 == 0 {
			continue
		}
		if spec, serr := w.resolver.ResolveExecutionSpecByVersion(ctx, ev.InstalledPackID, ev.EventKind, ev.RouteVersion.Int64); serr == nil && spec.FlowExecutionFilter != nil {
			continue
		}
		env, derr := DecodeQueueEnvelope(ev.Payload)
		if derr != nil || env.ObjectKind == "" || env.SourceID == "" {
			// Best-effort only — decode errors here do NOT fail the row.
			// prepareEvent (post-claim) is the sole full decode+fail path.
			continue
		}
		if seenByKind[env.ObjectKind] == nil {
			seenByKind[env.ObjectKind] = make(map[string]bool)
		}
		if !seenByKind[env.ObjectKind][env.SourceID] {
			seenByKind[env.ObjectKind][env.SourceID] = true
			// Append in PEEK ORDER — legacy demux order contract.
			sourceIDsByKind[env.ObjectKind] = append(sourceIDsByKind[env.ObjectKind], env.SourceID)
		}
	}
	prefetchTiming := make(map[string]time.Duration)
	prefetchSourceErrors := make(map[string]map[string]error)
	for objectKind, ids := range sourceIDsByKind {
		outcome, elapsed, rerr := w.batchReadOutcomeWithStageTiming(ctx, objectKind, ids)
		if rerr == nil {
			cache[objectKind] = outcome.Details
			prefetchSourceErrors[objectKind] = outcome.SourceErrors
			prefetchTiming[objectKind] = elapsed
		}
		// A batch-global error leaves cache[objectKind] absent and preserves the
		// per-row post-claim fallback. Source-scoped errors stay attached to their
		// exact IDs and are finalized only after each row wins its claim.
	}

	// CLAIM + PROCESS + FINALIZE (per-row, single-event semantics).
	// Iterate peeked in PEEK ORDER to preserve legacy demux order.
	for _, ev := range peeked {
		// Reset the loose-claim tracker at the TOP of each iteration: it points at
		// the current CAS-won row only between the claim below and its handoff to
		// the batch group / executeEvent, so the recover backstop never
		// double-finalizes a row already owned by pendingBatch or already executed.
		inflight = nil
		if drainRequested(drain) {
			break // stop claiming more; nothing half-claimed
		}
		if w.cfg.MaxClaims > 0 && processed+claimed >= w.cfg.MaxClaims {
			break
		}
		if w.deferPendingBatchSameDelivery(ctx, pendingBatch, ev) {
			break
		}
		// Claim with a drain-cancelled child ctx so a drain firing between the
		// drainRequested check above and the DB UPDATE cannot still win a claim.
		// ClaimByID is CAS-only (shape 1): no post-CAS load, so a drain after the
		// CAS commit never strands an already-claimed row.
		claimCtx, cancelClaim := context.WithCancel(ctx)
		stopClaimDrain := cancelContextOnDrain(cancelClaim, drain)
		token, claimedAt, ok, cerr := w.events.ClaimByID(claimCtx, ev.ID, w.cfg.WorkerID,
			ev.InstalledPackID,
			sql.NullString{String: ev.DeliveryId, Valid: ev.DeliveryId != ""},
			ev.DeliverySequence)
		stopClaimDrain()
		cancelClaim()
		if cerr != nil || !ok {
			continue // lost race / already claimed / drain-cancelled — skip silently
		}
		claimed++
		// Stamp the fresh CAS token onto a local copy of the peeked row.
		// payload/RouteVersion/Companion are immutable post-enqueue, so the
		// peeked row's content is authoritative; only the token needs updating.
		// NO loadByID — nothing to error/strand after the CAS win.
		cev := *ev
		cev.ClaimToken = token
		cev.ClaimedAt = &claimedAt
		inflight = &cev // loose claimed row until handed off to batch / executeEvent

		sameDeliveryPendingBatch := sameDeliveryBatch(pendingBatch, &cev)
		var prep preparedEvent
		var perr error
		if sameDeliveryPendingBatch {
			prep, perr = w.prepareEventWithoutFinalizing(ctx, &cev)
		} else {
			prep, perr = w.prepareEvent(ctx, &cev)
		}
		if perr != nil {
			if sameDeliveryPendingBatch {
				if relErr := w.releaseClaim(&cev, "batch candidate prepare", perr); relErr != nil {
					_ = w.failBatchWithSyncState(ctx, pendingBatch, "batch candidate release", relErr)
				}
				break
			}
			// prepareEvent already called failTerminal/failWithSyncState
			// (incl. the companion incarnation gate for cross-pack rows).
			continue
		}
		var coalesced int
		cevPtr := &cev
		cevPtr, prep, coalesced, perr = w.coalesceLatestBySource(ctx, cevPtr, prep)
		claimed += coalesced
		if perr != nil {
			continue
		}
		cev = *cevPtr

		var details map[string]json.RawMessage
		eventCtx := w.stageTimingContext(ctx, &cev, prep, 0)
		if cev.CompanionFlowVersionID != "" {
			// CROSS-PACK: the companion gate already PASSED in prepareEvent. Read its
			// enrichment NOW — a single read AFTER the companion check (it was NOT
			// prefetched, per the companion-before-read contract). NEVER reuse the
			// same-pack prefetch cache for a cross-pack row, even if cache[kind] is
			// present from same-pack siblings sharing the objectKind.
			// Heartbeat the claimed row across this post-claim cross-pack read (same
			// reclaim hazard as the claim-first path).
			single, _, deadlineHit, serr := w.batchReadHeartbeated(eventCtx, leaseRefsFor(&cev), prep.objectKind, []string{prep.sourceID})
			if serr != nil {
				if sameDeliveryPendingBatch {
					if relErr := w.releaseClaim(&cev, "batch read deferred by same-delivery ordering", serr); relErr != nil {
						_ = w.failBatchWithSyncState(ctx, pendingBatch, "batch candidate release", relErr)
					}
					break
				}
				if deadlineHit {
					if relErr := w.releaseClaimForRetry(&cev, "batch read deadline", errBatchReadDeadlineExceeded); relErr != nil {
						w.logStageTiming(eventCtx, &cev, prep, "error", relErr)
					}
					continue
				}
				if errors.Is(serr, ErrSourceRecordInactive) {
					// drainPeekedBatch has no per-row error channel and deliberately
					// keeps draining after finalizer failures. The helper emits an
					// unconditional structured error if MarkSkipped does not persist.
					_ = w.markConfirmedInactiveSkipped(eventCtx, &cev, prep)
					continue
				}
				err := w.failWithSyncState(eventCtx, &cev, prep.objectKind, "batch read", serr)
				w.logStageTiming(eventCtx, &cev, prep, "error", err)
				continue
			}
			details = single
		} else {
			// SAME-PACK: KIND-LEVEL cache lookup — NOT sourceID-level. If prefetch
			// for this kind SUCCEEDED (cache[kind] present), pass the map through even
			// if this sourceID is absent — a legal payload-only enrichment
			// (buildFlowInputs runs no `record`); requiredEnrichmentBatchReadClient is
			// the sole authority on a missing REQUIRED enrichment. Single-read fallback
			// ONLY when prefetch for this kind ERRORED (cache[kind] absent).
			var kindPrefetched bool
			details, kindPrefetched = cache[prep.objectKind]
			if !kindPrefetched {
				// Heartbeat the claimed row across this post-claim same-pack fallback
				// read (the prefetch at peek time held no lease; this read does).
				single, _, deadlineHit, serr := w.batchReadHeartbeated(eventCtx, leaseRefsFor(&cev), prep.objectKind, []string{prep.sourceID})
				if serr != nil {
					if sameDeliveryPendingBatch {
						if relErr := w.releaseClaim(&cev, "batch read deferred by same-delivery ordering", serr); relErr != nil {
							_ = w.failBatchWithSyncState(ctx, pendingBatch, "batch candidate release", relErr)
						}
						break
					}
					if deadlineHit {
						if relErr := w.releaseClaimForRetry(&cev, "batch read deadline", errBatchReadDeadlineExceeded); relErr != nil {
							w.logStageTiming(eventCtx, &cev, prep, "error", relErr)
						}
						continue
					}
					if errors.Is(serr, ErrSourceRecordInactive) {
						// See the cross-pack branch above: finalizer failure is logged by
						// the helper and the claimed row remains recoverable by lease.
						_ = w.markConfirmedInactiveSkipped(eventCtx, &cev, prep)
						continue
					}
					err := w.failWithSyncState(eventCtx, &cev, prep.objectKind, "batch read", serr)
					w.logStageTiming(eventCtx, &cev, prep, "error", err)
					continue
				}
				details = single
			} else {
				if sourceErr := prefetchSourceErrors[prep.objectKind][prep.sourceID]; sourceErr != nil {
					if sameDeliveryPendingBatch {
						if relErr := w.releaseClaim(&cev, "batch read outcome deferred by same-delivery ordering", sourceErr); relErr != nil {
							_ = w.failBatchWithSyncState(ctx, pendingBatch, "batch candidate release", relErr)
						}
						break
					}
					if errors.Is(sourceErr, ErrSourceRecordInactive) {
						_ = w.markConfirmedInactiveSkipped(eventCtx, &cev, prep)
						continue
					}
					err := w.failWithSyncState(eventCtx, &cev, prep.objectKind, "batch read", sourceErr)
					w.logStageTiming(eventCtx, &cev, prep, "error", err)
					continue
				}
				stagetiming.Add(eventCtx, stagetiming.StageBatchRead, prefetchTiming[prep.objectKind])
			}
		}
		if w.shouldBatchEvent(&cev, prep) {
			flushed := pendingBatch
			pendingBatch, perr = w.appendPendingBatch(eventCtx, pendingBatch, preparedClaimedEvent{ev: &cev, prep: prep, details: details})
			inflight = nil // cev now owned by pendingBatch (finalized as a group member)
			if perr != nil && sameDeliveryBatch(flushed, &cev) {
				pendingBatch = nil
				break
			}
		} else {
			flushed := pendingBatch
			pendingBatch, perr = w.flushPendingBatch(ctx, pendingBatch)
			if perr != nil && sameDeliveryBatch(flushed, &cev) {
				break
			}
			_ = w.executeEvent(eventCtx, &cev, prep, details)
			inflight = nil // cev finalized by executeEvent
		}
	}
	_, _ = w.flushPendingBatch(ctx, pendingBatch)
	return claimed
}

// processOne claims at most one event from the queue and processes it.
//
// Returns (true, nil) on successful processing (MarkProcessed called).
// Returns (true, err) on processing failure (MarkFailed already called; err is
// the cause for caller logging).
// Returns (false, nil) when the queue is empty.
func (w *Worker) processOne(ctx context.Context) (bool, error) {
	claimed, _, err := w.processOneWithDrain(ctx, nil, w.cfg.BatchSize, false)
	return claimed, err
}

// RunClaimFirstUntilDrained is a cooperative drain loop for fan-out workers.
// It claims each event before BatchRead, so sibling workers cannot duplicate
// external enrichment reads against rows they have not won.
func (w *Worker) RunClaimFirstUntilDrained(ctx context.Context, drain <-chan struct{}) error {
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	return schedulerOrDefault(w.scheduler).run(ctx, drain, w)
}

func schedulerOrDefault(candidate scheduler) scheduler {
	if candidate == nil {
		return compatibilityScheduler{}
	}
	value := reflect.ValueOf(candidate)
	switch value.Kind() {
	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
		if value.IsNil() {
			return compatibilityScheduler{}
		}
	}
	return candidate
}

func (w *Worker) schedulerConfig() schedulerConfig {
	return schedulerConfig{
		MaxClaims:             w.cfg.MaxClaims,
		BatchSize:             w.cfg.BatchSize,
		PollInterval:          w.cfg.PollInterval,
		ClaimPath:             normalizeClaimPath(w.cfg.ClaimPath),
		QueueClasses:          append([]string(nil), w.cfg.QueueClasses...),
		PreferredQueueClasses: append([]string(nil), w.workItemPreferredQueueClasses...),
	}
}

func (w *Worker) requestClaimPermit(ctx context.Context, drain <-chan struct{}, batchSize int) (ClaimLimiterDecision, bool, error) {
	return w.waitForClaimPermit(ctx, drain, batchSize)
}

func (w *Worker) requestClaimPermitForScheduler(ctx context.Context, batchSize int) (ClaimLimiterDecision, bool, time.Duration, error) {
	if w.cfg.ClaimLimiter == nil {
		return ClaimLimiterDecision{Allow: true}, true, 0, nil
	}
	decision, err := w.cfg.ClaimLimiter.BeforeClaim(ctx, w.claimLimiterRequest(batchSize))
	if err != nil {
		return decision, false, w.cfg.PollInterval, err
	}
	decision = w.capClaimLimiterDecision(decision)
	if decision.Allow {
		return decision, true, 0, nil
	}
	delay := decision.Delay
	if delay <= 0 {
		delay = w.cfg.PollInterval
	}
	return decision, false, delay, nil
}

func (w *Worker) capClaimLimiterDecision(decision ClaimLimiterDecision) ClaimLimiterDecision {
	if normalizeClaimPath(w.cfg.ClaimPath) != ClaimPathWorkItem {
		return decision
	}
	if w.cfg.BatchSize > 1 {
		return decision
	}
	if decision.MaxClaims == 0 || decision.MaxClaims > 1 {
		decision.MaxClaims = 1
	}
	return decision
}

func (w *Worker) claimNextForScheduler(ctx context.Context, staleOnly bool) (*connector_events.ConnectorEvent, bool, error) {
	return w.claimNext(ctx, staleOnly)
}

func (w *Worker) peekClaimCandidatesForScheduler(ctx context.Context, limit int, staleOnly bool) ([]schedulerClaimCandidate, error) {
	if limit <= 0 {
		return nil, nil
	}
	if svc, ok := w.events.(interface {
		peekClaimCandidatesForScheduler(ctx context.Context, installedPackID string, queueClasses, preferredQueueClasses []string, limit int, staleOnly bool) ([]schedulerClaimCandidate, error)
	}); ok {
		return svc.peekClaimCandidatesForScheduler(ctx, w.cfg.PackInstallID, w.cfg.QueueClasses, w.workItemPreferredQueueClasses, limit, staleOnly)
	}
	events, err := w.peekBatch(ctx, limit, staleOnly)
	if err != nil || len(events) == 0 {
		return nil, err
	}
	candidates := make([]schedulerClaimCandidate, 0, len(events))
	for _, ev := range events {
		if ev == nil {
			continue
		}
		candidates = append(candidates, schedulerClaimCandidate{
			event:      ev,
			staleOnly:  staleOnly,
			eligibleAt: eventEligibleAt(ev),
		})
	}
	return candidates, nil
}

func (w *Worker) claimByIDForScheduler(ctx context.Context, candidate schedulerClaimCandidate) (*connector_events.ConnectorEvent, bool, error) {
	if candidate.event == nil {
		return nil, false, nil
	}
	observation := startSchedulerClaimObservation(w.cfg, candidate)
	outcome := "panic"
	defer func() { observation.finish(outcome) }()
	event, claimed, err := w.claimCandidateByID(ctx, candidate)
	outcome = schedulerClaimOutcome(claimed, err)
	return event, claimed, err
}

func (w *Worker) claimCandidateByID(ctx context.Context, candidate schedulerClaimCandidate) (*connector_events.ConnectorEvent, bool, error) {
	if candidate.workType != "" {
		if svc, ok := w.events.(interface {
			rememberPeekedWorkType(eventID, workType string)
		}); ok {
			svc.rememberPeekedWorkType(candidate.event.ID, candidate.workType)
		}
	}
	deliveryID := sql.NullString{String: candidate.event.DeliveryId, Valid: candidate.event.DeliveryId != ""}
	if projected, handled, claimed, err := w.claimProjectedWorkItemEventByID(ctx, workItemClaimByIDInput{
		eventID: candidate.event.ID, workerID: w.cfg.WorkerID, installedPackID: w.cfg.PackInstallID,
		queueClasses: w.cfg.QueueClasses, generation: candidate.generation, generationResolved: candidate.generationResolved,
		staleOnly: candidate.staleOnly,
	}); handled {
		return projected, claimed, err
	}
	if svc, ok := w.events.(interface {
		ClaimByIDForClassesWithMode(ctx context.Context, eventID, workerID, installedPackID string, deliveryID sql.NullString, deliverySequence int, queueClasses []string, generation int64, generationResolved, staleOnly bool) (string, time.Time, bool, error)
	}); ok {
		token, claimedAt, ok, err := svc.ClaimByIDForClassesWithMode(ctx, candidate.event.ID, w.cfg.WorkerID, w.cfg.PackInstallID, deliveryID, candidate.event.DeliverySequence, w.cfg.QueueClasses, candidate.generation, candidate.generationResolved, candidate.staleOnly)
		if err != nil || !ok {
			return nil, ok, err
		}
		claimed := *candidate.event
		claimed.ClaimToken = token
		claimed.ClaimedAt = &claimedAt
		return &claimed, true, nil
	}
	if svc, ok := w.events.(interface {
		ClaimByIDForClasses(ctx context.Context, eventID, workerID, installedPackID string, deliveryID sql.NullString, deliverySequence int, queueClasses []string, generation int64, generationResolved bool) (string, time.Time, bool, error)
	}); ok {
		token, claimedAt, ok, err := svc.ClaimByIDForClasses(ctx, candidate.event.ID, w.cfg.WorkerID, w.cfg.PackInstallID, deliveryID, candidate.event.DeliverySequence, w.cfg.QueueClasses, candidate.generation, candidate.generationResolved)
		if err != nil || !ok {
			return nil, ok, err
		}
		claimed := *candidate.event
		claimed.ClaimToken = token
		claimed.ClaimedAt = &claimedAt
		return &claimed, true, nil
	}
	token, claimedAt, ok, err := w.events.ClaimByID(ctx, candidate.event.ID, w.cfg.WorkerID, w.cfg.PackInstallID, deliveryID, candidate.event.DeliverySequence)
	if err != nil || !ok {
		return nil, ok, err
	}
	claimed := *candidate.event
	claimed.ClaimToken = token
	claimed.ClaimedAt = &claimedAt
	return &claimed, true, nil
}

// claimProjectedWorkItemEventByID returns the adapter-owned execution
// projection after a WorkItem CAS. The handled bit leaves only non-executor
// compatibility adapters on their public token/time path.
func (w *Worker) claimProjectedWorkItemEventByID(ctx context.Context, in workItemClaimByIDInput) (*connector_events.ConnectorEvent, bool, bool, error) {
	svc, ok := w.events.(workItemProjectedEventClaimService)
	if !ok {
		return nil, false, false, nil
	}
	return svc.claimProjectedWorkItemEventByID(ctx, in)
}

func (w *Worker) claimBatchCandidate(ctx context.Context, candidate *connector_events.ConnectorEvent) (*connector_events.ConnectorEvent, bool, error) {
	if candidate == nil {
		return nil, false, nil
	}
	if projected, handled, claimed, err := w.claimProjectedWorkItemEventByID(ctx, workItemClaimByIDInput{
		eventID: candidate.ID, workerID: w.cfg.WorkerID, installedPackID: w.cfg.PackInstallID,
	}); handled {
		return projected, claimed, err
	}
	token, claimedAt, ok, err := w.events.ClaimByID(ctx, candidate.ID, w.cfg.WorkerID, w.cfg.PackInstallID, sql.NullString{String: candidate.DeliveryId, Valid: candidate.DeliveryId != ""}, candidate.DeliverySequence)
	if err != nil || !ok {
		return nil, ok, err
	}
	claimed := *candidate
	claimed.ClaimToken = token
	claimed.ClaimedAt = &claimedAt
	return &claimed, true, nil
}

func (w *Worker) executeClaim(ctx context.Context, drain <-chan struct{}, request claimExecutionRequest) (claimExecutionResult, error) {
	if request.HasCandidate {
		claimCtx, cancelClaim := context.WithCancel(ctx)
		stopDrainCancel := cancelContextOnDrain(cancelClaim, drain)
		started := time.Now()
		ev, ok, claimErr := w.claimByIDForScheduler(claimCtx, request.Candidate)
		stopDrainCancel()
		cancelClaim()
		result := claimExecutionResult{
			Claimed:        ok,
			ClaimAttempted: true,
			ClaimLatency:   time.Since(started),
			ClaimEligible:  request.Candidate.eligibleAt,
			ClaimErr:       claimErr,
		}
		if claimErr != nil || !ok {
			return result, claimErr
		}
		claimedRows, err := w.processClaimedWithDrain(ctx, drain, ev, request.MaxClaims)
		result.ClaimedRows = claimedRows
		return result, err
	}
	if request.Event != nil {
		claimedRows, err := w.processClaimedWithDrain(ctx, drain, request.Event, request.MaxClaims)
		return claimExecutionResult{Claimed: true, ClaimedRows: claimedRows}, err
	}
	claimed, claimedRows, err := w.processOneWithDrain(ctx, drain, request.MaxClaims, request.StaleReclaimOnly)
	return claimExecutionResult{Claimed: claimed, ClaimedRows: claimedRows}, err
}

func (w *Worker) releaseClaimPermitFromScheduler(decision ClaimLimiterDecision, claimedRows int) {
	w.releaseClaimPermit(decision, claimedRows)
}

func eventEligibleAt(ev *connector_events.ConnectorEvent) time.Time {
	if ev == nil {
		return time.Time{}
	}
	if ev.CreatedDate.IsZero() {
		return time.Time{}
	}
	return ev.CreatedDate
}

func (compatibilityScheduler) run(ctx context.Context, drain <-chan struct{}, executor executor) error {
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	cfg := executor.schedulerConfig()
	var processed int
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if drainRequested(drain) {
			return nil
		}
		if cfg.MaxClaims > 0 && processed >= cfg.MaxClaims {
			return nil
		}

		permit, allowed, err := executor.requestClaimPermit(ctx, drain, cfg.BatchSize)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			continue
		}
		if !allowed {
			continue
		}

		claimLimit := permit.MaxClaims
		if claimLimit <= 0 {
			claimLimit = cfg.BatchSize
		}
		var result claimExecutionResult
		func() {
			// Deferred release so a panic that escapes processOneWithDrain's own
			// recover backstop still frees the reservation (claimReservation has no
			// expiry/sweep). Captures the FINAL claimedRows count, not the zero value.
			defer func() { executor.releaseClaimPermitFromScheduler(permit, result.ClaimedRows) }()
			result, err = executor.executeClaim(ctx, drain, claimExecutionRequest{
				MaxClaims:        claimLimit,
				StaleReclaimOnly: permit.StaleReclaimOnly,
			})
		}()
		if result.Claimed {
			claimedRows := result.ClaimedRows
			if claimedRows <= 0 {
				// A successful single-claim executor predating ClaimedRows still
				// consumes one MaxClaims slot. Batch-aware executors report the
				// exact number of rows they finalized.
				claimedRows = 1
			}
			processed += claimedRows
			// Per-event processing errors have already finalized the row through
			// failTerminal/failWithSyncState; keep draining other claimable work.
			continue
		}
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			if drainRequested(drain) {
				return nil
			}
			// ClaimNext/load failures are retryable polling failures for fan-out
			// loops. Returning here would stop this worker while the supervisor
			// still records the pack handle as running.
			timer := time.NewTimer(cfg.PollInterval)
			select {
			case <-ctx.Done():
				stopTimer(timer)
				return ctx.Err()
			case <-drain:
				stopTimer(timer)
				return nil
			case <-timer.C:
			}
			continue
		}

		timer := time.NewTimer(cfg.PollInterval)
		select {
		case <-ctx.Done():
			stopTimer(timer)
			return ctx.Err()
		case <-drain:
			stopTimer(timer)
			return nil
		case <-timer.C:
		}
	}
}

func (s *CentralClaimScheduler) run(ctx context.Context, drain <-chan struct{}, executor executor) error {
	if s == nil {
		return compatibilityScheduler{}.run(ctx, drain, executor)
	}
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	reg := &centralClaimRegistration{
		executor: executor,
		work:     make(chan centralClaimWork, 1),
	}
	done := s.register(reg, ctx, drain)
	defer s.unregister(reg)

	for {
		select {
		case work := <-reg.work:
			s.executeRegisteredWork(ctx, drain, reg, executor, work)
			continue
		default:
		}
		select {
		case work := <-reg.work:
			s.executeRegisteredWork(ctx, drain, reg, executor, work)
		case <-done:
			select {
			case work := <-reg.work:
				s.executeRegisteredWork(ctx, drain, reg, executor, work)
				continue
			default:
			}
			return s.loopErr()
		}
	}
}

func (s *CentralClaimScheduler) executeRegisteredWork(ctx context.Context, drain <-chan struct{}, reg *centralClaimRegistration, executor executor, work centralClaimWork) {
	var result claimExecutionResult
	var panicValue any
	defer func() {
		if r := recover(); r != nil {
			panicValue = r
		}
		executor.releaseClaimPermitFromScheduler(work.permit, result.ClaimedRows)
		for _, permit := range work.additionalPermits {
			executor.releaseClaimPermitFromScheduler(permit, result.ClaimedRows)
		}
		resultCh := reg.resultCh
		if resultCh == nil {
			resultCh = s.resultCh
		}
		resultCh <- centralClaimResult{
			reg:        reg,
			result:     result,
			panicValue: panicValue,
		}
		if panicValue != nil {
			panic(panicValue)
		}
	}()
	result, result.ProcessingErr = executor.executeClaim(ctx, drain, work.request)
}

func (s *CentralClaimScheduler) register(reg *centralClaimRegistration, ctx context.Context, drain <-chan struct{}) <-chan struct{} {
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.running {
		s.regs = nil
		s.next = 0
		s.stats = CentralClaimSchedulerStats{StartedAt: time.Now()}
		s.done = make(chan struct{})
		s.wakeCh = make(chan struct{}, 1)
		s.resultCh = make(chan centralClaimResult, 1024)
		s.err = nil
	}
	reg.resultCh = s.resultCh
	s.regs = append(s.regs, reg)
	if !s.running {
		s.running = true
		logCentralClaimSchedulerStart(reg.executor)
		go s.claimLoop(ctx, drain)
	} else {
		s.signalWakeLocked()
	}
	return s.done
}

func logCentralClaimSchedulerStart(executor executor) {
	if w, ok := executor.(*Worker); ok && w != nil {
		log.Printf("connector inbound central claim scheduler started pack_install_id=%s worker_id=%s claim_path=%s batch_size=%d queue_classes=%d",
			w.cfg.PackInstallID, w.cfg.WorkerID, w.cfg.ClaimPath, w.cfg.BatchSize, len(w.cfg.QueueClasses))
		zlog.Info().
			Str("pack_install_id", w.cfg.PackInstallID).
			Str("worker_id", w.cfg.WorkerID).
			Str("claim_path", string(w.cfg.ClaimPath)).
			Int("batch_size", w.cfg.BatchSize).
			Int("queue_classes", len(w.cfg.QueueClasses)).
			Msg("connector inbound central claim scheduler started")
		return
	}
	log.Printf("connector inbound central claim scheduler started")
	zlog.Info().Msg("connector inbound central claim scheduler started")
}

func (s *CentralClaimScheduler) signalWakeLocked() {
	if s.wakeCh == nil {
		return
	}
	select {
	case s.wakeCh <- struct{}{}:
	default:
	}
}

func (s *CentralClaimScheduler) unregister(reg *centralClaimRegistration) {
	s.mu.Lock()
	defer s.mu.Unlock()
	reg.closed = true
	for i, candidate := range s.regs {
		if candidate == reg {
			s.regs = append(s.regs[:i], s.regs[i+1:]...)
			if s.next > i {
				s.next--
			}
			break
		}
	}
	if s.next >= len(s.regs) {
		s.next = 0
	}
}

func (s *CentralClaimScheduler) loopErr() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.err
}

func (s *CentralClaimScheduler) finish(err error) {
	s.mu.Lock()
	if !s.running {
		s.mu.Unlock()
		return
	}
	s.err = err
	s.running = false
	stats := s.stats
	regs := len(s.regs)
	close(s.done)
	s.mu.Unlock()

	event := zlog.Info().
		Int("registered_workers", regs).
		Int64("claim_attempts", stats.ClaimAttempts).
		Int64("claim_success", stats.ClaimSuccess).
		Int64("claim_empty", stats.ClaimEmpty).
		Int64("peek_candidates", stats.PeekCandidateCount).
		Int64("peek_errors", stats.PeekErrors).
		Int64("permit_denied", stats.PermitDenied).
		Int64("permit_errors", stats.PermitErrors)
	if err != nil {
		event = event.Err(err)
	}
	if err != nil {
		log.Printf("connector inbound central claim scheduler stopped registered_workers=%d claim_attempts=%d claim_success=%d claim_empty=%d peek_candidates=%d peek_errors=%d permit_denied=%d permit_errors=%d err=%v",
			regs, stats.ClaimAttempts, stats.ClaimSuccess, stats.ClaimEmpty, stats.PeekCandidateCount, stats.PeekErrors, stats.PermitDenied, stats.PermitErrors, err)
	} else {
		log.Printf("connector inbound central claim scheduler stopped registered_workers=%d claim_attempts=%d claim_success=%d claim_empty=%d peek_candidates=%d peek_errors=%d permit_denied=%d permit_errors=%d",
			regs, stats.ClaimAttempts, stats.ClaimSuccess, stats.ClaimEmpty, stats.PeekCandidateCount, stats.PeekErrors, stats.PermitDenied, stats.PermitErrors)
	}
	event.Msg("connector inbound central claim scheduler stopped")
}

func (s *CentralClaimScheduler) claimLoop(ctx context.Context, drain <-chan struct{}) {
	defer func() {
		if r := recover(); r != nil {
			s.finish(fmt.Errorf("%w: %v", errCentralClaimSchedulerPanic, r))
		}
	}()
	for {
		if ctx.Err() != nil {
			s.finish(ctx.Err())
			return
		}
		if drainRequested(drain) {
			s.finish(nil)
			return
		}
		s.drainClaimResults()
		now := time.Now()
		if s.fillIdleClaimCandidates(ctx, drain, now) {
			continue
		}
		reg := s.nextAvailableRegistration(now)
		if reg == nil {
			if s.allRegistrationsDone() {
				s.finish(nil)
				return
			}
			if s.sleepOrReceiveResult(ctx, drain, s.nextPollDelay(now)) {
				continue
			}
			return
		}
		if s.registrationDone(reg) {
			continue
		}
		cfg := reg.executor.schedulerConfig()
		permit, allowed, delay, err := reg.executor.requestClaimPermitForScheduler(ctx, cfg.BatchSize)
		if err != nil {
			if ctx.Err() != nil {
				s.finish(ctx.Err())
				return
			}
			s.recordPermitError(err)
			s.deferRegistration(reg, cfg.PollInterval)
			continue
		}
		if !allowed {
			s.recordPermitDenied(permit, delay)
			s.deferRegistration(reg, delay)
			continue
		}
		claimLimit := permit.MaxClaims
		if claimLimit <= 0 {
			claimLimit = cfg.BatchSize
		}
		claimCtx, cancelClaim := context.WithCancel(ctx)
		stopDrainCancel := cancelContextOnDrain(cancelClaim, drain)
		ev, ok, err := s.claimNextWithPermitReleaseOnPanic(reg, claimCtx, permit)
		stopDrainCancel()
		cancelClaim()
		if err != nil {
			reg.executor.releaseClaimPermitFromScheduler(permit, 0)
			if ctx.Err() != nil {
				s.finish(ctx.Err())
				return
			}
			s.deferRegistration(reg, cfg.PollInterval)
			continue
		}
		if !ok {
			reg.executor.releaseClaimPermitFromScheduler(permit, 0)
			s.deferRegistration(reg, cfg.PollInterval)
			continue
		}
		s.dispatchClaim(reg, centralClaimWork{
			permit: permit,
			request: claimExecutionRequest{
				MaxClaims:        claimLimit,
				StaleReclaimOnly: permit.StaleReclaimOnly,
				Event:            ev,
			},
		})
	}
}

type centralClaimSlot struct {
	reg             *centralClaimRegistration
	executor        schedulerBatchCandidateExecutor
	permit          ClaimLimiterDecision
	permitBatchSize int
}

const centralClaimWorkItemPermitBatchCap = 8

func (s *CentralClaimScheduler) fillIdleClaimCandidates(ctx context.Context, drain <-chan struct{}, now time.Time) bool {
	regs := s.availableRegistrations(now)
	if len(regs) == 0 {
		return false
	}
	batchRegs := make([]*centralClaimRegistration, 0, len(regs))
	for _, reg := range regs {
		if _, ok := reg.executor.(schedulerBatchCandidateExecutor); ok {
			batchRegs = append(batchRegs, reg)
		}
	}
	if len(batchRegs) == 0 {
		return false
	}

	// Select one queue shape before reserving permits for its peers. An empty
	// shape must not delay another ready shape: release its permits, then try
	// each remaining shape in this refill before returning to the claim loop.
	// Reserving every idle executor and filtering by shape afterwards briefly
	// consumed the entire process-local budget on work that could not be
	// dispatched. Under a shared database budget, multiple runtime processes
	// amplified that phantom reservation pressure and left real executor capacity
	// idle.
	for len(batchRegs) > 0 {
		slots := make([]centralClaimSlot, 0, len(batchRegs))
		firstIndex := -1
		for i, reg := range batchRegs {
			executor := reg.executor.(schedulerBatchCandidateExecutor)
			cfg := reg.executor.schedulerConfig()
			permitBatchSize := centralClaimPermitBatchSize(cfg)
			permit, allowed, delay, err := reg.executor.requestClaimPermitForScheduler(ctx, permitBatchSize)
			if err != nil {
				if ctx.Err() != nil {
					s.finish(ctx.Err())
					return true
				}
				s.recordPermitError(err)
				s.deferRegistration(reg, cfg.PollInterval)
				continue
			}
			if !allowed {
				s.recordPermitDenied(permit, delay)
				s.deferRegistration(reg, delay)
				continue
			}
			slots = append(slots, centralClaimSlot{
				reg:             reg,
				executor:        executor,
				permit:          permit,
				permitBatchSize: permitBatchSize,
			})
			firstIndex = i
			break
		}
		if len(slots) == 0 {
			return true
		}
		staleOnly := slots[0].permit.StaleReclaimOnly
		queueShape := slots[0].reg.executor.schedulerConfig()
		for _, reg := range batchRegs[firstIndex+1:] {
			cfg := reg.executor.schedulerConfig()
			if !sameSchedulerQueueShape(cfg, queueShape, staleOnly) {
				continue
			}
			executor := reg.executor.(schedulerBatchCandidateExecutor)
			permitBatchSize := centralClaimPermitBatchSize(cfg)
			permit, allowed, delay, err := reg.executor.requestClaimPermitForScheduler(ctx, permitBatchSize)
			if err != nil {
				if ctx.Err() != nil {
					for _, slot := range slots {
						slot.reg.executor.releaseClaimPermitFromScheduler(slot.permit, 0)
					}
					s.finish(ctx.Err())
					return true
				}
				s.recordPermitError(err)
				s.deferRegistration(reg, cfg.PollInterval)
				continue
			}
			if !allowed {
				s.recordPermitDenied(permit, delay)
				s.deferRegistration(reg, delay)
				continue
			}
			if permit.StaleReclaimOnly != staleOnly {
				reg.executor.releaseClaimPermitFromScheduler(permit, 0)
				continue
			}
			slots = append(slots, centralClaimSlot{
				reg:             reg,
				executor:        executor,
				permit:          permit,
				permitBatchSize: permitBatchSize,
			})
		}

		inflightCandidateKeys := s.inflightCandidateKeys()
		peekLimit := len(slots) + len(inflightCandidateKeys)
		if peekLimit < len(slots) {
			peekLimit = len(slots)
		}
		peekCtx, cancelPeek := context.WithCancel(ctx)
		stopDrainCancel := cancelContextOnDrain(cancelPeek, drain)
		candidates, err := slots[0].executor.peekClaimCandidatesForScheduler(peekCtx, peekLimit, staleOnly)
		stopDrainCancel()
		cancelPeek()
		s.recordPeekCandidates(len(candidates))
		if err != nil {
			s.recordPeekError(err)
			for _, slot := range slots {
				slot.reg.executor.releaseClaimPermitFromScheduler(slot.permit, 0)
				slotCfg := slot.reg.executor.schedulerConfig()
				s.deferRegistration(slot.reg, slotCfg.PollInterval)
			}
			if ctx.Err() != nil {
				s.finish(ctx.Err())
			}
			return true
		}
		if len(candidates) == 0 {
			for _, slot := range slots {
				slot.reg.executor.releaseClaimPermitFromScheduler(slot.permit, 0)
				slotCfg := slot.reg.executor.schedulerConfig()
				// A completely empty queue is not a partial batch refill. Honor the
				// configured empty-queue poll interval; producer and backlog wakes
				// still clear this deadline immediately when new work becomes visible.
				s.deferRegistration(slot.reg, slotCfg.PollInterval)
			}
			s.recordClaimEmpty(int64(len(slots)))
			remaining := batchRegs[:0]
			for _, reg := range batchRegs {
				if sameSchedulerQueueShape(reg.executor.schedulerConfig(), queueShape, staleOnly) {
					continue
				}
				remaining = append(remaining, reg)
			}
			batchRegs = remaining
			continue
		}

		slotIdx := 0
		dispatchedCandidateKeys := make(map[string]struct{}, len(slots))
		for _, candidate := range candidates {
			if slotIdx >= len(slots) {
				break
			}
			key := schedulerCandidateKey(candidate)
			if key != "" {
				if _, exists := inflightCandidateKeys[key]; exists {
					continue
				}
				if _, exists := dispatchedCandidateKeys[key]; exists {
					continue
				}
				dispatchedCandidateKeys[key] = struct{}{}
			}
			if ctx.Err() != nil {
				for ; slotIdx < len(slots); slotIdx++ {
					slots[slotIdx].reg.executor.releaseClaimPermitFromScheduler(slots[slotIdx].permit, 0)
				}
				s.finish(ctx.Err())
				return true
			}
			if drainRequested(drain) {
				for ; slotIdx < len(slots); slotIdx++ {
					slots[slotIdx].reg.executor.releaseClaimPermitFromScheduler(slots[slotIdx].permit, 0)
				}
				s.finish(nil)
				return true
			}
			slot := slots[slotIdx]
			slotIdx++
			claimLimit := slot.permit.MaxClaims
			if claimLimit <= 0 {
				claimLimit = slot.permitBatchSize
			}
			if claimLimit <= 0 {
				claimLimit = 1
			}
			additionalPermits, claimLimit := s.expandCandidatePermit(ctx, slot, candidate, claimLimit)
			s.dispatchClaim(slot.reg, centralClaimWork{
				permit:            slot.permit,
				additionalPermits: additionalPermits,
				request: claimExecutionRequest{
					MaxClaims:        claimLimit,
					StaleReclaimOnly: slot.permit.StaleReclaimOnly,
					Candidate:        candidate,
					HasCandidate:     true,
				},
			})
		}
		for ; slotIdx < len(slots); slotIdx++ {
			slot := slots[slotIdx]
			slot.reg.executor.releaseClaimPermitFromScheduler(slot.permit, 0)
			slotCfg := slot.reg.executor.schedulerConfig()
			s.deferRegistration(slot.reg, centralClaimRefillDelay(slotCfg.PollInterval))
		}
		return true
	}
	return true
}

func (s *CentralClaimScheduler) expandCandidatePermit(
	ctx context.Context,
	slot centralClaimSlot,
	candidate schedulerClaimCandidate,
	claimLimit int,
) ([]ClaimLimiterDecision, int) {
	if slot.permit.StaleReclaimOnly {
		return nil, claimLimit
	}
	sizer, ok := slot.reg.executor.(schedulerCandidatePermitSizer)
	if !ok {
		return nil, claimLimit
	}
	desired := schedulerCandidatePermitBatchSize(ctx, sizer, candidate, claimLimit)
	if desired <= claimLimit {
		return nil, claimLimit
	}
	if slot.permit.ReservationID == 0 && slot.permit.MaxClaims <= 0 {
		return nil, desired
	}

	additionalSize := desired - claimLimit
	permit, allowed, delay, err := slot.reg.executor.requestClaimPermitForScheduler(ctx, additionalSize)
	if err != nil {
		s.recordPermitError(err)
		return nil, claimLimit
	}
	if !allowed || permit.StaleReclaimOnly {
		if permit.ReservationID != 0 {
			slot.reg.executor.releaseClaimPermitFromScheduler(permit, 0)
		}
		if !allowed {
			s.recordPermitDenied(permit, delay)
		}
		return nil, claimLimit
	}
	expanded := permit.MaxClaims
	if expanded <= 0 {
		expanded = additionalSize
	}
	if expanded > additionalSize {
		expanded = additionalSize
	}
	return []ClaimLimiterDecision{permit}, claimLimit + expanded
}

func schedulerCandidatePermitBatchSize(
	ctx context.Context,
	sizer schedulerCandidatePermitSizer,
	candidate schedulerClaimCandidate,
	fallback int,
) (desired int) {
	desired = fallback
	defer func() {
		if recovered := recover(); recovered != nil {
			eventID := ""
			if candidate.event != nil {
				eventID = candidate.event.ID
			}
			log.Printf("connector inbound candidate permit sizing panicked; using single-row permit event_id=%s panic=%v", eventID, recovered)
			zlog.Error().
				Interface("panic", recovered).
				Str("event_id", eventID).
				Msg("connector inbound candidate permit sizing panicked; using single-row permit")
			desired = fallback
		}
	}()
	return sizer.claimPermitBatchSizeForScheduler(ctx, candidate)
}

func centralClaimPermitBatchSize(_ schedulerConfig) int {
	// The batch feeder dispatches exactly one candidate to each idle executor.
	// Reserve that concrete claim here; route-aware batch expansion is handled
	// separately after a candidate is selected.
	return 1
}

type schedulerPermitModeCacheKey struct {
	installedPackID string
	eventKind       string
	routeVersion    int64
}

func (w *Worker) claimPermitBatchSizeForScheduler(ctx context.Context, candidate schedulerClaimCandidate) int {
	if w == nil || normalizeClaimPath(w.cfg.ClaimPath) != ClaimPathWorkItem || w.cfg.BatchSize <= 1 || w.resolver == nil || candidate.event == nil {
		return 1
	}
	event := candidate.event
	if !event.RouteVersion.Valid || event.RouteVersion.Int64 <= 0 {
		return 1
	}
	key := schedulerPermitModeCacheKey{
		installedPackID: event.InstalledPackID,
		eventKind:       event.EventKind,
		routeVersion:    event.RouteVersion.Int64,
	}
	batchCollection, ok := w.schedulerPermitModeCache.Load(key)
	if !ok {
		spec, err := w.resolver.ResolveExecutionSpecByVersion(ctx, event.InstalledPackID, event.EventKind, event.RouteVersion.Int64)
		if err != nil {
			return 1
		}
		batchCollection = spec.InvocationMode == connector_inbound_routing.InvocationModeBatchCollection
		w.schedulerPermitModeCache.Store(key, batchCollection)
	}
	isBatchCollection, _ := batchCollection.(bool)
	if isBatchCollection {
		return min(w.cfg.BatchSize, centralClaimWorkItemPermitBatchCap)
	}
	return 1
}

func schedulerCandidateKey(candidate schedulerClaimCandidate) string {
	if candidate.event == nil {
		return ""
	}
	return strings.TrimSpace(candidate.event.ID)
}

func (s *CentralClaimScheduler) inflightCandidateKeys() map[string]struct{} {
	s.mu.Lock()
	defer s.mu.Unlock()
	keys := make(map[string]struct{})
	for _, reg := range s.regs {
		if reg == nil || !reg.busy || reg.inflightCandidateKey == "" {
			continue
		}
		keys[reg.inflightCandidateKey] = struct{}{}
	}
	return keys
}

func sameSchedulerQueueShape(a, b schedulerConfig, staleOnly bool) bool {
	if normalizeClaimPath(a.ClaimPath) != normalizeClaimPath(b.ClaimPath) {
		return false
	}
	if len(a.QueueClasses) != len(b.QueueClasses) {
		return false
	}
	for i := range a.QueueClasses {
		if strings.TrimSpace(a.QueueClasses[i]) != strings.TrimSpace(b.QueueClasses[i]) {
			return false
		}
	}
	if staleOnly {
		return true
	}
	if len(a.PreferredQueueClasses) != len(b.PreferredQueueClasses) {
		return false
	}
	for i := range a.PreferredQueueClasses {
		if strings.TrimSpace(a.PreferredQueueClasses[i]) != strings.TrimSpace(b.PreferredQueueClasses[i]) {
			return false
		}
	}
	return true
}

func (s *CentralClaimScheduler) claimNextWithPermitReleaseOnPanic(reg *centralClaimRegistration, ctx context.Context, permit ClaimLimiterDecision) (ev *connector_events.ConnectorEvent, ok bool, err error) {
	defer func() {
		if r := recover(); r != nil {
			reg.executor.releaseClaimPermitFromScheduler(permit, 0)
			panic(r)
		}
	}()
	return reg.executor.claimNextForScheduler(ctx, permit.StaleReclaimOnly)
}

func (s *CentralClaimScheduler) recordPeekCandidates(count int) {
	s.mu.Lock()
	s.stats.PeekCandidateCount += int64(count)
	s.mu.Unlock()
}

func (s *CentralClaimScheduler) recordPeekError(err error) {
	s.mu.Lock()
	s.stats.PeekErrors++
	total := s.stats.PeekErrors
	s.mu.Unlock()
	if total <= 16 {
		log.Printf("connector inbound central claim scheduler peek error count=%d err=%v", total, err)
		zlog.Warn().Err(err).Int64("error_count", total).Msg("connector inbound central claim scheduler peek error")
	}
}

func (s *CentralClaimScheduler) recordPermitDenied(decision ClaimLimiterDecision, delay time.Duration) {
	s.mu.Lock()
	s.stats.PermitDenied++
	total := s.stats.PermitDenied
	s.mu.Unlock()
	if total <= 32 {
		log.Printf("connector inbound central claim scheduler permit denied count=%d reason=%s delay_ms=%d max_claims=%d stale_reclaim_only=%t reservation_id=%d",
			total, decision.Reason, delay.Milliseconds(), decision.MaxClaims, decision.StaleReclaimOnly, decision.ReservationID)
		zlog.Info().
			Int64("denied_count", total).
			Str("reason", decision.Reason).
			Dur("delay", delay).
			Int("max_claims", decision.MaxClaims).
			Bool("stale_reclaim_only", decision.StaleReclaimOnly).
			Int64("reservation_id", decision.ReservationID).
			Msg("connector inbound central claim scheduler permit denied")
	}
}

func (s *CentralClaimScheduler) recordPermitError(err error) {
	s.mu.Lock()
	s.stats.PermitErrors++
	total := s.stats.PermitErrors
	s.mu.Unlock()
	if total <= 16 {
		log.Printf("connector inbound central claim scheduler permit error count=%d err=%v", total, err)
		zlog.Warn().Err(err).Int64("error_count", total).Msg("connector inbound central claim scheduler permit error")
	}
}

func (s *CentralClaimScheduler) recordClaimEmpty(count int64) {
	s.mu.Lock()
	s.stats.ClaimEmpty += count
	total := s.stats.ClaimEmpty
	s.mu.Unlock()
	if total <= 64 {
		log.Printf("connector inbound central claim scheduler found no claimable WorkItems empty_executors=%d empty_total=%d", count, total)
		zlog.Info().
			Int64("empty_executors", count).
			Int64("empty_total", total).
			Msg("connector inbound central claim scheduler found no claimable WorkItems")
	}
}

func (s *CentralClaimScheduler) recordClaimAttempt(latency time.Duration, eligibleAt time.Time, ok bool, err error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.stats.ClaimAttempts++
	s.stats.ClaimLatencySamples++
	s.stats.ClaimLatencyTotal += latency
	if err != nil {
		s.stats.ClaimErrors++
		return
	}
	if !ok {
		s.stats.ClaimCASLost++
		return
	}
	s.stats.ClaimSuccess++
	if !eligibleAt.IsZero() {
		eligibleToClaim := time.Since(eligibleAt)
		if eligibleToClaim < 0 {
			eligibleToClaim = 0
		}
		s.stats.EligibleToClaimLatencySamples++
		s.stats.EligibleToClaimLatencyTotal += eligibleToClaim
		if eligibleToClaim > s.stats.EligibleToClaimLatencyMax {
			s.stats.EligibleToClaimLatencyMax = eligibleToClaim
		}
	}
}

func (s *CentralClaimScheduler) drainClaimResults() {
	for {
		select {
		case result := <-s.resultCh:
			s.handleClaimResult(result)
		default:
			return
		}
	}
}

func (s *CentralClaimScheduler) handleClaimResult(result centralClaimResult) {
	if result.reg == nil {
		return
	}
	if result.result.ClaimAttempted {
		s.recordClaimAttempt(result.result.ClaimLatency, result.result.ClaimEligible, result.result.Claimed, result.result.ClaimErr)
	}
	s.mu.Lock()
	if result.panicValue != nil {
		result.reg.busy = false
		result.reg.inflightCandidateKey = ""
		result.reg.closed = true
		s.mu.Unlock()
		return
	}
	if result.reg.busy {
		result.reg.busy = false
	}
	result.reg.inflightCandidateKey = ""
	transientProcessingErr := isCentralClaimTransientClaimError(result.result.ProcessingErr)
	if result.result.Claimed {
		if !transientProcessingErr {
			if result.result.ClaimedRows > 0 {
				result.reg.processed += result.result.ClaimedRows
			} else {
				result.reg.processed++
			}
		}
		s.clearDeferredRegistrationsLocked()
		s.signalWakeLocked()
	} else if result.result.ClaimAttempted && result.result.ClaimErr == nil {
		s.clearDeferredRegistrationsLocked()
		s.signalWakeLocked()
	} else if result.result.ClaimAttempted && isCentralClaimTransientClaimError(result.result.ClaimErr) {
		s.clearDeferredRegistrationsLocked()
		s.signalWakeLocked()
	} else if transientProcessingErr {
		s.clearDeferredRegistrationsLocked()
		s.signalWakeLocked()
	}
	s.mu.Unlock()
}

func isCentralClaimTransientClaimError(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "invalid connection") ||
		strings.Contains(msg, "driver: bad connection") ||
		strings.Contains(msg, "connection reset") ||
		strings.Contains(msg, "connection refused") ||
		strings.Contains(msg, "context deadline exceeded") ||
		strings.Contains(msg, "i/o timeout")
}

func centralClaimRefillDelay(pollInterval time.Duration) time.Duration {
	if pollInterval > 0 && pollInterval < defaultCentralClaimRefillDelay {
		return pollInterval
	}
	return defaultCentralClaimRefillDelay
}

func (s *CentralClaimScheduler) clearDeferredRegistrationsLocked() {
	for _, reg := range s.regs {
		if reg == nil || reg.closed || reg.busy {
			continue
		}
		reg.nextPoll = time.Time{}
	}
}

func (s *CentralClaimScheduler) deferRegistration(reg *centralClaimRegistration, delay time.Duration) {
	if delay <= 0 {
		delay = defaultPollInterval
	}
	s.mu.Lock()
	reg.nextPoll = time.Now().Add(delay)
	s.mu.Unlock()
}

func (s *CentralClaimScheduler) nextAvailableRegistration(now time.Time) *centralClaimRegistration {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.regs) == 0 {
		return nil
	}
	for checked := 0; checked < len(s.regs); checked++ {
		idx := (s.next + checked) % len(s.regs)
		reg := s.regs[idx]
		if reg.closed || reg.busy {
			continue
		}
		if !reg.nextPoll.IsZero() && reg.nextPoll.After(now) {
			continue
		}
		cfg := reg.executor.schedulerConfig()
		if cfg.MaxClaims > 0 && reg.processed >= cfg.MaxClaims {
			continue
		}
		s.next = (idx + 1) % len(s.regs)
		return reg
	}
	return nil
}

func (s *CentralClaimScheduler) availableRegistrations(now time.Time) []*centralClaimRegistration {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.regs) == 0 {
		return nil
	}
	regs := make([]*centralClaimRegistration, 0, len(s.regs))
	for checked := 0; checked < len(s.regs); checked++ {
		idx := (s.next + checked) % len(s.regs)
		reg := s.regs[idx]
		if reg.closed || reg.busy {
			continue
		}
		if !reg.nextPoll.IsZero() && reg.nextPoll.After(now) {
			continue
		}
		cfg := reg.executor.schedulerConfig()
		if cfg.MaxClaims > 0 && reg.processed >= cfg.MaxClaims {
			continue
		}
		regs = append(regs, reg)
	}
	if len(regs) > 0 {
		for i, candidate := range s.regs {
			if candidate == regs[len(regs)-1] {
				s.next = (i + 1) % len(s.regs)
				break
			}
		}
	}
	return regs
}

func (s *CentralClaimScheduler) nextPollDelay(now time.Time) time.Duration {
	s.mu.Lock()
	defer s.mu.Unlock()
	var delay time.Duration
	for _, reg := range s.regs {
		if reg.closed || reg.busy {
			continue
		}
		cfg := reg.executor.schedulerConfig()
		if cfg.MaxClaims > 0 && reg.processed >= cfg.MaxClaims {
			continue
		}
		if reg.nextPoll.IsZero() || !reg.nextPoll.After(now) {
			return 0
		}
		candidate := reg.nextPoll.Sub(now)
		if delay == 0 || candidate < delay {
			delay = candidate
		}
	}
	if delay <= 0 {
		return defaultPollInterval
	}
	return delay
}

func (s *CentralClaimScheduler) registrationDone(reg *centralClaimRegistration) bool {
	cfg := reg.executor.schedulerConfig()
	if cfg.MaxClaims <= 0 {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	return reg.processed >= cfg.MaxClaims
}

func (s *CentralClaimScheduler) allRegistrationsDone() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.regs) == 0 {
		return false
	}
	for _, reg := range s.regs {
		if reg.closed || reg.busy {
			return false
		}
		cfg := reg.executor.schedulerConfig()
		if cfg.MaxClaims <= 0 || reg.processed < cfg.MaxClaims {
			return false
		}
	}
	return true
}

func (s *CentralClaimScheduler) dispatchClaim(reg *centralClaimRegistration, work centralClaimWork) {
	s.mu.Lock()
	reg.busy = true
	reg.inflightCandidateKey = ""
	if work.request.HasCandidate {
		reg.inflightCandidateKey = schedulerCandidateKey(work.request.Candidate)
	}
	s.dispatchLogCount++
	dispatchLogCount := s.dispatchLogCount
	s.mu.Unlock()
	if dispatchLogCount <= 16 {
		logCentralClaimSchedulerDispatch(reg.executor, work, dispatchLogCount)
	}
	reg.work <- work
}

func logCentralClaimSchedulerDispatch(executor executor, work centralClaimWork, count int) {
	packInstallID := ""
	workerID := ""
	claimPath := ClaimPath("")
	event := zlog.Info().
		Int("dispatch_count", count).
		Int("max_claims", work.request.MaxClaims).
		Bool("stale_reclaim_only", work.request.StaleReclaimOnly).
		Bool("has_candidate", work.request.HasCandidate)
	if work.request.HasCandidate && work.request.Candidate.event != nil {
		event = event.
			Str("event_id", work.request.Candidate.event.ID).
			Str("work_type", work.request.Candidate.workType)
	}
	if w, ok := executor.(*Worker); ok && w != nil {
		packInstallID = w.cfg.PackInstallID
		workerID = w.cfg.WorkerID
		claimPath = normalizeClaimPath(w.cfg.ClaimPath)
		event = event.
			Str("pack_install_id", w.cfg.PackInstallID).
			Str("worker_id", w.cfg.WorkerID).
			Str("claim_path", string(w.cfg.ClaimPath))
	}
	log.Printf("connector inbound central claim scheduler dispatched WorkItem claim dispatch_count=%d max_claims=%d stale_reclaim_only=%t has_candidate=%t pack_install_id=%s worker_id=%s claim_path=%s",
		count, work.request.MaxClaims, work.request.StaleReclaimOnly, work.request.HasCandidate, packInstallID, workerID, claimPath)
	event.Msg("connector inbound central claim scheduler dispatched WorkItem claim")
}

func (s *CentralClaimScheduler) sleepOrReceiveResult(ctx context.Context, drain <-chan struct{}, d time.Duration) bool {
	if d <= 0 {
		d = defaultPollInterval
	}
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case result := <-s.resultCh:
		s.handleClaimResult(result)
		return true
	case <-s.wakeCh:
		return true
	case <-ctx.Done():
		s.finish(ctx.Err())
		return false
	case <-drain:
		s.finish(nil)
		return false
	case <-timer.C:
		return true
	}
}

// preparedEvent holds the decoded + resolved fields extracted from a claimed
// event before batch-read.  All fields are guaranteed non-empty on success.
type preparedEvent struct {
	objectKind               string
	sourceID                 string
	userPayload              json.RawMessage
	flowName                 string
	spec                     connector_inbound_routing.RouteExecutionSpec
	requiresEnrichmentRecord bool
	coalescedPayloads        []coalescedPayloadInput
}

type coalescedPayloadInput struct {
	eventID        string
	eventKind      string
	namespace      string
	objectKind     string
	sourceID       string
	idempotencyKey string
	payload        json.RawMessage
}

var errEventSkipped = errors.New("connector inbound event skipped by route flowExecutionFilter")

// prepareEvent decodes the QueueEnvelope, validates RouteVersion, resolves the
// flow name via ResolveFlowByVersion, and performs the companion incarnation
// check.  On any failure it calls the correct failTerminal/failWithSyncState
// branch (exactly as processOneWithDrain did) and returns a non-nil error so
// the caller skips this event.
func (w *Worker) prepareEvent(ctx context.Context, ev *connector_events.ConnectorEvent) (preparedEvent, error) {
	return w.prepareEventWithFinalization(ctx, ev, true)
}

func (w *Worker) prepareEventWithoutFinalizing(ctx context.Context, ev *connector_events.ConnectorEvent) (preparedEvent, error) {
	return w.prepareEventWithFinalization(ctx, ev, false)
}

func (w *Worker) prepareEventWithFinalization(ctx context.Context, ev *connector_events.ConnectorEvent, finalize bool) (preparedEvent, error) {
	env, err := DecodeQueueEnvelope(ev.Payload)
	if err != nil {
		if !finalize {
			return preparedEvent{}, fmt.Errorf("decode queue envelope: %w", err)
		}
		return preparedEvent{}, w.failTerminal(ctx, ev, "decode queue envelope", err)
	}
	objectKind := env.ObjectKind
	sourceID := env.SourceID
	userPayload := env.Payload

	if !ev.RouteVersion.Valid || ev.RouteVersion.Int64 == 0 {
		cause := fmt.Errorf("RouteVersion missing/zero — pre-R22 or unrouted; run migration runbook")
		if !finalize {
			return preparedEvent{}, fmt.Errorf("no immutable RouteVersion snapshot: %w", cause)
		}
		return preparedEvent{}, w.failTerminal(ctx, ev, "no immutable RouteVersion snapshot",
			cause)
	}
	spec, rerr := w.resolver.ResolveExecutionSpecByVersion(ctx, ev.InstalledPackID, ev.EventKind, ev.RouteVersion.Int64)
	if rerr != nil {
		if !finalize {
			return preparedEvent{}, fmt.Errorf("resolve route by version: %w", rerr)
		}
		if errors.Is(rerr, connector_inbound_routing.ErrRouteVersionSwept) ||
			errors.Is(rerr, connector_inbound_routing.ErrRouteNotDeclared) {
			return preparedEvent{}, w.failTerminal(ctx, ev, "resolve route by version", rerr)
		}
		return preparedEvent{}, w.failWithSyncState(ctx, ev, objectKind, "resolve route by version", rerr)
	}
	flowName := spec.FlowDeveloperName

	if spec.FlowExecutionFilter != nil {
		matches, ferr := spec.FlowExecutionFilter.Matches(userPayload)
		if ferr != nil {
			if !finalize {
				return preparedEvent{}, fmt.Errorf("flow execution filter: %w", ferr)
			}
			return preparedEvent{}, w.failTerminal(ctx, ev, "flow execution filter", ferr)
		}
		if !matches {
			if !finalize {
				return preparedEvent{}, errEventSkipped
			}
			if err := w.markSkipped(ev); err != nil {
				return preparedEvent{}, fmt.Errorf("MarkSkipped after flow execution filter: %w", err)
			}
			return preparedEvent{}, errEventSkipped
		}
	}

	if ev.CompanionFlowVersionID != "" {
		var err error
		if finalize {
			err = w.checkCompanionIncarnation(ctx, ev)
		} else {
			err = w.validateCompanionIncarnation(ctx, ev)
			if err != nil {
				err = fmt.Errorf("companion incarnation check: %w", err)
			}
		}
		if err != nil {
			return preparedEvent{}, err
		}
	}

	requiresEnrichmentRecord := false
	switch spec.InvocationMode {
	case connector_inbound_routing.InvocationModeLatestBySourceFullRecord:
		userPayload = json.RawMessage(`{}`)
		requiresEnrichmentRecord = true
	case connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayload,
		connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled:
		requiresEnrichmentRecord = true
	}

	return preparedEvent{
		objectKind:               objectKind,
		sourceID:                 sourceID,
		userPayload:              userPayload,
		flowName:                 flowName,
		spec:                     spec,
		requiresEnrichmentRecord: requiresEnrichmentRecord,
	}, nil
}

func preservesCoalescedPayloads(spec connector_inbound_routing.RouteExecutionSpec) bool {
	return spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayload ||
		spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled
}

func coalescedPayloadInputForEvent(ev *connector_events.ConnectorEvent, prep preparedEvent) coalescedPayloadInput {
	return coalescedPayloadInput{
		eventID:        ev.ID,
		eventKind:      ev.EventKind,
		namespace:      ev.Namespace,
		objectKind:     prep.objectKind,
		sourceID:       prep.sourceID,
		idempotencyKey: ev.IdempotencyKey,
		payload:        prep.userPayload,
	}
}

func settledProjectionPayloadInputs(ctx context.Context, events EventsService, selected *connector_events.ConnectorEvent, prep preparedEvent) ([]coalescedPayloadInput, error) {
	loader, ok := events.(settledProjectionPayloadEventsService)
	if !ok {
		return nil, fmt.Errorf("settled projection requires durable WorkItem payload membership")
	}

	rows, err := loader.LoadSettledProjectionPayloads(ctx, selected.ID)
	if err != nil {
		return nil, fmt.Errorf("load settled projection payloads: %w", err)
	}
	if len(rows) == 0 {
		return nil, fmt.Errorf("load settled projection payloads: selected event %q was absent", selected.ID)
	}

	inputs := make([]coalescedPayloadInput, 0, len(rows))
	selectedFound := false
	for _, row := range rows {
		if row == nil {
			return nil, fmt.Errorf("load settled projection payloads: selected event %q returned a nil row", selected.ID)
		}
		env, err := DecodeQueueEnvelope(row.Payload)
		if err != nil {
			return nil, fmt.Errorf("load settled projection payloads: decode event %q: %w", row.ID, err)
		}
		if env.ObjectKind != prep.objectKind || env.SourceID != prep.sourceID {
			return nil, fmt.Errorf("load settled projection payloads: event %q escaped source lane %q/%q", row.ID, prep.objectKind, prep.sourceID)
		}
		inputs = append(inputs, coalescedPayloadInput{
			eventID:        row.ID,
			eventKind:      row.EventKind,
			namespace:      row.Namespace,
			objectKind:     env.ObjectKind,
			sourceID:       env.SourceID,
			idempotencyKey: row.IdempotencyKey,
			payload:        env.Payload,
		})
		selectedFound = selectedFound || row.ID == selected.ID
	}
	if !selectedFound {
		return nil, fmt.Errorf("load settled projection payloads: selected event %q was not in its source-projection generation", selected.ID)
	}
	return inputs, nil
}

// checkCompanionIncarnation re-checks that the frozen companion app-pack
// incarnation (PackInstallId) is still active. It returns nil when the
// incarnation is eligible, and calls failTerminal (returning the resulting
// error) when the incarnation is absent/inactive/deleted or the checker is not
// wired.
// Same-pack/legacy events (CompanionFlowVersionID=="") must never call this.
func (w *Worker) checkCompanionIncarnation(ctx context.Context, ev *connector_events.ConnectorEvent) error {
	if err := w.validateCompanionIncarnation(ctx, ev); err != nil {
		return w.failTerminal(ctx, ev, "companion incarnation check", err)
	}
	return nil
}

func (w *Worker) validateCompanionIncarnation(ctx context.Context, ev *connector_events.ConnectorEvent) error {
	if w.checkCompanionInstalled == nil {
		return fmt.Errorf("worker has no companion incarnation checker wired; refusing to execute cross-pack event without incarnation re-check")
	}
	present, checkErr := w.checkCompanionInstalled(ctx, ev.CompanionPackInstallID)
	if checkErr != nil {
		return fmt.Errorf("companion incarnation re-check DB error: %w", checkErr)
	}
	if !present {
		return fmt.Errorf("companion incarnation is not active since enqueue: PackInstallId %q no longer eligible (companion was reborn or removed)", ev.CompanionPackInstallID)
	}
	return nil
}

// safeExecuteFlow invokes FlowInvoker.ExecuteFlow with a recover backstop. A
// panic raised inside the flow runtime over connector-controlled record/payload
// maps (nil-map write, slice OOB, nil-pointer interface deref) is converted to a
// nil result plus errFlowPanic so the caller's existing nil-result routing
// finalizes the event on the genuine-failure budget (MarkFailed) instead of
// crashing the in-process worker goroutine (which would take down the whole
// backend, REST/webhook server included). The wrapper is the precise blast
// radius — the shared FlowExecutor (UI preview / subflow / serverruntime) is
// intentionally NOT wrapped here. The stack lands in the returned error only
// (no payload), so MarkFailed's LastError stays within the secret-leak boundary.
func (w *Worker) safeExecuteFlow(ctx context.Context, flowName string, inputs map[string]any, user *models.UserSession) (result *FlowResult, err error) {
	defer func() {
		if r := recover(); r != nil {
			result = nil
			err = fmt.Errorf("%w: flow %q: %v\n%s", errFlowPanic, flowName, r, debug.Stack())
		}
	}()
	return w.flow.ExecuteFlow(ctx, flowName, inputs, user)
}

// safeExecuteFlowVersion is safeExecuteFlow for the cross-pack companion
// (frozen flow version) path. Same recover-to-(nil, errFlowPanic) contract.
func (w *Worker) safeExecuteFlowVersion(ctx context.Context, flowName, companionFlowVersionID string, inputs map[string]any, user *models.UserSession) (result *FlowResult, err error) {
	defer func() {
		if r := recover(); r != nil {
			result = nil
			err = fmt.Errorf("%w: flow %q: %v\n%s", errFlowPanic, flowName, r, debug.Stack())
		}
	}()
	return w.flow.ExecuteFlowVersion(ctx, flowName, companionFlowVersionID, inputs, user)
}

func (w *Worker) safeExecuteFlowOutcome(ctx context.Context, flowName string, inputs map[string]any, user *models.UserSession) (outcome *FlowDependencyOutcome, err error) {
	if invoker, ok := w.flow.(FlowDependencyOutcomeInvoker); ok {
		defer func() {
			if r := recover(); r != nil {
				outcome = nil
				err = fmt.Errorf("%w: flow %q: %v\n%s", errFlowPanic, flowName, r, debug.Stack())
			}
		}()
		outcome, err := invoker.ExecuteFlowDependencyOutcome(ctx, flowName, inputs, user)
		return normalizeFlowDependencyOutcome(outcome), err
	}
	invoker, ok := w.flow.(FlowOutcomeInvoker)
	if !ok {
		result, err := w.safeExecuteFlow(ctx, flowName, inputs, user)
		return &FlowDependencyOutcome{FlowOutcome: &FlowOutcome{Result: result}}, err
	}
	defer func() {
		if r := recover(); r != nil {
			outcome = nil
			err = fmt.Errorf("%w: flow %q: %v\n%s", errFlowPanic, flowName, r, debug.Stack())
		}
	}()
	legacy, err := invoker.ExecuteFlowOutcome(ctx, flowName, inputs, user)
	return normalizeFlowDependencyOutcome(&FlowDependencyOutcome{FlowOutcome: legacy}), err
}

func (w *Worker) safeExecuteFlowVersionOutcome(ctx context.Context, flowName, companionFlowVersionID string, inputs map[string]any, user *models.UserSession) (outcome *FlowDependencyOutcome, err error) {
	if invoker, ok := w.flow.(FlowDependencyOutcomeInvoker); ok {
		defer func() {
			if r := recover(); r != nil {
				outcome = nil
				err = fmt.Errorf("%w: flow %q: %v\n%s", errFlowPanic, flowName, r, debug.Stack())
			}
		}()
		outcome, err := invoker.ExecuteFlowVersionDependencyOutcome(ctx, flowName, companionFlowVersionID, inputs, user)
		return normalizeFlowDependencyOutcome(outcome), err
	}
	invoker, ok := w.flow.(FlowOutcomeInvoker)
	if !ok {
		result, err := w.safeExecuteFlowVersion(ctx, flowName, companionFlowVersionID, inputs, user)
		return &FlowDependencyOutcome{FlowOutcome: &FlowOutcome{Result: result}}, err
	}
	defer func() {
		if r := recover(); r != nil {
			outcome = nil
			err = fmt.Errorf("%w: flow %q: %v\n%s", errFlowPanic, flowName, r, debug.Stack())
		}
	}()
	legacy, err := invoker.ExecuteFlowVersionOutcome(ctx, flowName, companionFlowVersionID, inputs, user)
	return normalizeFlowDependencyOutcome(&FlowDependencyOutcome{FlowOutcome: legacy}), err
}

func normalizeFlowDependencyOutcome(outcome *FlowDependencyOutcome) *FlowDependencyOutcome {
	if outcome == nil || outcome.FlowOutcome == nil {
		return &FlowDependencyOutcome{FlowOutcome: &FlowOutcome{Result: FlowResultFromModel(nil)}}
	}
	return outcome
}

func flowResultFromOutcome(outcome *FlowDependencyOutcome) *FlowResult {
	if outcome == nil {
		return nil
	}
	return outcome.Result
}

func flowOutcomeIsRetriable(outcome *FlowDependencyOutcome) bool {
	if outcome == nil {
		return false
	}
	if outcome.Cause != nil && outcome.Cause.Retriable {
		return true
	}
	return outcome.Result != nil && outcome.Result.IsRetriable
}

func withFlowAttemptOutcome(ctx context.Context, outcome *FlowDependencyOutcome) context.Context {
	if outcome == nil || strings.TrimSpace(outcome.ExecutionID) == "" {
		return ctx
	}
	presentedCode := ""
	if outcome.Result != nil && outcome.Result.FirstError != nil {
		presentedCode = outcome.Result.FirstError.StatusCode
	}
	return connector_workqueue.WithWorkItemAttemptOutcome(ctx, connector_workqueue.WorkItemAttemptOutcome{
		FlowExecutionID:    outcome.ExecutionID,
		PresentedErrorCode: presentedCode,
		Cause:              outcome.Cause,
	})
}

// executeEvent builds flow inputs from the prefetched details map, then
// executes the flow (companion or same-pack path) and marks the event
// processed or failed.  The failure branches are byte-identical to
// processOneWithDrain.  details is keyed by sourceID.
func (w *Worker) executeEvent(ctx context.Context, ev *connector_events.ConnectorEvent, prep preparedEvent, details map[string]json.RawMessage) (err error) {
	status := "success"
	defer func() {
		if err != nil {
			status = "error"
		}
		w.logStageTiming(ctx, ev, prep, status, err)
	}()
	inputs, err := w.buildFlowInputs(ctx, ev, prep.objectKind, prep.sourceID, prep.userPayload, details, prep.requiresEnrichmentRecord, prep.coalescedPayloads)
	if err != nil {
		if errors.Is(err, ErrBatchReadRecordShape) {
			return w.failWithSyncState(ctx, ev, prep.objectKind, "batch read", err)
		}
		return w.failTerminal(ctx, ev, "build flow inputs", err)
	}
	ctx = context.WithValue(ctx, constants.ContextKeyConnectorSyncOrigin, w.cfg.Namespace)
	ctx = w.withConnectorInboundExecutionTrace(ctx, ev, prep, inputs)
	if w.runtimeCommands != nil {
		// Runtime.AcceptCommand transfers durable execution ownership, so it must
		// consume the same last-moment companion-incarnation fence as the legacy
		// Flow invocation below. prepareEvent checked once, but the incarnation
		// may have rotated during BatchRead.
		if ev.CompanionFlowVersionID != "" {
			if err := w.checkCompanionIncarnation(ctx, ev); err != nil {
				return err
			}
		}
		return w.executeRuntimeCommandHandoff(ctx, ev, prep, inputs)
	}
	user := w.flow.SystemUser(ctx)

	// Lease heartbeat: renew this in-flight event's lease while the (possibly slow)
	// flow runs so it never crosses the stale-reclaim cutoff and gets double-executed.
	// Stopped deterministically (blocks for the goroutine to exit) before any finalize.
	stopHeartbeat := w.startLeaseHeartbeat(leaseRefsFor(ev))
	defer stopHeartbeat()

	if ev.CompanionFlowVersionID != "" {
		// Re-check companion incarnation immediately before exec.  prepareEvent
		// already checked once (batch Phase 1), but the incarnation may have
		// rotated during Phase 2 (BatchRead) or while waiting for prior events
		// in Phase 3.  Fail closed rather than executing against a stale pack.
		if err := w.checkCompanionIncarnation(ctx, ev); err != nil {
			return err
		}
		ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, ev.CompanionFlowVersionID)
		flowCtx, cancelFlow := w.flowDeadlineContext(ctx)
		flowStarted := time.Now()
		outcome, err := w.safeExecuteFlowVersionOutcome(flowCtx, prep.flowName, ev.CompanionFlowVersionID, inputs, user)
		ctx = withFlowAttemptOutcome(ctx, outcome)
		result := flowResultFromOutcome(outcome)
		stagetiming.Add(ctx, stagetiming.StageFlowExec, time.Since(flowStarted))
		deadlineHit := isFlowDeadlineExceeded(flowCtx, err)
		cancelFlow()
		if outcome != nil && outcome.Dependency != nil && err == nil && result != nil && result.IsSuccess {
			return w.blockOnIdentityConvergence(ctx, ev, prep, *outcome.Dependency)
		}
		if err != nil || result == nil || !result.IsSuccess {
			// An opt-in FlowDeadline timeout re-drives via ReleaseClaimForRetry (NOT
			// MarkFailed) so a hung-but-not-dead flow does not burn the dead-letter
			// budget — same self-heal contract as a retriable fault.
			if deadlineHit {
				return w.releaseClaimForRetry(ev, "flow exec (companion) deadline", errFlowDeadlineExceeded, ctx)
			}
			// A retriable flow fault (FlowRetriableFault -> IsRetriable on a
			// failed result) is re-driven with a backoff via ReleaseClaimForRetry
			// instead of MarkFailed, so a concurrent-loser waiting on an in-flight
			// winner self-heals rather than dead-lettering on the genuine-failure
			// budget. See releaseClaimForRetry.
			if flowOutcomeIsRetriable(outcome) {
				return w.releaseClaimForRetry(ev, "flow exec (companion)", errors.New(flowRetriableMessage(prep.flowName, result)), ctx)
			}
			msg := flowFailureMessage(prep.flowName, result, err)
			return w.failWithSyncState(ctx, ev, prep.objectKind, "flow exec (companion)", errors.New(msg))
		}
		if flowOutcomeIsRetriable(outcome) {
			msg := flowRetriableMessage(prep.flowName, result)
			return w.failWithSyncState(ctx, ev, prep.objectKind, "flow exec (companion)", errors.New(msg))
		}
		return w.markProcessed(ev, ctx)
	}

	flowCtx, cancelFlow := w.flowDeadlineContext(ctx)
	flowStarted := time.Now()
	outcome, err := w.safeExecuteFlowOutcome(flowCtx, prep.flowName, inputs, user)
	ctx = withFlowAttemptOutcome(ctx, outcome)
	result := flowResultFromOutcome(outcome)
	stagetiming.Add(ctx, stagetiming.StageFlowExec, time.Since(flowStarted))
	deadlineHit := isFlowDeadlineExceeded(flowCtx, err)
	cancelFlow()
	if outcome != nil && outcome.Dependency != nil && err == nil && result != nil && result.IsSuccess {
		return w.blockOnIdentityConvergence(ctx, ev, prep, *outcome.Dependency)
	}
	if err != nil || result == nil || !result.IsSuccess {
		if deadlineHit {
			return w.releaseClaimForRetry(ev, "flow exec deadline", errFlowDeadlineExceeded, ctx)
		}
		// A retriable flow fault (FlowRetriableFault -> IsRetriable on a failed
		// result, e.g. a concurrent create-vs-adopt loser waiting on an in-flight
		// winner) must NOT burn the dead-letter budget: re-drive it with a backoff
		// via ReleaseClaimForRetry instead of MarkFailed so it self-heals once the
		// winner finalizes. See releaseClaimForRetry.
		if flowOutcomeIsRetriable(outcome) {
			return w.releaseClaimForRetry(ev, "flow exec", errors.New(flowRetriableMessage(prep.flowName, result)), ctx)
		}
		msg := flowFailureMessage(prep.flowName, result, err)
		return w.failWithSyncState(ctx, ev, prep.objectKind, "flow exec", errors.New(msg))
	}
	if flowOutcomeIsRetriable(outcome) {
		msg := flowRetriableMessage(prep.flowName, result)
		return w.failWithSyncState(ctx, ev, prep.objectKind, "flow exec", errors.New(msg))
	}
	return w.markProcessed(ev, ctx)
}

func (w *Worker) executeRuntimeCommandHandoff(
	ctx context.Context,
	ev *connector_events.ConnectorEvent,
	prep preparedEvent,
	inputs map[string]any,
) error {
	scope := w.runtimeCommandHandoffScope(ctx, ev)
	if err := scope.validate(); err != nil {
		return w.failTerminal(ctx, ev, "Runtime command handoff authority", err)
	}
	stopHeartbeat := w.startLeaseHeartbeat(leaseRefsFor(ev))
	defer stopHeartbeat()

	handoffCtx, cancelHandoff := w.flowDeadlineContext(ctx)
	err := w.runtimeCommands.AcceptRuntimeCommand(
		handoffCtx,
		RuntimeCommandHandoffRequest{
			Scope:             scope,
			DefinitionRef:     prep.flowName,
			DefinitionVersion: ev.CompanionFlowVersionID,
			Inputs:            inputs,
			DependencySource: flowcommandv2.DependencySourceScope{
				ObjectKind: prep.objectKind,
				SourceID:   prep.sourceID,
			},
		},
	)
	deadlineHit := isFlowDeadlineExceeded(handoffCtx, err)
	cancelHandoff()
	return w.finishRuntimeCommandHandoff(
		ctx,
		ev,
		prep,
		err,
		deadlineHit,
	)
}

func (w *Worker) resumeRuntimeCommandHandoff(
	ctx context.Context,
	ev *connector_events.ConnectorEvent,
) (bool, error) {
	scope := w.runtimeCommandHandoffScope(ctx, ev)
	if err := scope.validate(); err != nil {
		return true, w.failTerminal(
			ctx,
			ev,
			"Runtime command handoff authority",
			err,
		)
	}
	stopHeartbeat := w.startLeaseHeartbeat(leaseRefsFor(ev))
	defer stopHeartbeat()

	handoffCtx, cancelHandoff := w.flowDeadlineContext(ctx)
	found, err := w.runtimeCommands.ResumePreparedRuntimeCommand(
		handoffCtx,
		scope,
	)
	deadlineHit := isFlowDeadlineExceeded(handoffCtx, err)
	cancelHandoff()
	if !found && err == nil {
		return false, nil
	}
	if err == nil {
		return true, w.markProcessed(ev, ctx)
	}
	if deadlineHit {
		return true, w.releaseClaimForRetry(
			ev,
			"Runtime command receipt resume deadline",
			errFlowDeadlineExceeded,
			ctx,
		)
	}
	if runtimeCommandHandoffIsRetriable(err) {
		return true, w.releaseClaimForRetry(
			ev,
			"Runtime command receipt resume",
			err,
			ctx,
		)
	}
	return true, w.failTerminal(
		ctx,
		ev,
		"Runtime command receipt resume",
		err,
	)
}

func (w *Worker) finishRuntimeCommandHandoff(
	ctx context.Context,
	ev *connector_events.ConnectorEvent,
	prep preparedEvent,
	err error,
	deadlineHit bool,
) error {
	if err == nil {
		return w.markProcessed(ev, ctx)
	}
	if deadlineHit {
		return w.releaseClaimForRetry(
			ev,
			"Runtime command handoff deadline",
			errFlowDeadlineExceeded,
			ctx,
		)
	}
	if runtimeCommandHandoffIsRetriable(err) {
		return w.releaseClaimForRetry(
			ev,
			"Runtime command handoff",
			err,
			ctx,
		)
	}
	return w.failWithSyncState(
		ctx,
		ev,
		prep.objectKind,
		"Runtime command handoff",
		err,
	)
}

func (w *Worker) blockOnIdentityConvergence(ctx context.Context, ev *connector_events.ConnectorEvent, prep preparedEvent, request models.FlowExecutionDependencyRequest) error {
	if ev == nil {
		return fmt.Errorf("connector_inbound: invalid identity convergence dependency")
	}
	dependency, err := models.BindFlowIdentityConvergenceDependency(request, w.cfg.Namespace, prep.objectKind, prep.sourceID)
	if err != nil {
		return w.failWithSyncState(
			ctx,
			ev,
			prep.objectKind,
			"identity convergence dependency",
			fmt.Errorf("invalid dependency request: %w", err),
		)
	}
	blocker, ok := w.events.(identityConvergenceBlockingEventsService)
	if !ok || !blocker.SupportsIdentityConvergenceBlocking() {
		// Rolling-version backstop: never burn FailureCount or mark success when
		// the executor cannot persist the typed dependency yet.
		return w.releaseClaimForRetry(ev, "identity convergence dependency", errors.New("durable dependency blocker is not available"))
	}
	return blocker.BlockOnIdentityConvergence(ctx, ev.ID, ev.ClaimToken, IdentityConvergenceProbe{
		Namespace:  w.cfg.Namespace,
		ObjectKind: prep.objectKind,
		SourceID:   prep.sourceID,
		Dependency: dependency,
	})
}

func (w *Worker) claimNext(ctx context.Context, staleOnly bool) (*connector_events.ConnectorEvent, bool, error) {
	if staleOnly {
		sr, ok := w.events.(staleReclaimingEventsService)
		if !ok {
			// No stale-only support: reclaim nothing this tick rather than fall back
			// to a general claim that could admit a fresh row past the cap.
			return nil, false, nil
		}
		if len(w.cfg.QueueClasses) == 0 {
			return sr.ReclaimStaleNext(ctx, w.cfg.PackInstallID, w.cfg.WorkerID)
		}
		return sr.ReclaimStaleNextForClasses(ctx, w.cfg.PackInstallID, w.cfg.WorkerID, w.cfg.QueueClasses)
	}
	if len(w.cfg.QueueClasses) == 0 {
		return w.events.ClaimNext(ctx, w.cfg.PackInstallID, w.cfg.WorkerID)
	}
	if classAware, ok := w.events.(classAwareEventsService); ok {
		return classAware.ClaimNextForClasses(ctx, w.cfg.PackInstallID, w.cfg.WorkerID, w.cfg.QueueClasses)
	}
	return w.events.ClaimNext(ctx, w.cfg.PackInstallID, w.cfg.WorkerID)
}

func (w *Worker) peekBatch(ctx context.Context, n int, staleOnly bool) ([]*connector_events.ConnectorEvent, error) {
	if staleOnly {
		sr, ok := w.events.(staleReclaimingEventsService)
		if !ok {
			return nil, nil // no stale-only support → peek nothing this limp tick
		}
		if len(w.cfg.QueueClasses) == 0 {
			return sr.PeekStaleBatch(ctx, w.cfg.PackInstallID, n)
		}
		return sr.PeekStaleBatchForClasses(ctx, w.cfg.PackInstallID, w.cfg.QueueClasses, n)
	}
	if len(w.cfg.QueueClasses) == 0 {
		return w.events.PeekBatch(ctx, w.cfg.PackInstallID, n)
	}
	if classAware, ok := w.events.(classAwareEventsService); ok {
		return classAware.PeekBatchForClasses(ctx, w.cfg.PackInstallID, w.cfg.QueueClasses, n)
	}
	return w.events.PeekBatch(ctx, w.cfg.PackInstallID, n)
}

func (w *Worker) processOneWithDrain(ctx context.Context, drain <-chan struct{}, claimLimit int, staleOnly bool) (claimed bool, claimedRows int, err error) {
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	claimCtx, cancelClaim := context.WithCancel(ctx)
	stopDrainCancel := cancelContextOnDrain(cancelClaim, drain)
	ev, ok, err := w.claimNext(claimCtx, staleOnly)
	stopDrainCancel()
	cancelClaim()
	if err != nil || !ok {
		return false, 0, err
	}
	claimedRows, err = w.processClaimedWithDrain(ctx, drain, ev, claimLimit)
	return true, claimedRows, err
}

func (w *Worker) processClaimedWithDrain(ctx context.Context, drain <-chan struct{}, ev *connector_events.ConnectorEvent, claimLimit int) (claimedRows int, err error) {
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	// Recover backstop for the single-event claim-first path: a panic OUTSIDE the
	// wrapped flow call (prepare/coalesce/batchRead/buildFlowInputs) must finalize
	// the one claimed row instead of crashing the worker goroutine. The batch path
	// below delegates to executeClaimFirstBatchCollection, which finalizes its own
	// multi-row group under its own recover, so this backstop never double-acts on
	// the anchor row.
	defer func() {
		if r := recover(); r != nil {
			w.finalizeAfterPanic(ctx, nil, ev, r)
			claimedRows, err = 1, fmt.Errorf("%w: %v", errWorkerPanic, r)
		}
	}()
	if w.runtimeCommands != nil {
		resumed, resumeErr := w.resumeRuntimeCommandHandoff(ctx, ev)
		if resumed || resumeErr != nil {
			return 1, resumeErr
		}
	}

	prep, prepErr := w.prepareEvent(ctx, ev)
	if prepErr != nil {
		// prepareEvent already called failTerminal/failWithSyncState.
		if errors.Is(prepErr, errEventSkipped) {
			return 1, nil
		}
		return 1, prepErr
	}
	if w.runtimeCommands != nil {
		// Runtime handoff owns one immutable WorkItem. Per-event work maps directly
		// to that identity. The settled source-projection mode is also safe because
		// its WorkItem already owns the complete immutable projection generation;
		// coalesceLatestBySource only reconstructs that generation and never replaces
		// the claimed WorkItem. Other coalescing and collection modes can replace or
		// combine WorkItems before ownership transfer and need a different command
		// contract, so keep them fail-closed.
		workItemID, _ := w.runtimeCommandHandoffWorkItemID(ev)
		if !runtimeCommandHandoffSupportsClaimedWorkItem(
			w.cfg.ClaimPath,
			workItemID,
			prep.spec.InvocationMode,
		) {
			return 1, w.failTerminal(
				ctx,
				ev,
				"Runtime command handoff invocation mode",
				fmt.Errorf(
					"invocation mode %q is not supported by the Runtime command handoff for WorkItem %q",
					prep.spec.InvocationMode,
					workItemID,
				),
			)
		}
	}
	var coalesced int
	ev, prep, coalesced, prepErr = w.coalesceLatestBySource(ctx, ev, prep)
	_ = coalesced
	if prepErr != nil {
		if errors.Is(prepErr, errEventSkipped) {
			return 1, nil
		}
		return 1, prepErr
	}

	if w.shouldBatchEvent(ev, prep) {
		claimedRows, err := w.executeClaimFirstBatchCollection(ctx, drain, ev, prep, claimLimit)
		return claimedRows, err
	}

	// Single-event BatchRead (same as before).
	eventCtx := w.stageTimingContext(ctx, ev, prep, 0)
	// The post-claim vendor read holds the lease but runs OUTSIDE executeEvent's
	// flow heartbeat. A retry-backed BatchRead can exceed StaleProcessingSeconds,
	// so heartbeat the held lease across the read (batchReadHeartbeated defers the
	// stop, so a recovered BatchRead panic cannot leak the renewer goroutine).
	details, _, deadlineHit, batchErr := w.batchReadHeartbeated(eventCtx, leaseRefsFor(ev), prep.objectKind, []string{prep.sourceID})
	if batchErr != nil {
		if deadlineHit {
			// A wedged post-claim read is released for retry (NOT MarkFailed) so a
			// hung-but-not-dead vendor read self-heals on backoff — same contract as
			// the flow-deadline path.
			return 1, w.releaseClaimForRetry(ev, "batch read deadline", errBatchReadDeadlineExceeded)
		}
		if errors.Is(batchErr, ErrSourceRecordInactive) {
			return 1, w.markConfirmedInactiveSkipped(eventCtx, ev, prep)
		}
		err := w.failWithSyncState(eventCtx, ev, prep.objectKind, "batch read", batchErr)
		w.logStageTiming(eventCtx, ev, prep, "error", err)
		return 1, err
	}

	return 1 + coalesced, w.executeEvent(eventCtx, ev, prep, details)
}

func (w *Worker) executeClaimFirstBatchCollection(ctx context.Context, drain <-chan struct{}, ev *connector_events.ConnectorEvent, prep preparedEvent, claimLimit int) (claimedRows int, err error) {
	limit := batchGroupLimit(w.cfg.BatchSize, prep.spec)
	if claimLimit > 0 && claimLimit < limit {
		limit = claimLimit
	}
	if limit < 1 {
		limit = 1
	}
	group := &pendingBatchGroup{
		key:  batchGroupKey(ev, prep.objectKind, prep.spec),
		spec: prep.spec,
		events: []preparedClaimedEvent{{
			ev:   ev,
			prep: prep,
		}},
	}
	// Recover backstop: a panic OUTSIDE the wrapped flow call (candidate
	// peek/claim/prepare, batch read, or executeBatchCollection assembly) while
	// this multi-row group is in flight must finalize the WHOLE group — every
	// claimed member, not just the anchor — so none is stranded in 'processing'.
	// inflightCandidate covers the window where a candidate is CAS-claimed but not
	// yet appended to the group; a panic in its prepareEvent would otherwise strand
	// it, since it is not yet a group member.
	var inflightCandidate *connector_events.ConnectorEvent
	defer func() {
		if r := recover(); r != nil {
			w.finalizeAfterPanic(ctx, group, inflightCandidate, r)
			claimedRows = len(group.events)
			if inflightCandidate != nil {
				claimedRows++
			}
			err = fmt.Errorf("%w: %v", errWorkerPanic, r)
		}
	}()
	if limit > 1 {
		peeked, err := w.peekBatch(ctx, limit-1, false)
		if err != nil {
			return len(group.events), w.failBatchWithSyncState(ctx, group, "batch peek", err)
		}
		defer w.forgetPeekedCandidates(peeked)
		for _, candidate := range peeked {
			if len(group.events) >= limit || drainRequested(drain) {
				break
			}
			if !sameBatchCollectionRoute(ev, candidate) {
				continue
			}
			candidateEnv, envErr := DecodeQueueEnvelope(candidate.Payload)
			if envErr != nil || candidateEnv.ObjectKind != prep.objectKind {
				continue
			}
			if prep.spec.FlowExecutionFilter != nil && sameDeliveryEvent(ev, candidate) {
				break
			}
			claimed, ok, err := w.claimBatchCandidate(ctx, candidate)
			if err != nil {
				return len(group.events), w.failBatchWithSyncState(ctx, group, "batch claim", err)
			}
			if !ok {
				continue
			}
			inflightCandidate = claimed // CAS-claimed, not yet a group member
			sameDeliveryCandidate := sameDeliveryEvent(ev, candidate)
			var candidatePrep preparedEvent
			var prepErr error
			if sameDeliveryCandidate {
				candidatePrep, prepErr = w.prepareEventWithoutFinalizing(ctx, claimed)
			} else {
				candidatePrep, prepErr = w.prepareEvent(ctx, claimed)
			}
			if prepErr != nil {
				if sameDeliveryCandidate {
					if relErr := w.releaseClaim(claimed, "batch candidate prepare", prepErr); relErr != nil {
						return len(group.events), w.failBatchWithSyncState(ctx, group, "batch candidate release", relErr)
					}
				}
				// prepareEvent (non-same-delivery) already finalized the candidate, or
				// releaseClaim (same-delivery) released it — no longer in flight.
				inflightCandidate = nil
				continue
			}
			if candidatePrep.objectKind != prep.objectKind || batchGroupKey(claimed, candidatePrep.objectKind, candidatePrep.spec) != group.key {
				if sameDeliveryCandidate {
					if relErr := w.releaseClaim(claimed, "batch collection group mismatch",
						fmt.Errorf("claimed candidate does not match anchor batch group")); relErr != nil {
						return len(group.events), w.failBatchWithSyncState(ctx, group, "batch candidate release", relErr)
					}
					inflightCandidate = nil
					continue
				}
				_ = w.failTerminal(ctx, claimed, "batch collection group mismatch",
					fmt.Errorf("claimed candidate does not match anchor batch group"))
				inflightCandidate = nil
				continue
			}
			group.events = append(group.events, preparedClaimedEvent{ev: claimed, prep: candidatePrep})
			inflightCandidate = nil // now a group member; the group recover covers it
		}
	}

	sourceIDs := make([]string, 0, len(group.events))
	for _, item := range group.events {
		sourceIDs = append(sourceIDs, item.prep.sourceID)
	}
	batchCtx := w.stageTimingContext(ctx, ev, prep, 0)
	// Heartbeat the whole claimed group across the post-claim vendor read (same
	// reclaim hazard as the single-event path); executeBatchCollection starts its
	// own heartbeat for the flow span.
	outcome, _, deadlineHit, batchErr := w.batchReadOutcomeHeartbeated(batchCtx, leaseRefsForGroup(group), prep.objectKind, sourceIDs)
	if deadlineHit {
		// A wedged post-claim read re-drives the WHOLE claimed group via
		// ReleaseClaimForRetry (no dead-letter burn), the batch mirror of the
		// single-event read-deadline path. A deadline may be source-scoped when it
		// occurs during one confirmation probe, but the shared read context makes
		// the remaining outcome set incomplete.
		byID := make(map[string]preparedClaimedEvent, len(group.events))
		for _, item := range group.events {
			byID[item.ev.ID] = item
		}
		return len(group.events), w.finalizeBatchRetriableOrdered(batchCtx, group, byID, "batch read deadline", errBatchReadDeadlineExceeded)
	}
	if batchErr != nil {
		return len(group.events), w.failBatchWithSyncState(batchCtx, group, "batch read", batchErr)
	}
	for i := range group.events {
		group.events[i].details = outcome.Details
		group.events[i].batchReadErr = outcome.SourceErrors[group.events[i].prep.sourceID]
	}
	return len(group.events), w.executeBatchCollection(batchCtx, group)
}

func (w *Worker) forgetPeekedCandidates(events []*connector_events.ConnectorEvent) {
	forgetter, ok := w.events.(peekedCandidateForgetter)
	if !ok {
		return
	}
	for _, ev := range events {
		if ev != nil {
			forgetter.ForgetPeekedCandidate(ev.ID)
		}
	}
}

func (w *Worker) checkBatchCompanionIncarnation(ctx context.Context, group *pendingBatchGroup) error {
	if group == nil || len(group.events) == 0 || group.events[0].ev.CompanionFlowVersionID == "" {
		return nil
	}
	if w.checkCompanionInstalled == nil {
		return w.failBatchTerminal(ctx, group, "companion incarnation check",
			fmt.Errorf("worker has no companion incarnation checker wired; refusing to execute cross-pack event without incarnation re-check"))
	}
	companionPackInstallID := group.events[0].ev.CompanionPackInstallID
	present, checkErr := w.checkCompanionInstalled(ctx, companionPackInstallID)
	if checkErr != nil {
		return w.failBatchTerminal(ctx, group, "companion incarnation check",
			fmt.Errorf("companion incarnation re-check DB error: %w", checkErr))
	}
	if !present {
		return w.failBatchTerminal(ctx, group, "companion incarnation check",
			fmt.Errorf("companion incarnation is not active since enqueue: PackInstallId %q no longer eligible (companion was reborn or removed)", companionPackInstallID))
	}
	return nil
}

func (w *Worker) failBatchWithSyncState(ctx context.Context, group *pendingBatchGroup, stage string, cause error) error {
	blockedDeliveries := make(map[string]bool)
	var first error
	for _, item := range group.events {
		if item.ev.DeliveryId != "" && blockedDeliveries[item.ev.DeliveryId] {
			if rerr := w.releaseClaim(item.ev, stage+" deferred by same-delivery ordering", errors.New("earlier same-delivery batch member did not reach terminal success")); rerr != nil && first == nil {
				first = rerr
			}
			continue
		}
		if ferr := w.failWithSyncState(ctx, item.ev, item.prep.objectKind, stage, cause); ferr != nil && first == nil {
			first = ferr
		}
		if item.ev.DeliveryId != "" {
			blockedDeliveries[item.ev.DeliveryId] = true
		}
	}
	if first != nil {
		return first
	}
	return fmt.Errorf("%s: %w", stage, cause)
}

func (w *Worker) failBatchTerminal(ctx context.Context, group *pendingBatchGroup, stage string, cause error) error {
	var first error
	for _, item := range group.events {
		if ferr := w.failTerminal(ctx, item.ev, stage, cause); ferr != nil && first == nil {
			first = ferr
		}
	}
	if first != nil {
		return first
	}
	return fmt.Errorf("%s: %w", stage, cause)
}

func sameBatchCollectionRoute(a, b *connector_events.ConnectorEvent) bool {
	if a == nil || b == nil || !a.RouteVersion.Valid || !b.RouteVersion.Valid {
		return false
	}
	return a.InstalledPackID == b.InstalledPackID &&
		a.EventKind == b.EventKind &&
		a.RouteVersion.Int64 == b.RouteVersion.Int64 &&
		a.CompanionPackInstallID == b.CompanionPackInstallID &&
		a.CompanionFlowVersionID == b.CompanionFlowVersionID
}

func (w *Worker) batchReadWithStageTiming(ctx context.Context, objectKind string, sourceIDs []string) (map[string]json.RawMessage, time.Duration, error) {
	if !w.cfg.EnableStageTiming {
		details, err := w.batchRead.BatchRead(ctx, w.cfg.Namespace, objectKind, sourceIDs)
		return details, 0, err
	}
	started := time.Now()
	details, err := w.batchRead.BatchRead(ctx, w.cfg.Namespace, objectKind, sourceIDs)
	elapsed := time.Since(started)
	stagetiming.Add(ctx, stagetiming.StageBatchRead, elapsed)
	return details, elapsed, err
}

func (w *Worker) batchReadOutcomeWithStageTiming(ctx context.Context, objectKind string, sourceIDs []string) (BatchReadOutcome, time.Duration, error) {
	if !w.cfg.EnableStageTiming {
		outcome, err := ReadBatchOutcome(ctx, w.batchRead, w.cfg.Namespace, objectKind, sourceIDs)
		return outcome, 0, err
	}
	started := time.Now()
	outcome, err := ReadBatchOutcome(ctx, w.batchRead, w.cfg.Namespace, objectKind, sourceIDs)
	elapsed := time.Since(started)
	stagetiming.Add(ctx, stagetiming.StageBatchRead, elapsed)
	return outcome, elapsed, err
}

func (w *Worker) stageTimingContext(ctx context.Context, ev *connector_events.ConnectorEvent, prep preparedEvent, batchReadElapsed time.Duration) context.Context {
	if !w.cfg.EnableStageTiming {
		return ctx
	}
	collector := stagetiming.NewCollector()
	if batchReadElapsed > 0 {
		collector.Add(stagetiming.StageBatchRead, batchReadElapsed)
	}
	if ctx == nil {
		ctx = workerBackgroundContext()
	}
	ctx = stagetiming.WithCollector(ctx, collector)
	ctx = context.WithValue(ctx, constants.ContextKeyPerfTiming, true)
	if _, ok := ctx.Value(constants.ContextKeyRequestID).(string); !ok && ev != nil && ev.ID != "" {
		ctx = context.WithValue(ctx, constants.ContextKeyRequestID, "connector-inbound:"+ev.ID)
	}
	return ctx
}

func (w *Worker) logStageTiming(ctx context.Context, ev *connector_events.ConnectorEvent, prep preparedEvent, status string, err error) {
	if !w.cfg.EnableStageTiming || stagetiming.FromContext(ctx) == nil || ev == nil {
		return
	}
	if status == "" {
		status = "success"
	}
	snapshot := stagetiming.FromContext(ctx).Snapshot()
	claimWaitMS := connectorInboundClaimWaitMS(ev, time.Now())
	event := zlog.Info().
		Str("event", "connector_inbound_stage_timing").
		Str("request_id", connectorInboundRequestID(ctx)).
		Str("namespace", w.cfg.Namespace).
		Str("installed_pack_id", ev.InstalledPackID).
		Str("connector_event_id", ev.ID).
		Str("event_kind", ev.EventKind).
		Str("object_kind", prep.objectKind).
		Str("flow_name", prep.flowName).
		Str("source_hash", connectorInboundSourceHash(w.cfg.Namespace, prep.objectKind, prep.sourceID)).
		Str("status", status).
		Bool("has_error", err != nil).
		Int64("claim_wait_ms", claimWaitMS).
		Int64("batch_read_ms", snapshot.Milliseconds(stagetiming.StageBatchRead)).
		Int64("flow_exec_ms", snapshot.Milliseconds(stagetiming.StageFlowExec)).
		Int64("record_lookup_ms", snapshot.Milliseconds(stagetiming.StageRecordLookup)).
		Int64("record_mutation_ms", snapshot.Milliseconds(stagetiming.StageRecordMutation)).
		Int64("mapping_ms", snapshot.Milliseconds(stagetiming.StageMapping)).
		Int("batch_read_count", snapshot.Count(stagetiming.StageBatchRead)).
		Int("record_lookup_count", snapshot.Count(stagetiming.StageRecordLookup)).
		Int("record_mutation_count", snapshot.Count(stagetiming.StageRecordMutation)).
		Int("mapping_count", snapshot.Count(stagetiming.StageMapping))
	if err != nil {
		event = event.Err(err)
	}
	event.Msg("Connector inbound stage timing")
}

func (w *Worker) logBatchStageTiming(ctx context.Context, group *pendingBatchGroup, status string, err error) {
	if !w.cfg.EnableStageTiming || stagetiming.FromContext(ctx) == nil || group == nil || len(group.events) == 0 {
		return
	}
	if status == "" {
		status = "success"
	}
	anchor := group.events[0]
	if anchor.ev == nil {
		return
	}
	snapshot := stagetiming.FromContext(ctx).Snapshot()
	execTiming := connectorInboundExecutionTimingFromContext(ctx)
	totalMS := snapshot.Milliseconds(stagetiming.StageBatchRead) +
		snapshot.Milliseconds(stagetiming.StageFlowExec) +
		snapshot.Milliseconds(stagetiming.StageRecordLookup) +
		snapshot.Milliseconds(stagetiming.StageRecordMutation) +
		snapshot.Milliseconds(stagetiming.StageMapping)
	event := zlog.Info().
		Str("event", "connector_inbound_batch_stage_timing").
		Int64("elapsed_ms", totalMS).
		Str("request_id", connectorInboundRequestID(ctx)).
		Str("namespace", w.cfg.Namespace).
		Str("installed_pack_id", anchor.ev.InstalledPackID).
		Str("worker_id", w.cfg.WorkerID).
		Strs("queue_classes", w.cfg.QueueClasses).
		Str("claim_path", string(w.cfg.ClaimPath)).
		Str("anchor_connector_event_id", anchor.ev.ID).
		Str("event_kind", anchor.ev.EventKind).
		Str("queue_class", anchor.ev.QueueClass).
		Str("object_kind", anchor.prep.objectKind).
		Str("flow_name", group.spec.FlowDeveloperName).
		Int("batch_group_size", len(group.events)).
		Str("status", status).
		Bool("has_error", err != nil).
		Int64("batch_read_ms", snapshot.Milliseconds(stagetiming.StageBatchRead)).
		Int64("flow_exec_ms", snapshot.Milliseconds(stagetiming.StageFlowExec)).
		Int64("record_lookup_ms", snapshot.Milliseconds(stagetiming.StageRecordLookup)).
		Int64("record_mutation_ms", snapshot.Milliseconds(stagetiming.StageRecordMutation)).
		Int64("mapping_ms", snapshot.Milliseconds(stagetiming.StageMapping)).
		Int("batch_read_count", snapshot.Count(stagetiming.StageBatchRead)).
		Int("record_lookup_count", snapshot.Count(stagetiming.StageRecordLookup)).
		Int("record_mutation_count", snapshot.Count(stagetiming.StageRecordMutation)).
		Int("mapping_count", snapshot.Count(stagetiming.StageMapping))
	if execTiming != nil {
		event = event.
			Str("flow_path", execTiming.flowPath).
			Int64("build_inputs_ms", execTiming.buildInputsMS).
			Int64("flow_call_ms", execTiming.flowCallMS).
			Int64("finalize_ms", execTiming.finalizeMS).
			Int64("heartbeat_stop_ms", execTiming.heartbeatStopMS)
	}
	if err != nil {
		event = event.Err(err)
	}
	event.Msg("Connector inbound batch stage timing")
}

func connectorInboundClaimWaitMS(ev *connector_events.ConnectorEvent, fallback time.Time) int64 {
	if ev == nil || ev.CreatedDate.IsZero() {
		return 0
	}
	claimedAt := fallback
	if ev.ClaimedAt != nil && !ev.ClaimedAt.IsZero() {
		claimedAt = *ev.ClaimedAt
	}
	elapsed := claimedAt.Sub(ev.CreatedDate)
	if elapsed < 0 {
		return 0
	}
	return elapsed.Milliseconds()
}

func connectorInboundRequestID(ctx context.Context) string {
	if id, ok := ctx.Value(constants.ContextKeyRequestID).(string); ok {
		return id
	}
	return ""
}

func connectorInboundSourceHash(namespace, objectKind, sourceID string) string {
	if sourceID == "" {
		return ""
	}
	h := sha256.New()
	_, _ = h.Write([]byte(namespace))
	_, _ = h.Write([]byte{0})
	_, _ = h.Write([]byte(objectKind))
	_, _ = h.Write([]byte{0})
	_, _ = h.Write([]byte(sourceID))
	return hex.EncodeToString(h.Sum(nil))
}

func (w *Worker) withConnectorInboundExecutionTrace(
	ctx context.Context,
	ev *connector_events.ConnectorEvent,
	prep preparedEvent,
	inputs map[string]any,
) context.Context {
	if ev == nil {
		return ctx
	}
	item := w.connectorInboundExecutionTraceItem(ev, prep, inputs)
	return constants.WithConnectorInboundTrace(ctx, constants.ConnectorInboundTrace{
		WorkItemID:         item.WorkItemID,
		EventID:            item.EventID,
		Attempt:            item.Attempt,
		Namespace:          w.cfg.Namespace,
		ObjectKind:         item.ObjectKind,
		SourceIDHash:       item.SourceIDHash,
		HydratedRecordHash: item.HydratedRecordHash,
		HydratedFieldCount: item.HydratedFieldCount,
	})
}

func (w *Worker) runtimeCommandHandoffScope(
	ctx context.Context,
	ev *connector_events.ConnectorEvent,
) RuntimeCommandHandoffScope {
	if ev == nil {
		return RuntimeCommandHandoffScope{}
	}
	trace, _ := constants.ConnectorInboundTraceFromContext(ctx)
	workItemID := trace.WorkItemID
	if workItemID == "" {
		workItemID, _ = w.runtimeCommandHandoffWorkItemID(ev)
	}
	var routeVersion uint64
	if ev.RouteVersion.Valid && ev.RouteVersion.Int64 > 0 {
		routeVersion = uint64(ev.RouteVersion.Int64)
	}
	return RuntimeCommandHandoffScope{
		WorkItemID:      workItemID,
		EventID:         ev.ID,
		LeaseToken:      ev.ClaimToken,
		Namespace:       ev.Namespace,
		InstalledPackID: ev.InstalledPackID,
		EventKind:       ev.EventKind,
		SourcePartition: ev.SourceLaneKey,
		RouteVersion:    routeVersion,
	}.normalized()
}

func (w *Worker) runtimeCommandHandoffWorkItemID(
	ev *connector_events.ConnectorEvent,
) (string, bool) {
	if w == nil || ev == nil {
		return "", false
	}
	resolver, ok := w.events.(connectorInboundWorkItemTraceResolver)
	if !ok {
		return "", false
	}
	return resolver.WorkItemIDForEvent(ev.ID)
}

func runtimeCommandHandoffSupportsClaimedWorkItem(
	claimPath ClaimPath,
	workItemID string,
	invocationMode string,
) bool {
	if !RuntimeCommandHandoffSupportsInvocationMode(invocationMode) {
		return false
	}
	switch invocationMode {
	case connector_inbound_routing.InvocationModePerEvent:
		return true
	case connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled:
		return normalizeClaimPath(claimPath) == ClaimPathWorkItem &&
			connector_workqueue.IsSourceProjectionWorkID(workItemID)
	default:
		return false
	}
}

func (w *Worker) connectorInboundExecutionTraceItem(
	ev *connector_events.ConnectorEvent,
	prep preparedEvent,
	inputs map[string]any,
) constants.ConnectorInboundTraceItem {
	if ev == nil {
		return constants.ConnectorInboundTraceItem{}
	}
	workItemID := ""
	attempt := 0
	if resolver, ok := w.events.(connectorInboundWorkItemTraceResolver); ok {
		workItemID, _ = resolver.WorkItemIDForEvent(ev.ID)
	}
	if resolver, ok := w.events.(connectorInboundWorkItemAttemptTraceResolver); ok {
		attempt, _ = resolver.WorkItemAttemptForEvent(ev.ID)
	}
	hydratedHash, hydratedFields := connectorInboundHydratedRecordFingerprint(inputs)
	return constants.ConnectorInboundTraceItem{
		WorkItemID:         workItemID,
		EventID:            ev.ID,
		Attempt:            attempt,
		ObjectKind:         prep.objectKind,
		SourceIDHash:       connectorInboundRedactedHash(prep.sourceID),
		HydratedRecordHash: hydratedHash,
		HydratedFieldCount: hydratedFields,
	}
}

func (w *Worker) withConnectorInboundBatchExecutionTrace(
	ctx context.Context,
	items []constants.ConnectorInboundTraceItem,
) context.Context {
	if len(items) == 0 {
		return ctx
	}
	primary := items[0]
	related := append([]constants.ConnectorInboundTraceItem(nil), items[1:]...)
	return constants.WithConnectorInboundTrace(ctx, constants.ConnectorInboundTrace{
		WorkItemID:         primary.WorkItemID,
		EventID:            primary.EventID,
		Attempt:            primary.Attempt,
		Namespace:          w.cfg.Namespace,
		ObjectKind:         primary.ObjectKind,
		SourceIDHash:       primary.SourceIDHash,
		HydratedRecordHash: primary.HydratedRecordHash,
		HydratedFieldCount: primary.HydratedFieldCount,
		Related:            related,
	})
}

func connectorInboundHydratedRecordFingerprint(inputs map[string]any) (string, int) {
	record, ok := inputs["record"]
	if !ok || record == nil {
		return "", 0
	}
	encoded, err := json.Marshal(record)
	if err != nil {
		return "", 0
	}
	fieldCount := 0
	if fields, ok := record.(map[string]any); ok {
		fieldCount = len(fields)
	}
	return connectorInboundRedactedHash(string(encoded)), fieldCount
}

func connectorInboundRedactedHash(value string) string {
	if value == "" {
		return ""
	}
	sum := sha256.Sum256([]byte(value))
	return "sha256:" + hex.EncodeToString(sum[:])
}

func (w *Worker) markProcessed(ev *connector_events.ConnectorEvent, parent ...context.Context) error {
	ctx, cancel := w.finalizeContext(parent...)
	defer cancel()
	if err := w.events.MarkProcessed(ctx, ev.ID, ev.ClaimToken); err != nil {
		return err
	}
	w.wakeSchedulerAfterFinalize()
	return nil
}

func (w *Worker) markSkipped(ev *connector_events.ConnectorEvent) error {
	ctx, cancel := w.finalizeContext()
	defer cancel()
	if err := w.events.MarkSkipped(ctx, ev.ID, ev.ClaimToken); err != nil {
		return err
	}
	w.wakeSchedulerAfterFinalize()
	return nil
}

func (w *Worker) markConfirmedInactiveSkipped(ctx context.Context, ev *connector_events.ConnectorEvent, prep preparedEvent) error {
	if err := w.markSkipped(ev); err != nil {
		wrapped := fmt.Errorf("MarkSkipped after confirmed inactive source record: %w", err)
		logEvent := zlog.Error().
			Str("event", "connector_inbound_confirmed_inactive_finalize_failed").
			Str("namespace", w.cfg.Namespace).
			Str("worker_id", w.cfg.WorkerID).
			Str("object_kind", prep.objectKind).
			Str("source_hash", connectorInboundSourceHash(w.cfg.Namespace, prep.objectKind, prep.sourceID)).
			Err(wrapped)
		if ev != nil {
			logEvent = logEvent.
				Str("installed_pack_id", ev.InstalledPackID).
				Str("connector_event_id", ev.ID).
				Str("event_kind", ev.EventKind)
		}
		logEvent.Msg("Connector inbound could not finalize confirmed inactive source as skipped; leaving the lease for normal recovery")
		w.logStageTiming(ctx, ev, prep, "error", wrapped)
		return wrapped
	}
	w.logStageTiming(ctx, ev, prep, "skipped", nil)
	return nil
}

func (w *Worker) releaseClaim(ev *connector_events.ConnectorEvent, stage string, cause error) error {
	releaser, ok := w.events.(claimReleasingEventsService)
	if !ok {
		return fmt.Errorf("ReleaseClaim unavailable after %s: %v", stage, cause)
	}
	ctx, cancel := w.finalizeContext()
	defer cancel()
	if err := releaser.ReleaseClaim(ctx, ev.ID, ev.ClaimToken); err != nil {
		return fmt.Errorf("ReleaseClaim after %s: %w (original: %v)", stage, err, cause)
	}
	w.wakeSchedulerAfterFinalize()
	return nil
}

// releaseClaimForRetry re-drives a retriable flow fault (IsRetriable on a failed
// result) without consuming the genuine-failure retry budget. It returns the
// leased row to the queue with a backoff (events.ReleaseClaimForRetry sets
// 'failed' + NextAttemptAt) instead of calling MarkFailed, so a concurrent
// create-vs-adopt loser waits for the in-flight winner and self-heals on the
// next claim rather than dead-lettering at MaxRetryAttempts before the winner's
// processing lease expires. Returns the underlying cause so the caller surfaces
// a non-nil error (the event was not finalized this attempt).
func (w *Worker) releaseClaimForRetry(ev *connector_events.ConnectorEvent, stage string, cause error, parent ...context.Context) error {
	releaser, ok := w.events.(claimReleasingEventsService)
	if !ok {
		return fmt.Errorf("ReleaseClaimForRetry unavailable after %s: %v", stage, cause)
	}
	ctx, cancel := w.finalizeContext(parent...)
	defer cancel()
	if err := releaser.ReleaseClaimForRetry(ctx, ev.ID, ev.ClaimToken, cause.Error()); err != nil {
		return fmt.Errorf("ReleaseClaimForRetry after %s: %w (original: %v)", stage, err, cause)
	}
	w.wakeSchedulerAfterFinalize()
	return fmt.Errorf("%s: %w", stage, cause)
}

// failTerminal calls MarkFailed without writing SyncState. Used for
// parser-bug–class failures where the ObjectKind cannot be determined (so
// there is no object identity to record an error against in SyncState).
func (w *Worker) failTerminal(ctx context.Context, ev *connector_events.ConnectorEvent, stage string, cause error) error {
	msg := fmt.Sprintf("terminal failure at %s: %v", stage, cause)
	finalizeCtx, cancel := w.finalizeContext(ctx)
	defer cancel()
	if err := w.events.MarkFailed(finalizeCtx, ev.ID, ev.ClaimToken, msg); err != nil {
		return fmt.Errorf("MarkFailed after terminal %s: %w (original: %v)", stage, err, cause)
	}
	w.wakeSchedulerAfterFinalize()
	return fmt.Errorf("%s: %w", stage, cause)
}

// failWithSyncState writes a SyncState error record (best-effort) then calls
// MarkFailed (authoritative). Both failures are reported in the returned error.
//
// SyncState write is best-effort: if it fails, MarkFailed still runs and the
// queue row is the SSoT for failure state.
func (w *Worker) failWithSyncState(ctx context.Context, ev *connector_events.ConnectorEvent, objectKind, stage string, cause error) error {
	msg := fmt.Sprintf("worker %s: %s: %v", w.cfg.WorkerID, stage, cause)

	// Best-effort SyncState write.
	_ = w.syncstate.RecordError(ctx, w.cfg.Namespace, objectKind, "inbound", msg)

	finalizeCtx, cancel := w.finalizeContext(ctx)
	defer cancel()
	if err := w.events.MarkFailed(finalizeCtx, ev.ID, ev.ClaimToken, msg); err != nil {
		return fmt.Errorf("MarkFailed after %s: %w (original: %v)", stage, err, cause)
	}
	w.wakeSchedulerAfterFinalize()
	return fmt.Errorf("%s: %w", stage, cause)
}

func (w *Worker) wakeSchedulerAfterFinalize() {
	if w == nil || w.scheduler == nil {
		return
	}
	wakeable, ok := schedulerOrDefault(w.scheduler).(wakeableScheduler)
	if !ok {
		return
	}
	wakeable.Wake()
}

// finalizeAfterPanic is the recover backstop's finalizer for a panic raised
// OUTSIDE the wrapped flow call (prepare/coalesce/buildFlowInputs/batchRead/
// decode). It best-effort finalizes the WHOLE in-flight unit so no claimed row
// is left in 'processing' to re-surface at the 300s stale-reclaim window and
// crash-loop the process:
//   - a batch group in flight → EVERY member is MarkFailed via
//     failBatchWithSyncState (NOT just the row the panic happened on, which
//     would strand the other N-1 for 300s);
//   - a loose single claimed row → MarkFailed via failWithSyncState.
//
// MarkFailed (the budget path, dead at MaxRetryAttempts) is used deliberately so
// a deterministic poison dead-letters instead of re-firing forever. The log
// carries the stack + event id/namespace/ordinal only — NEVER the raw payload
// (public-core + secret-leak boundary).
func (w *Worker) finalizeAfterPanic(ctx context.Context, pendingBatch *pendingBatchGroup, inflight *connector_events.ConnectorEvent, r any) {
	cause := fmt.Errorf("%w: %v", errWorkerPanic, r)

	logEvent := zlog.Error().
		Str("event", "connector_inbound_worker_panic_recovered").
		Str("namespace", w.cfg.Namespace).
		Str("worker_id", w.cfg.WorkerID).
		Str("panic", fmt.Sprintf("%v", r)).
		Bytes("stack", debug.Stack())
	if inflight != nil {
		logEvent = logEvent.
			Str("inflight_event_id", inflight.ID).
			Int("inflight_ordinal", inflight.DeliverySequence)
	}
	if pendingBatch != nil && len(pendingBatch.events) > 0 {
		ids := make([]string, 0, len(pendingBatch.events))
		for _, item := range pendingBatch.events {
			ids = append(ids, item.ev.ID)
		}
		logEvent = logEvent.Strs("batch_event_ids", ids)
	}
	logEvent.Msg("connector inbound worker recovered from panic; finalizing in-flight unit")

	if pendingBatch != nil && len(pendingBatch.events) > 0 {
		_ = w.failBatchWithSyncState(ctx, pendingBatch, "panic recovery", cause)
	}
	if inflight != nil {
		// A same-delivery loose candidate (an earlier sibling is still pending in the
		// batch/group and has not reached terminal success) must preserve the
		// release-until-earlier-success ordering contract: release the claim — no
		// retry-budget burn, no premature dead-letter — mirroring the non-panic
		// candidate-prepare-error path (RunUntilDrained and claim-first batch
		// collection both releaseClaim a same-delivery candidate). Only a genuinely
		// standalone in-flight row is MarkFailed for poison dead-lettering.
		if pendingBatch != nil && len(pendingBatch.events) > 0 && sameDeliveryBatch(pendingBatch, inflight) {
			if relErr := w.releaseClaim(inflight, "panic recovery (same-delivery candidate)", cause); relErr != nil {
				_ = w.failWithSyncState(ctx, inflight, "", "panic recovery (same-delivery release failed)", relErr)
			}
		} else {
			_ = w.failWithSyncState(ctx, inflight, "", "panic recovery", cause)
		}
	}
}

func (w *Worker) finalizeContext(parent ...context.Context) (context.Context, context.CancelFunc) {
	timeout := w.cfg.FinalizeTimeout
	if timeout <= 0 {
		timeout = defaultFinalizeTimeout
	}
	if len(parent) == 0 || parent[0] == nil {
		return context.WithTimeout(workerBackgroundContext(), timeout)
	}
	return context.WithTimeout(context.WithoutCancel(parent[0]), timeout)
}

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

func drainRequested(drain <-chan struct{}) bool {
	if drain == nil {
		return false
	}
	select {
	case <-drain:
		return true
	default:
		return false
	}
}

func cancelContextOnDrain(cancel context.CancelFunc, drain <-chan struct{}) func() {
	if drain == nil {
		return func() {}
	}
	done := make(chan struct{})
	go func() {
		select {
		case <-drain:
			cancel()
		case <-done:
		}
	}()
	return func() { close(done) }
}

func stopTimer(timer *time.Timer) {
	if timer.Stop() {
		return
	}
	select {
	case <-timer.C:
	default:
	}
}

// buildFlowInputs constructs the Flow input map passed to [FlowInvoker.ExecuteFlow].
// The map always includes the raw event fields; details (from BatchRead) are
// merged in when non-nil. sourceID is the external object ID extracted from the
// QueueEnvelope — it is passed as the "sourceID" input and used as the lookup
// key into the details map returned by BatchRead.
//
// userPayload is the inner payload bytes from the QueueEnvelope. payload and record are
// unmarshalled to map[string]any so that Flow formulas can navigate into nested
// fields (e.g. {!payload.objectId}). Passing a JSON string to FlowExecutor would
// require an extra decode step inside the formula engine and is not supported.
// Returns an error when userPayload is not valid JSON — this indicates a
// parser/dispatcher bug and the event should be failed terminally.
func buildFlowInputs(ev *connector_events.ConnectorEvent, objectKind, sourceID string, userPayload json.RawMessage, details map[string]json.RawMessage, requireRecord bool, coalescedPayloads []coalescedPayloadInput) (map[string]any, error) {
	var payloadMap map[string]any
	if err := json.Unmarshal(userPayload, &payloadMap); err != nil {
		return nil, fmt.Errorf("payload unmarshal: %w", err)
	}
	inputs := map[string]any{
		"eventID":        ev.ID,
		"eventKind":      ev.EventKind,
		"namespace":      ev.Namespace,
		"objectKind":     objectKind,
		"sourceID":       sourceID,
		"idempotencyKey": ev.IdempotencyKey,
		"payload":        payloadMap,
	}
	if len(coalescedPayloads) > 0 {
		values, err := buildCoalescedPayloadInputs(coalescedPayloads)
		if err != nil {
			return nil, err
		}
		inputs["coalescedPayloads"] = values
	}
	if details != nil {
		if rec, ok := details[sourceID]; ok {
			var recordMap map[string]any
			if err := json.Unmarshal(rec, &recordMap); err != nil {
				// Wrap with ErrBatchReadRecordShape so processOne can distinguish
				// this INT-10 boundary fault (retriable) from payload corruption
				// (terminal). The dispatcher validated payload shape before enqueue,
				// so a decode failure here is at the batch-read boundary.
				return nil, fmt.Errorf("%w: source=%q: %v", ErrBatchReadRecordShape, sourceID, err)
			}
			// json.Unmarshal of JSON null succeeds but yields a nil map. A present-but-null
			// entry is NOT the same as an absent entry ("record not fetchable" is represented
			// by absence from the map, not by a null value). A null record from BatchRead is
			// an INT-10 boundary fault — fail retriably so the queue retries.
			if recordMap == nil {
				return nil, fmt.Errorf("%w: source=%q: batch read returned null record (absent entry means not-fetchable; null is INT-10 boundary fault)", ErrBatchReadRecordShape, sourceID)
			}
			inputs["record"] = recordMap
		}
	}
	if requireRecord {
		if _, ok := inputs["record"]; !ok {
			return nil, fmt.Errorf("%w: source=%q: required enrichment record missing", ErrBatchReadRecordShape, sourceID)
		}
	}
	return inputs, nil
}

func buildCoalescedPayloadInputs(coalescedPayloads []coalescedPayloadInput) ([]any, error) {
	values := make([]any, 0, len(coalescedPayloads))
	for _, item := range coalescedPayloads {
		var payloadMap map[string]any
		if err := json.Unmarshal(item.payload, &payloadMap); err != nil {
			return nil, fmt.Errorf("coalesced payload unmarshal: eventID=%q: %w", item.eventID, err)
		}
		values = append(values, map[string]any{
			"eventID":        item.eventID,
			"eventKind":      item.eventKind,
			"namespace":      item.namespace,
			"objectKind":     item.objectKind,
			"sourceID":       item.sourceID,
			"idempotencyKey": item.idempotencyKey,
			"payload":        payloadMap,
		})
	}
	return values, nil
}

// flowFailureMessage builds a human-readable failure string from flow result.
func flowFailureMessage(flowName string, result *FlowResult, err error) string {
	if err != nil {
		return fmt.Sprintf("flow %q returned error: %v", flowName, err)
	}
	if result == nil {
		return fmt.Sprintf("flow %q returned nil result", flowName)
	}
	return fmt.Sprintf("flow %q failed: %s", flowName, result.failureMessage())
}

// flowRetriableMessage builds a human-readable message for a retriable-success
// outcome (IsSuccess=true, IsRetriable=true). The worker uses this as the
// MarkFailed error message so operators can distinguish retriable no-ops from
// permanent failures in LastError / SyncState.
func flowRetriableMessage(flowName string, result *FlowResult) string {
	if result != nil && result.FirstError != nil {
		return fmt.Sprintf("flow %q signaled retriable: %s: %s", flowName, result.FirstError.StatusCode, result.FirstError.Message)
	}
	return fmt.Sprintf("flow %q signaled retriable condition (dependency not yet satisfied)", flowName)
}

// Ensure Worker satisfies no exported interface (it is used by value, not interface).
// Ensure EventsService is satisfied by *connector_events.Service at compile time.
var _ EventsService = (*connector_events.Service)(nil)

// systemUserStub is a compile-time check that FlowInvoker.SystemUser returns
// *models.UserSession — present only to satisfy the import.
var _ *models.UserSession = (*models.UserSession)(nil)