package connector_outboundintent

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/google/uuid"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/dependencyretry"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

const (
	dispatchCompletionReserve        = time.Second
	safeRetryInitialDelay            = time.Second
	safeRetryMaximumDelay            = 5 * time.Minute
	defaultDependencyReconcileBudget = time.Second
)

type commandStore interface {
	ClaimNext(context.Context, Scope, string, string, time.Duration) (Command, bool, error)
	ClaimWriteback(context.Context, Scope, string, string, time.Duration) (Command, bool, error)
	MarkDispatching(context.Context, Claim) (Command, error)
	Retry(context.Context, Claim, time.Time, string) (Command, error)
	RetryWriteback(context.Context, Claim, time.Time, string) (Command, error)
	Reject(context.Context, Claim, int, string) (Command, error)
	CheckpointProviderSuccess(context.Context, Claim, int, RemoteLinkFinalization, string) (Command, error)
	RecordProviderSuccess(context.Context, Claim, int, RemoteLinkFinalization) (Command, error)
	RecordProviderSuccessForRetry(context.Context, Claim, int, RemoteLinkFinalization, time.Time, string) (Command, error)
	RecordProviderSuccessTerminal(context.Context, Claim, int, RemoteLinkFinalization, string) (Command, error)
	FinalizeWriteback(context.Context, Claim) (Command, error)
	Uncertain(context.Context, Claim, int, string) (Command, error)
}

type dependencyCommandStore interface {
	waitForDependency(context.Context, Claim, commandDependency, string) (Command, error)
}

type dependencyReconcileStore interface {
	// Returns the cursor, whether work remains, successful wakes, and inspected waiters.
	reconcileDependencyWaits(context.Context, Scope, string, int) (string, bool, int, int, error)
}

type postDispatchIsolationStore interface {
	IsolatePostDispatch(context.Context, Claim, int, string, string) (Command, error)
}

// The installed-authority proof stays private from preparation through every
// provider-success transition. Public finalization values carry only snapshots.
type legacyRemoteLinkStore interface {
	markDispatchingWithLegacyProof(context.Context, Claim, *preparedLegacyRemoteLink) (Command, error)
	checkpointProviderSuccessWithLegacyProof(context.Context, Claim, int, RemoteLinkFinalization, string, *preparedLegacyRemoteLink) (Command, error)
	recordProviderSuccessWithLegacyProof(context.Context, Claim, int, RemoteLinkFinalization, *preparedLegacyRemoteLink) (Command, error)
	recordProviderSuccessForRetryWithLegacyProof(context.Context, Claim, int, RemoteLinkFinalization, time.Time, string, *preparedLegacyRemoteLink) (Command, error)
	recordProviderSuccessTerminalWithLegacyProof(context.Context, Claim, int, RemoteLinkFinalization, string, *preparedLegacyRemoteLink) (Command, error)
}

// PreparedDispatchProof is the opaque, process-local snapshot that must be
// consumed atomically with a provider command's dispatch transition. Callers
// may forward this value but cannot inspect or construct its authority.
type PreparedDispatchProof struct {
	scoped *preparedScopedRemoteLink
	legacy *preparedLegacyRemoteLink
}

// PreparedDispatchStore is the optional capability required when a Preparer
// returns an effect backed by a scoped remote-link snapshot. Repository
// implements it, and wrappers may forward or promote the method without
// exposing the proof internals.
type PreparedDispatchStore interface {
	MarkDispatchingPrepared(context.Context, Claim, PreparedDispatchProof) (Command, error)
}

type commandWorkHintStore interface {
	HasWorkHint(context.Context, Scope) (bool, error)
}

type commandScheduleStore interface {
	NextWorkAt(context.Context, Scope) (time.Time, bool, error)
}

type completionResultStore interface {
	finalizeWritebackWithCompletion(context.Context, Claim, writebackCompletion) (Command, error)
}

