package connector_outboundintent

import (
	"context"
	"crypto/sha256"
	"database/sql/driver"
	"encoding/hex"
	"errors"
	"reflect"
	"testing"
	"time"

	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type memoryCommandStore struct {
	command        Command
	typed          storedTypedTerminalResult
	typedAvailable *bool
	now            func() time.Time
	retryCount     int
	dependency     *commandDependency
	dependencyAt   *time.Time
	waitErr        error
}

type directWorkerDependencyReconcileStore struct {
	*memoryCommandStore
	cancel         context.CancelFunc
	reconcileCalls int
	claimCalls     int
}

func (s *directWorkerDependencyReconcileStore) reconcileDependencyWaits(
	context.Context,
	Scope,
	string,
	int,
) (string, bool, int, int, error) {
	s.reconcileCalls++
	if s.command.State != stateWaitingDependency {
		return "", false, 0, 0, nil
	}
	s.command.State = StateRetry
	now := s.currentTime()
	s.command.NextAttemptAt = &now
	s.dependency = nil
	s.dependencyAt = nil
	return "", false, 1, 1, nil
}

func (s *directWorkerDependencyReconcileStore) ClaimNext(
	ctx context.Context,
	scope Scope,
	workerID string,
	token string,
	lease time.Duration,
) (Command, bool, error) {
	command, found, err := s.memoryCommandStore.ClaimNext(ctx, scope, workerID, token, lease)
	if found {
		s.claimCalls++
		if s.cancel != nil {
			s.cancel()
		}
	}
	return command, found, err
}

// releasedCommandStore exposes only the released commandStore surface. It
// models wrappers compiled before the optional dependency-wait capability was
// added, even though its delegate is the full in-memory test store.
type releasedCommandStore struct {
	commandStore
}

func (s *memoryCommandStore) currentTime() time.Time {
	if s.now != nil {
		return s.now().UTC()
	}
	return time.Now().UTC()
}

type providerSuccessIdentityConflictStore struct {
	*memoryCommandStore
}

type providerSuccessCheckpointIdentityConflictStore struct {
	*memoryCommandStore
}

type writebackIdentityConflictStore struct {
	*memoryCommandStore
}

type unavailableIdentityConflictIsolationStore struct {
	*providerSuccessIdentityConflictStore
	err error
}

func (s *providerSuccessIdentityConflictStore) RecordProviderSuccess(
	context.Context,
	Claim,
	int,
	RemoteLinkFinalization,
) (Command, error) {
	return Command{}, ErrIdentityConflict
}

func (s *providerSuccessCheckpointIdentityConflictStore) CheckpointProviderSuccess(
	context.Context,
	Claim,
	int,
	RemoteLinkFinalization,
	string,
) (Command, error) {
	return Command{}, &DurableIdentityConflictError{
		Stage:           "remote_link_finalize",
		DifferingFields: []string{"MappingFingerprint"},
	}
}

func (s *writebackIdentityConflictStore) FinalizeWriteback(context.Context, Claim) (Command, error) {
	return Command{}, ErrIdentityConflict
}

func (s *unavailableIdentityConflictIsolationStore) IsolatePostDispatch(
	context.Context,
	Claim,
	int,
	string,
	string,
) (Command, error) {
	return Command{}, s.err
}

func (s *memoryCommandStore) typedResultsAvailable(context.Context) (bool, error) {
	if s.typedAvailable == nil {
		return true, nil
	}
	return *s.typedAvailable, nil
}

type hintingMemoryCommandStore struct {
	*memoryCommandStore
	hint      bool
	hintErr   error
	hintCalls int
	claims    int
}

func (s *memoryCommandStore) NextWorkAt(_ context.Context, _ Scope) (time.Time, bool, error) {
	switch s.command.State {
	case StatePending:
		return s.currentTime(), true, nil
	case StateRetry, StateWriteback:
		if s.command.NextAttemptAt != nil {
			return *s.command.NextAttemptAt, true, nil
		}
		return s.currentTime(), true, nil
	case StateClaimed, StateDispatching:
		if s.command.LeaseExpiresAt != nil {
			return *s.command.LeaseExpiresAt, true, nil
		}
	}
	return time.Time{}, false, nil
}

func (s *hintingMemoryCommandStore) HasWorkHint(context.Context, Scope) (bool, error) {
	s.hintCalls++
	return s.hint, s.hintErr
}

func (s *hintingMemoryCommandStore) ClaimWriteback(
	ctx context.Context, scope Scope, workerID, token string, lease time.Duration,
) (Command, bool, error) {
	s.claims++
	return s.memoryCommandStore.ClaimWriteback(ctx, scope, workerID, token, lease)
}

func (s *hintingMemoryCommandStore) ClaimNext(
	ctx context.Context, scope Scope, workerID, token string, lease time.Duration,
) (Command, bool, error) {
	s.claims++
	return s.memoryCommandStore.ClaimNext(ctx, scope, workerID, token, lease)
}

func (s *memoryCommandStore) ClaimNext(_ context.Context, _ Scope, workerID, token string, lease time.Duration) (Command, bool, error) {
	if s.command.State != StatePending && s.command.State != StateRetry {
		return Command{}, false, nil
	}
	now := s.currentTime()
	if s.command.NextAttemptAt != nil && s.command.NextAttemptAt.After(now) {
		return Command{}, false, nil
	}
	s.command.State, s.command.ClaimedBy, s.command.ClaimToken = StateClaimed, workerID, token
	s.command.ClaimedAt = &now
	expires := now.Add(lease)
	s.command.LeaseExpiresAt = &expires
	return s.command, true, nil
}

func (s *memoryCommandStore) ClaimWriteback(_ context.Context, _ Scope, workerID, token string, lease time.Duration) (Command, bool, error) {
	if s.command.State != StateWriteback || s.command.ClaimToken != "" {
		return Command{}, false, nil
	}
	now := s.currentTime()
	if s.command.NextAttemptAt != nil && s.command.NextAttemptAt.After(now) {
		return Command{}, false, nil
	}
	s.command.ClaimedBy, s.command.ClaimToken = workerID, token
	s.command.ClaimedAt = &now
	expires := now.Add(lease)
	s.command.LeaseExpiresAt = &expires
	return s.command, true, nil
}

func (s *memoryCommandStore) MarkDispatching(_ context.Context, claim Claim) (Command, error) {
	if s.command.State != StateClaimed || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	now := s.currentTime()
	s.command.State, s.command.DispatchStarted = StateDispatching, &now
	s.command.AttemptCount++
	return s.command, nil
}

func (s *memoryCommandStore) waitForDependency(
	_ context.Context,
	claim Claim,
	dependency commandDependency,
	code string,
) (Command, error) {
	if s.waitErr != nil {
		return Command{}, s.waitErr
	}
	if s.command.State != StateClaimed || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	now := s.currentTime()
	s.command.State = stateWaitingDependency
	s.command.NextAttemptAt = nil
	s.dependency = &dependency
	s.dependencyAt = &now
	if _, checkpointed := checkpointedAssociationPhase(s.command); !checkpointed {
		s.command.FailureCode = code
	}
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) Retry(_ context.Context, claim Claim, next time.Time, code string) (Command, error) {
	if s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	now := s.currentTime()
	retryAt := now.Add(safeRetryDelay(s.command.CommandID, s.retryCount, next.Sub(now)))
	s.command.State, s.command.NextAttemptAt = StateRetry, &retryAt
	s.retryCount++
	if _, checkpointed := checkpointedAssociationPhase(s.command); !checkpointed {
		s.command.FailureCode = code
	}
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) RetryWriteback(_ context.Context, claim Claim, next time.Time, code string) (Command, error) {
	if s.command.State != StateWriteback || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	now := s.currentTime()
	retryAt := now.Add(safeRetryDelay(s.command.CommandID, s.retryCount, next.Sub(now)))
	s.command.NextAttemptAt, s.command.FailureCode = &retryAt, code
	s.retryCount++
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) Reject(_ context.Context, claim Claim, status int, code string) (Command, error) {
	if s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.State, s.command.ResponseStatus, s.command.FailureCode = StateRejected, status, code
	s.retryCount = 0
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) CheckpointProviderSuccess(
	_ context.Context,
	claim Claim,
	status int,
	value RemoteLinkFinalization,
	code string,
) (Command, error) {
	if s.command.State != StateDispatching || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.AttemptCount = 1
	s.retryCount = 0
	s.command.ResponseStatus, s.command.RemoteID, s.command.FailureCode = status, value.RemoteID, code
	return s.command, nil
}

func (s *memoryCommandStore) RecordProviderSuccess(_ context.Context, claim Claim, status int, value RemoteLinkFinalization) (Command, error) {
	if s.command.State != StateDispatching || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.State, s.command.ResponseStatus, s.command.RemoteID = StateWriteback, status, value.RemoteID
	s.retryCount = 0
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) RecordProviderSuccessForRetry(_ context.Context, claim Claim, status int, value RemoteLinkFinalization, next time.Time, code string) (Command, error) {
	if s.command.State != StateDispatching || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	newSeries := s.command.RemoteID == ""
	seriesCount := s.retryCount
	if newSeries {
		seriesCount = 0
	}
	now := s.currentTime()
	retryAt := now.Add(safeRetryDelay(s.command.CommandID, seriesCount, next.Sub(now)))
	s.command.State, s.command.ResponseStatus, s.command.RemoteID = StateRetry, status, value.RemoteID
	s.command.NextAttemptAt, s.command.FailureCode = &retryAt, code
	if newSeries {
		s.retryCount = 1
	} else {
		s.retryCount++
	}
	s.command.DispatchStarted = nil
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) RecordProviderSuccessTerminal(_ context.Context, claim Claim, status int, value RemoteLinkFinalization, code string) (Command, error) {
	if s.command.State != StateDispatching || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.State, s.command.ResponseStatus, s.command.RemoteID = StateUncertain, status, value.RemoteID
	s.command.FailureCode = code
	s.retryCount = 0
	s.command.DispatchStarted = nil
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) recordTypedResult(_ context.Context, claim Claim, status int, typed storedTypedTerminalResult) (Command, error) {
	if s.command.State != StateDispatching || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.State, s.command.ResponseStatus, s.command.RemoteID = StateWriteback, status, ""
	s.retryCount = 0
	s.typed = typed
	s.typed.canonical = append([]byte(nil), typed.canonical...)
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) loadTypedResult(_ context.Context, claim Claim) (storedTypedTerminalResult, error) {
	if s.command.State != StateWriteback || s.command.ClaimToken != claim.Token || s.typed.formatVersion == 0 {
		return storedTypedTerminalResult{}, ErrCommandNotFound
	}
	result := s.typed
	result.canonical = append([]byte(nil), result.canonical...)
	return result, nil
}

func (s *memoryCommandStore) FinalizeWriteback(_ context.Context, claim Claim) (Command, error) {
	if s.command.State != StateWriteback || s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.State = StateSucceeded
	s.retryCount = 0
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) IsolatePostDispatch(
	_ context.Context,
	claim Claim,
	status int,
	remoteID string,
	code string,
) (Command, error) {
	if (s.command.State != StateDispatching && s.command.State != StateWriteback) ||
		s.command.ClaimToken != claim.Token ||
		(s.command.ResponseStatus != 0 && status != 0 && s.command.ResponseStatus != status) ||
		(s.command.RemoteID != "" && remoteID != "" && s.command.RemoteID != remoteID) {
		return Command{}, ErrLeaseLost
	}
	s.command.State, s.command.FailureCode = StateUncertain, code
	s.retryCount = 0
	if s.command.ResponseStatus == 0 {
		s.command.ResponseStatus = status
	}
	if s.command.RemoteID == "" {
		s.command.RemoteID = remoteID
	}
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) Uncertain(_ context.Context, claim Claim, status int, code string) (Command, error) {
	if s.command.ClaimToken != claim.Token {
		return Command{}, ErrLeaseLost
	}
	s.command.State, s.command.ResponseStatus, s.command.FailureCode = StateUncertain, status, code
	s.retryCount = 0
	s.clearClaim()
	return s.command, nil
}

func (s *memoryCommandStore) clearClaim() {
	s.command.ClaimedBy, s.command.ClaimToken = "", ""
	s.command.ClaimedAt, s.command.LeaseExpiresAt = nil, nil
}

type countingPreparer struct {
	dispatches int
	result     EffectResult
	typed      *TypedTerminalResult
}

func TestWorkerNegativeHintSkipsAuthoritativeClaimTransactions(t *testing.T) {
	store := &hintingMemoryCommandStore{memoryCommandStore: &memoryCommandStore{}}
	worker, err := NewWorker(store, &countingPreparer{}, &scriptedCommitter{}, testWorkerCommand().Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	processed, err := worker.ProcessOne(t.Context())
	if err != nil || processed {
		t.Fatalf("ProcessOne = %v, %v, want false, nil", processed, err)
	}
	if store.hintCalls != 1 || store.claims != 0 {
		t.Fatalf("hint calls = %d, claim calls = %d, want 1, 0", store.hintCalls, store.claims)
	}
}

func TestWorkerHintErrorFallsThroughToAuthoritativeClaims(t *testing.T) {
	command := testWorkerCommand()
	store := &hintingMemoryCommandStore{
		memoryCommandStore: &memoryCommandStore{command: command},
		hintErr:            errors.New("hint unavailable"),
	}
	worker, err := NewWorker(store, &countingPreparer{}, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	processed, err := worker.ProcessOne(t.Context())
	if err != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v, want true, nil", processed, err)
	}
	if store.hintCalls != 1 || store.claims == 0 {
		t.Fatalf("hint calls = %d, claim calls = %d, want 1 and authoritative fallback", store.hintCalls, store.claims)
	}
}

type failingPreparer struct{ err error }

func (p failingPreparer) Prepare(context.Context, Command) (PreparedEffect, error) {
	return nil, p.err
}

type countingRetryablePreparer struct{ calls int }

func (p *countingRetryablePreparer) Prepare(context.Context, Command) (PreparedEffect, error) {
	p.calls++
	return nil, &PreparationError{
		Code: "binding_unavailable", Retryable: true, Cause: errors.New("catalog temporarily unavailable"),
	}
}

func TestSafeRetryDelayIsDeterministicExponentialCappedAndHonorsMinimum(t *testing.T) {
	commandID := "integration-outbound-retry-policy"
	if got := safeRetryDelay(commandID, 0, 0); got != time.Second {
		t.Fatalf("first retry delay = %s, want 1s", got)
	}
	if got := safeRetryDelay(commandID, 0, -time.Second); got != time.Second {
		t.Fatalf("negative provider delay = %s, want safe 1s default", got)
	}
	previous := time.Duration(0)
	for retryCount := 1; retryCount <= 20; retryCount++ {
		got := safeRetryDelay(commandID, retryCount, 0)
		if got != safeRetryDelay(commandID, retryCount, 0) {
			t.Fatalf("retry %d delay is not deterministic", retryCount+1)
		}
		if got < safeRetryInitialDelay || got > safeRetryMaximumDelay {
			t.Fatalf("retry %d delay = %s, want [1s, %s]", retryCount+1, got, safeRetryMaximumDelay)
		}
		if retryCount < 9 && got <= previous {
			t.Fatalf("retry %d delay = %s, previous %s; uncapped series did not increase", retryCount+1, got, previous)
		}
		previous = got
	}
	minimum := 11 * time.Minute
	if got := safeRetryDelay(commandID, 20, minimum); got != minimum {
		t.Fatalf("provider minimum delay = %s, want %s", got, minimum)
	}
}

func TestWorkerPreparationRetryPersistsBackoffAcrossRestartAndBoundsQuietClaims(t *testing.T) {
	start := time.Date(2026, 8, 27, 8, 0, 0, 0, time.UTC)
	now := start
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command, now: func() time.Time { return now }}
	preparer := &countingRetryablePreparer{}
	newWorker := func() *Worker {
		worker, err := NewWorker(store, preparer, &scriptedCommitter{}, command.Scope, "worker-1")
		if err != nil {
			t.Fatal(err)
		}
		worker.now = func() time.Time { return now }
		return worker
	}
	worker := newWorker()

	processed, err := worker.ProcessOne(t.Context())
	if err != nil || !processed {
		t.Fatalf("first retry = %v, %v", processed, err)
	}
	if store.retryCount != 1 || store.command.AttemptCount != 0 ||
		store.command.NextAttemptAt == nil || !store.command.NextAttemptAt.Equal(start.Add(time.Second)) {
		t.Fatalf("first durable retry = %#v", store.command)
	}
	if processed, err := worker.ProcessOne(t.Context()); err != nil || processed {
		t.Fatalf("claim before exact due = %v, %v, want false, nil", processed, err)
	}

	// Reconstructing the worker must consume the same durable retry series; a
	// process restart cannot return the command to a one-second loop.
	worker = newWorker()
	windowEnd := start.Add(6 * time.Minute)
	for store.command.NextAttemptAt != nil && !store.command.NextAttemptAt.After(windowEnd) {
		now = *store.command.NextAttemptAt
		processed, err := worker.ProcessOne(t.Context())
		if err != nil || !processed {
			t.Fatalf("retry at %s = %v, %v", now, processed, err)
		}
	}
	if preparer.calls > 10 {
		t.Fatalf("six-minute unavailable window prepared %d times, want at most 10", preparer.calls)
	}
	if preparer.calls < 5 || store.retryCount != preparer.calls || store.command.AttemptCount != 0 {
		t.Fatalf("retry series calls=%d retry_count=%d command=%#v", preparer.calls, store.retryCount, store.command)
	}
}

func TestWorkerSafeRetryPolicyCoversDefinitelyNotDispatchedEffects(t *testing.T) {
	now := time.Date(2026, 8, 27, 9, 0, 0, 0, time.UTC)
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command, now: func() time.Time { return now }}
	preparer := &countingPreparer{result: EffectResult{
		Outcome: EffectNotDispatched, FailureCode: "mail_oauth_unavailable",
	}}
	worker, err := NewWorker(store, preparer, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	worker.now = func() time.Time { return now }

	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("first no-dispatch retry = %v, %v", processed, err)
	}
	if store.retryCount != 1 || store.command.AttemptCount != 1 || store.command.NextAttemptAt == nil ||
		!store.command.NextAttemptAt.Equal(now.Add(time.Second)) {
		t.Fatalf("first no-dispatch disposition retry_count=%d command=%#v", store.retryCount, store.command)
	}
	now = *store.command.NextAttemptAt
	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("second no-dispatch retry = %v, %v", processed, err)
	}
	delay := store.command.NextAttemptAt.Sub(now)
	if store.retryCount != 2 || store.command.AttemptCount != 2 || delay < 1500*time.Millisecond || delay > 2*time.Second {
		t.Fatalf("second no-dispatch delay=%s retry_count=%d command=%#v", delay, store.retryCount, store.command)
	}
}

