package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"strings"

	"github.com/nexuscrm/persistence"
)

// migrationIDVerifyConnectorWorkItemOnlyExecutionCutover is the release
// receipt consumed by the WorkItem-only connector supervisor. It is separate
// from the index-removal receipt: an index can be absent while an old process
// can still enqueue Events without a WorkItem.
const (
	migrationIDVerifyConnectorWorkItemOnlyExecutionCutover          = "20260724_verify_connector_workitem_only_execution_cutover_v2"
	migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover     = 2
	migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover = "Atomically hand off connector execution to WorkItems after terminalizing Event-owned backlog while ingress is durably quiesced (NexusCRM #3980)."
)

// connectorIngressPackCutoverLockSQL takes every inbound pack row lock in one
// pessimistic transaction. Active rows use the same lock as the connector
// enqueue transaction before it writes an Event and its WorkItem; non-active
// rows cannot admit ingress but can preserve abandoned Event-owned backlog.
// Retained Event namespaces are selected even when their route was deleted, so
// a known pack cannot strand historical Events behind the global receipt.
const connectorIngressPackCutoverLockSQL = `
SELECT p.Name, p.PackInstallId, p.Status
  FROM _System_InstalledPack p
 WHERE COALESCE(p.IsDeleted, 0) = 0
   AND (
       EXISTS (
           SELECT 1
            FROM _System_ConnectorInboundRoute r
           WHERE COALESCE(r.IsDeleted, 0) = 0
             AND LOWER(TRIM(r.Namespace)) = LOWER(TRIM(p.Name))
       ) OR EXISTS (
           SELECT 1
            FROM _System_ConnectorEvent e
       WHERE e.ExecutionOwner = 'event'
             AND e.Status IN ('pending', 'failed', 'processing')
             AND LOWER(TRIM(e.Namespace)) = LOWER(TRIM(p.Name))
       ) OR EXISTS (
           SELECT 1
            FROM _System_ConnectorWorkItem w
           WHERE w.InstalledPackId = p.PackInstallId
             AND w.WorkType IN ('connector_event', 'workflow_action_apply')
             AND w.Attempt > 0
             AND COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')
       )
   )
 ORDER BY p.Name
 FOR UPDATE`

// connectorIngressControlCutoverLockSQL locks every admission control row that
// can affect an active inbound connector pack. The cutover requires one
// namespace-wide non-active '*' row and rejects any active override, so no
// webhook can pass admission while the pack locks and receipt are held.
const connectorIngressControlCutoverLockSQL = `
SELECT COALESCE(PackInstallId, ''), Channel, Mode
  FROM _System_ConnectorIngressControl
 WHERE LOWER(TRIM(Namespace)) = LOWER(TRIM(?))
   AND (PackInstallId = ? OR COALESCE(PackInstallId, '') = '')
   AND COALESCE(IsDeleted, 0) = 0
   AND (ExpiresAt IS NULL OR ExpiresAt > UTC_TIMESTAMP())
 FOR UPDATE`

const connectorLegacyEventCutoverLockSQL = `
SELECT Id, Status
  FROM _System_ConnectorEvent
 WHERE LOWER(TRIM(Namespace)) = LOWER(TRIM(?))
   AND ExecutionOwner = 'event'
   AND Status IN ('pending', 'failed', 'processing')
 FOR UPDATE`

// connectorLegacyWorkItemCutoverLockSQL retires only the legacy Event-owned
// side of the rolling authority model. A missing or stale Event link is
// ambiguous rather than disposable and is rejected by the global residual
// proof; an already WorkItem-owned row remains native execution authority.
const connectorLegacyWorkItemCutoverLockSQL = `
SELECT w.WorkId, w.State
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
 WHERE w.WorkType IN ('connector_event', 'workflow_action_apply')
   AND COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')
	AND COALESCE(w.Attempt, 0) = 0
   AND e.ExecutionOwner = 'event'
   AND (
       (
           w.InstalledPackId = ?
           AND e.Status IN ('pending', 'failed')
       ) OR (
           LOWER(TRIM(e.Namespace)) = LOWER(TRIM(?))
           AND e.Status IN ('pending', 'failed')
       )
   )
 FOR UPDATE`