type noEffectResultStore interface {
	recordNoEffectResult(context.Context, Claim, int, string, outboundintentv1.ResourceNoEffectReason) (Command, error)
}

type typedResultStore interface {
	typedResultsAvailable(context.Context) (bool, error)
	recordTypedResult(context.Context, Claim, int, storedTypedTerminalResult) (Command, error)
	loadTypedResult(context.Context, Claim) (storedTypedTerminalResult, error)
}

type Worker struct {
	store                       commandStore
	preparer                    Preparer
	committer                   ResultCommitter
	scope                       Scope
	workerID                    string
	lease                       time.Duration
	pollInterval                time.Duration
	dependencyReconcileInterval time.Duration
	dependencyReconcileBudget   time.Duration
	now                         func() time.Time
}

func NewWorker(store commandStore, preparer Preparer, committer ResultCommitter, scope Scope, workerID string) (*Worker, error) {
	if store == nil || preparer == nil || committer == nil || scope.validate() != nil ||
		strings.TrimSpace(workerID) == "" || workerID != strings.TrimSpace(workerID) {
		return nil, ErrInvalidInput
	}
	return &Worker{
		store: store, preparer: preparer, committer: committer, scope: scope, workerID: workerID,
		lease: DefaultClaimLease, pollInterval: 250 * time.Millisecond,
		dependencyReconcileInterval: defaultDependencyReconcileInterval, now: time.Now,
	}, nil
}

// safeRetryDelay is the one domain policy for every retry that is known not to
// repeat an uncertain provider effect. AttemptCount deliberately remains the
// provider-dispatch counter; RetryCount supplies this independently durable
// series across process restarts. The deterministic equal-jitter tail prevents
// commands accepted together from converging on the same future claim instant
// without making restart timing depend on process-local randomness.
func safeRetryDelay(commandID string, retryCount int, minimum time.Duration) time.Duration {
	if retryCount < 0 {
		retryCount = 0
	}
	delay := safeRetryInitialDelay
	for remaining := retryCount; remaining > 0 && delay < safeRetryMaximumDelay; remaining-- {
		if delay >= safeRetryMaximumDelay/2 {
			delay = safeRetryMaximumDelay
			break
		}
		delay *= 2
	}
	if retryCount > 0 {
		spread := delay / 4
		digest := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%d", commandID, retryCount+1)))
		sample := uint64(digest[0])<<24 | uint64(digest[1])<<16 | uint64(digest[2])<<8 | uint64(digest[3])
		delay -= time.Duration(sample % uint64(spread+1))
	}
	if minimum > delay {
		return minimum
	}
	return delay
}

func retryMinimumAt(now time.Time, minimum time.Duration) time.Time {
	if minimum <= 0 {
		minimum = safeRetryInitialDelay
	}
	return now.UTC().Add(minimum)
}

func isLockNowaitContention(err error) bool {
	var mysqlErr *mysql.MySQLError
	return errors.As(err, &mysqlErr) && mysqlErr.Number == 3572
}

func (w *Worker) ProcessOne(ctx context.Context) (bool, error) {
	return w.processOne(ctx, true)
}