func TestWorkerWritebackRetryBacksOffAcrossRestartThenResetsAtTerminal(t *testing.T) {
	now := time.Date(2026, 8, 27, 10, 0, 0, 0, time.UTC)
	command := testWorkerCommand()
	command.State, command.RemoteID = StateWriteback, "remote-1"
	store := &memoryCommandStore{command: command, now: func() time.Time { return now }}
	committer := &scriptedCommitter{failures: 2}
	newWorker := func() *Worker {
		worker, err := NewWorker(store, &countingPreparer{}, committer, command.Scope, "worker-1")
		if err != nil {
			t.Fatal(err)
		}
		worker.now = func() time.Time { return now }
		return worker
	}
	worker := newWorker()
	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("first writeback retry = %v, %v", processed, err)
	}
	if store.retryCount != 1 || store.command.NextAttemptAt == nil ||
		!store.command.NextAttemptAt.Equal(now.Add(time.Second)) {
		t.Fatalf("first writeback retry_count=%d command=%#v", store.retryCount, store.command)
	}
	if processed, err := worker.ProcessOne(t.Context()); err != nil || processed {
		t.Fatalf("writeback claimed before due = %v, %v", processed, err)
	}

	now = *store.command.NextAttemptAt
	worker = newWorker()
	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("second writeback retry = %v, %v", processed, err)
	}
	secondDelay := store.command.NextAttemptAt.Sub(now)
	if store.retryCount != 2 || secondDelay < 1500*time.Millisecond || secondDelay > 2*time.Second {
		t.Fatalf("second writeback delay=%s retry_count=%d command=%#v", secondDelay, store.retryCount, store.command)
	}

	now = *store.command.NextAttemptAt
	worker = newWorker()
	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("successful writeback = %v, %v", processed, err)
	}
	if store.command.State != StateSucceeded || store.retryCount != 0 || committer.calls != 3 {
		t.Fatalf("terminal writeback retry_count=%d calls=%d command=%#v", store.retryCount, committer.calls, store.command)
	}
}