const connectorLegacyWorkItemCutoverRetireSQL = `
UPDATE _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   SET w.State = 'skipped',
       w.LeaseOwner = NULL,
       w.LeaseToken = NULL,
       w.LeaseUntil = NULL,
       w.TerminalAt = COALESCE(w.TerminalAt, UTC_TIMESTAMP()),
       w.LastModifiedDate = UTC_TIMESTAMP()
 WHERE w.WorkType IN ('connector_event', 'workflow_action_apply')
   AND COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead', 'processing')
	AND COALESCE(w.Attempt, 0) = 0
   AND e.ExecutionOwner = 'event'
   AND (
       (
           w.InstalledPackId = ?
           AND e.Status IN ('pending', 'failed')
       ) OR (
           LOWER(TRIM(e.Namespace)) = LOWER(TRIM(?))
           AND e.Status IN ('pending', 'failed')
       )
   )`

const connectorLegacyEventCutoverRetireSQL = `
UPDATE _System_ConnectorEvent
   SET Status = 'skipped',
       ExecutionOwner = 'work_item',
       NextAttemptAt = NULL,
       ClaimedBy = NULL,
       ClaimedAt = NULL,
       ClaimToken = NULL,
       LastModifiedDate = UTC_TIMESTAMP()
 WHERE LOWER(TRIM(Namespace)) = LOWER(TRIM(?))
   AND ExecutionOwner = 'event'
   AND Status IN ('pending', 'failed')`

var connectorEventAuthorityResidualCutoverLockSQL = `
SELECT e.Id, e.Namespace
  FROM _System_ConnectorEvent e
  LEFT JOIN _System_InstalledPack p
    ON p.PackInstallId = e.InstalledPackId
   AND COALESCE(p.IsDeleted, 0) = 0
 WHERE ` + connectorNonterminalEventPredicate("e") + `
   AND (
       COALESCE(e.ExecutionOwner, '') <> 'work_item'
       OR NULLIF(TRIM(COALESCE(e.InstalledPackId, '')), '') IS NULL
       OR p.PackInstallId IS NULL
       OR NOT EXISTS (
           SELECT 1
             FROM _System_ConnectorWorkItem w
            WHERE w.EventId = e.Id
              AND w.InstalledPackId = e.InstalledPackId
              AND w.WorkType IN ('connector_event', 'workflow_action_apply')
              AND COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')
       )
   )
 FOR UPDATE`

var connectorWorkItemAuthorityResidualCutoverLockSQL = `
SELECT w.WorkId, COALESCE(e.Namespace, p.Name, '')
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
  LEFT JOIN _System_InstalledPack p
    ON p.PackInstallId = w.InstalledPackId
   AND COALESCE(p.IsDeleted, 0) = 0
 WHERE w.WorkType IN ('connector_event', 'workflow_action_apply')
   AND ` + connectorNonterminalWorkItemPredicate("w") + `
   AND ` + connectorWorkItemExecutionAuthorityPredicate("w", "e") + `
   AND (
       NULLIF(TRIM(COALESCE(w.InstalledPackId, '')), '') IS NULL
       OR p.PackInstallId IS NULL
       OR e.Id IS NULL
       OR COALESCE(e.ExecutionOwner, '') <> 'work_item'
   )
 FOR UPDATE`

type connectorIngressCutoverPack struct {
	namespace     string
	packInstallID string
	status        string
}

// VerifyConnectorWorkItemOnlyExecutionCutover stamps the WorkItem-only
// execution receipt only while inbound admission is durably quiesced. It
// retires pre-cutover Event-owned rows that are not being executed in the same
// transaction as the receipt. This is a one-time abandonment, not a
// compatibility executor: it does not replay, materialize, or otherwise make
// historical Events claimable after the handoff.
func VerifyConnectorWorkItemOnlyExecutionCutover(ctx context.Context, db *sql.DB) (err error) {
	return verifyConnectorWorkItemOnlyExecutionCutover(ctx, db, true)
}