func (w *Worker) processOne(ctx context.Context, useHint bool) (bool, error) {
	if w == nil || w.store == nil || w.preparer == nil || w.committer == nil || ctx == nil {
		return false, ErrDependency
	}
	if hinted, ok := w.store.(commandWorkHintStore); useHint && ok {
		found, hintErr := hinted.HasWorkHint(ctx, w.scope)
		if hintErr == nil && !found {
			return false, nil
		}
		// A hint failure is never an availability failure. Fall through to the
		// authoritative transactions, which preserve the pre-optimization path.
	}
	if command, found, err := w.store.ClaimWriteback(ctx, w.scope, w.workerID, uuid.NewString(), w.lease); err != nil || found {
		if err != nil || !found {
			return false, err
		}
		return true, w.commitWriteback(ctx, command)
	}
	command, found, err := w.store.ClaimNext(ctx, w.scope, w.workerID, uuid.NewString(), w.lease)
	if err != nil || !found {
		return false, err
	}
	claim := Claim{Command: command, Token: command.ClaimToken}
	prepared, err := w.preparer.Prepare(ctx, command)
	if err != nil {
		var dependencyPreparation *dependencyPreparationError
		if errors.As(err, &dependencyPreparation) && dependencyPreparation != nil {
			store, supported := w.store.(dependencyCommandStore)
			if !safeCode(dependencyPreparation.Code) ||
				validateDependency(dependencyPreparation.Dependency) != nil {
				_, finishErr := w.store.Reject(ctx, claim, 0, "invalid_execution_binding")
				return true, finishErr
			}
			if !supported {
				_, retryErr := w.store.Retry(
					ctx, claim, retryMinimumAt(w.now(), 0), dependencyPreparation.Code,
				)
				return true, retryErr
			}
			_, waitErr := store.waitForDependency(
				ctx, claim, dependencyPreparation.Dependency, dependencyPreparation.Code,
			)
			if waitErr != nil {
				return true, waitErr
			}
			return true, nil
		}
		var preparation *PreparationError
		if !errors.As(err, &preparation) || preparation == nil || !safeCode(preparation.Code) {
			_, finishErr := w.rejectBeforeDispatch(ctx, claim, "invalid_execution_binding")
			return true, errors.Join(err, finishErr)
		}
		if preparation.Retryable {
			_, retryErr := w.store.Retry(
				ctx, claim, retryMinimumAt(w.now(), 0), preparation.Code,
			)
			// Preparation is a handled command outcome. Only failure to
			// persist that outcome belongs to the owner's recovery boundary.
			return true, retryErr
		}
		_, finishErr := w.rejectBeforeDispatch(ctx, claim, preparation.Code)
		return true, finishErr
	}
	if prepared == nil {
		_, finishErr := w.rejectBeforeDispatch(ctx, claim, "invalid_execution_binding")
		return true, finishErr
	}
	if _, typed := prepared.(TypedTerminalResultEffect); typed {
		store, ok := w.store.(typedResultStore)
		if !ok {
			_, finishErr := w.rejectBeforeDispatch(ctx, claim, "typed_result_schema_unavailable")
			return true, finishErr
		}
		available, capabilityErr := store.typedResultsAvailable(ctx)
		if capabilityErr != nil {
			_, retryErr := w.store.Retry(
				ctx, claim, retryMinimumAt(w.now(), 0), "typed_result_schema_unavailable",
			)
			if retryErr != nil {
				return true, errors.Join(capabilityErr, retryErr)
			}
			return true, nil
		}
		if !available {
			_, finishErr := w.rejectBeforeDispatch(ctx, claim, "typed_result_schema_unavailable")
			return true, finishErr
		}
	}
	if command.LeaseExpiresAt == nil ||
		!command.LeaseExpiresAt.After(w.now().UTC().Add(dispatchCompletionReserve)) {
		_, retryErr := w.store.Retry(
			ctx, claim, retryMinimumAt(w.now(), 0), "dispatch_window_unavailable",
		)
		return true, retryErr
	}
	var legacyProof *preparedLegacyRemoteLink
	var legacyStore legacyRemoteLinkStore
	var scopedLinkProof *preparedScopedRemoteLink
	if effect, ok := prepared.(interface {
		scopedRemoteLinkProof() *preparedScopedRemoteLink
	}); ok {
		scopedLinkProof = effect.scopedRemoteLinkProof()
	}
	if effect, ok := prepared.(interface {
		legacyRemoteLinkProof() *preparedLegacyRemoteLink
	}); ok && effect.legacyRemoteLinkProof() != nil {
		var available bool
		legacyProof = effect.legacyRemoteLinkProof()
		legacyStore, available = w.store.(legacyRemoteLinkStore)
		if !available {
			_, finishErr := w.rejectBeforeDispatch(ctx, claim, "remote_link_fence_unavailable")
			return true, finishErr
		}
	}
	if scopedLinkProof != nil {
		dispatchStore, available := w.store.(PreparedDispatchStore)
		if !available {
			// MarkDispatching has no prepared-proof parameter. Falling back would
			// discard the snapshot and permit stale provider I/O if the link changed
			// after Prepare, so every unexpected proof-bearing composition fails
			// closed before dispatch.
			_, finishErr := w.rejectBeforeDispatch(ctx, claim, "remote_link_fence_unavailable")
			return true, finishErr
		}
		command, err = dispatchStore.MarkDispatchingPrepared(
			ctx,
			claim,
			PreparedDispatchProof{scoped: scopedLinkProof, legacy: legacyProof},
		)
	} else if legacyProof != nil {
		command, err = legacyStore.markDispatchingWithLegacyProof(ctx, claim, legacyProof)
	} else {
		command, err = w.store.MarkDispatching(ctx, claim)
	}
	if err != nil {
		if isLockNowaitContention(err) {
			_, retryErr := w.store.Retry(
				ctx, claim, retryMinimumAt(w.now(), 0), "remote_link_fence_contended",
			)
			if retryErr != nil {
				return true, errors.Join(err, retryErr)
			}
			return true, nil
		}
		if errors.Is(err, ErrIdentityConflict) {
			// A prepared link may have been retired or changed before dispatch.
			// This is a command-scoped no-I/O rejection, not a service failure.
			_, rejectErr := w.rejectBeforeDispatch(ctx, claim, "remote_link_conflict")
			return true, rejectErr
		}
		return true, err
	}
	claim.Command = command
	dispatchDeadline := command.LeaseExpiresAt.Add(-dispatchCompletionReserve)
	dispatchCtx, cancel := context.WithDeadline(ctx, dispatchDeadline)
	var result EffectResult
	var checkpointErr error
	var providerCheckpointStatus int
	var providerCheckpointRemoteID string
	providerCheckpointCalled := false
	typedCheckpointCalled := false
	if typed, ok := prepared.(TypedTerminalResultEffect); ok {
		result = typed.DispatchWithTypedTerminalResultCheckpoint(
			dispatchCtx,
			func(checkpointCtx context.Context, status int, terminal TypedTerminalResult) error {
				if typedCheckpointCalled {
					checkpointErr = ErrIdentityConflict
					return checkpointErr
				}
				typedCheckpointCalled = true
				canonical := append([]byte(nil), terminal.CanonicalResult...)
				sum := sha256.Sum256(canonical)
				stored := storedTypedTerminalResult{
					formatVersion: terminal.ResultFormatVersion,
					digest:        "sha256:" + hex.EncodeToString(sum[:]),
					canonical:     canonical,
				}
				request := outboundintentv1.CommitResultRequest{
					IntentID: command.Intent.IntentID, PayloadDigest: command.Intent.PayloadDigest,
					CatalogRevision: command.Scope.CatalogRevision, IntegrationCommandID: command.CommandID,
					ResultFormatVersion: stored.formatVersion, ResultDigest: stored.digest,
					CanonicalResult: append([]byte(nil), stored.canonical...),
				}
				if (status != 0 && (status < 100 || status > 599)) ||
					outboundintentv1.ValidateCommitResultRequestForIntent(command.Intent, request) != nil {
					checkpointErr = ErrIdentityConflict
					return checkpointErr
				}
				store, storeOK := w.store.(typedResultStore)
				if !storeOK {
					checkpointErr = ErrSchemaUnverified
					return checkpointErr
				}
				_, checkpointErr = store.recordTypedResult(checkpointCtx, claim, status, stored)
				return checkpointErr
			},
		)
	} else if checkpointed, ok := prepared.(ProviderSuccessCheckpointEffect); ok {
		result = checkpointed.DispatchWithProviderSuccessCheckpoint(
			dispatchCtx,
			func(
				checkpointCtx context.Context,
				status int,
				finalization RemoteLinkFinalization,
				code string,
			) error {
				providerCheckpointStatus = status
				providerCheckpointRemoteID = finalization.RemoteID
				if providerCheckpointCalled {
					checkpointErr = ErrIdentityConflict
					return checkpointErr
				}
				providerCheckpointCalled = true
				var checkpointed Command
				if legacyProof != nil {
					checkpointed, checkpointErr = legacyStore.checkpointProviderSuccessWithLegacyProof(
						checkpointCtx, claim, status, finalization, code, legacyProof,
					)
				} else {
					checkpointed, checkpointErr = w.store.CheckpointProviderSuccess(
						checkpointCtx, claim, status, finalization, code,
					)
				}
				if checkpointErr == nil {
					claim.Command = checkpointed
				}
				return checkpointErr
			},
		)
	} else {
		result = prepared.Dispatch(dispatchCtx)
	}
	cancel()
	if checkpointErr != nil {
		if providerCheckpointCalled && errors.Is(checkpointErr, ErrIdentityConflict) {
			return true, w.isolatePostDispatchIdentityConflict(
				ctx, claim, providerCheckpointStatus, providerCheckpointRemoteID,
				"remote_link_identity_conflict", checkpointErr,
			)
		}
		// The checkpoint transaction may have committed even when its response
		// was lost. Leave the dispatch claim untouched; lease recovery inspects
		// the durable provider identity or typed terminal result and resumes only
		// the safe Runtime writeback phase.
		return true, fmt.Errorf("connector outbound intent durable effect checkpoint: %w", checkpointErr)
	}
	if typedCheckpointCalled {
		if result.Outcome != EffectTypedTerminal {
			return true, ErrIdentityConflict
		}
		return true, nil
	}
	if result.Outcome == EffectTypedTerminal {
		_, err = w.store.Uncertain(ctx, claim, result.ResponseStatus, "typed_result_checkpoint_missing")
		return true, err
	}
	if !safeCode(result.FailureCode) && result.Outcome != EffectSucceeded {
		result = EffectResult{Outcome: EffectUncertain, FailureCode: "invalid_effect_result"}
	}
	switch result.Outcome {
	case EffectSucceeded:
		if result.LinkFinalization == nil {
			_, err = w.store.Uncertain(ctx, claim, result.ResponseStatus, "missing_link_finalization")
			break
		}
		if legacyProof != nil {
			_, err = legacyStore.recordProviderSuccessWithLegacyProof(ctx, claim, result.ResponseStatus, *result.LinkFinalization, legacyProof)
		} else {
			_, err = w.store.RecordProviderSuccess(ctx, claim, result.ResponseStatus, *result.LinkFinalization)
		}
	case EffectRejectedNoEffect:
		if !providerCheckpointCalled && command.RemoteID == "" {
			_, err = w.rejectNoEffect(ctx, claim, result.ResponseStatus, result.FailureCode,
				outboundintentv1.ResourceNoEffectDefinitiveNoEffect)
		} else if store, ok := w.store.(postDispatchIsolationStore); ok {
			_, err = store.IsolatePostDispatch(ctx, claim, 0, claim.Command.RemoteID, "invalid_no_effect_evidence")
		} else {
			_, err = w.store.Uncertain(ctx, claim, result.ResponseStatus, "invalid_no_effect_evidence")
		}
	case EffectRejected:
		_, err = w.store.Reject(ctx, claim, result.ResponseStatus, result.FailureCode)
	case EffectNotDispatched:
		_, err = w.store.Retry(
			ctx, claim, retryMinimumAt(w.now(), result.RetryAfter), result.FailureCode,
		)
	case EffectRetryableAfterSuccess:
		if result.LinkFinalization == nil {
			_, err = w.store.Uncertain(ctx, claim, result.ResponseStatus, "missing_link_finalization")
			break
		}
		if legacyProof != nil {
			_, err = legacyStore.recordProviderSuccessForRetryWithLegacyProof(
				ctx, claim, result.ResponseStatus, *result.LinkFinalization,
				retryMinimumAt(w.now(), result.RetryAfter), result.FailureCode, legacyProof,
			)
		} else {
			_, err = w.store.RecordProviderSuccessForRetry(
				ctx, claim, result.ResponseStatus, *result.LinkFinalization,
				retryMinimumAt(w.now(), result.RetryAfter), result.FailureCode,
			)
		}
	case EffectTerminalAfterSuccess:
		if result.LinkFinalization == nil {
			_, err = w.store.Uncertain(ctx, claim, result.ResponseStatus, "missing_link_finalization")
			break
		}
		if legacyProof != nil {
			_, err = legacyStore.recordProviderSuccessTerminalWithLegacyProof(
				ctx, claim, result.ResponseStatus, *result.LinkFinalization, result.FailureCode, legacyProof,
			)
		} else {
			_, err = w.store.RecordProviderSuccessTerminal(
				ctx, claim, result.ResponseStatus, *result.LinkFinalization, result.FailureCode,
			)
		}
	case EffectUncertain:
		_, err = w.store.Uncertain(ctx, claim, result.ResponseStatus, result.FailureCode)
	default:
		_, err = w.store.Uncertain(ctx, claim, 0, "invalid_effect_result")
	}
	if err != nil && errors.Is(err, ErrIdentityConflict) && result.LinkFinalization != nil {
		switch result.Outcome {
		case EffectSucceeded, EffectRetryableAfterSuccess, EffectTerminalAfterSuccess:
			err = w.isolatePostDispatchIdentityConflict(
				ctx, claim, result.ResponseStatus, result.LinkFinalization.RemoteID,
				"remote_link_identity_conflict", err,
			)
		}
	}
	return true, err
}