func (p *countingPreparer) Prepare(context.Context, Command) (PreparedEffect, error) {
	if p.typed != nil {
		return &countingTypedEffect{preparer: p}, nil
	}
	return preparedEffectFunc(func(context.Context) EffectResult {
		p.dispatches++
		return p.result
	}), nil
}

type countingTypedEffect struct{ preparer *countingPreparer }

func (effect *countingTypedEffect) Dispatch(context.Context) EffectResult {
	effect.preparer.dispatches++
	return EffectResult{Outcome: EffectUncertain, FailureCode: "typed_checkpoint_owner_missing"}
}

func (effect *countingTypedEffect) DispatchWithTypedTerminalResultCheckpoint(
	ctx context.Context,
	checkpoint TypedTerminalResultCheckpoint,
) EffectResult {
	effect.preparer.dispatches++
	result := *effect.preparer.typed
	result.CanonicalResult = append([]byte(nil), result.CanonicalResult...)
	if err := checkpoint(ctx, effect.preparer.result.ResponseStatus, result); err != nil {
		return EffectResult{Outcome: EffectUncertain, FailureCode: "typed_result_checkpoint_failed"}
	}
	return effect.preparer.result
}

type preparedEffectFunc func(context.Context) EffectResult

func (f preparedEffectFunc) Dispatch(ctx context.Context) EffectResult { return f(ctx) }

