package worker

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

	driversql "github.com/go-sql-driver/mysql"
	"github.com/google/uuid"
	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/service"
	reposync "github.com/pingcap/agent-git-service/internal/sync"
	"github.com/pingcap/agent-git-service/internal/sync/canonical"
	"github.com/pingcap/agent-git-service/internal/sync/gittransport"
)

const (
	defaultPollInterval = 5 * time.Second
	defaultLeaseTTL     = 2 * time.Minute

	// Renew well before the lease boundary. A full inventory can spend minutes
	// in Git or a provider API call, so the lease keeper must have enough time
	// to retry a transient TiDB row-lock wait without allowing the lease to
	// expire underneath an otherwise healthy Run.
	minLeaseHeartbeatInterval = 5 * time.Second
	maxLeaseHeartbeatInterval = 20 * time.Second
	minLeaseRenewalTimeout    = 3 * time.Second
	// TiDB's lock-wait timeout is 50s in staging. A 10s local deadline turns a
	// recoverable wait into a cancelled database transaction before TiDB can
	// return its retryable 1205 error. Leave room for one real lock wait and a
	// retry while still bounding the heartbeat operation.
	maxLeaseRenewalTimeout = time.Minute
	leaseRenewalAttempts   = 4
	// Stop renewal attempts before the fenced lease itself expires. This keeps a
	// stalled worker from performing provider writes after another worker can
	// safely reclaim the Binding.
	leaseExpirySafety = 5 * time.Second
)

var errRepositorySyncCancelled = errors.New("repository sync run cancelled")

type Options struct {
	WorkerID     string
	PollInterval time.Duration
	LeaseTTL     time.Duration
	BatchLimits  reposync.BatchLimits
}

type Worker struct {
	service *service.Service
	factory reposync.ProviderFactory
	options Options
}

func New(svc *service.Service, factory reposync.ProviderFactory, options Options) (*Worker, error) {
	if svc == nil || factory == nil {
		return nil, fmt.Errorf("repository sync worker requires service and provider factory")
	}
	if strings.TrimSpace(options.WorkerID) == "" {
		options.WorkerID = "sync-" + uuid.NewString()
	}
	if options.PollInterval <= 0 {
		options.PollInterval = defaultPollInterval
	}
	if options.LeaseTTL <= 0 {
		options.LeaseTTL = defaultLeaseTTL
	}
	return &Worker{service: svc, factory: factory, options: options}, nil
}