// rejectBeforeDispatch reports a terminal no-effect result only if this command
// has not already checkpointed a provider mutation in an earlier attempt.
func (w *Worker) rejectBeforeDispatch(ctx context.Context, claim Claim, code string) (Command, error) {
	if claim.Command.RemoteID != "" {
		return w.store.Reject(ctx, claim, 0, code)
	}
	return w.rejectNoEffect(ctx, claim, 0, code, outboundintentv1.ResourceNoEffectRejectedBeforeDispatch)
}

func (w *Worker) rejectNoEffect(ctx context.Context, claim Claim, status int, code string, reason outboundintentv1.ResourceNoEffectReason) (Command, error) {
	if claim.Command.Intent.Operation != outboundintentv1.OperationUpsert && claim.Command.Intent.Operation != outboundintentv1.OperationDelete {
		return w.store.Reject(ctx, claim, status, code)
	}
	store, ok := w.store.(noEffectResultStore)
	if !ok {
		// Preserve the public NewWorker store interface. The durable Repository
		// supports no-effect receipts; older custom stores retain their behavior.
		return w.store.Reject(ctx, claim, status, code)
	}
	command, err := store.recordNoEffectResult(ctx, claim, status, code, reason)
	if errors.Is(err, ErrSchemaUnverified) {
		// A schema rollout must not turn a safe refusal into an unacknowledged
		// terminal command that holds Runtime's owner fence permanently.
		return w.store.Retry(ctx, claim, retryMinimumAt(w.now(), 0), "typed_result_schema_unavailable")
	}
	return command, err
}