type checkpointingPreparer struct{ effect *checkpointingEffect }

func (p checkpointingPreparer) Prepare(context.Context, Command) (PreparedEffect, error) {
	return p.effect, nil
}

type checkpointingEffect struct {
	checkpointed bool
	fallbacks    int
}

func (e *checkpointingEffect) Dispatch(context.Context) EffectResult {
	e.fallbacks++
	return EffectResult{Outcome: EffectUncertain, FailureCode: "checkpoint_owner_missing"}
}

func (e *checkpointingEffect) DispatchWithProviderSuccessCheckpoint(
	ctx context.Context,
	checkpoint ProviderSuccessCheckpoint,
) EffectResult {
	finalization := RemoteLinkFinalization{RemoteID: "remote-1"}
	if err := checkpoint(ctx, 201, finalization, "association_create_outcome_uncertain"); err != nil {
		return EffectResult{Outcome: EffectUncertain, FailureCode: "provider_success_checkpoint_failed"}
	}
	e.checkpointed = true
	return EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 201, RemoteID: "remote-1",
		LinkFinalization: &finalization,
	}
}

type scriptedCommitter struct {
	calls             int
	failures          int
	malformedReceipts bool
	lastRequest       outboundintentv1.CommitResultRequest
}

func (c *scriptedCommitter) CommitOutboundResult(_ context.Context, request outboundintentv1.CommitResultRequest) (outboundintentv1.CommitResultResponse, error) {
	c.calls++
	c.lastRequest = request
	if c.calls <= c.failures {
		return outboundintentv1.CommitResultResponse{}, errors.New("Runtime unavailable")
	}
	response := outboundintentv1.CommitResultResponse{
		IntentID: request.IntentID,
		Receipts: outboundintentv1.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: outboundintentv1.CapabilityName,
			CatalogRevision: request.CatalogRevision, SchemaGeneration: 7,
			EngineGeneration: 9, SecurityEpoch: 13,
		},
	}
	if c.malformedReceipts {
		response.Receipts.SecurityEpoch = 0
	}
	return response, nil
}