// The range protocol reuses the same locked handoff proof but is never allowed
// to retire Event-owned backlog. Terminal retained history needs no retirement.
func verifyConnectorWorkItemOnlyExecutionCutover(ctx context.Context, db *sql.DB, retireLegacyBacklog bool) (err error) {
	if db == nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover requires a database")
	}

	applied, err := workItemCutoverMigrationApplied(ctx, db)
	if err != nil {
		return err
	}
	if applied {
		log.Printf("   VerifyConnectorWorkItemOnlyExecutionCutover: already applied; skipping")
		return nil
	}
	workItemReady, err := connectorWorkItemCutoverSchemaReady(ctx, db)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover check WorkItem schema: %w", err)
	}
	if !workItemReady {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: ConnectorWorkItem execution schema is not ready; run a full setup apply before retiring Event-owned backlog")
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, db)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover begin handoff transaction: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		if cleanupErr := cleanup(); err == nil && cleanupErr != nil {
			err = fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover restore transaction session: %w", cleanupErr)
		}
	}()

	packs, err := lockInboundConnectorPacksForWorkItemCutover(ctx, tx)
	if err != nil {
		return err
	}
	applied, err = workItemCutoverMigrationApplied(ctx, tx)
	if err != nil {
		return err
	}
	if applied {
		if err := tx.Commit(); err != nil {
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover commit already-applied handoff: %w", err)
		}
		committed = true
		log.Printf("   VerifyConnectorWorkItemOnlyExecutionCutover: already applied after pack handoff lock; skipping")
		return nil
	}

	for _, pack := range packs {
		if pack.status == installedPackStatusActive {
			if err := requireInboundConnectorCutoverFence(ctx, tx, pack); err != nil {
				return err
			}
		}
		if retireLegacyBacklog {
			if err := retireLegacyConnectorEventsForWorkItemCutover(ctx, tx, pack); err != nil {
				return err
			}
		}
	}
	if err := requireNoResidualConnectorEventAuthorityForWorkItemCutover(ctx, tx); err != nil {
		return err
	}
	if err := requireNoResidualConnectorWorkItemAuthorityForWorkItemCutover(ctx, tx); err != nil {
		return err
	}

	if err := recordWorkItemCutoverMigrationApplied(ctx, tx); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover stamp: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover commit handoff: %w", err)
	}
	committed = true
	log.Printf("   VerifyConnectorWorkItemOnlyExecutionCutover: ingress-fenced handoff committed and migration stamped")
	return nil
}

// retireLegacyConnectorEventsForWorkItemCutover drains active legacy Event
// executors before terminalizing the abandoned backlog. The row locks make a
// concurrent old claimant serialize behind the handoff; if it has already
// claimed an Event, the cutover refuses to stamp until that executor has
// finished (or has been stopped by the deployment). This keeps an old worker
// from continuing side effects after its token is cleared.
//
// Once every selected row is locked and none is processing, pending/failed
// rows are terminalized while those locks remain held. The selection covers
// every install identity in the pack namespace, so a PackInstallId warm-upgrade
// cannot strand historical Events. The old claimant's CAS then loses after
// commit, and the linked attempt-zero WorkItem observes the terminal Event
// compatibility fence rather than becoming executable.
func retireLegacyConnectorEventsForWorkItemCutover(ctx context.Context, tx *sql.Tx, pack connectorIngressCutoverPack) error {
	rows, err := tx.QueryContext(ctx, connectorLegacyEventCutoverLockSQL, pack.namespace)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover lock legacy events for %s: %w", pack.namespace, err)
	}

	processing := 0
	for rows.Next() {
		var id, status string
		if err := rows.Scan(&id, &status); err != nil {
			_ = rows.Close()
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover scan legacy event for %s: %w", pack.namespace, err)
		}
		if strings.EqualFold(strings.TrimSpace(status), "processing") {
			processing++
		}
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover iterate legacy events for %s: %w", pack.namespace, err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover close legacy event lock for %s: %w", pack.namespace, err)
	}
	if processing > 0 {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: %d legacy Event executor(s) are still processing for %s; drain or stop the old Event worker before handoff", processing, pack.namespace)
	}

	// Keep Event -> WorkItem lock order: it matches the legacy worker's
	// ownership path while closing attempt-zero compatibility mirrors.
	if err := retireLegacyConnectorWorkItemsForWorkItemCutover(ctx, tx, pack); err != nil {
		return err
	}

	_, err = tx.ExecContext(ctx, connectorLegacyEventCutoverRetireSQL, pack.namespace)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover retire legacy events for %s: %w", pack.namespace, err)
	}
	return nil
}