func (w *Worker) isolatePostDispatchIdentityConflict(
	ctx context.Context,
	claim Claim,
	status int,
	remoteID string,
	code string,
	cause error,
) error {
	if !errors.Is(cause, ErrIdentityConflict) {
		return cause
	}
	store, ok := w.store.(postDispatchIsolationStore)
	if !ok {
		return errors.Join(cause, ErrDependency)
	}
	_, isolateErr := store.IsolatePostDispatch(ctx, claim, status, remoteID, code)
	if isolateErr != nil {
		return errors.Join(cause, isolateErr)
	}
	log.Printf(
		"WARN connector outbound intent command isolated after post-dispatch identity conflict failure_code=%s response_status=%d remote_id_present=%t: %v",
		code, status, strings.TrimSpace(remoteID) != "", cause,
	)
	return nil
}

func (w *Worker) nextWorkAt(ctx context.Context) (time.Time, bool, error) {
	if w == nil || w.store == nil || ctx == nil {
		return time.Time{}, false, ErrDependency
	}
	scheduled, ok := w.store.(commandScheduleStore)
	if !ok {
		return time.Time{}, false, nil
	}
	return scheduled.NextWorkAt(ctx, w.scope)
}

func (w *Worker) commitWriteback(ctx context.Context, command Command) error {
	claim := Claim{Command: command, Token: command.ClaimToken}
	request := outboundintentv1.CommitResultRequest{
		IntentID: command.Intent.IntentID, PayloadDigest: command.Intent.PayloadDigest,
		CatalogRevision:      command.Scope.CatalogRevision,
		IntegrationCommandID: command.CommandID, RemoteID: command.RemoteID,
	}
	if command.RemoteID == "" {
		store, ok := w.store.(typedResultStore)
		if !ok {
			_, err := w.store.RetryWriteback(
				ctx, claim, retryMinimumAt(w.now(), 0), "typed_result_store_unavailable",
			)
			return err
		}
		typed, err := store.loadTypedResult(ctx, claim)
		if err != nil {
			_, retryErr := w.store.RetryWriteback(
				ctx, claim, retryMinimumAt(w.now(), 0), "typed_result_unavailable",
			)
			return errors.Join(err, retryErr)
		}
		request.ResultFormatVersion, request.ResultDigest = typed.formatVersion, typed.digest
		request.CanonicalResult = append([]byte(nil), typed.canonical...)
	}
	if command.LeaseExpiresAt == nil || !command.LeaseExpiresAt.After(w.now().UTC().Add(dispatchCompletionReserve)) {
		_, err := w.store.RetryWriteback(
			ctx, claim, retryMinimumAt(w.now(), 0), "writeback_window_unavailable",
		)
		return err
	}
	commitCtx, cancel := context.WithDeadline(ctx, command.LeaseExpiresAt.Add(-dispatchCompletionReserve))
	response, err := w.committer.CommitOutboundResult(commitCtx, request)
	cancel()
	if err != nil {
		_, retryErr := w.store.RetryWriteback(
			ctx, claim, retryMinimumAt(w.now(), 0), "runtime_projection_unavailable",
		)
		if retryErr != nil {
			return errors.Join(err, retryErr)
		}
		return nil
	}
	if outboundintentv1.ValidateCommitResultResponse(request, response) != nil {
		_, retryErr := w.store.RetryWriteback(
			ctx, claim, retryMinimumAt(w.now(), 0), "runtime_projection_invalid",
		)
		if retryErr != nil {
			return errors.Join(ErrIdentityConflict, retryErr)
		}
		return nil
	}
	if store, ok := w.store.(completionResultStore); ok {
		completion := writebackCompletion{request: request, response: response}
		completion.request.CanonicalResult = append([]byte(nil), request.CanonicalResult...)
		_, err = store.finalizeWritebackWithCompletion(ctx, claim, completion)
	} else {
		_, err = w.store.FinalizeWriteback(ctx, claim)
	}
	if errors.Is(err, ErrIdentityConflict) {
		return w.isolatePostDispatchIdentityConflict(
			ctx, claim, command.ResponseStatus, command.RemoteID,
			"writeback_identity_conflict", err,
		)
	}
	return err
}