func TestWorkerUsesExistingCommandLifecycleForTypedTerminalResult(t *testing.T) {
	command := testTypedWorkerCommand()
	canonical := []byte(`{"version":1,"outcome":"DELIVERED","remoteMessageId":"remote-1","senderAddress":"sender@example.com"}`)
	store := &memoryCommandStore{command: command}
	preparer := &countingPreparer{
		result: EffectResult{Outcome: EffectTypedTerminal, ResponseStatus: 200},
		typed: &TypedTerminalResult{
			ResultFormatVersion: outboundintentv1.MailResultFormatV1,
			CanonicalResult:     canonical,
		},
	}
	committer := &scriptedCommitter{}
	worker, err := NewWorker(store, preparer, committer, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	worker.now = time.Now

	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("provider pass = %v, %v", processed, err)
	}
	if store.command.State != StateWriteback || preparer.dispatches != 1 || committer.calls != 0 {
		t.Fatalf("after typed checkpoint: command=%#v dispatches=%d commits=%d", store.command, preparer.dispatches, committer.calls)
	}
	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("writeback = %v, %v", processed, err)
	}
	if store.command.State != StateSucceeded || preparer.dispatches != 1 || committer.calls != 1 {
		t.Fatalf("after typed writeback: command=%#v dispatches=%d commits=%d", store.command, preparer.dispatches, committer.calls)
	}
	if committer.lastRequest.RemoteID != "" ||
		committer.lastRequest.ResultFormatVersion != outboundintentv1.MailResultFormatV1 ||
		string(committer.lastRequest.CanonicalResult) != string(canonical) ||
		committer.lastRequest.ResultDigest == "" {
		t.Fatalf("typed Runtime writeback = %#v", committer.lastRequest)
	}
}

func TestWorkerRejectsTypedEffectBeforeDispatchWhenResultSchemaIsUnavailable(t *testing.T) {
	command := testTypedWorkerCommand()
	available := false
	store := &memoryCommandStore{command: command, typedAvailable: &available}
	preparer := &countingPreparer{
		result: EffectResult{Outcome: EffectTypedTerminal},
		typed: &TypedTerminalResult{
			ResultFormatVersion: outboundintentv1.MailResultFormatV1,
			CanonicalResult: []byte(
				`{"version":1,"outcome":"REJECTED","failureCode":"PROVIDER_REJECTED"}`,
			),
		},
	}
	worker, err := NewWorker(store, preparer, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := worker.ProcessOne(t.Context()); err != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v", processed, err)
	}
	if preparer.dispatches != 0 || store.command.State != StateRejected ||
		store.command.FailureCode != "typed_result_schema_unavailable" {
		t.Fatalf("schema gate = command=%#v dispatches=%d", store.command, preparer.dispatches)
	}
}

func TestWorkerNeverReplaysProviderAfterDurableProviderSuccess(t *testing.T) {
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command}
	preparer := &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 201, RemoteID: "remote-1",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-1"},
	}}
	committer := &scriptedCommitter{failures: 1}
	worker, err := NewWorker(store, preparer, committer, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	worker.now = time.Now

	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("provider pass = %v, %v", processed, err)
	}
	if store.command.State != StateWriteback || preparer.dispatches != 1 || committer.calls != 0 {
		t.Fatalf("after provider: state=%s dispatches=%d commits=%d", store.command.State, preparer.dispatches, committer.calls)
	}

	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("first writeback = %v, %v", processed, err)
	}
	if store.command.State != StateWriteback || preparer.dispatches != 1 || committer.calls != 1 {
		t.Fatalf("after failed writeback: state=%s dispatches=%d commits=%d", store.command.State, preparer.dispatches, committer.calls)
	}
	store.command.NextAttemptAt = nil

	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("reconciled writeback = %v, %v", processed, err)
	}
	if store.command.State != StateSucceeded || preparer.dispatches != 1 || committer.calls != 2 {
		t.Fatalf("final: state=%s dispatches=%d commits=%d", store.command.State, preparer.dispatches, committer.calls)
	}
}