// Run continuously schedules, claims, and executes durable sync Runs for one
// tenant context. A panic is deliberately not swallowed; the durable lease
// expires and another process reconstructs work from RunItems.
func (w *Worker) Run(ctx context.Context) error {
	ticker := time.NewTicker(w.options.PollInterval)
	defer ticker.Stop()
	for {
		if err := w.DrainOnce(ctx); err != nil && !errors.Is(err, context.Canceled) {
			slog.ErrorContext(ctx, "repository sync worker iteration failed", "worker_id", w.options.WorkerID, "error", err)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func (w *Worker) DrainOnce(ctx context.Context) error {
	if _, err := w.service.EnqueueDueRepositorySyncRuns(ctx, time.Now().UTC(), 100); err != nil {
		return fmt.Errorf("enqueue schedules: %w", err)
	}
	claim, ok, err := w.service.ClaimNextRepositorySyncRun(ctx, w.options.WorkerID, w.options.LeaseTTL)
	if err != nil || !ok {
		return err
	}
	err = w.executeClaim(ctx, claim)
	// A process shutdown must leave the still-valid lease in place. It will
	// expire and be fenced by the next Worker; releasing it while the Run is
	// still running created the historical "running + empty lease" orphan.
	if errors.Is(err, context.Canceled) && ctx.Err() != nil {
		return err
	}

	finalizeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second)
	defer cancel()
	if err == nil {
		releaseErr := w.service.ReleaseRepositorySyncLease(finalizeCtx, claim.Binding.ID, claim.LeaseOwner, claim.LeaseGeneration)
		return errors.Join(err, releaseErr)
	}
	if errors.Is(err, reposync.ErrRateLimited) {
		_, deferErr := w.service.DeferRepositorySyncRun(finalizeCtx, claim, rateLimitRetryAt(err), terminalErrorCode(err), err.Error())
		if deferErr != nil {
			return errors.Join(err, deferErr)
		}
		releaseErr := w.service.ReleaseRepositorySyncLease(finalizeCtx, claim.Binding.ID, claim.LeaseOwner, claim.LeaseGeneration)
		return errors.Join(err, releaseErr)
	}
	_, terminateErr := w.service.TerminateRepositorySyncRun(finalizeCtx, claim, terminalStatus(err), terminalErrorCode(err), err.Error())
	if terminateErr != nil {
		// Do not clear a lease for a Run we could not terminalize. Either this
		// worker lost its generation or a later Worker will reclaim it on expiry.
		return errors.Join(err, terminateErr)
	}
	releaseErr := w.service.ReleaseRepositorySyncLease(finalizeCtx, claim.Binding.ID, claim.LeaseOwner, claim.LeaseGeneration)
	return errors.Join(err, releaseErr)
}

func rateLimitRetryAt(err error) time.Time {
	var limited *reposync.RateLimitError
	if errors.As(err, &limited) && limited != nil {
		return limited.RetryAt
	}
	return time.Time{}
}

func (w *Worker) executeClaim(parent context.Context, claim service.RepositorySyncClaim) (resultErr error) {
	ctx, cancel := context.WithCancelCause(parent)
	heartbeatErr := make(chan error, 1)
	heartbeatDone := make(chan struct{})
	heartbeatReady := make(chan struct{})
	go func() {
		defer close(heartbeatDone)
		w.heartbeat(ctx, claim, heartbeatErr, cancel, heartbeatReady)
	}()
	// Do not enter a potentially long Git inventory before the lease keeper has
	// installed its timer. In staging, a large local Git probe could delay a
	// newly-started goroutine long enough for the first renewal to happen at the
	// lease boundary, leaving no room to recover from a transient TiDB lock.
	select {
	case <-heartbeatReady:
	case <-ctx.Done():
		return context.Cause(ctx)
	}
	// Do not finalize or release a Run while a heartbeat may still be
	// updating the same sync_bindings row. That overlap was a direct source of
	// the TiDB 50-second row-lock waits observed in staging. Cancellation is
	// propagated to the in-flight renewal and the wait establishes a clean
	// handoff to the terminal/pause transaction below.
	defer func() {
		cancel(nil)
		<-heartbeatDone
	}()
	defer func() {
		if errors.Is(resultErr, context.Canceled) {
			if cause := context.Cause(ctx); cause != nil && !errors.Is(cause, context.Canceled) {
				resultErr = cause
			}
		}
	}()
	if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageCheckingAccess); err != nil {
		return err
	}

	sourceSide, targetSide := claim.Run.SourceSide, claim.Run.TargetSide
	if claim.Run.Kind == service.SyncRunKindHandoff && claim.Run.Phase == service.SyncRunPhaseHandoffReverseDry {
		sourceSide, targetSide = targetSide, sourceSide
	}
	sourceEndpoint, err := endpointForSide(claim.Endpoints, sourceSide)
	if err != nil {
		return err
	}
	targetEndpoint, err := endpointForSide(claim.Endpoints, targetSide)
	if err != nil {
		return err
	}
	source, err := w.factory.Open(ctx, sourceEndpoint)
	if err != nil {
		return fmt.Errorf("open source provider: %w", err)
	}
	defer source.Close()
	if resolved := source.Endpoint(); resolved.RepositoryFullName != "" && resolved.RepositoryFullName != sourceEndpoint.RepositoryFullName {
		if err := w.service.RefreshRepositorySyncEndpointLocator(ctx, claim.Binding.ID, sourceSide, sourceEndpoint.RepositoryID, resolved.RepositoryFullName); err != nil {
			return err
		}
		sourceEndpoint.RepositoryFullName = resolved.RepositoryFullName
	}
	target, err := w.factory.Open(ctx, targetEndpoint)
	if err != nil {
		return fmt.Errorf("open target provider: %w", err)
	}
	defer target.Close()
	if resolved := target.Endpoint(); resolved.RepositoryFullName != "" && resolved.RepositoryFullName != targetEndpoint.RepositoryFullName {
		if err := w.service.RefreshRepositorySyncEndpointLocator(ctx, claim.Binding.ID, targetSide, targetEndpoint.RepositoryID, resolved.RepositoryFullName); err != nil {
			return err
		}
		targetEndpoint.RepositoryFullName = resolved.RepositoryFullName
	}
	if sourceVisibility, ok := source.(reposync.VisibilityReporter); ok {
		if targetVisibility, targetOK := target.(reposync.VisibilityReporter); targetOK {
			confirmation := claim.Run.VisibilityConfirmation
			if claim.Run.Kind == service.SyncRunKindHandoff && claim.Run.Phase == service.SyncRunPhaseHandoffReverseDry {
				confirmation = claim.Run.HandoffVisibilityConfirmation
			}
			if err := w.service.AssertRepositorySyncVisibility(ctx, claim, sourceSide, sourceVisibility.Visibility(), targetVisibility.Visibility(), confirmation); err != nil {
				return err
			}
		}
	}

	if err := w.service.AssertRepositorySyncLease(ctx, claim); err != nil {
		return err
	}
	if err := w.checkCancellation(ctx, claim.Run.ID); err != nil {
		return err
	}
	links, err := w.service.RepositorySyncLinkViews(ctx, claim.Binding.ID, sourceSide)
	if err != nil {
		return err
	}
	if aware, ok := source.(reposync.ObjectLinkAware); ok {
		aware.SetObjectLinks(links, true)
	}
	if aware, ok := target.(reposync.ObjectLinkAware); ok {
		aware.SetObjectLinks(links, false)
	}
	var existing int64
	if err := w.service.DBForCtx(ctx).Model(&db.SyncRunItem{}).Where("run_id = ? AND phase = ?", claim.Run.ID, claim.Run.Phase).Count(&existing).Error; err != nil {
		return err
	}
	var sourceDigest, targetDigest string
	plannerMode := claim.Run.PlannerMode
	// RunItems are written in short planning transactions. They are executable
	// only after PersistRepositorySyncPlan records a non-auto planner mode as
	// its final completion marker. A reclaimed worker must not mistake a
	// partially persisted plan for a complete plan.
	planPersisted := existing > 0 && plannerMode != service.SyncPlannerModeAuto
	if !planPersisted {
		var sourceInventory, targetInventory []reposync.InventoryObject
		targetConfirmedEmpty := false
		checkpointPlan := false
		checkpointSkipped := 0
		if checker, ok := target.(reposync.EmptyChecker); ok {
			if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageInventoryTarget); err != nil {
				return err
			}
			targetConfirmedEmpty, err = checker.IsEmpty(ctx)
			if err != nil {
				return fmt.Errorf("check target emptiness: %w", err)
			}
		}
		firstBaseline := claim.Run.AuditMode == service.SyncAuditModeFull && claim.Binding.LastConvergedRunID == nil && len(links) == 0
		if targetConfirmedEmpty || firstBaseline {
			if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageInventorySource); err != nil {
				return err
			}
			var digest string
			sourceInventory, digest, err = source.Inventory(ctx)
			if err != nil {
				return fmt.Errorf("source inventory: %w", err)
			}
			sourceDigest = digest
			sourceInventory, err = canonical.MapForTarget(sourceInventory, sourceEndpoint.Provider, targetEndpoint.Provider)
			if err != nil {
				return fmt.Errorf("map source inventory: %w", err)
			}
		}
		if targetConfirmedEmpty {
			targetDigest = reposync.EmptyInventoryDigest()
		} else if firstBaseline {
			// Preserve the existing non-empty first-migration behavior: the first
			// baseline performs a complete diff so matching Git refs can be adopted
			// safely and all unmanaged collisions are visible.
			if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageInventoryTarget); err != nil {
				return err
			}
			var inventoryDigest string
			targetInventory, inventoryDigest, err = target.Inventory(ctx)
			if err != nil {
				return fmt.Errorf("target inventory: %w", err)
			}
			targetDigest = inventoryDigest
			targetConfirmedEmpty = len(targetInventory) == 0
		} else {
			checkpointPlan = true
			if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageInventorySource); err != nil {
				return err
			}
			sourceIndex, digest, indexErr := readInventoryIndex(ctx, source)
			if indexErr != nil {
				return fmt.Errorf("source inventory index: %w", indexErr)
			}
			sourceDigest = digest
			fullAudit := claim.Run.AuditMode == service.SyncAuditModeFull
			sourceIndex, links, checkpointSkipped, err = reposync.SelectCheckpointCandidates(sourceIndex, links, claim.Run.MapperVersion, fullAudit)
			if err != nil {
				return fmt.Errorf("select checkpoint candidates: %w", err)
			}
			sourceInventory, err = readSourceCandidateDetails(ctx, source, sourceIndex)
			if err != nil {
				return fmt.Errorf("read source candidate details: %w", err)
			}
			sourceInventory, err = canonical.MapForTarget(sourceInventory, sourceEndpoint.Provider, targetEndpoint.Provider)
			if err != nil {
				return fmt.Errorf("map source candidates: %w", err)
			}
			if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageInventoryTarget); err != nil {
				return err
			}
			targetInventory, err = readManagedTargetCandidates(ctx, target, links)
			if err != nil {
				return fmt.Errorf("read managed target candidates: %w", err)
			}
			collisions, collisionErr := readNewTargetCollisions(ctx, target, sourceInventory, links)
			if collisionErr != nil {
				return fmt.Errorf("read new-object target collisions: %w", collisionErr)
			}
			targetInventory = appendDistinctInventory(targetInventory, collisions...)
			targetDigest = inventoryDigest(targetInventory)
		}
		fastForwardUnlinkedRefs, ancestryErr := detectUnlinkedFastForwardRefs(ctx, source, target, sourceInventory, targetInventory, links)
		if ancestryErr != nil {
			// Ancestry is an opt-in safety proof, not a prerequisite for the
			// normal conflict-safe plan. If a ref moved or a transient transport
			// error prevents this extra read, preserve the target and surface the
			// normal conflict rather than failing the entire sync run.
			slog.WarnContext(ctx, "repository sync could not verify unlinked ref ancestry", "run_id", claim.Run.ID, "error", ancestryErr)
			fastForwardUnlinkedRefs = map[string]bool{}
		}
		confirmedUnlinkedRefs := map[string]bool{}
		if claim.Binding.RefConflictPolicy == service.SyncRefConflictPolicyUseSource {
			confirmedUnlinkedRefs = unlinkedRefCollisions(sourceInventory, targetInventory, links)
		}
		if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStagePlanning); err != nil {
			return err
		}
		planResult, persistErr := w.service.PersistRepositorySyncPlan(ctx, claim.Run.ID, service.PersistRepositorySyncPlanInput{
			Source: sourceInventory, Target: targetInventory, Links: links,
			SourceInventoryDigest: sourceDigest, TargetInventoryDigest: targetDigest,
			TargetConfirmedEmpty: targetConfirmedEmpty, CheckpointPlan: checkpointPlan,
			CheckpointSkipped: checkpointSkipped, FastForwardUnlinkedRefs: fastForwardUnlinkedRefs,
			ConfirmedUnlinkedRefs: confirmedUnlinkedRefs,
			BatchLimits:           w.options.BatchLimits,
			LeaseOwner:            claim.LeaseOwner, LeaseGeneration: claim.LeaseGeneration,
		})
		if persistErr != nil {
			return fmt.Errorf("persist plan: %w", persistErr)
		}
		plannerMode = planResult.PlannerMode
	}
	if sourceDigest == "" {
		if claim.Run.Kind == service.SyncRunKindHandoff && claim.Run.Phase == service.SyncRunPhaseHandoffReverseDry {
			sourceDigest = claim.Run.TargetInventoryDigest
		} else {
			sourceDigest = claim.Run.SourceInventoryDigest
		}
	}
	if targetDigest == "" {
		if claim.Run.Kind == service.SyncRunKindHandoff && claim.Run.Phase == service.SyncRunPhaseHandoffReverseDry {
			targetDigest = claim.Run.SourceInventoryDigest
		} else {
			targetDigest = claim.Run.TargetInventoryDigest
		}
	}

	batches, rows, err := w.service.LoadRepositorySyncExecutionPlan(ctx, claim.Run.ID, claim.Run.Phase, w.options.BatchLimits)
	if err != nil {
		return err
	}
	if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageApplying); err != nil {
		return err
	}
	for _, batch := range batches {
		batch.Bootstrap = plannerMode == reposync.PlannerModeBootstrap
		if err := w.checkCancellation(ctx, claim.Run.ID); err != nil {
			return err
		}
		if err := w.service.ResolveRepositorySyncBatchRelations(ctx, claim.Binding.ID, sourceSide, &batch); err != nil {
			return err
		}
		if err := w.checkHeartbeat(heartbeatErr); err != nil {
			return err
		}
		if err := w.service.AssertRepositorySyncLease(ctx, claim); err != nil {
			return err
		}
		sourceStates, err := readSourceBatch(ctx, source, batch.Items, sourceEndpoint.Provider, targetEndpoint.Provider)
		if err != nil {
			return fmt.Errorf("refresh source %s batch: %w", batch.ObjectType, err)
		}
		targetStates, err := recoverTargetBatch(ctx, target, batch.Items)
		if err != nil {
			return fmt.Errorf("recover target %s batch: %w", batch.ObjectType, err)
		}
		if err := w.checkHeartbeat(heartbeatErr); err != nil {
			return err
		}
		if err := w.service.AssertRepositorySyncLease(ctx, claim); err != nil {
			return err
		}
		ready := batch.Items[:0]
		for _, item := range batch.Items {
			row, ok := rows[runItemKey(item)]
			if !ok {
				return fmt.Errorf("durable RunItem missing for %s", reposync.ExecutionKey(item))
			}
			if item.SourceIdentity != "" {
				current, found := sourceStates[item.SourceIdentity]
				sourceChanged := false
				if item.Action == reposync.ActionDelete {
					// A delete is valid only while the linked authority identity is
					// still absent. If it reappeared after planning, the next Run
					// must decide whether to update/reopen/recreate the target.
					sourceChanged = found
				} else {
					sourceChanged = !found || current.Revision != item.SourceRevision || current.ContentHash != item.DesiredHash
				}
				if sourceChanged {
					if claim.Run.Kind == service.SyncRunKindHandoff {
						return fmt.Errorf("authority object changed during handoff: %s", item.ObjectKey)
					}
					if err := w.service.SupersedeRepositorySyncRunItem(ctx, row.ID, "authority object changed after planning"); err != nil {
						return err
					}
					continue
				}
			}
			recovered, found := targetStates[reposync.ExecutionKey(item)]
			if !found {
				return fmt.Errorf("recover target %s returned no object state", item.ObjectKey)
			}
			if row.ErrorCode == service.SyncRunItemErrorApplyResultUnknown {
				confirmed := item.Action == reposync.ActionDelete && !recovered.Exists
				if item.Action != reposync.ActionDelete && recovered.Exists && recovered.Hash == item.DesiredHash {
					confirmed = item.Action != reposync.ActionCreate || recovered.Managed || item.ObjectType == canonical.TypeGitRef || item.ObjectType == canonical.TypeWikiRef
				}
				if confirmed {
					if _, err := w.service.CompleteRepositorySyncRunItem(ctx, row.ID, completedItemInput(recovered, claim)); err != nil {
						return err
					}
				} else {
					_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{
						ErrorCode:    service.SyncRunItemErrorApplyResultUnknown,
						ErrorMessage: "target read-back still cannot prove the previous external operation outcome",
						Blocked:      true,
					})
				}
				continue
			}
			if item.Action == reposync.ActionCreate && recovered.Exists {
				if row.State == service.SyncRunItemStatePlanned {
					if (item.ObjectType == canonical.TypeGitRef || item.ObjectType == canonical.TypeWikiRef) && recovered.Hash == item.DesiredHash {
						if _, err := w.service.CompleteRepositorySyncRunItem(ctx, row.ID, completedItemInput(recovered, claim)); err != nil {
							return err
						}
						continue
					}
					_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "target_key_occupied", ErrorMessage: "standby object occupied the create key after planning", Blocked: true})
					continue
				}
				if !recovered.Managed || recovered.Hash != item.DesiredHash {
					_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "create_recovery_mismatch", ErrorMessage: "uncertain create resolved to an object that does not match the durable intent", Blocked: true})
					continue
				}
			}
			if item.Action == reposync.ActionDelete && !recovered.Exists {
				if _, err := w.service.CompleteRepositorySyncRunItem(ctx, row.ID, service.CompleteRepositorySyncRunItemInput{TargetIdentity: item.TargetIdentity, LeaseOwner: claim.LeaseOwner, LeaseGeneration: claim.LeaseGeneration}); err != nil {
					return err
				}
				continue
			}
			if item.Action != reposync.ActionDelete && recovered.Exists && recovered.Hash == item.DesiredHash {
				if _, err := w.service.CompleteRepositorySyncRunItem(ctx, row.ID, completedItemInput(recovered, claim)); err != nil {
					return err
				}
				continue
			}
			if recovered.Exists && item.ExpectedTargetHash != "" && recovered.Hash != item.ExpectedTargetHash {
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "target_changed", ErrorMessage: "standby object changed after planning", Blocked: true})
				continue
			}
			ready = append(ready, item)
		}
		batch.Items = ready
		if len(batch.Items) == 0 {
			continue
		}
		if err := w.checkHeartbeat(heartbeatErr); err != nil {
			return err
		}
		if err := w.service.AssertRepositorySyncLease(ctx, claim); err != nil {
			return err
		}
		for _, item := range batch.Items {
			row := rows[runItemKey(item)]
			if _, err := w.service.BeginRepositorySyncRunItem(ctx, row.ID); err != nil {
				return err
			}
		}
		results, applyErr := target.ApplyBatch(ctx, batch, source)
		if heartbeatFailure := w.checkHeartbeat(heartbeatErr); heartbeatFailure != nil {
			// Do not record or start another side effect after losing the
			// generation-checked Binding lease. A replacement worker recovers
			// every applying RunItem from target read-back.
			return heartbeatFailure
		}
		if err := w.service.AssertRepositorySyncLease(ctx, claim); err != nil {
			return err
		}
		resultByKey := make(map[string]reposync.ApplyResult, len(results))
		for _, result := range results {
			resultByKey[result.ObjectType+"\x00"+result.ObjectKey] = result
		}
		for _, item := range batch.Items {
			row := rows[runItemKey(item)]
			result, ok := resultByKey[reposync.ExecutionKey(item)]
			if !ok {
				message := "provider returned no object result"
				if applyErr != nil {
					message = applyErr.Error()
				}
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "apply_result_missing", ErrorMessage: message, Blocked: true})
				continue
			}
			if result.Unknown {
				// A non-idempotent provider request may have succeeded even when
				// the response was lost. Recover by durable marker/natural key
				// before blocking; never repeat the create blindly.
				recovered, recoverErr := target.Recover(ctx, item)
				if recoverErr == nil && recovered.Exists && recovered.Hash == item.DesiredHash {
					if _, completeErr := w.service.CompleteRepositorySyncRunItem(ctx, row.ID, completedItemInput(recovered, claim)); completeErr != nil {
						return completeErr
					}
					continue
				}
				message := result.ErrorMessage
				if recoverErr != nil {
					message += "; recovery: " + recoverErr.Error()
				}
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: service.SyncRunItemErrorApplyResultUnknown, ErrorMessage: message, Blocked: true})
				continue
			}
			if result.ErrorCode != "" {
				blocked := result.ErrorCode == "permission_denied" || result.ErrorCode == "rate_limited" || result.ErrorCode == "quota_exceeded"
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: result.ErrorCode, ErrorMessage: result.ErrorMessage, Blocked: blocked})
				continue
			}
			if err := w.service.BeginRepositorySyncRunItemVerification(ctx, row.ID); err != nil {
				return err
			}
			state, err := target.Verify(ctx, item, result)
			if err != nil {
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "verify_failed", ErrorMessage: err.Error()})
				continue
			}
			if item.Action == reposync.ActionDelete && state.Exists {
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "delete_not_observed", ErrorMessage: "target object still exists after delete"})
				continue
			}
			if item.Action != reposync.ActionDelete && (!state.Exists || state.Hash != item.DesiredHash) {
				_, _ = w.service.FailRepositorySyncRunItem(ctx, row.ID, service.FailRepositorySyncRunItemInput{ErrorCode: "verify_mismatch", ErrorMessage: "target read-back does not match desired canonical hash"})
				continue
			}
			if _, err := w.service.CompleteRepositorySyncRunItem(ctx, row.ID, completedItemInput(state, claim)); err != nil {
				return err
			}
		}
		if applyErr != nil {
			slog.WarnContext(ctx, "repository sync physical batch partially failed", "run_id", claim.Run.ID, "batch", batch.Key, "error", applyErr)
		}
	}
	if err := w.checkCancellation(ctx, claim.Run.ID); err != nil {
		return err
	}
	if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageVerifying); err != nil {
		return err
	}

	finalSourceDigest, finalTargetDigest := sourceDigest, targetDigest
	// Handoff deliberately retains the stronger quiet-authority proof. Normal
	// eventual-convergence runs avoid a second full repository scan; object
	// revisions are rechecked at each physical batch boundary instead.
	if claim.Run.Kind == service.SyncRunKindHandoff {
		_, finalSourceDigest, err = source.Inventory(ctx)
		if err != nil {
			return err
		}
		_, finalTargetDigest, err = target.Inventory(ctx)
		if err != nil {
			return err
		}
	}
	if targetDigest == "" {
		targetDigest = finalTargetDigest
	}
	if err := w.checkHeartbeat(heartbeatErr); err != nil {
		return err
	}
	if err := w.service.AssertRepositorySyncLease(ctx, claim); err != nil {
		return err
	}
	if err := w.service.UpdateRepositorySyncRunStage(ctx, claim, service.SyncRunStageFinalizing); err != nil {
		return err
	}
	report, counters, attention, blocked, failed, pending, err := w.buildReport(ctx, claim.Run.ID, claim.Run.Phase)
	if err != nil {
		return err
	}
	stable := sourceDigest == finalSourceDigest
	if claim.Run.Kind == service.SyncRunKindHandoff {
		if claim.Run.Phase == service.SyncRunPhaseHandoffReverseDry {
			// During reverse dry-run, target is the old authority. It must still
			// match the digest captured by the final forward sync.
			stable = stable && claim.Run.SourceInventoryDigest != "" && finalTargetDigest == claim.Run.SourceInventoryDigest
		}
		if attention || blocked || failed || pending || !stable {
			return fmt.Errorf("handoff did not converge: attention=%t blocked=%t failed=%t pending=%t authority_stable=%t", attention, blocked, failed, pending, stable)
		}
		_, err = w.service.CompleteRepositorySyncHandoffPhase(ctx, claim.Run.ID, service.CompleteRepositorySyncHandoffPhaseInput{
			ExpectedPhase: claim.Run.Phase, AuthorityStable: stable, SourceInventoryDigest: finalSourceDigest,
			TargetInventoryDigest: finalTargetDigest, CountersJSON: counters, ReportJSON: report,
			LeaseOwner: claim.LeaseOwner, LeaseGeneration: claim.LeaseGeneration,
		})
		return err
	}
	status := service.SyncRunStatusCompleted
	if blocked {
		status = service.SyncRunStatusBlocked
	} else if attention {
		status = service.SyncRunStatusRequiresAttention
	} else if failed {
		status = service.SyncRunStatusFailed
	} else if !stable || pending {
		status = service.SyncRunStatusCompletedWithPendingChanges
	}
	_, err = w.service.CompleteRepositorySyncRun(ctx, claim.Run.ID, service.CompleteRepositorySyncRunInput{
		Status: status, AuthorityStable: stable, AllManagedObjectsVerified: !attention && !blocked && !failed,
		SourceInventoryDigest: finalSourceDigest, TargetInventoryDigest: finalTargetDigest,
		CountersJSON: counters, ReportJSON: report,
		LeaseOwner: claim.LeaseOwner, LeaseGeneration: claim.LeaseGeneration,
	})
	return err
}