func retireLegacyConnectorWorkItemsForWorkItemCutover(ctx context.Context, tx *sql.Tx, pack connectorIngressCutoverPack) error {
	rows, err := tx.QueryContext(ctx, connectorLegacyWorkItemCutoverLockSQL, pack.packInstallID, pack.namespace)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover lock legacy WorkItems for %s: %w", pack.namespace, err)
	}

	processing := 0
	for rows.Next() {
		var id, state string
		if err := rows.Scan(&id, &state); err != nil {
			_ = rows.Close()
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover scan legacy WorkItem for %s: %w", pack.namespace, err)
		}
		if strings.EqualFold(strings.TrimSpace(state), "processing") {
			processing++
		}
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover iterate legacy WorkItems for %s: %w", pack.namespace, err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover close legacy WorkItem lock for %s: %w", pack.namespace, err)
	}
	if processing > 0 {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: %d WorkItem executor(s) are still processing for %s; drain or stop the worker before handoff", processing, pack.namespace)
	}

	if _, err := tx.ExecContext(ctx, connectorLegacyWorkItemCutoverRetireSQL, pack.packInstallID, pack.namespace); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover retire legacy WorkItems for %s: %w", pack.namespace, err)
	}
	return nil
}

// requireNoResidualConnectorEventAuthorityForWorkItemCutover proves that every
// nonterminal Event has one exact WorkItem executor and one exact live pack
// owner. Legacy, blank, malformed, and unknown owners all fail closed.
func requireNoResidualConnectorEventAuthorityForWorkItemCutover(ctx context.Context, tx *sql.Tx) error {
	rows, err := tx.QueryContext(ctx, connectorEventAuthorityResidualCutoverLockSQL)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover lock residual Event authority: %w", err)
	}
	defer rows.Close()

	var count int
	var namespace string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id, &namespace); err != nil {
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover scan residual Event authority: %w", err)
		}
		count++
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover iterate residual Event authority: %w", err)
	}
	if count > 0 {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: %d nonterminal Event row(s) lack exact WorkItem-only authority (for example namespace %s); require a full setup apply before handoff", count, strings.TrimSpace(namespace))
	}
	return nil
}

// requireNoResidualConnectorWorkItemAuthorityForWorkItemCutover applies the
// same authority definition as reapply admission: Attempt>0 or an Event that
// already names WorkItem ownership. Every such row must resolve to the exact
// Event and one non-deleted installed-pack incarnation.
func requireNoResidualConnectorWorkItemAuthorityForWorkItemCutover(ctx context.Context, tx *sql.Tx) error {
	rows, err := tx.QueryContext(ctx, connectorWorkItemAuthorityResidualCutoverLockSQL)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover lock residual WorkItem authority: %w", err)
	}
	defer rows.Close()

	var count int
	var namespace string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id, &namespace); err != nil {
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover scan residual WorkItem authority: %w", err)
		}
		count++
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover iterate residual WorkItem authority: %w", err)
	}
	if count > 0 {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: %d nonterminal authoritative WorkItem row(s) lack an exact Event or installed-pack owner (for example namespace %s); require a full setup apply before handoff", count, strings.TrimSpace(namespace))
	}
	return nil
}