func TestWorkerDurablyIsolatesProviderSuccessIdentityConflict(t *testing.T) {
	command := testWorkerCommand()
	store := &providerSuccessIdentityConflictStore{
		memoryCommandStore: &memoryCommandStore{command: command},
	}
	preparer := &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 200,
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-conflicted"},
	}}
	worker, err := NewWorker(store, preparer, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}

	processed, processErr := worker.ProcessOne(t.Context())
	if processErr != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v, want true, nil", processed, processErr)
	}
	if store.command.State != StateUncertain ||
		store.command.FailureCode != "remote_link_identity_conflict" ||
		store.command.ResponseStatus != 200 || store.command.RemoteID != "remote-conflicted" ||
		store.command.ClaimToken != "" || store.command.LeaseExpiresAt != nil {
		t.Fatalf("isolated command = %#v", store.command)
	}
}

func TestWorkerDurablyIsolatesProviderSuccessCheckpointIdentityConflict(t *testing.T) {
	command := testWorkerCommand()
	store := &providerSuccessCheckpointIdentityConflictStore{
		memoryCommandStore: &memoryCommandStore{command: command},
	}
	effect := &checkpointingEffect{}
	worker, err := NewWorker(
		store, checkpointingPreparer{effect: effect}, &scriptedCommitter{},
		command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}

	processed, processErr := worker.ProcessOne(t.Context())
	if processErr != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v, want true, nil", processed, processErr)
	}
	if effect.checkpointed || effect.fallbacks != 0 {
		t.Fatalf("checkpoint effect = %#v", effect)
	}
	if store.command.State != StateUncertain ||
		store.command.FailureCode != "remote_link_identity_conflict" ||
		store.command.ResponseStatus != 201 || store.command.RemoteID != "remote-1" ||
		store.command.ClaimToken != "" || store.command.LeaseExpiresAt != nil {
		t.Fatalf("isolated command = %#v", store.command)
	}
}

func TestWorkerDurablyIsolatesWritebackIdentityConflict(t *testing.T) {
	command := testWorkerCommand()
	command.State = StateWriteback
	command.ResponseStatus = 200
	command.RemoteID = "remote-1"
	store := &writebackIdentityConflictStore{
		memoryCommandStore: &memoryCommandStore{command: command},
	}
	worker, err := NewWorker(store, &countingPreparer{}, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}

	processed, processErr := worker.ProcessOne(t.Context())
	if processErr != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v, want true, nil", processed, processErr)
	}
	if store.command.State != StateUncertain ||
		store.command.FailureCode != "writeback_identity_conflict" ||
		store.command.ResponseStatus != 200 || store.command.RemoteID != "remote-1" ||
		store.command.ClaimToken != "" || store.command.LeaseExpiresAt != nil {
		t.Fatalf("isolated command = %#v", store.command)
	}
}

func TestWorkerStillFailsClosedWhenIdentityConflictCannotBeDurablyIsolated(t *testing.T) {
	command := testWorkerCommand()
	isolationErr := errors.New("isolation store unavailable")
	store := &unavailableIdentityConflictIsolationStore{
		providerSuccessIdentityConflictStore: &providerSuccessIdentityConflictStore{
			memoryCommandStore: &memoryCommandStore{command: command},
		},
		err: isolationErr,
	}
	worker, err := NewWorker(store, &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 200, RemoteID: "remote-conflicted",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-conflicted"},
	}}, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}

	processed, processErr := worker.ProcessOne(t.Context())
	if !processed || !errors.Is(processErr, ErrIdentityConflict) || !errors.Is(processErr, isolationErr) {
		t.Fatalf("ProcessOne = %v, %v, want joined identity and isolation errors", processed, processErr)
	}
	if store.command.State != StateDispatching || store.command.ClaimToken == "" ||
		store.command.LeaseExpiresAt == nil {
		t.Fatalf("uncontained command lost its safety fence: %#v", store.command)
	}
}

func TestWorkerRetriesMalformedWritebackReceiptWithoutFinalizing(t *testing.T) {
	command := testWorkerCommand()
	command.State = StateWriteback
	command.RemoteID = "remote-1"
	store := &memoryCommandStore{command: command}
	committer := &scriptedCommitter{malformedReceipts: true}
	worker, err := NewWorker(store, &countingPreparer{}, committer, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	worker.now = time.Now

	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("malformed writeback = %v, %v", processed, err)
	}
	if store.command.State != StateWriteback || store.command.FailureCode != "runtime_projection_invalid" ||
		committer.calls != 1 {
		t.Fatalf("malformed receipt finalized command: %#v calls=%d", store.command, committer.calls)
	}

	committer.malformedReceipts = false
	store.command.NextAttemptAt = nil
	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("valid writeback retry = %v, %v", processed, err)
	}
	if store.command.State != StateSucceeded || committer.calls != 2 {
		t.Fatalf("valid receipt did not finalize command: %#v calls=%d", store.command, committer.calls)
	}
}