func readSourceBatch(ctx context.Context, source reposync.Provider, items []reposync.PlannedItem, sourceProvider, targetProvider string) (map[string]reposync.InventoryObject, error) {
	identities := make([]string, 0, len(items))
	objectType := ""
	for _, item := range items {
		if item.SourceIdentity == "" {
			continue
		}
		objectType = item.ObjectType
		identities = append(identities, item.SourceIdentity)
	}
	if len(identities) == 0 {
		return map[string]reposync.InventoryObject{}, nil
	}
	if reader, ok := source.(reposync.BatchReader); ok {
		objects, err := reader.ReadBatch(ctx, objectType, identities)
		if err != nil {
			return nil, err
		}
		return mapSourceBatch(objects, sourceProvider, targetProvider)
	}
	result := make(map[string]reposync.InventoryObject, len(identities))
	for _, identity := range identities {
		object, err := source.Read(ctx, objectType, identity)
		if errors.Is(err, reposync.ErrObjectNotFound) {
			continue
		}
		if err != nil {
			return nil, err
		}
		result[identity] = object
	}
	return mapSourceBatch(result, sourceProvider, targetProvider)
}

func readInventoryIndex(ctx context.Context, source reposync.Provider) ([]reposync.InventoryObject, string, error) {
	if indexer, ok := source.(reposync.InventoryIndexer); ok {
		return indexer.InventoryIndex(ctx)
	}
	objects, digest, err := source.Inventory(ctx)
	if err != nil {
		return nil, "", err
	}
	for index := range objects {
		objects[index].Payload = nil
		objects[index].PayloadBytes = 0
	}
	return objects, digest, nil
}