func (w *Worker) Run(ctx context.Context, started func()) error {
	if w == nil || w.store == nil || w.preparer == nil || w.committer == nil || started == nil || ctx == nil {
		return ErrDependency
	}
	// Preserve the standalone entrypoint's first-turn startup contract.
	// Production readiness is owned by Service.Run's schema/authority probes.
	if _, err := w.processWithRecovery(ctx, true); err != nil {
		return fmt.Errorf("connector outbound intent startup probe: %w", err)
	}
	started()
	var dependencyTicker *time.Ticker
	var dependencyTick <-chan time.Time
	if _, ok := w.store.(dependencyReconcileStore); ok {
		interval := w.dependencyReconcileInterval
		if interval <= 0 {
			interval = defaultDependencyReconcileInterval
		}
		dependencyTicker = time.NewTicker(interval)
		dependencyTick = dependencyTicker.C
		defer dependencyTicker.Stop()
	}
	dependencyCursor := ""
	reconcileDependencies := func() error {
		nextCursor, more, _, err := w.reconcileDependencies(ctx, dependencyCursor)
		if err != nil {
			return err
		}
		if more {
			dependencyCursor = nextCursor
		} else {
			dependencyCursor = ""
		}
		return nil
	}
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-dependencyTick:
			if err := reconcileDependencies(); err != nil {
				return err
			}
		default:
		}
		processed, err := w.processWithRecovery(ctx, true)
		if err != nil {
			return err
		}
		if processed {
			continue
		}
		timer := time.NewTimer(w.pollInterval)
		select {
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			return ctx.Err()
		case <-dependencyTick:
			if !timer.Stop() {
				select {
				case <-timer.C:
				default:
				}
			}
			if err := reconcileDependencies(); err != nil {
				return err
			}
		case <-timer.C:
		}
	}
}