func TestWorkerExpectedPreparationOutcomesDoNotStopLifecycle(t *testing.T) {
	for _, test := range []struct {
		name       string
		retryable  bool
		dependency *commandDependency
		wantState  string
	}{
		{
			name: "dependency",
			dependency: &commandDependency{
				Kind: dependencyKindAssociationRemoteLink,
				Key:  "integration_association_dependency_v1:dependency",
				ReadinessKeys: []string{
					"integration_link_v2:dependency",
				},
			},
			wantState: stateWaitingDependency,
		},
		{name: "retryable", retryable: true, wantState: StateRetry},
		{name: "terminal", wantState: StateRejected},
	} {
		t.Run(test.name, func(t *testing.T) {
			command := testWorkerCommand()
			store := &memoryCommandStore{command: command}
			worker, err := NewWorker(
				store,
				failingPreparer{err: func() error {
					if test.dependency != nil {
						return &dependencyPreparationError{
							Code: "association_link_pending", Dependency: *test.dependency, Cause: ErrDependency,
						}
					}
					return &PreparationError{
						Code: "association_link_pending", Retryable: test.retryable, Cause: ErrDependency,
					}
				}()},
				&scriptedCommitter{}, command.Scope, "worker-1",
			)
			if err != nil {
				t.Fatal(err)
			}
			if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
				t.Fatalf("ProcessOne = %v, %v", processed, err)
			}
			if store.command.State != test.wantState {
				t.Fatalf("state = %q, want %q", store.command.State, test.wantState)
			}
			if test.dependency != nil &&
				(store.dependency == nil || !reflect.DeepEqual(*store.dependency, *test.dependency) ||
					store.dependencyAt == nil || store.command.NextAttemptAt != nil) {
				t.Fatalf("dependency wait = command %#v dependency %#v, want %#v without deadline", store.command, store.dependency, *test.dependency)
			}
		})
	}
}

func TestWorkerDependencyOutcomeFallsBackToRetryForReleasedStoreSurface(t *testing.T) {
	command := testWorkerCommand()
	delegate := &memoryCommandStore{command: command}
	store := &releasedCommandStore{commandStore: delegate}
	worker, err := NewWorker(
		store,
		failingPreparer{err: &dependencyPreparationError{
			Code: "association_link_pending",
			Dependency: commandDependency{
				Kind: dependencyKindAssociationRemoteLink,
				Key:  "integration_association_dependency_v1:released-store",
				ReadinessKeys: []string{
					"integration_link_v2:released-store",
				},
			},
			Cause: ErrDependency,
		}},
		&scriptedCommitter{}, command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	worker.now = func() time.Time { return time.Unix(100, 0).UTC() }
	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v", processed, err)
	}
	if delegate.command.State != StateRetry || delegate.command.FailureCode != "association_link_pending" ||
		delegate.retryCount != 1 || delegate.dependency != nil {
		t.Fatalf("released store fallback = command %#v retry_count=%d dependency=%#v",
			delegate.command, delegate.retryCount, delegate.dependency)
	}
}

func TestWorkerDependencyRegistrationTransientFailureStaysInsideRecoveryBoundary(t *testing.T) {
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command, waitErr: driver.ErrBadConn}
	worker, err := NewWorker(
		store,
		failingPreparer{err: &dependencyPreparationError{
			Code: "association_link_pending",
			Dependency: commandDependency{
				Kind: dependencyKindAssociationRemoteLink,
				Key:  "integration_association_dependency_v1:transient-registration",
				ReadinessKeys: []string{
					"integration_link_v2:transient-registration",
				},
			},
			Cause: ErrDependency,
		}},
		&scriptedCommitter{}, command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, err := worker.processWithRecovery(t.Context(), false)
	if err != nil {
		t.Fatalf("transient dependency registration stopped worker: %v", err)
	}
	if processed || store.command.State != StateClaimed {
		t.Fatalf("recovered registration turn = processed=%v state=%q, want durable claim left for lease recovery", processed, store.command.State)
	}
}

func TestWorkerRunRecoversMissedDependencyWake(t *testing.T) {
	command := testWorkerCommand()
	command.State = stateWaitingDependency
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	store := &directWorkerDependencyReconcileStore{
		memoryCommandStore: &memoryCommandStore{command: command},
		cancel:             cancel,
	}
	worker, err := NewWorker(
		store, &countingPreparer{}, &scriptedCommitter{}, command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	worker.pollInterval = time.Hour
	worker.dependencyReconcileInterval = 2 * time.Millisecond
	started := false
	err = worker.Run(ctx, func() { started = true })
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("Worker.Run error = %v, want context cancellation after recovered claim", err)
	}
	if !started || store.reconcileCalls == 0 || store.claimCalls != 1 ||
		store.command.State == stateWaitingDependency {
		t.Fatalf(
			"direct Worker recovery = started=%v reconcile=%d claims=%d state=%q",
			started, store.reconcileCalls, store.claimCalls, store.command.State,
		)
	}
}

func TestWorkerInvalidDependencyIsTerminalizedWithoutStoppingLifecycle(t *testing.T) {
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command}
	worker, err := NewWorker(
		store,
		failingPreparer{err: &dependencyPreparationError{
			Code: "unsafe code",
			Dependency: commandDependency{
				Kind: dependencyKindAssociationRemoteLink,
				Key:  "malformed",
				ReadinessKeys: []string{
					"integration_link_v2:malformed",
				},
			},
			Cause: ErrDependency,
		}},
		&scriptedCommitter{}, command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, err := worker.ProcessOne(t.Context())
	if err != nil || !processed || store.command.State != StateRejected {
		t.Fatalf("invalid dependency outcome = processed=%v state=%q err=%v", processed, store.command.State, err)
	}
}

func TestWorkerPreparationRetryPreservesCheckpointedAssociationLane(t *testing.T) {
	command := testWorkerCommand()
	command.State = StateRetry
	command.RemoteID = "remote-1"
	command.ResponseStatus = 201
	command.FailureCode = "association_create_outcome_uncertain"
	store := &memoryCommandStore{command: command}
	worker, err := NewWorker(
		store,
		failingPreparer{err: &PreparationError{
			Code: "association_profile_unavailable", Retryable: true, Cause: ErrDependency,
		}},
		&scriptedCommitter{}, command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("ProcessOne = %v, %v", processed, err)
	}
	phase, checkpointed := checkpointedAssociationPhase(store.command)
	if store.command.State != StateRetry || !checkpointed || phase != "create" ||
		store.command.FailureCode != "association_create_outcome_uncertain" {
		t.Fatalf("retry lost recovery lane: %#v", store.command)
	}
}