func readSourceCandidateDetails(ctx context.Context, source reposync.Provider, candidates []reposync.InventoryObject) ([]reposync.InventoryObject, error) {
	if len(candidates) == 0 {
		return nil, nil
	}
	byType := make(map[string][]string)
	for _, candidate := range candidates {
		byType[candidate.ObjectType] = append(byType[candidate.ObjectType], candidate.Identity)
	}
	resolved := make(map[string]reposync.InventoryObject, len(candidates))
	for objectType, identities := range byType {
		if reader, ok := source.(reposync.BatchReader); ok {
			objects, err := reader.ReadBatch(ctx, objectType, identities)
			if err != nil {
				return nil, err
			}
			for identity, object := range objects {
				resolved[objectType+"\x00"+identity] = object
			}
			continue
		}
		for _, identity := range identities {
			object, err := source.Read(ctx, objectType, identity)
			if errors.Is(err, reposync.ErrObjectNotFound) {
				continue
			}
			if err != nil {
				return nil, err
			}
			resolved[objectType+"\x00"+identity] = object
		}
	}
	details := make([]reposync.InventoryObject, 0, len(candidates))
	for _, candidate := range candidates {
		if object, exists := resolved[candidate.ObjectType+"\x00"+candidate.Identity]; exists {
			object.SourceHash = object.ContentHash
			details = append(details, object)
		}
	}
	return details, nil
}