func (w *Worker) reconcileDependencies(
	ctx context.Context,
	cursor string,
) (string, bool, int, error) {
	if w == nil || w.store == nil {
		return "", false, 0, ErrDependency
	}
	store, ok := w.store.(dependencyReconcileStore)
	if !ok {
		return "", false, 0, nil
	}
	budget := w.dependencyReconcileBudget
	if budget <= 0 {
		budget = defaultDependencyReconcileBudget
	}
	turnCtx, cancel := context.WithTimeout(ctx, budget)
	defer cancel()
	nextCursor, more, woken := cursor, true, 0
	remaining := maxDependencyReconcileBatch
	err := dependencyretry.Do(turnCtx, "outbound dependency reconciliation", func() error {
		if remaining == 0 {
			return nil
		}
		attemptCursor, attemptMore, attemptWoken, inspected, attemptErr := store.reconcileDependencyWaits(
			turnCtx, w.scope, nextCursor, remaining,
		)
		if inspected < 0 || inspected > remaining {
			return ErrSchemaUnverified
		}
		remaining -= inspected
		// A partially completed page reports its last fully checked waiter.
		// Retain both its cursor and successful CAS wakes when the budget expires.
		nextCursor, more = attemptCursor, attemptMore
		woken += attemptWoken
		return attemptErr
	})
	if ctx.Err() != nil {
		return nextCursor, more, woken, ctx.Err()
	}
	if errors.Is(turnCtx.Err(), context.DeadlineExceeded) {
		// Budget exhaustion is a yield, not an owner failure. A timed-out CAS
		// may have committed; the service drains after an unfinished page.
		// Rechecking the unfinished waiter remains idempotent.
		return nextCursor, true, woken, nil
	}
	if err != nil {
		return nextCursor, more, woken, err
	}
	return nextCursor, more, woken, nil
}

func (w *Worker) processWithRecovery(ctx context.Context, useHint bool) (bool, error) {
	var processed bool
	err := dependencyretry.Do(ctx, "outbound durable worker turn", func() error {
		var err error
		// Re-enter at the durable claim, never replay a PreparedEffect. A lost
		// dispatch/checkpoint response stays claimed until lease recovery chooses
		// writeback or uncertainty; it is not permission to call the provider twice.
		processed, err = w.processOne(ctx, useHint)
		return err
	})
	return processed, err
}