func TestWorkerOwnsProviderSuccessCheckpointBeforeEffectCompletion(t *testing.T) {
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command}
	effect := &checkpointingEffect{}
	worker, err := NewWorker(
		store, checkpointingPreparer{effect: effect}, &scriptedCommitter{},
		command.Scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("process = %v, %v", processed, err)
	}
	if !effect.checkpointed || effect.fallbacks != 0 || store.command.State != StateWriteback ||
		store.command.RemoteID != "remote-1" {
		t.Fatalf("checkpoint effect=%#v command=%#v", effect, store.command)
	}
}

func TestWorkerDurablyCheckpointsMainSuccessBeforeAssociationRetry(t *testing.T) {
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command}
	preparer := &countingPreparer{result: EffectResult{
		Outcome: EffectRetryableAfterSuccess, ResponseStatus: 201,
		RemoteID: "remote-1", FailureCode: "association_create_outcome_uncertain",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-1"},
	}}
	worker, err := NewWorker(store, preparer, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	worker.now = time.Now
	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("process = %v, %v", processed, err)
	}
	if store.command.State != StateRetry || store.command.RemoteID != "remote-1" ||
		store.command.FailureCode != "association_create_outcome_uncertain" || preparer.dispatches != 1 {
		t.Fatalf("checkpoint = %#v dispatches=%d", store.command, preparer.dispatches)
	}
}

func TestWorkerDurablyRecordsRemoteIdentityForTerminalPartialSuccess(t *testing.T) {
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command}
	preparer := &countingPreparer{result: EffectResult{
		Outcome: EffectTerminalAfterSuccess, ResponseStatus: 201,
		RemoteID: "remote-1", FailureCode: "association_create_rejected",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-1"},
	}}
	worker, err := NewWorker(store, preparer, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	worker.now = time.Now
	if processed, err := worker.ProcessOne(context.Background()); err != nil || !processed {
		t.Fatalf("process = %v, %v", processed, err)
	}
	if store.command.State != StateUncertain || store.command.RemoteID != "remote-1" ||
		store.command.FailureCode != "association_create_rejected" || preparer.dispatches != 1 {
		t.Fatalf("terminal checkpoint = %#v dispatches=%d", store.command, preparer.dispatches)
	}
}

func testWorkerCommand() Command {
	scope := Scope{OrgID: "org-1", AppInstallationID: "app-1", AuthorizedWorkloadID: "integration-1", CatalogRevision: 41}
	return Command{
		CommandID: "command-1", LinkKey: "link-1", Scope: scope, State: StatePending,
		Intent: outboundintentv1.Intent{
			IntentID: "intent-1", PayloadDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
			CatalogRevision: 41,
		},
	}
}

func testTypedWorkerCommand() Command {
	payload := []byte(`{"version":1,"authorizedActorId":"user-1","toAddresses":["recipient@example.com"],"subject":"Hello","textBody":"Body"}`)
	payloadSum := sha256.Sum256(payload)
	command := testWorkerCommand()
	command.Intent.Operation = outboundintentv1.OperationSendMail
	command.Intent.PayloadFormatVersion = outboundintentv1.MailPayloadFormatV1
	command.Intent.PayloadDigest = "sha256:" + hex.EncodeToString(payloadSum[:])
	command.Intent.CanonicalPayload = payload
	command.Intent.IdempotencyKey = "request-1"
	command.Intent.SourceObject = "Contact"
	command.Intent.SourceRecordID = "contact-1"
	command.Intent.ConnectorOwnerKey = "mail-owner"
	command.Intent.FlowDeveloperName = "Std_SendRecordEmail"
	command.Intent.FlowActionName = "sendRecordEmail"
	command.Intent.ActionSlotID = "send-record-email"
	command.Intent.DestinationSlotID = "connected-mailbox"
	command.Intent.Bindings = []outboundintentv1.BindingReference{{
		Kind: catalogsemantic.BindingConnectorCalloutProfile, OwnerPack: "mail-owner",
		PackInstallationID: "install-1",
		PackPayloadDigest:  "nexus-pack-sha256-v1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		BusinessKey:        "mail/delivery", DefinitionVersion: 1,
	}}
	return command
}

type retryBudgetStore struct {
	*memoryCommandStore
	limits  []int
	cursors []string
}

func (s *retryBudgetStore) reconcileDependencyWaits(_ context.Context, _ Scope, cursor string, limit int) (string, bool, int, int, error) {
	s.limits = append(s.limits, limit)
	s.cursors = append(s.cursors, cursor)
	if len(s.limits) == 1 {
		return "first-progress", true, 1, 600, driver.ErrBadConn
	}
	return "second-progress", true, 1, limit, driver.ErrBadConn
}
func TestDependencySafetyRetriesShareItemBudgetAndResumeCursor(t *testing.T) {
	store := &retryBudgetStore{memoryCommandStore: &memoryCommandStore{}}
	worker := &Worker{store: store, dependencyReconcileBudget: 2 * time.Second}
	cursor, more, woken, err := worker.reconcileDependencies(t.Context(), "start")
	if err != nil || cursor != "second-progress" || !more || woken != 2 || len(store.limits) != 2 || store.limits[0] != 1000 || store.limits[1] != 400 || store.cursors[1] != "first-progress" {
		t.Fatalf("cursor=%s more=%v woken=%d limits=%v cursors=%v err=%v", cursor, more, woken, store.limits, store.cursors, err)
	}
}