func readManagedTargetCandidates(ctx context.Context, target reposync.Provider, links []reposync.LinkView) ([]reposync.InventoryObject, error) {
	byType := make(map[string][]string)
	for _, link := range links {
		if link.TargetIdentity != "" {
			byType[link.ObjectType] = append(byType[link.ObjectType], link.TargetIdentity)
		}
	}
	types := make([]string, 0, len(byType))
	for objectType := range byType {
		types = append(types, objectType)
	}
	sort.Strings(types)
	objects := make([]reposync.InventoryObject, 0, len(links))
	for _, objectType := range types {
		identities := byType[objectType]
		if reader, ok := target.(reposync.BatchReader); ok {
			found, err := reader.ReadBatch(ctx, objectType, identities)
			if err != nil {
				return nil, err
			}
			for _, identity := range identities {
				if object, exists := found[identity]; exists {
					objects = append(objects, object)
				}
			}
			continue
		}
		for _, identity := range identities {
			object, err := target.Read(ctx, objectType, identity)
			if errors.Is(err, reposync.ErrObjectNotFound) {
				continue
			}
			if err != nil {
				return nil, err
			}
			objects = append(objects, object)
		}
	}
	return objects, nil
}

func readNewTargetCollisions(ctx context.Context, target reposync.Provider, source []reposync.InventoryObject, links []reposync.LinkView) ([]reposync.InventoryObject, error) {
	linked := make(map[string]bool, len(links))
	pendingLinkIDs := make(map[string]uint)
	for _, link := range links {
		key := link.ObjectType + "\x00" + link.SourceIdentity
		if link.TargetIdentity != "" {
			linked[key] = true
		} else {
			pendingLinkIDs[key] = link.ID
		}
	}
	byType := make(map[string][]reposync.PlannedItem)
	uniqueKeys := make(map[string]string)
	for _, object := range source {
		if linked[object.ObjectType+"\x00"+object.Identity] || object.UniqueKey == "" || object.Representation == "archive_only" {
			continue
		}
		item := reposync.PlannedItem{
			ObjectType: object.ObjectType, ObjectKey: object.ObjectKey, Action: reposync.ActionCreate,
			SourceIdentity: object.Identity, SourceRevision: object.Revision, SourceHash: object.SourceHash,
			DesiredHash: object.ContentHash, PayloadBytes: object.PayloadBytes, Payload: object.Payload,
			Representation: object.Representation,
		}
		byType[object.ObjectType] = append(byType[object.ObjectType], item)
		uniqueKeys[reposync.ExecutionKey(item)] = object.UniqueKey
	}
	result := make([]reposync.InventoryObject, 0)
	for _, items := range byType {
		states, err := recoverTargetBatch(ctx, target, items)
		if err != nil {
			return nil, err
		}
		for _, item := range items {
			state, exists := states[reposync.ExecutionKey(item)]
			if !exists {
				return nil, fmt.Errorf("target collision lookup returned no state for %s", item.ObjectKey)
			}
			if !state.Exists {
				continue
			}
			managedLinkID := uint(0)
			if state.Managed {
				managedLinkID = pendingLinkIDs[item.ObjectType+"\x00"+item.SourceIdentity]
			}
			result = append(result, reposync.InventoryObject{
				ObjectType: item.ObjectType, Identity: state.Identity, ObjectKey: item.ObjectKey,
				UniqueKey: uniqueKeys[reposync.ExecutionKey(item)], Revision: state.Revision, ContentHash: state.Hash,
				ManagedLinkID: managedLinkID,
			})
		}
	}
	return result, nil
}