func lockInboundConnectorPacksForWorkItemCutover(ctx context.Context, tx *sql.Tx) ([]connectorIngressCutoverPack, error) {
	rows, err := tx.QueryContext(ctx, connectorIngressPackCutoverLockSQL)
	if err != nil {
		return nil, fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover lock inbound connector packs: %w", err)
	}
	defer rows.Close()

	var packs []connectorIngressCutoverPack
	for rows.Next() {
		var pack connectorIngressCutoverPack
		if err := rows.Scan(&pack.namespace, &pack.packInstallID, &pack.status); err != nil {
			return nil, fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover scan inbound connector pack: %w", err)
		}
		pack.namespace = strings.ToLower(strings.TrimSpace(pack.namespace))
		pack.packInstallID = strings.TrimSpace(pack.packInstallID)
		pack.status = strings.ToLower(strings.TrimSpace(pack.status))
		if pack.namespace == "" || pack.packInstallID == "" {
			return nil, fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover inbound connector pack has empty namespace or install identity")
		}
		packs = append(packs, pack)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover iterate inbound connector packs: %w", err)
	}
	return packs, nil
}

func requireInboundConnectorCutoverFence(ctx context.Context, tx *sql.Tx, pack connectorIngressCutoverPack) error {
	rows, err := tx.QueryContext(ctx, connectorIngressControlCutoverLockSQL, pack.namespace, pack.packInstallID)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover lock ingress fence for %s: %w", pack.namespace, err)
	}
	defer rows.Close()

	hasUniversalFence := false
	for rows.Next() {
		var packInstallID, channel, mode string
		if err := rows.Scan(&packInstallID, &channel, &mode); err != nil {
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover scan ingress fence for %s: %w", pack.namespace, err)
		}
		packInstallID = strings.TrimSpace(packInstallID)
		channel = strings.ToLower(strings.TrimSpace(channel))
		mode = strings.ToLower(strings.TrimSpace(mode))
		if mode == "active" {
			return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: inbound ingress is not quiesced for %s; active admission override %s/%s remains", pack.namespace, packInstallID, channel)
		}
		if packInstallID == "" && channel == "*" && (mode == "quiescing" || mode == "paused") {
			hasUniversalFence = true
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover iterate ingress fence for %s: %w", pack.namespace, err)
	}
	if !hasUniversalFence {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: inbound ingress is not quiesced for %s; require a non-expired namespace-wide '*' quiescing or paused admission fence", pack.namespace)
	}
	return nil
}

type workItemCutoverReceiptQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

func workItemCutoverMigrationApplied(ctx context.Context, querier workItemCutoverReceiptQuerier) (bool, error) {
	var version int
	var description string
	err := querier.QueryRowContext(ctx, `
SELECT Version, Description
  FROM _System_Migration
 WHERE Id = ?
 LIMIT 1`, migrationIDVerifyConnectorWorkItemOnlyExecutionCutover).Scan(&version, &description)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover read handoff receipt: %w", err)
	}
	if version != migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover || description != migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover {
		return false, fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: stale or malformed handoff receipt %q; require a full setup apply before handoff", migrationIDVerifyConnectorWorkItemOnlyExecutionCutover)
	}
	return true, nil
}

func recordWorkItemCutoverMigrationApplied(ctx context.Context, tx *sql.Tx) error {
	_, err := tx.ExecContext(ctx, `
INSERT INTO _System_Migration (Id, Version, Description, AppliedAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, NOW(), NOW(), NOW())
ON DUPLICATE KEY UPDATE LastModifiedDate = NOW()`,
		migrationIDVerifyConnectorWorkItemOnlyExecutionCutover,
		migrationVersionVerifyConnectorWorkItemOnlyExecutionCutover,
		migrationDescriptionVerifyConnectorWorkItemOnlyExecutionCutover,
	)
	if err != nil {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover record handoff receipt: %w", err)
	}
	applied, err := workItemCutoverMigrationApplied(ctx, tx)
	if err != nil {
		return err
	}
	if !applied {
		return fmt.Errorf("VerifyConnectorWorkItemOnlyExecutionCutover: handoff receipt was not persisted")
	}
	return nil
}