// detectUnlinkedFastForwardRefs identifies the narrow first-sync case where
// both repositories have the same branch name, but the target branch is just
// behind the authority branch. Ref names alone never establish ownership: Git
// must prove ancestry before the planner is allowed to adopt the target ref.
func detectUnlinkedFastForwardRefs(ctx context.Context, source, target reposync.Provider, sourceInventory, targetInventory []reposync.InventoryObject, links []reposync.LinkView) (map[string]bool, error) {
	result := make(map[string]bool)
	sourceRemote, sourceOK := source.(reposync.GitRemoteSource)
	targetRemote, targetOK := target.(reposync.GitRemoteSource)
	if !sourceOK || !targetOK {
		return result, nil
	}
	linkedSource := make(map[string]bool, len(links))
	for _, link := range links {
		linkedSource[link.ObjectType+"\x00"+link.SourceIdentity] = true
	}
	targetByUnique := make(map[string]reposync.InventoryObject, len(targetInventory))
	for _, object := range targetInventory {
		if object.ObjectType == canonical.TypeGitRef && object.UniqueKey != "" {
			targetByUnique[object.UniqueKey] = object
		}
	}
	candidates := make([]gittransport.RefAncestryCandidate, 0)
	for _, object := range sourceInventory {
		if object.ObjectType != canonical.TypeGitRef || object.UniqueKey == "" || !strings.HasPrefix(object.UniqueKey, "refs/heads/") {
			continue
		}
		if linkedSource[object.ObjectType+"\x00"+object.Identity] {
			continue
		}
		collision, exists := targetByUnique[object.UniqueKey]
		if !exists || collision.ContentHash == object.ContentHash {
			continue
		}
		candidates = append(candidates, gittransport.RefAncestryCandidate{Key: object.UniqueKey, Source: object, Target: collision})
	}
	if len(candidates) == 0 {
		return result, nil
	}
	sourceGit, err := sourceRemote.GitRemote(ctx, false)
	if err != nil {
		return nil, err
	}
	targetGit, err := targetRemote.GitRemote(ctx, false)
	if err != nil {
		return nil, err
	}
	return gittransport.FastForwardableRefs(ctx, sourceGit, targetGit, candidates)
}

// unlinkedRefCollisions returns only branch names that an administrator
// explicitly opted to replace. The normal git apply path still supplies the
// compared target SHA as --force-with-lease, so a concurrent target push wins
// instead of being overwritten.
func unlinkedRefCollisions(sourceInventory, targetInventory []reposync.InventoryObject, links []reposync.LinkView) map[string]bool {
	result := make(map[string]bool)
	linkedSource := make(map[string]bool, len(links))
	for _, link := range links {
		linkedSource[link.ObjectType+"\x00"+link.SourceIdentity] = true
	}
	targetByUnique := make(map[string]reposync.InventoryObject, len(targetInventory))
	for _, object := range targetInventory {
		if object.ObjectType == canonical.TypeGitRef && strings.HasPrefix(object.UniqueKey, "refs/heads/") {
			targetByUnique[object.UniqueKey] = object
		}
	}
	for _, object := range sourceInventory {
		if object.ObjectType != canonical.TypeGitRef || !strings.HasPrefix(object.UniqueKey, "refs/heads/") || linkedSource[object.ObjectType+"\x00"+object.Identity] {
			continue
		}
		if target, exists := targetByUnique[object.UniqueKey]; exists && target.ContentHash != object.ContentHash {
			result[object.UniqueKey] = true
		}
	}
	return result
}

func appendDistinctInventory(base []reposync.InventoryObject, additions ...reposync.InventoryObject) []reposync.InventoryObject {
	seen := make(map[string]bool, len(base)+len(additions))
	for _, object := range base {
		seen[object.ObjectType+"\x00"+object.Identity] = true
	}
	for _, object := range additions {
		key := object.ObjectType + "\x00" + object.Identity
		if !seen[key] {
			base = append(base, object)
			seen[key] = true
		}
	}
	return base
}

func inventoryDigest(objects []reposync.InventoryObject) string {
	ordered := append([]reposync.InventoryObject(nil), objects...)
	sort.Slice(ordered, func(i, j int) bool {
		if ordered[i].ObjectType != ordered[j].ObjectType {
			return ordered[i].ObjectType < ordered[j].ObjectType
		}
		return ordered[i].Identity < ordered[j].Identity
	})
	hash := sha256.New()
	for _, object := range ordered {
		_, _ = fmt.Fprintf(hash, "%s\x00%s\x00%s\x00", object.ObjectType, object.Identity, object.ContentHash)
	}
	return hex.EncodeToString(hash.Sum(nil))
}

func mapSourceBatch(objects map[string]reposync.InventoryObject, sourceProvider, targetProvider string) (map[string]reposync.InventoryObject, error) {
	result := make(map[string]reposync.InventoryObject, len(objects))
	for identity, object := range objects {
		mapped, err := canonical.MapObjectForTarget(object, sourceProvider, targetProvider)
		if err != nil {
			return nil, err
		}
		result[identity] = mapped
	}
	return result, nil
}

func recoverTargetBatch(ctx context.Context, target reposync.Provider, items []reposync.PlannedItem) (map[string]reposync.TargetState, error) {
	if recoverer, ok := target.(reposync.BatchRecoverer); ok {
		return recoverer.RecoverBatch(ctx, items)
	}
	result := make(map[string]reposync.TargetState, len(items))
	for _, item := range items {
		state, err := target.Recover(ctx, item)
		if err != nil {
			return nil, err
		}
		result[reposync.ExecutionKey(item)] = state
	}
	return result, nil
}

func (w *Worker) heartbeat(ctx context.Context, claim service.RepositorySyncClaim, result chan<- error, cancel context.CancelCauseFunc, ready chan<- struct{}) {
	ticker := time.NewTicker(leaseHeartbeatInterval(w.options.LeaseTTL))
	defer ticker.Stop()
	close(ready)
	leaseExpiresAt := time.Now().UTC().Add(w.options.LeaseTTL)
	if claim.Binding.LeaseExpiresAt != nil {
		leaseExpiresAt = claim.Binding.LeaseExpiresAt.UTC()
	}
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			renewedUntil, err := w.renewLease(ctx, claim, leaseExpiresAt)
			if err != nil {
				select {
				case result <- err:
				default:
				}
				cancel(err)
				return
			}
			leaseExpiresAt = renewedUntil
		}
	}
}

func leaseHeartbeatInterval(leaseTTL time.Duration) time.Duration {
	interval := leaseTTL / 6
	if interval < minLeaseHeartbeatInterval {
		return minLeaseHeartbeatInterval
	}
	if interval > maxLeaseHeartbeatInterval {
		return maxLeaseHeartbeatInterval
	}
	return interval
}

func leaseRenewalTimeout(leaseTTL time.Duration) time.Duration {
	timeout := leaseTTL / 2
	if timeout < minLeaseRenewalTimeout {
		return minLeaseRenewalTimeout
	}
	if timeout > maxLeaseRenewalTimeout {
		return maxLeaseRenewalTimeout
	}
	return timeout
}

func (w *Worker) renewLease(ctx context.Context, claim service.RepositorySyncClaim, leaseExpiresAt time.Time) (time.Time, error) {
	var lastErr error
	for attempt := 0; attempt < leaseRenewalAttempts; attempt++ {
		remaining := leaseRenewalBudget(time.Now(), leaseExpiresAt)
		if remaining <= 0 {
			if lastErr != nil {
				return time.Time{}, lastErr
			}
			return time.Time{}, fmt.Errorf("%w: repository sync lease renewal reached its expiry safety boundary", service.ErrConflict)
		}
		timeout := leaseRenewalTimeout(w.options.LeaseTTL)
		if remaining < timeout {
			timeout = remaining
		}
		renewCtx, cancel := context.WithTimeout(ctx, timeout)
		renewedUntil, err := w.service.RenewRepositorySyncLeaseUntil(renewCtx, claim.Binding.ID, claim.LeaseOwner, claim.LeaseGeneration, w.options.LeaseTTL)
		cancel()
		if err == nil {
			return renewedUntil, nil
		}
		lastErr = err
		if !retryableLeaseRenewalError(err) || ctx.Err() != nil || attempt == leaseRenewalAttempts-1 {
			return time.Time{}, err
		}
		slog.WarnContext(ctx, "repository sync lease renewal retrying",
			"binding_id", claim.Binding.ID,
			"run_id", claim.Run.ID,
			"attempt", attempt+1,
			"error", err,
		)
		backoff := time.Second << attempt
		select {
		case <-ctx.Done():
			return time.Time{}, context.Cause(ctx)
		case <-time.After(backoff):
		}
	}
	return time.Time{}, lastErr
}

func leaseRenewalBudget(now, leaseExpiresAt time.Time) time.Duration {
	return leaseExpiresAt.Sub(now) - leaseExpirySafety
}

func retryableLeaseRenewalError(err error) bool {
	if errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	// database/sql can combine a context timeout with its transaction-cleanup
	// error as an opaque string ("context deadline exceeded; sql: transaction
	// has already been committed or rolled back"). It no longer unwraps to
	// context.DeadlineExceeded, but it is still safe to retry the fenced lease
	// update while the parent Run context remains active.
	if strings.Contains(strings.ToLower(err.Error()), "context deadline exceeded") {
		return true
	}
	var mysqlErr *driversql.MySQLError
	return errors.As(err, &mysqlErr) && (mysqlErr.Number == 1205 || mysqlErr.Number == 1213)
}

func (w *Worker) checkHeartbeat(result <-chan error) error {
	select {
	case err := <-result:
		return err
	default:
		return nil
	}
}

func (w *Worker) checkCancellation(ctx context.Context, runID uint) error {
	cancelled, err := w.service.RepositorySyncRunCancellationRequested(ctx, runID)
	if err != nil {
		return err
	}
	if cancelled {
		return errRepositorySyncCancelled
	}
	return nil
}

func (w *Worker) buildReport(ctx context.Context, runID uint, phase string) (db.LargeText, db.LargeText, bool, bool, bool, bool, error) {
	var run db.SyncRun
	if err := w.service.DBForCtx(ctx).First(&run, "id = ?", runID).Error; err != nil {
		return "", "", false, false, false, false, err
	}
	var rows []db.SyncRunItem
	if err := w.service.DBForCtx(ctx).Where("run_id = ? AND phase = ?", runID, phase).Order("id ASC").Find(&rows).Error; err != nil {
		return "", "", false, false, false, false, err
	}
	counters := map[string]int{
		"scanned": len(rows), "checkpoint_skipped": 0, "created": 0, "updated": 0, "deleted": 0, "noop": 0,
		"preserved_unmanaged": 0, "attributed": 0, "projected": 0, "archive_only": 0,
		"conflicts": 0, "unsupported": 0, "failed": 0, "pending_changes": 0, "reconciled_labels": 0,
	}
	byObjectType := map[string]map[string]int{}
	var plannedCounters map[string]int
	if json.Unmarshal([]byte(run.CountersJSON), &plannedCounters) == nil {
		counters["scanned"] = plannedCounters["scanned"]
		counters["checkpoint_skipped"] = plannedCounters["checkpoint_skipped"]
	}
	problems := make([]map[string]any, 0)
	attention, blocked, failed, pending := false, false, false, false
	for _, row := range rows {
		typeCounters := byObjectType[row.ObjectType]
		if typeCounters == nil {
			typeCounters = map[string]int{"created": 0, "updated": 0, "deleted": 0, "unchanged": 0, "conflicts": 0, "blocked": 0, "failed": 0}
			byObjectType[row.ObjectType] = typeCounters
		}
		switch row.Action {
		case reposync.ActionCreate:
			if row.State == service.SyncRunItemStateCompleted {
				counters["created"]++
				typeCounters["created"]++
			}
		case reposync.ActionUpdate:
			if row.State == service.SyncRunItemStateCompleted {
				counters["updated"]++
				typeCounters["updated"]++
			}
		case reposync.ActionDelete:
			if row.State == service.SyncRunItemStateCompleted {
				counters["deleted"]++
				typeCounters["deleted"]++
			}
		case reposync.ActionNoop:
			counters["noop"]++
			typeCounters["unchanged"]++
		case reposync.ActionConflict:
			counters["conflicts"]++
			typeCounters["conflicts"]++
			attention = true
		}
		var intent reposync.PlannedItem
		_ = json.Unmarshal([]byte(row.IntentJSON), &intent)
		if intent.Reason == reposync.ReasonAdoptUnlinkedLabel {
			counters["reconciled_labels"]++
		}
		if intent.Representation != "" && intent.Representation != "native" {
			counters[intent.Representation]++
		}
		if row.Action == reposync.ActionReport && strings.Contains(intent.Reason, "preserve unmanaged") {
			counters["preserved_unmanaged"]++
		}
		switch row.State {
		case service.SyncRunItemStateBlocked:
			if row.Action == reposync.ActionConflict {
				attention = true
			} else {
				counters["blocked"]++
				typeCounters["blocked"]++
				blocked = true
			}
		case service.SyncRunItemStateFailed:
			failed = true
			counters["failed"]++
			typeCounters["failed"]++
		case service.SyncRunItemStateSuperseded:
			pending = true
			counters["pending_changes"]++
		}
		if row.State == service.SyncRunItemStateBlocked || row.State == service.SyncRunItemStateFailed {
			reason := string(row.ErrorMessage)
			if strings.TrimSpace(reason) == "" {
				reason = intent.Reason
			}
			problemType, suggestedAction, retryable := syncProblemPresentation(row.ErrorCode, row.Action)
			problems = append(problems, map[string]any{
				"object_type": row.ObjectType, "object_key": row.ObjectKey, "action": row.Action, "state": row.State,
				"source_identity": intent.SourceIdentity, "target_identity": intent.TargetIdentity,
				"error_code": row.ErrorCode, "problem_type": problemType, "reason": reason,
				"suggested_action": suggestedAction, "retryable": retryable,
			})
		}
	}
	direction := "github_to_ags"
	if run.SourceSide == service.SyncSideA {
		direction = "ags_to_github"
	}
	reportJSON, err := json.Marshal(map[string]any{
		"mode": run.Mode, "direction": direction, "audit_mode": run.AuditMode,
		"changes": map[string]int{
			"created": counters["created"], "updated": counters["updated"], "deleted": counters["deleted"],
			"unchanged": counters["noop"], "checkpoint_skipped": counters["checkpoint_skipped"],
		},
		"data_gaps": map[string]int{
			"represented_differently": counters["attributed"] + counters["projected"],
			"not_written_to_target":   counters["archive_only"],
		},
		"attention": map[string]int{
			"conflicts": counters["conflicts"], "blocked": counters["blocked"], "failed": counters["failed"],
		},
		"reconciled":     map[string]int{"labels": counters["reconciled_labels"]},
		"by_object_type": byObjectType,
		"counters":       counters, "problems": problems,
	})
	if err != nil {
		return "", "", false, false, false, false, err
	}
	countersJSON, err := json.Marshal(counters)
	if err != nil {
		return "", "", false, false, false, false, err
	}
	return db.LargeText(reportJSON), db.LargeText(countersJSON), attention, blocked, failed, pending, nil
}

func syncProblemPresentation(errorCode, action string) (string, string, bool) {
	switch strings.TrimSpace(errorCode) {
	case "permission_denied":
		return "authorization_required", "grant_repository_access", true
	case "rate_limited":
		return "temporarily_rate_limited", "retry_later", true
	case "quota_exceeded":
		return "quota_exceeded", "review_github_quota", true
	case service.SyncRunItemErrorApplyResultUnknown:
		return "external_result_unknown", "verify_result", true
	case "provider_conflict", "target_changed", "target_key_occupied", "create_recovery_mismatch":
		return "repository_conflict", "review_conflict", false
	case "verify_failed", "verify_mismatch", "delete_not_observed":
		return "verification_failed", "retry_sync", true
	}
	if action == reposync.ActionConflict {
		return "repository_conflict", "review_conflict", false
	}
	return "sync_failed", "view_details", true
}

func endpointForSide(endpoints []db.SyncBindingEndpoint, side string) (reposync.Endpoint, error) {
	for _, endpoint := range endpoints {
		if endpoint.Side != side {
			continue
		}
		result := reposync.Endpoint{Side: endpoint.Side, Provider: endpoint.Provider, RepositoryID: endpoint.RepositoryID, RepositoryFullName: endpoint.RepositoryFullName, ConnectionID: endpoint.ConnectionID}
		if endpoint.AGSRepositoryID != nil {
			result.AGSRepositoryID = *endpoint.AGSRepositoryID
		}
		return result, nil
	}
	return reposync.Endpoint{}, fmt.Errorf("sync binding has no endpoint for side %q", side)
}

func runItemKey(item reposync.PlannedItem) string {
	return fmt.Sprintf("%d\x00%s\x00%s\x00%s", item.DependencyPhase, item.ObjectType, item.Action, item.ObjectKey)
}

func completedItemInput(state reposync.TargetState, claim service.RepositorySyncClaim) service.CompleteRepositorySyncRunItemInput {
	return service.CompleteRepositorySyncRunItemInput{
		TargetIdentity: state.Identity, TargetRevision: state.Revision, TargetHash: state.Hash,
		TargetNumber: state.Number, TargetURL: state.URL,
		LeaseOwner: claim.LeaseOwner, LeaseGeneration: claim.LeaseGeneration,
	}
}

func terminalStatus(err error) string {
	if errors.Is(err, errRepositorySyncCancelled) {
		return service.SyncRunStatusCancelled
	}
	if errors.Is(err, reposync.ErrResultUnknown) || errors.Is(err, reposync.ErrPermission) || errors.Is(err, reposync.ErrRateLimited) || errors.Is(err, reposync.ErrQuotaExceeded) {
		return service.SyncRunStatusBlocked
	}
	return service.SyncRunStatusFailed
}

func terminalErrorCode(err error) string {
	switch {
	case errors.Is(err, errRepositorySyncCancelled):
		return "cancelled"
	case errors.Is(err, reposync.ErrResultUnknown):
		return service.SyncRunItemErrorApplyResultUnknown
	case errors.Is(err, reposync.ErrPermission):
		return "permission_denied"
	case errors.Is(err, reposync.ErrRateLimited):
		return "rate_limited"
	case errors.Is(err, reposync.ErrQuotaExceeded):
		return "quota_exceeded"
	default:
		return "sync_failed"
	}
}