// SPDX-License-Identifier: Apache-2.0

package git

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	gogit "github.com/go-git/go-git/v6"
	"github.com/go-git/go-git/v6/plumbing"
	gitclient "github.com/go-git/go-git/v6/plumbing/client"
	"github.com/go-logr/logr"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3"
	"github.com/ConfigButler/gitops-reverser/internal/sanitize"
	"github.com/ConfigButler/gitops-reverser/internal/telemetry"
	itypes "github.com/ConfigButler/gitops-reverser/internal/types"
	"github.com/ConfigButler/gitops-reverser/internal/typeset"
)

const (
	// DefaultCommitWindow is the default idle timeout that closes a commit window, used when
	// GitTarget.spec.commit.window.idleTimeout is unset or the target cannot be read.
	DefaultCommitWindow = 5 * time.Second

	// DefaultCommitWindowMaxDuration is the default maximum duration of a commit window, used when
	// GitTarget.spec.commit.window.maxDuration is unset or the target cannot be read. It keeps
	// continuous activity from postponing a commit until the memory limit forces one.
	DefaultCommitWindowMaxDuration = time.Minute

	// PushCooldown is the minimum interval between successful pushes. The cooldown
	// is intentionally fixed: commit cadence is a user concern (spec.commit.window on
	// the CRD); push cadence is an implementation/politeness concern.
	PushCooldown = 5 * time.Second
)

var (
	// These function vars are narrow test seams for classifying push failures
	// without duplicating the repository layer in tests.
	//nolint:gochecknoglobals
	pushAtomicFn = PushAtomic
	//nolint:gochecknoglobals
	fetchRemoteBranchHashFn = fetchRemoteBranchHash
	//nolint:gochecknoglobals
	syncToRemoteFn = syncToRemoteFrom
	// beforePushAdmission runs right before a push cycle compares its root's parent generation
	// with the current one. Nil in production; a test changes the parent here to land a change
	// just before admission.
	//nolint:gochecknoglobals
	beforePushAdmission func()
	//nolint:gochecknoglobals
	advertiseRemoteBranchFn = advertiseRemoteBranch
)

// BranchWorker processes events for a single repository branch, for the GitBranch holding it.
// It can serve multiple GitTargets that write to different paths in the same branch.
// This design ensures serialized commits per branch, preventing merge conflicts.
type BranchWorker struct {
	// Identity (immutable after creation). GitBranch is the claim holder this worker writes for, and
	// GitBranchUID its metadata.uid, which names the checkout; GitProviderRef and
	// GitProviderNamespace name that GitBranch's GitProvider, whose credential and signing key it
	// uses.
	GitBranch            types.NamespacedName
	GitBranchUID         types.UID
	GitProviderRef       string
	GitProviderNamespace string
	Branch               string

	// repo is the repository this worker is ABOUT: its clone, its base trust, its observations
	// and its retained writes all describe that one repository and no other.
	//
	// It is immutable for the same reason the three fields above are. The worker does not follow
	// its GitProvider to a new repository — meeting one is not a state change to be undone field
	// by field, it is a different subject — so the WorkerManager replaces the worker instead.
	// See RepoIdentity and WorkerManager.EnsureWorker.
	repo RepoIdentity

	// Dependencies
	Client        client.Client
	Log           logr.Logger
	contentWriter *contentWriter
	// mapper resolves manifest GVKs into resource identities while building the
	// GitTarget inventory, for the LOCAL cluster and as the fallback when no source
	// cluster is named. A nil mapper keeps the writer structure-only, so object-less
	// deletes have no resource index to target.
	mapper typeset.Lookup

	// clusterMapper resolves the GVK->GVR lookup for a NAMED source cluster: a folder that
	// mirrors a remote resolves its documents against that remote's type registry, never a
	// union of all clusters (two clusters can serve one GVK under different GVRs/scopes). Nil
	// in the CLI and in tests, which fall back to `mapper`; wired at startup by the
	// WorkerManager from the watch manager's ClusterTypeLookup.
	clusterMapper func(clusterID string) typeset.Lookup

	// sshHostKeys configures SSH host-key resolution (install-level default ConfigMap and the
	// dev-only missing-key opt-out) for this worker's credential reads. Set by the WorkerManager
	// before Start, on the same goroutine the event loop reads it from.
	sshHostKeys SSHHostKeyConfig
	// credentialPolicy controls explicit insecure opt-ins for Git credential transports.
	credentialPolicy CredentialTransportPolicy

	// pathRefusal surfaces a refused write plan as GitTarget GitPathAccepted=False. The
	// live-event paths have no result channel to carry the refusal back, so without it a
	// refused live write would abort the commit and leave the GitTarget looking healthy. Set
	// by the WorkerManager before Start; a nil reporter only drops the status transition.
	pathRefusal PathRefusalReporter

	// renderFidelityGate closes normal live writes while a target's current render-vs-live
	// epoch is pending or divergent. Resync remains allowed so it can measure and repair Git.
	// Set by WorkerManager before Start, alongside pathRefusal.
	renderFidelityGate *RenderFidelityGate

	// scanAcceptance publishes what a read-only scan of the folder found, so a refusal that
	// arrived in Git is visible before anything tries to write. Set by the WorkerManager before
	// Start, alongside pathRefusal; a nil reporter only drops the projection.
	scanAcceptance ScanAcceptanceReporter

	// layoutReporter publishes what each scan resolved about a target's folder, projected as
	// status.placement and the LayoutResolved condition. Set by the WorkerManager before Start,
	// alongside pathRefusal; a nil reporter only drops the projection.
	layoutReporter LayoutReporter

	// remoteReporter delivers each confirmed observation of the branch's remote state to the
	// worker manager, which keeps it for every GitTarget on the branch. Set by the WorkerManager
	// before Start, alongside pathRefusal; nil in the CLI and in tests, where an observation is
	// simply not shared.
	remoteReporter RemoteReporter

	// applyRequester delivers the apply requests this worker decides (applier_claim.go), and inbox
	// holds the reasons pending for the branch's appliers from what was observed about the remote,
	// and the reports of attempts that came back. Set by the
	// WorkerManager before Start; a nil requester switches the applier claim off (the CLI, and tests
	// that do not exercise it).
	applyRequester ApplyRequester
	// claimStatus writes each applier's operation to its applier claim's status; nil writes nothing.
	claimStatus ApplierClaimStatusWriter
	inbox       applyInbox
	// applyWake wakes the loop to service the appliers without queueing a work item: an attempt
	// reported back.
	applyWake chan struct{}
	// unresolvedOperations is how many of this branch's apply operations are Unresolved, as the
	// loop last published it, for the apply_operations_unresolved gauge.
	unresolvedOperations atomic.Int64
	// heldAppliers is how many of this branch's appliers the barrier holds a reason back for, as the
	// loop last published it, for the apply_appliers_held gauge (barrier.go).
	heldAppliers atomic.Int64
	// applyEpoch, appliersAhead and hasAppliers are the loop's ApplyState for the watch side, which
	// reads it when a snapshot's collection starts (apply_epoch.go).
	applyEpoch    atomic.Uint64
	appliersAhead atomic.Bool
	hasAppliers   atomic.Bool
	// liveEdits counts the live events the loop has mirrored, for the requester's goroutine to read: a
	// converge that only catches the applier's state up yields to an edit that arrives before its
	// resume (convergeHolds).
	liveEdits atomic.Uint64
	// initialSnapshots reports whether every collection a GitTarget watches has a settled
	// snapshot, and resyncRecheck asks the watch side for fresh snapshots of a GitTarget. Set by the
	// WorkerManager before Start; nil in the CLI and in tests (apply_epoch.go).
	initialSnapshots func(target itypes.ResourceReference) bool
	// applierAdmitted leaves out of the branch's appliers a GitTarget the dispatcher would refuse
	// (WorkerManager.applierAdmitted). Nil, in the CLI and in tests that build a worker by hand,
	// admits every GitTarget.
	applierAdmitted ApplierAdmission
	resyncRecheck   func(target itypes.ResourceReference)
	// applyBarrier lets an apply window wait for the watch after its operation completed
	// (apply_barrier.go). Set by the WorkerManager; nil in the CLI and in tests, where a window closes
	// after its grace alone.
	applyBarrier ApplyBarrierFunc
	// replacementPending reports whether this worker replaced one whose GitTargets have not been told
	// yet (WorkerManager.ReplacementPending): until they are, their snapshot record is the old
	// worker's, and no applier is asked for anything. Nil in the CLI and in tests.
	replacementPending func() bool

	// Event processing
	eventQueue chan WorkItem
	ctx        context.Context
	cancelFunc context.CancelFunc
	wg         sync.WaitGroup
	started    bool
	mu         sync.Mutex

	// stoppingState closes the queue to new work from the moment Stop is called.
	//
	// Shutdown drains whatever is still on the FIFO without processing it, so an item accepted
	// after that point is silently thrown away, while Enqueue's "true" has already told a watch
	// loop it may advance its durable cursor past the event. That cost nothing while Stop was
	// only ever called at process exit. A worker REPLACED because its GitProvider names a
	// different repository is stopped with the process still running, and the sibling GitTargets
	// on its branch go on holding streams that point at it until their own next reconcile, which
	// is minutes of accepted-and-discarded events.
	//
	// Refusing them turns a silent loss into the drop path every producer already handles: the
	// watch leaves its cursor where it is and redelivers to the replacement. Items already ON the
	// queue when Stop is called are still drained, which is the behaviour shutdown always had.
	//
	// It is written and read under pendingResyncsMu, the lock every enqueue already holds across
	// its send. Checking it outside that lock is not enough: an enqueue could pass the check,
	// block on the lock, and send after Stop had finished draining, into a queue with no loop left
	// to read it — and still answer "accepted".
	stoppingState bool
	// exitedState is set once the event loop has returned, for whatever reason, and the worker
	// has settled every CommitRequest it knew. From then on nothing can act on a request, so a
	// withdraw is answered at once instead of being queued for a loop that is gone. Guarded by
	// pendingResyncsMu, like stoppingState.
	exitedState bool

	// lastObservation is what this worker last PROVED about the target branch on the remote, and
	// when. See RemoteObservation: it replaces a fetch-only trio (branchExists, lastCommitSHA,
	// lastFetchTime) that went arbitrarily stale on exactly the busiest targets, because a push
	// never wrote to it.
	lastObservation atomic.Pointer[RemoteObservation]

	// repoMu serializes repository/worktree operations within this worker.
	repoMu sync.Mutex

	// baseTrustedState carries the invariant "the checkout's root sits at the remote tip of the
	// target branch, or the worker knows it does not". It is read through baseTrusted() and written
	// through setBaseTrusted and invalidateBase.
	//
	// It is an atomic rather than a repoMu-protected field because the loss points are spread
	// across every error return on the write path, and a lock there would put a second ordering
	// constraint on code whose only job is to record that something went wrong.
	//
	// See the package's design note in docs/design/push-notification-and-reconcile-trigger.md, Part 1.
	baseTrustedState atomic.Bool

	// checkoutApplied is how many of the loop's retained writes the checkout holds: the checkout is
	// the projection of the pending writes, its root plus one commit per write in
	// pendingWrites[:applied].
	// checkoutUnknown means a write failed part-way and the worktree may hold anything.
	//
	// It is the one answer to "may the loop commit on, or push, what the checkout holds?", which
	// is why it is a count and not a flag. A reset to the remote tip sets it to zero, which with
	// writes retained says their commits are gone and must be replayed: pushing then would find the
	// branch already at the remote tip, report success without sending anything, and settle work
	// that exists nowhere. A commit raises it, a replay sets it to every pending write, and a push
	// that publishes them returns it to zero. A failed write sets checkoutUnknown, and only a reset
	// clears that: a successful push says where the remote is, and nothing about leftovers in the
	// worktree.
	//
	// See docs/design/gittarget-branch-worker-pending-writes.md.
	checkoutApplied atomic.Int64

	// lastReapply records when this worker last decided a Refusal request for a GitTarget's applier, keyed
	// by "namespace/name", so refusalReapplyInterval can floor the rate. It is guarded by its own
	// mutex rather than repoMu: the decision is taken before any repository work starts, and
	// borrowing the repository lock for it would order the two for no reason.
	refusalMu   sync.Mutex
	lastReapply map[string]time.Time
	// coveredRefusal records, per GitTarget AND WATCHED COLLECTION, the refusal observation the last
	// reapply request was decided for: a digest of the objects that were refused together with the
	// issues raised about them. A recheck that observes exactly the same thing is already covered
	// by that request, so it is not another applier claim. It is dropped when a resync for that collection is
	// ACCEPTED — see refusalRecovered, which is what makes a live edit re-made after a revert
	// count as new, and what keeps one watched type's success from speaking for another's.
	// Guarded by refusalMu with the rate limit, because the two are read together.
	coveredRefusal map[refusalKey]string

	// branchBufferMaxBytes caps the retained in-memory event data; tripped on
	// event arrival, an immediate finalize bypasses the commit window.
	branchBufferMaxBytes int64

	// pushCycleRootHash is the hash the target branch was at on the remote
	// before the first commit of the current push cycle. pushCycleRootBranch is
	// the remote branch/ref that hash came from. For an existing worker branch
	// this is usually the same branch; for a brand new branch it is the default
	// branch we based the new branch on.
	//
	// PushAtomic uses this pair to detect concurrent updates: if the remote has
	// moved since, we hit the conflict path and replay from retained pending
	// writes.
	pushCycleRootBranch plumbing.ReferenceName
	// Cleared after a successful push. Protected by repoMu.
	pushCycleRootHash plumbing.Hash

	// newBranchParent and newBranchParentHash are the parent the local write branch was created
	// from, while the remote does not carry the write branch. They let a cycle that finds HEAD
	// already on the local write branch still root itself on the parent: a no-op cycle (a resync
	// or a live write with nothing to commit) creates the local branch and leaves HEAD on it, and
	// local HEAD alone cannot say that the branch is absent on the remote.
	//
	// Cleared as soon as the remote is known to carry the branch: a fetch that found it, or a push
	// that created or confirmed it. Protected by repoMu.
	newBranchParent     plumbing.ReferenceName
	newBranchParentHash plumbing.Hash

	// parentState is the GitTarget's spec.parentBranch, with a generation that SetParentBranch
	// raises on every change. It is set by the reconcile and read by the event loop, hence atomic.
	// See parentConfig.
	parentState atomic.Pointer[parentConfig]
	// baseParentGen is the generation of the parent configuration the checkout was last proven
	// against: the snapshot whose fetch produced it, or the current one once the checkout is known
	// not to depend on the parent (the write branch exists). Trust is scoped to it, so a push or a
	// fetch that finishes after a change can never vouch for the new configuration.
	baseParentGen atomic.Uint64
	// pushCycleRootGen is the parent generation pushCycleRootBranch was chosen under. The push
	// admission check compares it with the current one. Protected by repoMu.
	pushCycleRootGen uint64

	// firsts surfaces the first successful commit and push at default verbosity.
	firsts branchWorkerLogFirsts

	// retentionLoggedAt throttles the default-verbosity "prune policy retained documents"
	// line to one per GitTarget subtree per retentionLogInterval. Retaining is a steady
	// state, so the log must not scale with the reconcile rate. Owned by the event loop
	// goroutine, like firsts, so it carries no lock.
	retentionLoggedAt map[string]time.Time

	// hasUnpushedWork mirrors whether the event loop is currently holding a live
	// open window, any committed-but-not-yet-pushed pending writes, or live events
	// held at a pending render-fidelity gate (render_fidelity_hold.go). The event
	// loop is the only writer and, via syncUnpushedWorkFlag, the only reader; the
	// atomic guards the store/load and lets the depth gauge account for retained
	// work the channel length alone cannot see.
	hasUnpushedWork atomic.Bool

	// pushNotices folds the push notices reported while one is already queued into it. See
	// EnqueuePushNotification.
	pushNotices pendingPushNotice
	// lastPushNoticeAt is when a Git host last reported a push to this branch, in Unix nanoseconds,
	// zero before the first. See LastPushNotice.
	lastPushNoticeAt atomic.Int64

	// inflightItems counts work items accepted onto eventQueue but not yet fully
	// handled. It is incremented before the channel send (so it can never lag the
	// loop's receive) and decremented only after handleQueueItem returns. Using
	// len(eventQueue) instead would read 0 in the window between the loop
	// receiving an item and finishing it — a scrape there could satisfy a drain
	// gate while a commit/push is still in flight. The depth gauge sums this with
	// hasUnpushedWork, so it reads 0 only once every accepted item has been
	// handled and nothing is retained.
	inflightItems atomic.Int64

	// pendingResyncs coalesces queued resyncs by (GitTarget, scope). A resync is
	// state-based and idempotent — "make Git match this desired set for this
	// scope" — so a newer one wholly supersedes an older one still waiting. The
	// map holds the open marker per key, and the marker the current request; the
	// FIFO holds the marker, which is what preserves ordering against live events. Queue depth is
	// therefore bounded by the number of distinct scopes, not by the request
	// rate, so a storm of resyncs for one target can no longer fill the queue
	// and starve every other GitTarget on this branch.
	//
	// Coalescing reuses the marker's FIFO POSITION, so it is sound only while nothing for that
	// scope was queued behind it: running a newer snapshot at the older position would apply it
	// before writes it already contains, which would then overwrite it. The entry's tail flag
	// records that boundary.
	pendingResyncsMu sync.Mutex
	pendingResyncs   map[resyncKey]*resyncMarker
	// queuedResyncs counts, per GitTarget, the resync markers on the FIFO the loop has not taken.
	// Guarded by pendingResyncsMu. See queuedResyncsFor.
	queuedResyncs map[string]int

	// crOutcomes holds resolved CommitRequest outcomes for the controller to poll
	// via LookupCommitRequestOutcome. The event loop is the only writer (on its
	// goroutine), the controller the only reader (on a reconcile goroutine), so the
	// mutex guards the cross-goroutine handoff. Entries are GC'd by age. The
	// accessors live in commit_request_attach_loop.go alongside the attach logic.
	crOutcomesMu sync.Mutex
	crOutcomes   map[commitRequestID]commitRequestOutcomeEntry
	// crPhases holds where each unresolved CommitRequest stands, for the controller to report
	// while it waits. Guarded by crOutcomesMu; the event loop is the only writer.
	crPhases map[commitRequestID]CommitRequestPhase
	// crApplies holds where the apply each CommitRequest asked for stands (save_apply.go). Guarded by
	// crOutcomesMu; the event loop is the only writer.
	crApplies map[commitRequestID]saveApplyEntry
	// crQueuedAttaches counts, per request, the attaches on the FIFO that the loop has not handled
	// yet. An outcome is not collected while any are queued: the outcome is what turns a late attach
	// into a no-op, so collecting it first would let that attach register a withdrawn request
	// again, however long the outcome had been kept. Guarded by crOutcomesMu.
	crQueuedAttaches map[commitRequestID]int
	// crOwners is the manager's table of which worker holds each request. See commitRequestOwners.
	// Set by the WorkerManager before Start; nil for a worker built on its own.
	crOwners *commitRequestOwners

	// clock is the time source of the retry schedule and of the commit windows' timers: the idle and
	// max-duration deadlines and a CommitRequest's attach deadline. nil is time.Now. See now().
	clock func() time.Time
	// parentRecoveryOpen, parentRecoveryFound and parentProbeHold mirror the event loop's
	// recovery latch for readers outside it. See parent_recovery.go.
	parentRecoveryOpen  atomic.Bool
	parentRecoveryFound atomic.Bool
	parentProbeHold     atomic.Pointer[parentProbeHold]
	// intake decides whether new writes, saves and resyncs may enter the FIFO. See intake.go.
	intake intakeGate
	// publicationReporter is told when the worker's publication report changes. Set by the
	// WorkerManager before Start; nil in tests that do not assert on it. See publication.go.
	publicationReporter func()
	// commitRequestReporter is told when a CommitRequest's standing on this worker changes. Set by
	// the WorkerManager before Start; nil in tests that do not assert on it. See
	// commit_request_changes.go.
	commitRequestReporter func(CommitRequestChange)
	// publication is the loop's latest publication report, and backlog what it keeps, for readers
	// outside the loop. See publication.go.
	publication atomic.Pointer[PublicationStatus]
	backlog     publicationBacklog
}

// branchWorkerLogFirsts logs the first successful commit and push of a worker's
// lifetime at default verbosity, so an operator can confirm the git write path
// works end to end, without logging every subsequent commit/push cycle.
type branchWorkerLogFirsts struct {
	commit sync.Once
	push   sync.Once
}

type windowFinalizeReason string

const (
	windowFinalizeReasonUnspecified       windowFinalizeReason = "unspecified"
	windowFinalizeReasonIdleTimeout       windowFinalizeReason = "idle-timeout"
	windowFinalizeReasonMaxDuration       windowFinalizeReason = "max-duration"
	windowFinalizeReasonAttachNext        windowFinalizeReason = "attach-next"
	windowFinalizeReasonResyncBeforeApply windowFinalizeReason = "resync-before-apply"
	windowFinalizeReasonIdentityChange    windowFinalizeReason = "author-or-target-change"
	windowFinalizeReasonBufferLimit       windowFinalizeReason = "buffer-limit"
	windowFinalizeReasonShutdown          windowFinalizeReason = "shutdown"
	windowFinalizeReasonForeignCommit     windowFinalizeReason = "foreign-commit"
	windowFinalizeReasonApplyDue          windowFinalizeReason = "apply-due"
)

// NewBranchWorker creates a worker for a (provider, branch) combination. A zero
// BranchWorkerLimits gets the defaults for both knobs; see BranchWorkerLimits.
func NewBranchWorker(
	client client.Client,
	log logr.Logger,
	providerName, providerNamespace string,
	branch string,
	repo RepoIdentity,
	writer *contentWriter,
	limits BranchWorkerLimits,
) *BranchWorker {
	if writer == nil {
		writer = newContentWriter(itypes.SensitiveResourcePolicy{})
	}
	limits = limits.withDefaults()
	// Both flags start at their zero values, which is the state §3 requires of a new worker:
	// nothing has looked at the remote, so the base is untrusted, and nothing has written, so
	// the worktree is clean.
	return &BranchWorker{
		GitProviderRef:       providerName,
		GitProviderNamespace: providerNamespace,
		Branch:               branch,
		repo:                 repo,
		Client:               client,
		Log: log.WithValues(
			"provider", providerName,
			"namespace", providerNamespace,
			"branch", branch,
		),
		contentWriter:        writer,
		eventQueue:           make(chan WorkItem, limits.QueueDepth),
		applyWake:            make(chan struct{}, 1),
		pendingResyncs:       make(map[resyncKey]*resyncMarker),
		branchBufferMaxBytes: limits.MaxBufferBytes,
	}
}

// workerStateRoot is where every worker's on-disk state lives. It is a variable rather than a
// constant so a test can point it at its own directory: the tests that exercise reclaiming a
// retired worker's checkout delete directories, and doing that under the shared root would take
// out the state of anything else using it, including a developer's running operator.
//
//nolint:gochecknoglobals // a test seam, like pushAtomicFn above
var workerStateRoot = filepath.Join("/tmp", "gitops-reverser-workers")

// SetWorkerStateRoot points every worker's on-disk state at dir, for the whole process. The
// operator never calls it. A test binary that runs a WorkerManager calls it from TestMain, before
// any worker exists: the manager's sweep deletes every directory under the root that its own
// GitBranches do not name, so under the shared default it would delete the checkouts of every
// other process using that root.
func SetWorkerStateRoot(dir string) { workerStateRoot = dir }

// repoRootPath is where this worker keeps its checkout: one directory, named by the UID of the
// GitBranch it writes for.
//
// The GitBranch pins the branch (spec.branch is immutable), the claim gives a repository branch to
// one GitBranch at a time, and EnsureWorker keeps at most one live worker per GitBranch, so no two
// live workers share a directory. A UID is minted fresh for every object, so a GitBranch deleted
// and recreated under its name starts in a new directory. And because the branch is not part of
// the path, a branch name's slashes cannot nest one worker's checkout inside another's.
//
// The repository is the one thing the UID does not pin: the GitProvider is named, not owned, and
// can be recreated against another URL. EnsureWorker removes the old worker's checkout before the
// replacement starts in the same directory, and a checkout left behind by a restart is never
// trusted: a new worker's first cycle repoints origin, fetches and resets.
func (w *BranchWorker) repoRootPath() string {
	return filepath.Join(workerStateRoot, w.stateComponent())
}

// stateComponent names this worker's state directory.
//
// A worker built with no GitBranch behind it — tests — has no UID to be unique by. It falls back
// to a digest of its remote and branch, which is the only identity it has.
func (w *BranchWorker) stateComponent() string {
	if w.GitBranchUID != "" {
		return string(w.GitBranchUID)
	}
	sum := sha256.Sum256([]byte(strings.TrimSpace(w.repo.URL) + "\x00" + w.Branch))
	return "url-" + hex.EncodeToString(sum[:16])
}

// repoPath is the checkout of the repository this worker is ABOUT.
//
// The URL comes from the worker's own identity and never from the GitProvider as it reads NOW. The
// provider is still read on every cycle — for credentials, commit identity and signing — and it
// can name a different repository by then, because it is repointed by being recreated under its
// name, and the GitTargets find the worker by that name. Following it there is what produced the
// outage in #382:
// the cycle planned against trust earned on the old repository and every write failed at
// `open repository: repository does not exist`. A worker that cannot be pointed anywhere cannot
// have that happen to it; the WorkerManager replaces it instead.
func (w *BranchWorker) repoPath() string { return w.repoRootPath() }

// removeLocalState deletes everything this worker kept on disk: its GitBranch's directory.
//
// It is called only when a worker is being retired because nothing needs it any more, never on
// shutdown — a restart wants the clones it left behind, and re-cloning every branch on every
// restart is exactly the cost the on-disk cache exists to avoid.
//
// Only a worker with a GitBranch UID reclaims anything. That is every worker the WorkerManager
// starts; one a test builds by hand has a directory derived from whatever identity it was given,
// down to none at all, and deleting a directory nobody can name is how a live worker's goes too.
func (w *BranchWorker) removeLocalState() {
	if w.GitBranchUID == "" {
		w.Log.V(1).Info("Not removing worker state: the worker has no complete identity")
		return
	}
	path := w.repoRootPath()
	if err := os.RemoveAll(path); err != nil {
		// Worth knowing and not worth failing for: the worker is gone either way, and what is
		// left is disk rather than anything that can produce a wrong answer.
		w.Log.Error(err, "Could not remove the retired worker's checkout", "path", path)
		return
	}
	w.Log.Info("Removed the retired worker's checkout", "path", path)
}

// Start begins processing events.
func (w *BranchWorker) Start(parentCtx context.Context) error {
	w.mu.Lock()
	if w.started {
		w.mu.Unlock()
		return errors.New("worker already started")
	}
	w.ctx, w.cancelFunc = context.WithCancel(parentCtx)
	w.started = true
	w.mu.Unlock()

	w.Log.Info("Starting branch worker")

	w.wg.Add(1)
	go func() {
		defer w.wg.Done()
		w.processEvents()
		w.settleCommitRequestsAfterExit()
	}()

	return nil
}

// Stop gracefully shuts down the worker.
func (w *BranchWorker) Stop() {
	w.mu.Lock()
	if !w.started {
		w.mu.Unlock()
		return
	}
	w.mu.Unlock()

	// Before the cancel, and under the lock every enqueue holds across its send, so admission
	// closes at one instant: an enqueue either completed before this point or is refused.
	w.pendingResyncsMu.Lock()
	w.stoppingState = true
	w.pendingResyncsMu.Unlock()

	w.Log.Info("Stopping branch worker")
	w.cancelFunc()
	w.wg.Wait()
	w.Log.Info("Branch worker stopped")
}

// Enqueue adds a single live event to this worker's queue. It reports whether the
// event entered the FIFO; a false return means the queue was full and the event was
// dropped, so a caller advancing a durable watch cursor past this event must not treat
// the drop as success (see reconcile.GitTargetEventStream.OnWatchEvent).
func (w *BranchWorker) Enqueue(event Event) bool {
	return w.enqueueRequest(&WriteRequest{Events: []Event{event}})
}

// EnqueueWithdraw asks the loop to cancel a CommitRequest it has not acted on. It never blocks: a
// full queue or a stopping worker drops it, and the controller sends it again on its next poll. A
// worker whose loop has exited answers at once, because nothing is left that could act on it.
func (w *BranchWorker) EnqueueWithdraw(req *AttachCommitRequest) {
	if req == nil {
		return
	}
	w.pendingResyncsMu.Lock()
	if w.exitedState {
		w.pendingResyncsMu.Unlock()
		w.answerWithdrawAfterExit(req.id())
		return
	}
	defer w.pendingResyncsMu.Unlock()
	if w.stoppingLocked() {
		// Still draining: its last push may yet carry the request, so only the exited worker can
		// answer. The controller asks again on its next poll.
		return
	}
	w.inflightItems.Add(1)
	select {
	case w.eventQueue <- WorkItem{Withdraw: req}:
	default:
		w.releaseInflightItem()
		w.Log.V(1).Info("Queue full, CommitRequest withdraw dropped (the controller re-sends)",
			"request", req.Namespace+"/"+req.Name)
	}
}

// EnqueueAttach adds a CommitRequest attach to this worker's queue. Riding the
// same queue as resource events is what makes it process in audit order, after
// every earlier write. The attach is fire-and-forget: the controller polls the
// outcome via LookupCommitRequestOutcome and re-sends idempotently, so a queue-
// full drop is recovered by the next poll rather than a synchronous reply.
func (w *BranchWorker) EnqueueAttach(req *AttachCommitRequest) {
	if req == nil {
		return
	}
	charge := attachCharge(req.Message)

	// Increment before the send so inflightItems can never lag the loop's
	// receive; roll back if the queue is full and the item is dropped.
	w.inflightItems.Add(1)
	// An attach decides which commit window subsequent work joins, so a resync
	// must not be moved across one either. Attaches are controller-driven and
	// rare, so blocking a coalesce on them costs no storm protection. The mark
	// and the send are ONE critical section: see enqueueRequest.
	w.pendingResyncsMu.Lock()
	if w.stoppingLocked() {
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.recordQueueRefusal(queueRefusalAttach)
		w.Log.V(1).Info("Worker is stopping, CommitRequest attach refused (the controller re-sends)")
		return
	}
	if !w.admitPayload(charge) {
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.recordQueueRefusal(queueRefusalAttach)
		w.Log.V(1).Info("CommitRequest attach refused while the branch has paused intake "+
			"(the controller re-sends)", "request", req.Namespace+"/"+req.Name)
		return
	}
	id := req.id()
	if !w.crOwners.claim(id, w) {
		// The router routes to the owner, so only a request that moved workers between two polls
		// gets here. The owner answers for it; registering it twice could commit it twice.
		w.intake.release(charge)
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.Log.Info("CommitRequest attach refused: another branch worker holds the request",
			"request", req.Namespace+"/"+req.Name)
		return
	}
	w.markResyncTailForTargetLocked(req.GitTargetNamespace, req.GitTargetName)
	w.countQueuedAttach(id)
	select {
	case w.eventQueue <- WorkItem{Attach: req, charge: charge}:
		w.pendingResyncsMu.Unlock()
		w.Log.Info("CommitRequest attach enqueued",
			"request", req.Namespace+"/"+req.Name,
			"author", req.Author,
			"target", req.GitTargetNamespace+"/"+req.GitTargetName,
			"attach", string(req.Attach),
			"messageOverride", req.Message != "")
		// Nothing publishes depth here: the gauge reads inflightItems at scrape
		// time, so an enqueue is visible to the next scrape whether or not the
		// loop has woken to notice it.
	default:
		w.uncountQueuedAttach(id)
		w.intake.release(charge)
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.recordQueueRefusal(queueRefusalAttach)
		w.Log.Error(nil, "Event queue full, CommitRequest attach refused (controller will re-send)")
	}
}

// EnqueueResync adds a resync request to this worker's queue. Like a finalize
// signal it rides the same queue as resource events, so it is applied in order with
// live events: a resync enqueued during the snapshot window lands before the buffered
// live events that follow it. If the queue is full the request is refused and its
// caller is told immediately via the result channel, so it can gather the scope again.
//
// It reports whether the request actually entered the FIFO. A caller gating downstream state on
// the resync's ordering must not treat a refusal as success: it would mark the target
// reconciled-through-Hc with no reconcile ever queued.
func (w *BranchWorker) EnqueueResync(request *ResyncRequest) bool {
	if request == nil {
		return false
	}
	key := resyncKeyFor(request)
	charge := request.charge() // sized here, off the loop and outside the lock

	w.pendingResyncsMu.Lock()
	if w.stoppingLocked() {
		w.pendingResyncsMu.Unlock()
		w.recordQueueRefusal(queueRefusalResync)
		w.Log.V(1).Info("Worker is stopping, resync request refused",
			"gitTarget", request.GitTargetNamespace+"/"+request.GitTargetName)
		// Answered rather than dropped in silence: the caller is waiting on this channel, and the
		// shutdown drain does not answer what it throws away.
		request.reply(ResyncResult{Err: ErrFinalizeQueueFull})
		return false
	}
	if w.pendingResyncs == nil {
		w.pendingResyncs = make(map[resyncKey]*resyncMarker)
	}
	w.markResyncTailForResyncLocked(request, key)
	if pending, queued := w.pendingResyncs[key]; queued && !pending.tailPassed {
		// A marker for this key is already in the FIFO and nothing for this scope
		// was queued behind it, so the marker's position is still the right place
		// for a newer snapshot. Swap in the newer request; the loop reads whatever
		// is current when the marker comes up. Only the difference in size is new.
		if !w.admitPayload(charge - pending.request.charge()) {
			w.pendingResyncsMu.Unlock()
			return w.refuseResyncAtIntake(request, charge)
		}
		superseded := pending.request
		pending.request, pending.charge = request, charge
		w.pendingResyncsMu.Unlock()
		// The superseded request's caller is waiting on its reply channel. Answer
		// it so its drain ends, with a sentinel that says "a newer resync for the
		// same scope is queued", not "this failed".
		superseded.reply(ResyncResult{Err: ErrResyncSuperseded})
		w.Log.V(1).Info("Resync request coalesced into the one already queued",
			"resources", len(request.Desired),
			"gitTarget", request.GitTargetNamespace+"/"+request.GitTargetName,
			"sourceCollection", sourceCollectionForLog(request.SourceCollection))
		return true
	} else if queued {
		// A write inside this scope was queued behind the marker. Coalescing here
		// would run this snapshot ahead of those writes and let them overwrite it
		// with older state. Release the key instead: the queued marker keeps the
		// request it holds, and this request takes a fresh marker at the tail below.
		delete(w.pendingResyncs, key)
		w.Log.V(1).Info("Resync request not coalesced: writes are queued behind the pending marker",
			"scope", request.Scope.String(),
			"gitTarget", request.GitTargetNamespace+"/"+request.GitTargetName)
	}
	// Insert the entry and queue its marker in ONE critical section. Releasing the
	// lock between them would let a concurrent enqueue for the same key coalesce
	// into an entry whose marker does not exist yet -- and which this call is
	// about to remove on a full queue. That caller would be told enqueued=true
	// while its request never ran and its drain waited for a reply that never
	// came. The send is non-blocking, so holding the lock across it cannot
	// deadlock against the loop's takePendingResync.
	if !w.admitPayload(charge) {
		w.pendingResyncsMu.Unlock()
		return w.refuseResyncAtIntake(request, charge)
	}
	marker := &resyncMarker{request: request, charge: charge}
	w.pendingResyncs[key] = marker
	w.inflightItems.Add(1)
	select {
	case w.eventQueue <- WorkItem{Resync: marker}:
		if w.queuedResyncs == nil {
			w.queuedResyncs = map[string]int{}
		}
		w.queuedResyncs[resyncTargetKey(request)]++
		w.pendingResyncsMu.Unlock()
		w.Log.V(1).Info("Resync request enqueued",
			"resources", len(request.Desired),
			"gitTarget", request.GitTargetNamespace+"/"+request.GitTargetName,
			"sourceCollection", sourceCollectionForLog(request.SourceCollection))
		return true
	default:
		w.releaseInflightItem()
		delete(w.pendingResyncs, key)
		w.intake.release(charge)
		w.pendingResyncsMu.Unlock()
		w.recordQueueRefusal(queueRefusalResync)
		w.Log.Error(nil, "Event queue full, resync request refused; its caller gathers it again",
			"gitTarget", request.GitTargetNamespace+"/"+request.GitTargetName,
			"sourceCollection", sourceCollectionForLog(request.SourceCollection))
		request.reply(ResyncResult{Err: ErrFinalizeQueueFull})
		return false
	}
}

// refuseResyncAtIntake answers a resync the intake gate refused. It reports false: the request did
// not enter the FIFO.
func (w *BranchWorker) refuseResyncAtIntake(request *ResyncRequest, charge int64) bool {
	w.recordQueueRefusal(queueRefusalResync)
	err := w.intakeRefusal(charge)
	w.Log.V(1).Info("Resync refused while the branch has paused intake",
		"gitTarget", request.GitTargetNamespace+"/"+request.GitTargetName, "reason", err.Error())
	request.reply(ResyncResult{Err: err})
	return false
}

// markResyncTailForWriteLocked records, on every pending resync whose GitTarget and
// scope contain one of this write's events, that work for that scope is now queued
// behind its marker. A pending resync with a nil scope covers the whole GitTarget, so
// any write for that target marks it. The caller must hold pendingResyncsMu.
//
// The target is read from the EVENT, not the request: the live path leaves the request-level
// fields empty, so reading the request alone would silently never match the only path this fence
// exists for.
//
// The scope match is by object identity, not producing stream, because the collection that produced an
// event cannot be recovered from it. Over-matching is deliberate: it can forgo a coalesce, never
// wrongly permit one.
func (w *BranchWorker) markResyncTailForWriteLocked(request *WriteRequest) {
	for key, pending := range w.pendingResyncs {
		if pending.tailPassed {
			continue
		}
		for i := range request.Events {
			namespace, name := request.Events[i].GitTargetNamespace, request.Events[i].GitTargetName
			if key.namespace != namespace || key.name != name {
				continue
			}
			if pending.request.Scope.Matches(request.Events[i].Identifier) {
				pending.tailPassed = true
				break
			}
		}
	}
}

// markResyncTailForResyncLocked records, on every OTHER pending resync of the same GitTarget whose
// sweep boundary overlaps this one's, that a snapshot is now queued behind its marker. A snapshot
// is a write to its whole boundary, so it fences an earlier snapshot exactly as a live write does.
//
// Without it a selector changed A -> B -> A queued snapshots A1, B, A2 and coalesced A2 into A1's
// position: A2 ran first and the retired B snapshot then ran after it, undoing the current
// selection. With it A2 takes the tail, so the order stays A1, B, A2. The caller must hold
// pendingResyncsMu.
func (w *BranchWorker) markResyncTailForResyncLocked(request *ResyncRequest, key resyncKey) {
	for pendingKey, pending := range w.pendingResyncs {
		if pending.tailPassed || pendingKey == key {
			continue
		}
		if pendingKey.namespace == request.GitTargetNamespace && pendingKey.name == request.GitTargetName &&
			pending.request.Scope.Overlaps(request.Scope) {
			pending.tailPassed = true
		}
	}
}

// markResyncTailForTargetLocked records the same boundary for every pending resync of one
// GitTarget, regardless of scope. It is used for queued items that carry no resource
// identity of their own. The caller must hold pendingResyncsMu.
func (w *BranchWorker) markResyncTailForTargetLocked(namespace, name string) {
	for key, pending := range w.pendingResyncs {
		if key.namespace == namespace && key.name == name {
			pending.tailPassed = true
		}
	}
}

// takePendingResync returns the request a marker runs, which may be newer than the one it was queued
// with, and the charge it holds, and closes its key if it is still open, so the next enqueue queues a
// fresh marker. A key already released belongs to a later marker, which is left alone.
func (w *BranchWorker) takePendingResync(marker *resyncMarker) (*ResyncRequest, int64) {
	w.pendingResyncsMu.Lock()
	defer w.pendingResyncsMu.Unlock()
	key := resyncKeyFor(marker.request)
	if w.pendingResyncs[key] == marker {
		delete(w.pendingResyncs, key)
	}
	target := resyncTargetKey(marker.request)
	if queued, counted := w.queuedResyncs[target]; counted && queued > 1 {
		w.queuedResyncs[target] = queued - 1
	} else {
		delete(w.queuedResyncs, target)
	}
	return marker.request, marker.charge
}

// queuedResyncsFor reports whether a resync of the GitTarget is on the FIFO, not taken yet.
func (w *BranchWorker) queuedResyncsFor(target string) bool {
	w.pendingResyncsMu.Lock()
	defer w.pendingResyncsMu.Unlock()
	return w.queuedResyncs[target] > 0
}

// resyncTargetKey is the GitTarget a resync is for, as namespace/name.
func resyncTargetKey(request *ResyncRequest) string {
	return request.GitTargetNamespace + "/" + request.GitTargetName
}

// enqueueRequest places a write request on the FIFO and reports whether it was
// accepted. A false return means the item was refused (a full queue, paused intake, or a
// stopping worker), so a caller that gates durable state on the write (a watch cursor) must not
// treat the refusal as success, and offers the write again.
func (w *BranchWorker) enqueueRequest(request *WriteRequest) bool {
	if request == nil {
		return false
	}
	item := WorkItem{Request: request, charge: eventCharge(request)} // sized off the loop
	// Increment before the send so inflightItems can never lag the loop's
	// receive; roll back if the queue is full and the item is dropped.
	w.inflightItems.Add(1)
	// Mark the pending resyncs this write falls inside, and send, in ONE critical
	// section. Marking before an unlocked send would leave a window in which a
	// concurrent EnqueueResync sees the mark, declines to coalesce, and takes its
	// own tail position BEFORE this write reaches the FIFO -- the same inversion
	// the fence exists to prevent, one step removed. EnqueueResync holds the same
	// lock across its own send, so the two orderings agree. Both sends are
	// non-blocking, so holding the lock across them cannot deadlock. Marking a
	// write the full queue then drops only forgoes a coalesce, the safe direction.
	w.pendingResyncsMu.Lock()
	if w.stoppingLocked() {
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.recordQueueRefusal(queueRefusalWrite)
		w.Log.V(1).Info("Worker is stopping, request refused; the producer keeps it to redeliver",
			"events", len(request.Events),
			"gitTarget", request.gitTarget())
		return false
	}
	if !w.admitPayload(item.charge) {
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.recordQueueRefusal(queueRefusalWrite)
		w.Log.V(1).Info("Write refused while the branch has paused intake; "+
			"the producer keeps it to redeliver", "events", len(request.Events), "gitTarget", request.gitTarget())
		return false
	}
	w.markResyncTailForWriteLocked(request)
	select {
	case w.eventQueue <- item:
		w.pendingResyncsMu.Unlock()
		w.Log.V(1).Info("Write request enqueued",
			"events", len(request.Events),
			"gitTarget", request.gitTarget(),
			"sourceCollection", sourceCollectionForLog(request.sourceCollection()))
		// Nothing publishes depth here: the gauge reads inflightItems at scrape
		// time, so an enqueue is visible to the next scrape whether or not the
		// loop has woken to notice it.
		return true
	default:
		w.intake.release(item.charge)
		w.pendingResyncsMu.Unlock()
		w.releaseInflightItem()
		w.recordQueueRefusal(queueRefusalWrite)
		// Name the producing collection on a refusal. A saturated queue is diagnosed from what was
		// refused and by whom: the 595-in-16-seconds storm was one GitTarget, and the next
		// one may be one COLLECTION of one GitTarget.
		w.Log.Error(nil, "Event queue full, request refused; its producer offers it again",
			"events", len(request.Events),
			"gitTarget", request.gitTarget(),
			"sourceCollection", sourceCollectionForLog(request.sourceCollection()))
		return false
	}
}

// queueDepth reports this worker's pending work: accepted-but-unhandled items plus one when the
// loop is holding retained unpushed work. It reads 0 only on a full drain, so a drain gate cannot
// be satisfied mid-push.
//
// Read by the git_queue_depth observable gauge at scrape time. Both fields are atomics, so the
// callback takes no lock the loop could be holding across its slow work.
func (w *BranchWorker) queueDepth() int64 {
	depth := w.inflightItems.Load()
	if w.hasUnpushedWork.Load() {
		depth++
	}
	return depth
}

// processEvents is the main event processing loop.
//
// The loop owns one live commit-shaped event window. A commit-window timer
// detects silence for that open window, and a one-shot push timer enforces the
// PushCooldown between successful pushes. Commit and push are independent:
// local commits accumulate from retained pending writes and feed
// replay-on-conflict; only a successful push clears that retained queue.
func (w *BranchWorker) processEvents() {
	// The provider is no longer read for the commit window — that is now per-GitTarget — but it
	// remains a precondition for the worker: a branch whose GitProvider cannot be read has no
	// credential, no allowed-branch list and nowhere to push, so starting the loop would only
	// accumulate work that can never land.
	if _, err := w.getGitProvider(w.ctx); err != nil {
		w.Log.Error(err, "Failed to get GitProvider, worker exiting")
		return
	}

	loop := newBranchWorkerEventLoop(w, DefaultCommitWindow)
	w.Log.Info("Branch worker event loop configured",
		"defaultCommitWindow", DefaultCommitWindow.String(),
		"queueDepth", cap(w.eventQueue),
		"branchBufferMaxBytes", w.branchBufferMaxBytes)
	loop.run()
}

// branchWorkerEventLoop holds the per-branch event-loop state. Only the
// goroutine running run() may touch these fields, so no extra synchronisation
// is required.
type branchWorkerEventLoop struct {
	// commitWindows holds the live commit-shaped event windows and what they are charged. A window is
	// finalized when its idle or max-duration timer runs out, and eagerly on author/target changes,
	// resync arrivals, byte-cap trips, an attach: Next, and shutdown. Embedded, so loop.openWindow
	// names the one window it holds.
	commitWindows

	w *BranchWorker

	// barrierSeq numbers the barriers this loop's apply windows take (apply_barrier.go).
	barrierSeq uint64

	// defaultWindow is the operator-level default for a GitTarget's commit window timers, used
	// for any target that declares no spec.commit.window field and for one that cannot be read.
	defaultWindow commitWindowDefaults

	// pendingWrites holds the retained durability units for local commits that
	// have not yet successfully reached the remote. A successful push is the
	// only thing that clears this slice.
	pendingWrites      []PendingWrite
	pendingWritesBytes int64

	lastPushAt time.Time
	// cooldownHoldsSaves keeps a save's push behind the cooldown, like any other push. A test seam,
	// for tests that inspect a save's write before it is published; never set otherwise.
	cooldownHoldsSaves bool
	commitTimer        *time.Timer
	// commitTimerAt is the worker-clock time commitTimer is armed for, while it is armed. The timer
	// itself runs on real time; tests that step the clock deliver its wake at this time.
	commitTimerAt time.Time
	// pushTimer fires the next push when the success cooldown ends.
	pushTimer *time.Timer
	// retry is the one schedule of the next attempt at work a failed attempt left pending. See
	// retry.go.
	retry retrySchedule

	// snapshotRerequests are the GitTargets whose snapshot was discarded as stale, until a fresh one
	// is asked for; snapshotHeldSince is when each GitTarget's first one was, until one is used. See
	// apply_epoch.go.
	snapshotRerequests map[string]itypes.ResourceReference
	snapshotHeldSince  map[string]time.Time
	// awaitingFolders are the GitTargets whose folder the branch did not hold when their applier was
	// about to be asked, and which a fresh snapshot was asked for, until it is there (barrier.go).
	awaitingFolders map[itypes.ResourceReference]bool
	// ownershipLogged is the objects another applier owns already logged as kept out of a folder.
	// See apply_ownership.go.
	ownershipLogged map[string]struct{}

	// pendingCRs holds CommitRequests registered (via AttachCommitRequest) but not
	// yet resolved, keyed by identity. A request is parked here until a same-author
	// window opens (then it attaches to it) or its attach deadline passes.
	// Loop-goroutine only.
	pendingCRs map[commitRequestID]*pendingCommitRequest
	// savedCatchUps is the CommitRequests, by GitTarget, whose pushed commit asked their applier to
	// catch up at once (catchUp: AfterPush), with that commit, until the appliers are next listed
	// (takeReasons, save_apply.go).
	savedCatchUps map[itypes.ResourceReference]map[commitRequestID]string
	// crSeq numbers registrations, so competing requests attach first come, first served.
	crSeq uint64
	// refusalTimer fires at the earliest deadline in refusalPending. See refusal_reapply.go.
	refusalTimer *time.Timer
	// refusalPending holds one entry per GitTarget and watched collection with a coalesced reapply
	// request due.
	//
	// It is a MAP because one branch worker serves every target on its (provider, branch), while
	// the rate limit and the consent check are both per target. A single slot let one target
	// overwrite another's pending work: A queues, B replaces it, B is then suspended, and A's
	// authorized request is gone with nothing to report it.
	refusalPending map[refusalKey]pendingRefusalReapply
	// unsentRefusals holds the refusals the worker decided to send their target's applier a Refusal
	// request for, until an attempt carries them after a flush. It is keyed like refusalPending,
	// by target AND collection, so a collection that recovers cancels its own request and no other
	// (refusalRecovered). See applier_claim.go.
	unsentRefusals map[refusalKey]unsentRefusal

	// appliers holds the reasons pending for each applier named on this branch and how its attempts
	// are going, and nextApplyID numbers the attempts. applierTimer fires when the earliest waiting
	// attempt comes due. See applier_claim.go.
	appliers     map[ApplierRef]*applierState
	nextApplyID  uint64
	applierTimer *time.Timer
	// listRetryAt is when a failed listing of the branch's appliers is tried again, zero when the
	// last one succeeded (followRecordedOperations, takeReasons).
	listRetryAt time.Time
	// pushHinted records that a Git host reported a push while the branch was busy. The wake closes
	// the open window so its publication finds the move (closeWindowForPushHint).
	pushHinted bool
	// decisions numbers the decided writes (PendingWrite.decision). flushing is set while an apply is
	// due and waits for the writes up to flushMark, the last decision when it came due, to be
	// published (serviceAppliers). flushEdits holds the API-side edits decided since the mark, by
	// GitTarget and objectKey: the attempts the flush sends hold them in their windows (S1).
	decisions  uint64
	flushing   bool
	flushMark  uint64
	flushEdits map[itypes.ResourceReference]map[string]time.Time

	// attachTimer fires at the earliest attach deadline of a waiting CommitRequest, so a request
	// that no window reaches still resolves on time. An attached request needs no timer of its own:
	// its window's timers close it.
	attachTimer *time.Timer
	// attachTimerAt is the worker-clock time attachTimer is armed for, while it is armed.
	attachTimerAt time.Time

	// recovery is the open task of writing what a missing parent branch held back, and the probe
	// that drives it. See parent_recovery.go.
	recovery parentRecovery

	// fidelityDropsLogged is, per GitTarget, the gate status under which its last drop was logged at
	// Info, so a closed gate logs its first drop and counts the rest. See render_fidelity_drops.go.
	fidelityDropsLogged map[string]renderFidelityDropMark

	// held is, per GitTarget, the live events held while its render fidelity is pending; heldSeq
	// numbers them, heldEventBytes is what they are charged, and holdTimer looks at their gates again
	// (render_fidelity_hold.go).
	held           map[itypes.ResourceReference]*heldWrites
	heldSeq        uint64
	heldEventBytes int64
	holdTimer      *time.Timer
}

// commitWindowDefaults are a GitTarget's commit window timers when it declares none.
type commitWindowDefaults struct {
	idleTimeout time.Duration
	maxDuration time.Duration
}

func newBranchWorkerEventLoop(w *BranchWorker, defaultIdleTimeout time.Duration) *branchWorkerEventLoop {
	return &branchWorkerEventLoop{
		w: w,
		defaultWindow: commitWindowDefaults{
			idleTimeout: defaultIdleTimeout,
			maxDuration: DefaultCommitWindowMaxDuration,
		},
	}
}

func (l *branchWorkerEventLoop) run() {
	defer l.stopTimers()

	// Before the first live event can be mirrored: an operation an earlier process left in flight
	// is followed again, with its apply window open.
	l.followRecordedOperations(time.Now())
	l.publishClaimStatus()
	l.publishLoopState(0)
	for {
		commitC, pushC, attachC, refusalC := l.timerChannels()
		select {
		case <-l.w.ctx.Done():
			l.handleShutdown()
			l.publishLoopState(0)
			return
		case item := <-l.w.eventQueue:
			l.handleQueueItem(item)
			// Released after the wake has published what the item left behind: see queueDepth, and
			// decisionHolds, which reads busyWriting and this count the same way.
			l.w.releaseInflightItem()
			continue
		case <-commitC:
			l.commitTimer = nil
			l.closeOrArmWindow()
		case <-pushC:
			l.pushTimer = nil
			l.pushPending()
		case <-attachC:
			l.attachTimer = nil
			// The work (attach waiting requests, finalize due ones) is done by
			// serviceCommitRequests below.
		case <-refusalC:
			l.refusalTimer = nil
			l.flushPendingReapply()
		case <-l.retryTimerC():
			l.retry.timer = nil
			l.runRetry()
		case <-l.applierTimerC():
			l.applierTimer = nil
			// serviceAppliers, at the end of the wake, sends what has come due.
		case <-l.w.applyWake:
			// An attempt reported back; serviceAppliers settles and sends.
		case <-l.holdTimerC():
			l.holdTimer = nil
			// serviceHeldWrites, at the start of the wake, looks at the held GitTargets' gates.
		}
		l.endWake(0)
	}
}

// handleQueueItem handles one item from the FIFO as one wake of the loop.
func (l *branchWorkerEventLoop) handleQueueItem(item WorkItem) {
	l.endWake(l.dispatch(item))
}

// endWake is what every wake of the loop ends with, whatever woke it. released is the intake charge
// of the item the wake handled, or zero.
//
//   - Release the live events held for a GitTarget whose render fidelity is now established, or drop
//     them on a divergence or past the hold's bounds (serviceHeldWrites).
//   - Bind any waiting CommitRequest to an open window, resolve any whose attach deadline has
//     passed, and re-arm the deadline timer.
//   - Close the open window early if a Git host reported a push (closeWindowForPushHint).
//   - Drive the pending writes: commit what the wake decided, schedule its push (advance).
//   - Send the apply requests that are due, after a flush, once the GitTargets they cover have
//     their initial snapshots published (serviceAppliers).
//   - Ask again for the snapshots discarded as stale once Git is no longer ahead, and publish the
//     ApplyState (apply_epoch.go).
//   - Publish the result, moving the handled item's charge to what the loop holds.
func (l *branchWorkerEventLoop) endWake(released int64) {
	l.serviceHeldWrites()
	l.serviceCommitRequests()
	l.closeWindowForPushHint()
	l.advance()
	l.serviceAppliers()
	l.rerequestSnapshots()
	l.publishApplyState()
	l.publishLoopState(released)
}

// publishLoopState refreshes the worker's retained-work flag and the intake gate from the loop's
// authoritative state; the loop goroutine is the only writer of both. released is the charge of the
// item the wake handled: see intakeGate.sync.
//
// The flag is all the depth gauge needs from the loop: queueDepth() reads it and inflightItems at scrape
// time, so nothing here publishes a value that could go stale while the loop is busy. The order is
// the contract: the loop releases a handled item from the in-flight count only after this, because
// publishing the flag second would let both read empty while a window is open, and a scrape landing
// there would report a drained queue for a worker that has work.
func (l *branchWorkerEventLoop) publishLoopState(released int64) {
	l.w.hasUnpushedWork.Store(l.commitWindows.anyOpen() || len(l.pendingWrites) > 0 || len(l.held) > 0)
	l.syncAdmission(released)
}

func (l *branchWorkerEventLoop) timerChannels() (
	<-chan time.Time, <-chan time.Time, <-chan time.Time, <-chan time.Time,
) {
	var commitC, pushC, attachC, refusalC <-chan time.Time
	if l.commitTimer != nil {
		commitC = l.commitTimer.C
	}
	if l.pushTimer != nil {
		pushC = l.pushTimer.C
	}
	if l.attachTimer != nil {
		attachC = l.attachTimer.C
	}
	if l.refusalTimer != nil {
		refusalC = l.refusalTimer.C
	}
	return commitC, pushC, attachC, refusalC
}

// totalRetainedBytes is what the operator-level byte cap is enforced against:
// the open windows plus the locally-committed-but-not-yet-pushed events that
// we keep around for replay-on-conflict.
func (l *branchWorkerEventLoop) totalRetainedBytes() int64 {
	return l.commitWindows.windowBytes + l.pendingWritesBytes
}

// dispatch hands one item to its handler, and returns the intake charge the item was admitted with.
func (l *branchWorkerEventLoop) dispatch(item WorkItem) int64 {
	switch {
	case item.Attach != nil:
		l.handleAttachCommitRequest(item.Attach)
	case item.Withdraw != nil:
		l.handleWithdrawCommitRequest(item.Withdraw)
	case item.ApplyBarrier != nil:
		l.handleApplyBarrier(item.ApplyBarrier)
	case item.Resync != nil:
		// Take the CURRENT request for this scope: a newer one may have replaced it while this
		// marker waited in the FIFO. Running it here keeps the original queue position, so a resync
		// still lands before the live events buffered behind it.
		req, charge := l.w.takePendingResync(item.Resync)
		// Every live event held so far was queued before this snapshot was taken.
		req.heldUpTo = l.heldSeq
		l.handleResyncRequest(req)
		return charge
	case item.Refresh != nil:
		if item.Refresh.PushNotice {
			// Before the remote is read, so a push reported from here on queues a look of its own.
			if after, folded := l.w.pushNotices.take(); folded {
				item.Refresh.After = after
			}
		}
		l.handleRefreshRequest(item.Refresh)
	case item.Request == nil:
	default:
		l.handleLiveEvents(item.Request)
	}
	return item.charge
}

// handleLiveEvents appends one write request's live events to the commit window, splitting and
// finalizing it as identity, the byte cap or a zero window require.
func (l *branchWorkerEventLoop) handleLiveEvents(request *WriteRequest) {
	for _, event := range request.Events {
		if !l.mirrors(event) {
			continue
		}
		l.w.liveEdits.Add(1)
		l.appendLiveEvent(event)
	}
}

// appendLiveEvent appends one live event that is mirrored to the commit window, splitting and
// finalizing it as identity, the byte cap or a zero window require.
func (l *branchWorkerEventLoop) appendLiveEvent(event Event) {
	{
		for _, other := range l.commitWindows.blocking(event) {
			// Log the identity that broke the window so an unexpected split is
			// diagnosable: the common cause is an incoming event whose author is empty
			// (an unattributed update, e.g. a /status change that carries no audit fact)
			// arriving against a non-empty window author, which forces a finalize and
			// can split a CommitRequest collect window. windowAuthor vs eventAuthor makes
			// that visible at a glance.
			l.w.Log.Info("Window identity change forces finalize before appending event",
				"windowAuthor", other.Author,
				"eventAuthor", event.UserInfo.Username,
				"operation", event.Operation,
				"resource", event.Identifier.String(),
				"eventTarget", event.GitTargetNamespace+"/"+event.GitTargetName)
			l.finalizeWindow(other, windowFinalizeReasonIdentityChange)
		}

		window := l.commitWindows.forWrite(event)
		opened := false
		if window == nil {
			l.w.Log.Info("Opening commit window",
				"author", event.UserInfo.Username,
				"gitTarget", event.GitTargetNamespace+"/"+event.GitTargetName,
				"resource", event.Identifier.String())
			window = l.commitWindows.open(event, l.w.contentWriter, l.w.now())
			// One window, one GitTarget: snapshot that target's timers now, so a later edit to the
			// target never moves a deadline already running.
			timers := l.w.commitWindowFor(
				l.w.ctx, event.GitTargetName, event.GitTargetNamespace, l.defaultWindow)
			window.timers = windowTimers{
				idle:  timers.idleTimeout,
				maxAt: l.w.now().Add(timers.maxDuration),
			}
			opened = true
		}
		// The window keeps one event per path, so a replaced one stops being charged.
		if replaced, ok := window.add(event); ok {
			l.commitWindows.charge(window, -eventRetainedCharge(&replaced))
		}
		window.lastWriteAt = l.w.now()
		l.commitWindows.charge(window, eventRetainedCharge(&event))

		if opened {
			// A waiting CommitRequest gets the new window BEFORE anything can close it: before the
			// target's timers are applied, and before a buffer-limit flush. With idleTimeout or
			// maxDuration at 0s, or a write that trips the cap, the window would otherwise close in
			// this very step, and a waiting request would miss the write it was waiting for — or,
			// waiting on another author, never see the window that refuses it.
			l.noteForeignWindow()
			l.attachWaitingCommitRequests()
			if !l.commitWindows.isOpen(window) {
				return // the attached request's maxDuration: 0s closed it already
			}
		}

		if l.totalRetainedBytes() >= l.w.branchBufferMaxBytes {
			// Memory-pressure trip: drain immediately, ignoring the commit
			// window. The cap exists to bound pod memory, not to shape
			// commits. The push still respects the cooldown — a push that
			// fails on a stuck remote will not free memory, but that's the
			// known stuck-push pathology documented in the design.
			l.finalizeOpenWindowWithReason(windowFinalizeReasonBufferLimit)
			return
		}
		l.closeOrArmWindow()
	}
}

// mirrors reports whether a live event reaches the commit window now. It does not for an object
// another applier owns in a folder an applier claim applies (F4), for the applier's own write in its
// apply window (apply_window.go), and while the target's render fidelity is not established: on a
// divergence the event is dropped, and while the gate is pending it is held, once judged as on
// arrival, and committed in order when the gate opens (render_fidelity_hold.go). Events held for the
// target earlier go first, once its gate opened.
func (l *branchWorkerEventLoop) mirrors(event Event) bool {
	target := itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace)
	l.serviceHeld(target)
	status, closed := l.w.renderFidelityClosed(event.GitTargetName, event.GitTargetNamespace)
	if closed && status.State == RenderFidelityFalse {
		l.dropAtRenderFidelity(target, status, event.Identifier.String())
		return false
	}
	if l.ownedElsewhere(event.GitTargetName, event.GitTargetNamespace, event.Identifier.String(),
		event.AppliedBy) {
		return false
	}
	// Judged when it arrives, held or not: the apply windows and the flush record an edit when it
	// was made.
	if l.withholdLateApply(event) {
		return false
	}
	if closed || l.held[target] != nil {
		// Its adoption waits with it (F-c): an apply before the create is in a commit stamps nothing.
		l.holdAtRenderFidelity(target, event, status)
		return false
	}
	if event.Operation == string(itypes.OperationCreate) {
		l.noteCreated(event)
	}
	return true
}

// closeOrArmWindow finalizes every open window whose first deadline has passed, oldest first, and
// arms the commit timer for the earliest deadline of the windows left open. A zero timer therefore
// closes the window in the same step that collected the write.
func (l *branchWorkerEventLoop) closeOrArmWindow() {
	now := l.w.now()
	var next time.Time
	for _, window := range l.commitWindows.all() {
		closeAt, reason := window.closeAt()
		if !closeAt.After(now) {
			l.finalizeWindow(window, reason)
			continue
		}
		if next.IsZero() || closeAt.Before(next) {
			next = closeAt
		}
	}
	if next.IsZero() {
		l.stopCommitTimer()
		return
	}
	l.resetCommitTimer(next.Sub(now))
}

func (l *branchWorkerEventLoop) handleShutdown() {
	l.w.Log.Info("Handling shutdown, finalizing open window and pushing pending commits")
	l.finalizeOpenWindowWithReason(windowFinalizeReasonShutdown)
	l.dropHeldAtShutdown()
	if len(l.pendingWrites) > 0 {
		// Shutdown bypasses the cooldown and the retry: pending work needs to land before the
		// worker exits, even if a push was just sent. The push commits what was just decided.
		l.pushPending()
	}
	l.failUnpushedCommitRequests()
	l.drainUnhandledQueueItems()
}

// failUnpushedCommitRequests settles every CommitRequest still riding a write the exiting loop
// could not push.
//
// A request that reached a local commit is resolved by the push and by nothing else, so a
// shutdown whose final push failed would leave it in flight with no loop left to settle it. The
// controller would poll until its own safety window expired and then fail closed on a timeout,
// which says nothing about what happened. Say what happened instead.
func (l *branchWorkerEventLoop) failUnpushedCommitRequests() {
	for i := range l.pendingWrites {
		id := l.pendingWrites[i].CommitRequest
		if id == nil {
			continue
		}
		l.resolveCommitRequest(*id, FinalizeResult{
			Err: errors.New("worker stopped before the commit reached the remote"),
		})
	}
}

// drainUnhandledQueueItems clears items the exiting loop will never handle. See drainQueue.
func (l *branchWorkerEventLoop) drainUnhandledQueueItems() { l.w.drainQueue() }

// drainQueue clears items no loop will handle. Each was counted into inflightItems at enqueue and
// decremented only after handling, so without this the final depth would stay non-zero forever:
// the loop has stopped, so nothing republishes a corrected value. A buffered CommitRequest attach
// is dropped, and the request given back unless the worker knows it otherwise; the controller
// re-sends on its next poll, to whichever worker holds it then.
func (w *BranchWorker) drainQueue() {
	for {
		select {
		case item := <-w.eventQueue:
			if item.Attach != nil {
				w.uncountQueuedAttach(item.Attach.id())
			}
			charge := item.charge
			if item.Resync != nil {
				_, charge = w.takePendingResync(item.Resync)
			}
			w.intake.release(charge)
			w.releaseInflightItem()
		default:
			return
		}
	}
}

func (l *branchWorkerEventLoop) resetCommitTimer(delay time.Duration) {
	l.commitTimerAt = l.w.now().Add(delay)
	if l.commitTimer == nil {
		l.commitTimer = time.NewTimer(delay)
		return
	}
	if !l.commitTimer.Stop() {
		select {
		case <-l.commitTimer.C:
		default:
		}
	}
	l.commitTimer.Reset(delay)
}

// finalizeOpenWindowWithReason closes every open window, oldest first: the triggers that are about
// the branch rather than about one window (a resync, an apply flush, the byte cap, shutdown).
func (l *branchWorkerEventLoop) finalizeOpenWindowWithReason(reason windowFinalizeReason) {
	for _, window := range l.commitWindows.all() {
		l.finalizeWindow(window, reason)
	}
}

// finalizeWindow closes one live event window into one decided write. The attached CommitRequest
// message overrides the live template.
//
// The closed window is a decided write (decide): a remote that cannot be reached leaves it pending
// for the publication retry, and its CommitRequest held. Only a failure of the write itself
// drops it, and resolves that request Failed (settleFailed).
func (l *branchWorkerEventLoop) finalizeWindow(window *openWindow, reason windowFinalizeReason) {
	l.stopCommitTimer()
	// Counted here, before anything can fail or find no diff, so every window is counted once and
	// its collection time never includes the commit.
	l.w.recordWindowClosed(window, reason)
	events := window.orderedEvents()
	windowAuthor := window.Author
	targetName, targetNamespace := window.GitTarget, window.GitTargetNamespace
	windowTarget := targetNamespace + "/" + targetName
	pendingCR := window.pendingCR
	effectiveMessage := window.pendingMessage
	// The render-fidelity gate drops the whole window, request or not: its events were planned
	// against a render nobody has checked. Suspension is read from the write built below, because
	// a suspended target still scans. See write_gate.go.
	if err := l.w.targetWriteRefusal(targetName, targetNamespace, false); err != nil {
		l.w.Log.V(1).Info("Discarding open window while render fidelity is not established",
			"reason", string(reason), "gitTarget", targetNamespace+"/"+targetName)
		l.dropWindow(window, pendingCR, err)
		return
	}

	l.w.Log.Info("Finalizing open commit window",
		"reason", string(reason),
		"windowAuthor", windowAuthor,
		"windowTarget", windowTarget,
		"events", len(events),
		"windowBytes", window.bytes,
		"pendingWrites", len(l.pendingWrites),
		"messageOverride", effectiveMessage != "",
		"attachedCR", pendingCR != nil)

	pendingWrite, err := l.w.buildGroupedPendingWrite(l.w.ctx, events)
	if err != nil {
		l.w.recordCommitFailure(commitFailureError)
		l.w.Log.Error(err, "Failed to build pending write; dropping open window",
			"reason", string(reason),
			"windowAuthor", windowAuthor,
			"windowTarget", windowTarget,
			"events", len(events))
		l.dropWindow(window, pendingCR, fmt.Errorf("build pending write: %w", err))
		return
	}
	pendingWrite.CommitMessage = effectiveMessage
	pendingCR, proceed := l.gateBuiltWindow(window, pendingCR, pendingWrite.Target(), reason)
	if !proceed {
		return
	}
	l.carryRequestOnWrite(pendingWrite, pendingCR)

	l.commitWindows.remove(window)
	l.decide(*pendingWrite)

	l.w.Log.Info("Open commit window finalized",
		"reason", string(reason),
		"windowAuthor", windowAuthor,
		"windowTarget", windowTarget,
		"events", len(events),
		"pendingWrites", len(l.pendingWrites))
}

// carryRequestOnWrite carries the attached CommitRequest onto the write so its result follows the
// data: it is resolved Committed once this write is pushed (§6.5). The request's
// whenNothingToCommit rides along, so a window that changed nothing can still record its message.
func (l *branchWorkerEventLoop) carryRequestOnWrite(pendingWrite *PendingWrite, pendingCR *commitRequestID) {
	pendingWrite.CommitRequest = pendingCR
	if pendingCR == nil {
		return
	}
	if pcr := l.pendingCRs[*pendingCR]; pcr != nil {
		pendingWrite.AllowEmpty = pcr.commitEmpty
	}
}

// gateBuiltWindow asks the write gates once more for a window whose write is built, right before
// it commits, and returns the request the write still carries and whether to commit it at all.
//
// The render-fidelity gate is shared and moves concurrently, and recovering the checkout or
// resolving the target's metadata can take long enough for it to close after the finalize's first
// look. Whichever look sees it closed, the whole window stops: its request fails and nothing
// commits, so the two can never disagree. Suspension is different: the write still runs, so the
// target's scan stays fresh, but it commits nothing, and the push would report that as
// AlreadyPresent, so only the request fails. See write_gate.go.
func (l *branchWorkerEventLoop) gateBuiltWindow(
	window *openWindow, pendingCR *commitRequestID, target ResolvedTargetMetadata, reason windowFinalizeReason,
) (*commitRequestID, bool) {
	if err := l.w.targetWriteRefusal(target.Name, target.Namespace, false); err != nil {
		l.w.Log.V(1).Info("Discarding built window: render fidelity closed while it was finalized",
			"reason", string(reason), "gitTarget", target.Namespace+"/"+target.Name)
		l.dropWindow(window, pendingCR, err)
		return nil, false
	}
	if pendingCR != nil && target.Suspend {
		l.resolveCommitRequest(*pendingCR, FinalizeResult{Err: errTargetSuspended})
		return nil, true
	}
	return pendingCR, true
}

// dropWindow discards a window whose finalize failed before it was decided, resolving any
// attached CommitRequest as Failed so the controller does not poll forever.
func (l *branchWorkerEventLoop) dropWindow(window *openWindow, pendingCR *commitRequestID, cause error) {
	l.commitWindows.remove(window)
	if pendingCR != nil {
		l.resolveCommitRequest(*pendingCR, FinalizeResult{Branch: l.w.Branch, Err: cause})
	}
}

// maybeSchedulePush is advance's push step: it pushes immediately when the cooldown has elapsed (or
// has never fired), and otherwise schedules a one-shot pushTimer to fire when the cooldown expires.
// While the cooldown is active, additional commits accumulate locally; when the timer fires, all
// pending commits go to the remote in a single push. After a failed attempt the retry's deadline
// governs instead, so new commits do not each spend a push on a remote that just refused one.
func (l *branchWorkerEventLoop) maybeSchedulePush() {
	if len(l.pendingWrites) == 0 || l.awaitingRetry() {
		return
	}
	if l.lastPushAt.IsZero() || (l.flushing && l.pendingUpTo(l.flushMark)) {
		// An apply waits for this publication: it is made at once, cooldown or not.
		l.pushPending()
		return
	}
	if !l.cooldownHoldsSaves && l.pendingCarriesCommitRequest() {
		// A save waits for this publication, and a person waits for the save: it is made at once too.
		// Saves cannot burst the way plain writes can, since an author has one save in progress at a
		// time, so the cooldown still paces every stream of plain writes.
		l.w.Log.V(1).Info("Pushing at once: a CommitRequest waits for this push")
		l.pushPending()
		return
	}
	elapsed := time.Since(l.lastPushAt)
	if elapsed >= PushCooldown {
		l.pushPending()
		return
	}
	if l.pushTimer == nil {
		l.pushTimer = time.NewTimer(PushCooldown - elapsed)
	}
}

// pendingCarriesCommitRequest reports whether a pending write carries a CommitRequest, which resolves
// only once the write is pushed.
func (l *branchWorkerEventLoop) pendingCarriesCommitRequest() bool {
	for i := range l.pendingWrites {
		if l.pendingWrites[i].CommitRequest != nil {
			return true
		}
	}
	return false
}

// pushPending materializes the pending writes and publishes them. On success, pendingWrites is cleared,
// lastPushAt advances, and the outage, if there was one, is over. On failure (transient or after
// exhausting replay retries), pendingWrites stays in place, safe but unpublished, and the next
// attempt is scheduled (retry.go), so the work lands without another commit. Pending writes that
// all settle without a push are advance's to close.
func (l *branchWorkerEventLoop) pushPending() {
	if len(l.pendingWrites) == 0 {
		l.stopPushTimer()
		return
	}
	if l.w.awaitingParentProbe() {
		// The parent is known missing and the probe is not due: it publishes these when it is.
		l.stopPushTimer()
		return
	}

	// Commit any decided write that is not committed yet, and rebuild when a reset discarded the
	// local commits behind the others: pushing then would find the branch already at the remote
	// tip, report success without sending anything, and settle work that exists nowhere. Keep the
	// writes rather than publish a lie if that fails.
	if err := l.materialize(); err != nil {
		l.publicationFailed(err, "Cannot publish until the retained writes are committed; keeping them")
		return
	}
	if len(l.pendingWrites) == 0 {
		// Every decided write failed to commit, or committed nothing, so there is nothing to publish.
		l.stopPushTimer()
		return
	}

	err := l.w.pushPendingCommits(l.pendingWrites)
	// A rejected push replays the writes onto the moved remote, and the new tree can refuse some of
	// them. Those were not pushed and never will be, whatever the push did with the rest: they stop
	// being pending now, and are settled once the pending writes describe the push's outcome,
	// because settling one can decide more work.
	refused := l.takeReplayRefusals()
	if err != nil {
		// Leave pendingWrites in place; do NOT advance lastPushAt — the
		// design specifies lastPushAt only advances on a successful push. A
		// CommitRequest riding a retained write stays unresolved while we retry.
		l.publicationFailed(err, "Push failed; pending writes retained for retry")
		l.settleReplayRefusals(refused)
		return
	}

	// The writes are now on the remote: resolve every CommitRequest riding one with
	// the pushed commit's own SHA (§6.5) — "Committed" means "on the remote".
	l.resolvePushedCommitRequests()

	l.pendingWrites = nil
	l.pendingWritesBytes = 0
	l.lastPushAt = time.Now()
	l.stopPushTimer()
	l.endOutage()
	l.settleReplayRefusals(refused)
}

// publicationFailed keeps the writes for the next attempt, which it schedules: see retry.go.
func (l *branchWorkerEventLoop) publicationFailed(err error, message string) {
	l.stopPushTimer()
	l.noteParentUnavailable(err)
	l.scheduleRetry(err)
	l.w.Log.Error(err, message, "pendingWrites", len(l.pendingWrites), "retryAt", l.retry.due)
}

// resolvePushedCommitRequests settles every CommitRequest carried by a just-pushed write, now
// that the remote has spoken.
//
// Two outcomes, decided by whether the write ended up with a commit of its own. A real SHA is
// Committed, read per-write rather than from the branch HEAD because a batched push may stack a
// later commit on top. A zero SHA is AlreadyPresent: the plan found nothing to change AND the
// push confirmed the remote agreed, which is the difference between this and resolving at
// finalize time. If the remote had moved, the replay would have turned this write into a real
// commit before we got here, and it would resolve Committed instead.
func (l *branchWorkerEventLoop) resolvePushedCommitRequests() {
	for i := range l.pendingWrites {
		pw := l.pendingWrites[i]
		if pw.CommitRequest == nil {
			continue
		}
		if pw.CommitSHA.IsZero() {
			l.resolveCommitRequest(*pw.CommitRequest, FinalizeResult{Outcome: FinalizeAlreadyPresent})
			continue
		}
		if pw.emptyCommitted {
			// Nothing changed, and the request asked to record its message anyway. The outcome
			// keeps the cause, and the commit proves the message is in Git.
			l.resolveCommitRequest(*pw.CommitRequest, FinalizeResult{
				Outcome: pw.nothingToCommitOutcome(),
				Commit:  pw.CommitSHA.String(),
				Branch:  l.w.Branch,
			})
			continue
		}
		l.resolveCommitRequest(*pw.CommitRequest, FinalizeResult{
			Outcome: FinalizeCommitted,
			Commit:  pw.CommitSHA.String(),
			Branch:  l.w.Branch,
		})
	}
}

func (l *branchWorkerEventLoop) stopPushTimer() {
	if l.pushTimer == nil {
		return
	}
	if !l.pushTimer.Stop() {
		select {
		case <-l.pushTimer.C:
		default:
		}
	}
	l.pushTimer = nil
}

func (l *branchWorkerEventLoop) stopCommitTimer() {
	if l.commitTimer == nil {
		return
	}
	if !l.commitTimer.Stop() {
		select {
		case <-l.commitTimer.C:
		default:
		}
	}
	l.commitTimer = nil
	l.commitTimerAt = time.Time{}
}

func (l *branchWorkerEventLoop) stopTimers() {
	l.stopCommitTimer()
	l.stopPushTimer()
	l.stopAttachTimer()
	l.stopRefusalTimer()
	l.stopApplierTimer()
	l.stopHoldTimer()
	if l.retry.timer != nil {
		l.retry.timer.Stop()
		l.retry.timer = nil
	}
}

func (l *branchWorkerEventLoop) stopRefusalTimer() {
	if l.refusalTimer != nil {
		l.refusalTimer.Stop()
		l.refusalTimer = nil
	}
}

// baseTrusted reports whether the worktree is known to sit at the remote tip of the target
// branch.
//
// The failure direction is the safe one. A stale false costs one fetch. A stale true is caught by
// the compare-and-swap on the next push, which is the mechanism the write path already relies on.
//
// Trust is scoped to the parent configuration: a checkout proven under an older one is not
// trusted, whatever the flag says, because the flag can be set by a push or a fetch that started
// before the change.
func (w *BranchWorker) baseTrusted() bool {
	return w.baseTrustedState.Load() && w.baseParentGen.Load() == w.parentSnapshot().gen
}

// checkoutUnknown is the checkoutApplied value of a worktree a failed write may have left anything
// in. Only a reset clears it.
//
// It is NOT the same question as baseTrusted, and collapsing the two is a bug rather than a
// simplification. A write can fail part-way through executePendingWrites and leave staged changes
// behind while an earlier write is still retained; if the push that follows were allowed to
// declare everything well, the next cycle would plan on top of those leftovers and commit them
// under an unrelated author. So a successful push may make the base trusted, and must never make
// the worktree clean.
const checkoutUnknown = -1

// worktreeDirty reports whether the worktree may hold a partial write.
func (w *BranchWorker) worktreeDirty() bool { return w.checkoutApplied.Load() == checkoutUnknown }

// checkoutHolds reports whether the checkout is the projection of n retained writes: its root plus
// one commit for each. With n > 0 and anything else, the writes must be replayed before they are
// committed on or pushed.
func (w *BranchWorker) checkoutHolds(n int) bool { return w.checkoutApplied.Load() == int64(n) }

// setBaseTrusted records whether the base can be trusted.
//
// Trust is GAINED in exactly two places, and both are here rather than scattered: a reset to the
// fetched tip (updateBranchMetadataFromPullReport, whose every call site is a PrepareBranch or a
// syncToRemote, and which both fetch and reset), and a successful push (the uploaded commits are
// now the remote tip). fetchRemoteBranchHash pointedly does not call it: that one fetches without
// resetting, so it learns where the remote is without making the worktree match.
func (w *BranchWorker) setBaseTrusted(trusted bool) {
	w.baseTrustedState.Store(trusted)
}

// invalidateBase records that the worktree can no longer be assumed to sit at the remote tip.
//
// reason is logged rather than stored: what an operator needs is the sequence, and what the code
// needs is only the boolean.
func (w *BranchWorker) invalidateBase(reason string) {
	if w.baseTrustedState.Swap(false) {
		w.Log.V(1).Info("Base no longer trusted", "reason", reason, "branch", w.Branch)
	}
}

// markWorktreeDirty records that the worktree may hold a partial write. Only a reset clears it.
func (w *BranchWorker) markWorktreeDirty(reason string) {
	if w.checkoutApplied.Swap(checkoutUnknown) != checkoutUnknown {
		w.Log.V(1).Info("Worktree may hold a partial write", "reason", reason, "branch", w.Branch)
	}
}

// markCheckoutReset records that the checkout now holds none of the retained writes: a reset to
// the remote tip landed, or is about to. It is called BEFORE a reset that a replay will follow, as
// well as after every reset, because a reset moves the branch ref and then rewrites the worktree,
// so it can fail with the ref already moved and the commits behind the retained writes already
// unreachable. Marking too eagerly costs one replay.
func (w *BranchWorker) markCheckoutReset() { w.checkoutApplied.Store(0) }

// markCheckoutPublished records that a push published every retained write: unless a failed write
// left the worktree dirty, the checkout is its root again.
func (w *BranchWorker) markCheckoutPublished() {
	for {
		applied := w.checkoutApplied.Load()
		if applied == checkoutUnknown || w.checkoutApplied.CompareAndSwap(applied, 0) {
			return
		}
	}
}

// stoppingLocked reports that Stop has begun, so nothing new may enter the queue. The caller must
// hold pendingResyncsMu and must still be holding it when it sends: see stoppingState.
func (w *BranchWorker) stoppingLocked() bool { return w.stoppingState }

// ensureBaseForCycle performs the head-of-cycle fetch, which is now conditional.
//
// The push session reads the remote's ref advertisement on a connection the cycle was making
// anyway, and a cycle that commits nothing still reaches it, so a trusted base needs no fetch to
// plan against. That is the whole saving. See docs/design/push-notification-and-reconcile-trigger.md, Part 1.
//
// Only a checkout holding no retained writes may fetch here at all: a reset would destroy the
// local commits those writes produced. With writes retained, the event loop's materialize has
// already made the checkout their projection, resetting and replaying when it had to.
func (w *BranchWorker) ensureBaseForCycle(provider *configv1alpha3.GitProvider, repoPath string) error {
	if w.checkoutApplied.Load() > 0 {
		return nil
	}
	w.restampParentGenIfIndependent()
	if w.baseTrusted() && !w.worktreeDirty() {
		return nil
	}
	if w.awaitingParentProbe() {
		return errAwaitingParentProbe
	}

	// Resolve credentials only on the branch that touches the remote. A cycle planning on a
	// trusted base never reads the credentials Secret at all, which is a saved API GET per cycle
	// now that the Secret cache is disabled. See docs/rbac.md §5.
	auth, err := getAuthFromSecret(w.ctx, w.Client, provider, w.sshHostKeys, w.credentialPolicy)
	if err != nil {
		return fmt.Errorf("resolve auth: %w", err)
	}

	// A dirty worktree is a previous write that failed part-way; an untrusted base is the
	// ordinary "nothing has looked at the remote yet". The first is a bug report, the second is a
	// cost, so they are not the same series.
	reason := fetchReasonPublication
	if w.worktreeDirty() {
		reason = fetchReasonRecovery
	}
	w.recordFetch(reason)

	pullReport, err := w.prepareBranch(w.ctx, repoPath, auth)
	if err != nil {
		return fmt.Errorf("prepare repository: %w", err)
	}
	w.updateBranchMetadataFromPullReport(pullReport)
	return nil
}

// commitPendingWrites creates local commits for the provided pending writes without pushing them,
// on top of whatever retained writes the checkout already holds (checkoutApplied), and counts
// each write the loop retains into it. Only the first commit with nothing committed before it may
// fetch and reset; see ensureBaseForCycle, which decides whether that fetch is needed at all.
func (w *BranchWorker) commitPendingWrites(pendingWrites []PendingWrite) error {
	w.repoMu.Lock()
	defer w.repoMu.Unlock()

	if len(pendingWrites) == 0 {
		return nil
	}

	provider, err := w.getGitProvider(w.ctx)
	if err != nil {
		return fmt.Errorf("get GitProvider: %w", err)
	}

	repoPath := w.repoPath()
	if err := w.ensureBaseForCycle(provider, repoPath); err != nil {
		return err
	}

	repo, err := gogit.PlainOpen(repoPath)
	if err != nil {
		return fmt.Errorf("open repository: %w", err)
	}

	baseBranch, baseHash, err := w.ensureWriteBranch(repo)
	if err != nil {
		// From here on the worktree has been touched, so every failure leaves it in a state
		// nobody has verified. Dropping trust makes the next cycle fetch and reset, which is the
		// same cleanup the unconditional fetch performs today as a side effect.
		w.invalidateBase("ensure write branch failed")
		return err
	}

	if w.checkoutApplied.Load() == 0 {
		w.pushCycleRootBranch = baseBranch
		w.pushCycleRootHash = baseHash
		w.pushCycleRootGen = w.baseParentGen.Load()
	}

	commitsCreated, err := w.executePendingWrites(w.ctx, repo, pendingWrites)
	if err != nil {
		// A failed write undone locally left the checkout where it was, base included; one that
		// could not be undone has left a worktree nobody has verified.
		if w.worktreeDirty() {
			w.invalidateBase("execute pending writes failed")
		}
		return writeFailedError{fmt.Errorf("execute pending writes: %w", err)}
	}
	for i := range pendingWrites {
		if pendingWrites[i].retained() {
			w.checkoutApplied.Add(1)
		}
	}
	if commitsCreated == 0 {
		return nil
	}

	w.firsts.commit.Do(func() {
		w.Log.Info("First commit written to local repository",
			"branch", w.Branch,
			"commits", commitsCreated)
	})
	return nil
}

// pushPendingCommits publishes any local commits that have not yet reached the
// remote. On a conflict it resets to the latest remote tip, rebuilds from the
// retained pending writes, and retries. On a transient failure it leaves the
// local commits and retained pending writes in place for a later retry.
func (w *BranchWorker) pushPendingCommits(pendingWrites []PendingWrite) error {
	w.repoMu.Lock()
	defer w.repoMu.Unlock()

	if len(pendingWrites) == 0 {
		return nil
	}

	// One cycle, one terminal outcome, timed end to end. Closing the measurement HERE rather than
	// at each return inside the cycle is what makes "exactly once" structural: a return path added
	// to runPushCycle later cannot forget to record, and cannot record twice.
	started := time.Now()
	err := w.runPushCycle(pendingWrites)
	outcome := pushOutcomePushed
	if err != nil {
		outcome = pushOutcomeFailed
	}
	w.recordPushOutcome(outcome, started)
	if err == nil {
		// The published commits are the remote tip now, so they are the root the next write
		// commits on, and the checkout holds no pending write on top of it. A worktree a
		// failed write left dirty stays dirty: the push says nothing about its leftovers.
		w.markCheckoutPublished()
		// The commits are on the remote now, which is the only place they can honestly be counted,
		// and pendingWrites reflect whatever the final replay produced.
		w.publishCommitsForPush(pendingWrites)
	}
	return err
}

// runPushCycle is the push itself: try, and on a rejection caused by a moved remote, sync, rebuild
// the pending writes on the new head, and try again. The caller owns repoMu and the measurement.
//
// The whole cycle runs under one deadline, gitPublishTimeout, on top of each call's own bound
// (network_bound.go): three contention attempts, each with a fetch and a replay, must not each
// start a fresh budget while the loop, and every target on the branch, waits.
func (w *BranchWorker) runPushCycle(pendingWrites []PendingWrite) error {
	ctx, cancel := context.WithTimeout(w.ctx, gitPublishTimeout)
	defer cancel()

	provider, err := w.getGitProvider(ctx)
	if err != nil {
		return fmt.Errorf("get GitProvider: %w", err)
	}

	auth, err := getAuthFromSecret(ctx, w.Client, provider, w.sshHostKeys, w.credentialPolicy)
	if err != nil {
		return fmt.Errorf("resolve auth: %w", err)
	}

	repoPath := w.repoPath()
	repo, err := gogit.PlainOpen(repoPath)
	if err != nil {
		return fmt.Errorf("open repository: %w", err)
	}

	const maxRetries = 3
	var lastErr error

	for range maxRetries {
		// Re-read every attempt: a replay below re-roots the cycle.
		rootHash := w.pushCycleRootHash
		rootBranch := w.pushCycleRootBranch
		if rootBranch == "" {
			rootBranch = plumbing.NewBranchReferenceName(w.Branch)
		}

		// The admission boundary for a parent change: compared here, under repoMu and right before
		// the push, so a change made after the cycle was planned never leaves on the old root. A
		// change after this point is in flight, and the push is not recalled.
		if beforePushAdmission != nil {
			beforePushAdmission()
		}
		if !w.admitPushRoot(rootBranch) {
			w.Log.Info("Parent branch changed since this cycle was planned; rebuilding before the push",
				"branch", w.Branch, "parentBranch", w.ParentBranch())
			w.recordFetch(fetchReasonRecovery)
			if err := w.replayOntoRemote(ctx, repo, pendingWrites, auth); err != nil {
				return err
			}
			continue
		}

		outcome, err := pushAtomicFn(ctx, repo, rootHash, rootBranch, auth)
		if err == nil {
			w.notePushSucceeded(outcome, rootBranch, rootHash, len(pendingWrites))
			return nil
		}
		lastErr = err
		if retry, err := w.afterFailedPush(ctx, repo, pendingWrites, err, rootBranch, rootHash, auth); !retry {
			return err
		}
	}

	if lastErr == nil {
		return fmt.Errorf("push not admitted after %d attempts: the parent branch kept changing", maxRetries)
	}
	return fmt.Errorf("push failed after %d attempts: %w", maxRetries, lastErr)
}

// afterFailedPush decides what one failed push attempt means: the remote moved, and the writes are
// replayed onto it for another attempt (retry); the push landed although its answer was lost, and
// the cycle is done (nil); or anything else, which ends the cycle with the push's error. The caller
// owns repoMu.
func (w *BranchWorker) afterFailedPush(
	ctx context.Context,
	repo *gogit.Repository,
	pendingWrites []PendingWrite,
	pushErr error,
	rootBranch plumbing.ReferenceName,
	rootHash plumbing.Hash,
	auth []gitclient.Option,
) (bool, error) {
	// A rejection is a moved remote; any other push error may have died mid-upload. Either
	// way the remote state is no longer something we can claim to know.
	w.invalidateBase("push failed or was rejected")

	moved, landed := w.remoteMovedDuringPush(ctx, repo, pushErr, rootBranch, rootHash, auth)
	if landed != nil {
		// The push reached the remote and only its answer was lost (a deadline, a dropped
		// connection): the branch is at the commits we sent. Replaying them would plan the same
		// writes onto a tree that already holds them, and a save's empty commit would land twice.
		w.Log.Info("A push whose reply was lost had landed; the remote holds our commits",
			"branch", w.Branch, "head", landed.Head.String(), "err", pushErr.Error())
		w.notePushSucceeded(*landed, rootBranch, rootHash, len(pendingWrites))
		return false, nil
	}
	if !moved {
		return false, pushErr
	}
	w.recordPushRetry(pushRetryRemoteMoved)
	w.recordFetch(fetchReasonContention)
	if err := w.replayOntoRemote(ctx, repo, pendingWrites, auth); err != nil {
		return false, err
	}
	return true, nil
}

// notePushSucceeded records a push the remote took, or already had. The caller owns repoMu.
func (w *BranchWorker) notePushSucceeded(
	outcome PushOutcome, rootBranch plumbing.ReferenceName, rootHash plumbing.Hash, commits int,
) {
	// A push that returns without an error is an OBSERVATION of the remote, on the
	// connection it was opening anyway: the server took the ref update, the advertisement
	// already showed the branch at our head, or it did not carry the branch at all. All
	// three say where the branch is, so all three take trust — including the push that
	// CREATED the branch, which the old branchExists guard (written by fetches alone)
	// wrongly excluded.
	//
	// This must NOT clear worktreeDirty. An earlier write can have failed part-way
	// through executePendingWrites and left staged changes behind while this write was
	// retained; a successful push says where the remote is, and nothing about that.
	w.setBaseTrusted(true)
	// The half a fetch-only record could not carry: on an active branch the push is the
	// event that MOVES the revision.
	w.recordPushObservation(outcome, rootBranch, rootHash)
	w.endPushCycle(outcome.Kind)
	w.firsts.push.Do(func() {
		w.Log.Info("First push to remote completed",
			"branch", w.Branch,
			// String(), never the raw URL: spec.url takes userinfo, and this line is at
			// default verbosity.
			"repository", w.repo.String(),
			"commits", commits)
	})
}

// replayOntoRemote syncs to the remote, rebuilds the pending writes on what it found, and makes
// that the cycle's root. The caller owns repoMu and has recorded the fetch.
func (w *BranchWorker) replayOntoRemote(
	ctx context.Context,
	repo *gogit.Repository,
	pendingWrites []PendingWrite,
	auth []gitclient.Option,
) error {
	// Marked BEFORE the reset: see markCheckoutReset.
	w.markCheckoutReset()
	pullReport, err := w.syncToRemote(ctx, repo, auth)
	if err != nil {
		w.invalidateBase("sync before replay failed")
		return fmt.Errorf("sync remote before replay: %w", err)
	}
	w.updateBranchMetadataFromPullReport(pullReport)

	// The replay re-PLANS against the rebased worktree, so it can decide deletions the first apply
	// never made, under whatever policy was in force when the write was planned. Re-check the policy
	// before that happens, so an operator who tightened it in the meantime is obeyed.
	rootBranch, rootHash, err := w.ensureWriteBranch(repo)
	if err == nil {
		err = w.tightenPendingPruneModes(ctx, pendingWrites)
	}
	held := 0
	if err == nil {
		// A write the new tree refuses is skipped and stamped, not a failure of the replay: see
		// replayPendingWrites. The checkout holds the others, and the loop settles the skipped one.
		held, err = w.replayPendingWrites(ctx, repo, pendingWrites)
	}
	if err != nil {
		w.invalidateBase("rebuild before replay failed")
		return fmt.Errorf("rebuild pending writes: %w", err)
	}
	w.pushCycleRootBranch = rootBranch
	w.pushCycleRootHash = rootHash
	w.pushCycleRootGen = w.baseParentGen.Load()
	w.checkoutApplied.Store(int64(held))
	return nil
}

// admitPushRoot reports whether a push rooted on rootBranch may leave under the current parent
// configuration: the root was chosen under it, or the root is the write branch itself, which a
// parent change cannot move. In that second case it re-stamps the root, at no cost. The caller
// owns repoMu.
func (w *BranchWorker) admitPushRoot(rootBranch plumbing.ReferenceName) bool {
	gen := w.parentSnapshot().gen
	if w.pushCycleRootGen == gen {
		return true
	}
	if rootBranch != plumbing.NewBranchReferenceName(w.Branch) || !w.writeBranchProvenPresent() {
		return false
	}
	w.pushCycleRootGen = gen
	w.restampParentGenIfIndependent()
	return true
}

// writeBranchProvenPresent reports whether the last observation of the remote found the write
// branch. A parent change cannot alter a cycle rooted on an existing write branch, and the push's
// compare-and-swap still guards the branch if the observation went stale.
func (w *BranchWorker) writeBranchProvenPresent() bool {
	observed := w.lastObservation.Load()
	return observed != nil && observed.Commit != ""
}

// restampParentGenIfIndependent carries trust across a parent change when the checkout does not
// depend on the parent: the write branch exists, and HEAD is not a local branch still rooted on a
// parent. It costs no connection, which is what keeps an existing write branch's cycles free.
func (w *BranchWorker) restampParentGenIfIndependent() {
	if w.newBranchParent == "" && w.writeBranchProvenPresent() {
		w.baseParentGen.Store(w.parentSnapshot().gen)
	}
}

// rootParentStale reports whether the retained cycle's root was chosen under an older parent
// configuration, and the root depends on the parent.
func (w *BranchWorker) rootParentStale() bool {
	w.repoMu.Lock()
	defer w.repoMu.Unlock()
	if w.pushCycleRootGen == w.parentSnapshot().gen {
		return false
	}
	return w.pushCycleRootBranch != plumbing.NewBranchReferenceName(w.Branch) || !w.writeBranchProvenPresent()
}

// remoteMovedDuringPush reports whether a failed push was contention — the remote branch no
// longer sitting at the hash our commits were based on — which is the only push failure a replay
// can fix. When it finds the remote at the commits the push sent, it reports the push as landed
// instead: the failure lost only the answer.
//
// It prefers the rejection's own advertisement and reaches the network only when the push
// produced none: a dropped connection, a deadline, an auth failure, a server-side refusal. That
// fallback is a whole SmartFetch, which is why it is the exception rather than the path. It is
// also the only path that can find a landed push: a rejection's advertisement was compared
// against our head already, and an advertisement at our head is never a rejection.
//
// A fetch that itself fails is reported as "not moved". The caller surfaces the original push
// error either way, and replaying onto a tree nobody has read would be guessing at contention
// rather than observing it.
func (w *BranchWorker) remoteMovedDuringPush(
	ctx context.Context,
	repo *gogit.Repository,
	pushErr error,
	rootBranch plumbing.ReferenceName,
	rootHash plumbing.Hash,
	auth []gitclient.Option,
) (bool, *PushOutcome) {
	// A push that creates the write branch is rooted on its parent, so a rejection naming the
	// write branch itself is not about the root: somebody created the branch since we based our
	// work. That is contention all the same, and the replay builds on their commit.
	var moved *RemoteMovedError
	if errors.As(pushErr, &moved) && moved.Branch != rootBranch &&
		moved.Branch == plumbing.NewBranchReferenceName(w.Branch) {
		return true, nil
	}
	// An empty repository gained refs since the cycle was planned on it. The replay's fetch
	// resolves the new default branch, or refuses, and the rebuild re-plans from the retained
	// writes instead of reusing the orphan commits.
	var notEmpty *RepositoryNotEmptyError
	if errors.As(pushErr, &notEmpty) {
		return true, nil
	}

	remoteHash, known := advertisedRootHash(pushErr, rootBranch)
	if known {
		return remoteHash != rootHash, nil
	}
	if rootBranch != plumbing.NewBranchReferenceName(w.Branch) {
		return w.newBranchMovedDuringPush(ctx, repo, rootBranch, rootHash, auth)
	}
	w.recordFetch(fetchReasonPushFailureProbe)
	remoteHash, fetchErr := fetchRemoteBranchHashFn(ctx, repo, rootBranch, auth)
	if fetchErr != nil {
		w.invalidateBase("remote-state fetch failed")
		return false, nil
	}
	if landed := landedPush(repo, remoteHash, rootHash); landed != nil {
		return false, landed
	}
	return remoteHash != rootHash, nil
}

// landedPush reports a push that reached the remote although its caller saw it fail: the branch
// is now at our local head, and that head is not the root the cycle started from (a cycle with
// nothing to send has nothing to have landed). Only the exact head counts. A remote that moved on
// past our commits would also hold them, but proving that needs ancestry a depth-1 fetch does not
// carry, so that case replays, as any other contention does.
func landedPush(repo *gogit.Repository, remoteHash, rootHash plumbing.Hash) *PushOutcome {
	head := localHead(repo)
	if head.IsZero() || head == rootHash || remoteHash != head {
		return nil
	}
	return &PushOutcome{Kind: PushAccepted, Head: head}
}

// newBranchMovedDuringPush is the fallback for a push that was creating the write branch and failed
// without an advertisement to read: a dropped connection, or the server refusing an upload whose
// Old was zero because somebody created the branch after our advertisement. Probing the root alone
// would find the parent unmoved and conclude nothing happened, and the work would never replay. So
// it reads both refs from one advertisement, and a write branch that is now there is contention,
// exactly like the typed rejection naming it — unless it is there at our own head, which is our
// push having landed. It runs only on this exceptional path.
func (w *BranchWorker) newBranchMovedDuringPush(
	ctx context.Context,
	repo *gogit.Repository,
	rootBranch plumbing.ReferenceName,
	rootHash plumbing.Hash,
	auth []gitclient.Option,
) (bool, *PushOutcome) {
	advertisement, err := advertiseRemoteBranchFn(ctx, w.repo.URL, plumbing.NewBranchReferenceName(w.Branch),
		rootBranch.Short(), auth)
	if err != nil {
		w.invalidateBase("remote-state probe failed")
		return false, nil
	}
	if !advertisement.branch.IsZero() {
		if landed := landedPush(repo, advertisement.branch, rootHash); landed != nil {
			return false, landed
		}
		w.Log.Info("The write branch was created by somebody else during our push; replaying onto it",
			"branch", w.Branch)
		return true, nil
	}
	return advertisement.parent != rootHash, nil
}

// advertisedRootHash reads the remote's own answer out of a failed push, when the push got one.
//
// A compare-and-swap rejection is the remote telling us where the branch is: validatePushState
// compared the advertised hash for the cycle's root branch, so the number the replay needs was
// already on the wire. Learning it again with fetchRemoteBranchHash is a whole SmartFetch — two
// more requests to the Git host on the most expensive path in the system — for a fact we had.
//
// The branch check is not ceremony. The error carries the branch its advertisement was read for,
// and the caller's rootBranch can differ from it across a retry (a new branch is rooted on the
// default branch until the first push creates it), so a hash for a different ref would be a wrong
// answer rather than a missing one.
func advertisedRootHash(err error, rootBranch plumbing.ReferenceName) (plumbing.Hash, bool) {
	var moved *RemoteMovedError
	if !errors.As(err, &moved) || moved.Branch != rootBranch {
		return plumbing.ZeroHash, false
	}
	return moved.Advertised, true
}

// refreshRemoteAndRebuildPendingWrites moves the local checkout to the current remote tip, then
// replays retained pending writes on top of it without pushing, so whatever judges the tree next
// judges the newest remote one instead of a stale local clone.
//
// reason is the fetch series this costs. It is a parameter rather than a constant because three
// different things end up here — a forced recheck, a dirty-worktree recovery, and the snapshot a
// resync judges against — and hard-wiring one of them made the OTHER two report as forced
// rechecks. That split the `recovery` series in half on the one axis it must not depend on:
// whether writes happened to be retained. A dirty worktree recovered with work in hand and one
// recovered without it are the same event, and §12's "a climbing recovery series is a bug report"
// only means anything if they land in the same series.
func (w *BranchWorker) refreshRemoteAndRebuildPendingWrites(
	ctx context.Context,
	pendingWrites []PendingWrite,
	reason string,
) error {
	w.repoMu.Lock()
	defer w.repoMu.Unlock()

	if len(pendingWrites) == 0 {
		return nil
	}

	// Whatever brought us here, the base is no longer something we can claim to know. The sync
	// below re-establishes it.
	w.invalidateBase(reason)

	provider, err := w.getGitProvider(ctx)
	if err != nil {
		return fmt.Errorf("get GitProvider: %w", err)
	}

	auth, err := getAuthFromSecret(ctx, w.Client, provider, w.sshHostKeys, w.credentialPolicy)
	if err != nil {
		return fmt.Errorf("resolve auth: %w", err)
	}

	repoPath := w.repoPath()
	repo, err := gogit.PlainOpen(repoPath)
	if err != nil {
		return fmt.Errorf("open repository: %w", err)
	}

	w.recordFetch(reason)
	return w.replayOntoRemote(ctx, repo, pendingWrites, auth)
}

// tightenPendingPruneModes lowers every retained write's captured prune mode before replay.
// Tightening only: see PruneMode.MoreRestrictiveOf for why loosening must NOT propagate.
//
// It mutates the Targets map in place, which is the point: the map is shared with the retained
// PendingWrite, so one pass covers both deletion paths and survives every push attempt.
//
// The two failure modes differ on purpose. GONE is a definite answer, so the write replays under
// the most restrictive mode. A FAILED read is no answer, so the error leaves the writes retained
// for the next cycle — guessing either way would drop a legitimate delete or apply an
// unauthorized one.
func (w *BranchWorker) tightenPendingPruneModes(ctx context.Context, pendingWrites []PendingWrite) error {
	current := map[pendingTargetKey]configv1alpha3.PruneMode{}
	// Ranged by value on purpose: Targets is a map, so writing through this copy still updates the
	// retained write's own map — which is exactly the sharing this relies on.
	for _, pendingWrite := range pendingWrites {
		for key, md := range pendingWrite.Targets {
			mode, cached := current[key]
			if !cached {
				target, err := w.getGitTarget(ctx, key.Name, key.Namespace)
				switch {
				case apierrors.IsNotFound(err):
					mode = configv1alpha3.PruneNever
				case err != nil:
					return fmt.Errorf("re-read prune policy for %s/%s before replay: %w", key.Namespace, key.Name, err)
				default:
					mode = target.EffectivePruneMode()
				}
				current[key] = mode
			}
			tightened := md.PruneMode.MoreRestrictiveOf(mode)
			if tightened == md.PruneMode.OrDefault() {
				continue
			}
			w.Log.Info("prune policy tightened since this write was planned; replaying under the stricter mode",
				"gitTarget", key.Namespace+"/"+key.Name,
				"planned", string(md.PruneMode.OrDefault()), "replaying", string(tightened))
			md.PruneMode = tightened
			pendingWrite.Targets[key] = md
		}
	}
	return nil
}

func (w *BranchWorker) estimateEventsSize(events []Event) int64 {
	var total int64
	for i := range events {
		total += eventPayloadSize(&events[i])
	}
	return total
}

// estimateObjectSize approximates the serialized YAML size for an event's object.
func estimateObjectSize(ev *Event) int64 {
	if ev.Object == nil {
		return 0
	}
	if b, err := sanitize.MarshalToOrderedYAML(ev.Object); err == nil {
		return int64(len(b))
	}
	return 0
}

func (w *BranchWorker) getGitTarget(
	ctx context.Context,
	targetName string,
	targetNamespace string,
) (*configv1alpha3.GitTarget, error) {
	targetKey := types.NamespacedName{Name: targetName, Namespace: targetNamespace}
	var target configv1alpha3.GitTarget
	if err := w.Client.Get(ctx, targetKey, &target); err != nil {
		return nil, fmt.Errorf("failed to get GitTarget %s: %w", targetKey, err)
	}
	return &target, nil
}

func fetchRemoteBranchHash(
	ctx context.Context,
	repo *gogit.Repository,
	branch plumbing.ReferenceName,
	auth []gitclient.Option,
) (plumbing.Hash, error) {
	fetched, err := SmartFetch(ctx, repo, branch, auth)
	if err != nil {
		return plumbing.ZeroHash, err
	}
	if fetched != branch {
		// SmartFetch fell back to the remote's default branch, which is how it reports that the
		// target branch is not there. It built no refspec for it, so prune left
		// refs/remotes/origin/<branch> exactly where it was, and reading that ref now would
		// report a branch that no longer exists — at the hash it held before somebody deleted it.
		// Zero is what the remote is actually advertising.
		return plumbing.ZeroHash, nil
	}

	remoteRef, err := repo.Reference(plumbing.NewRemoteReferenceName("origin", branch.Short()), true)
	if errors.Is(err, plumbing.ErrReferenceNotFound) {
		return plumbing.ZeroHash, nil
	}
	if err != nil {
		return plumbing.ZeroHash, err
	}

	return remoteRef.Hash(), nil
}

func (w *BranchWorker) ensureWriteBranch(repo *gogit.Repository) (plumbing.ReferenceName, plumbing.Hash, error) {
	targetBranch := plumbing.NewBranchReferenceName(w.Branch)
	baseBranch, baseHash, err := GetCurrentBranch(repo)
	if err != nil {
		return "", plumbing.ZeroHash, fmt.Errorf("get current branch: %w", err)
	}

	if baseBranch != targetBranch {
		if err := switchOrCreateBranch(repo, targetBranch, w.Log, w.Branch, baseHash); err != nil {
			return "", plumbing.ZeroHash, err
		}
		// HEAD was on another branch, which happens only when a fetch found the write branch
		// absent and checked out its parent instead. Remember it, so the next cycle on this local
		// branch is still rooted on the parent.
		w.newBranchParent, w.newBranchParentHash = baseBranch, baseHash
		return baseBranch, baseHash, nil
	}

	// Already on the local write branch, with nothing committed on top of the parent it was
	// created from: the remote still does not carry it, so the parent remains the root. Only while
	// the parent configuration is the one that checkout was made under: after a change, the
	// remembered parent is the old one, and rooting on it would let the push create the branch
	// there. The root then names the local branch, which the admission check refuses.
	if w.newBranchParent != "" && baseHash == w.newBranchParentHash &&
		w.baseParentGen.Load() == w.parentSnapshot().gen {
		return w.newBranchParent, w.newBranchParentHash, nil
	}

	return baseBranch, baseHash, nil
}

// recordPushObservation records what a push that returned without an error observed.
func (w *BranchWorker) recordPushObservation(
	outcome PushOutcome, rootBranch plumbing.ReferenceName, rootHash plumbing.Hash,
) {
	revision := ""
	if !outcome.Head.IsZero() {
		revision = outcome.Head.String()
	}
	if outcome.Kind == PushNoBranch && rootBranch != plumbing.NewBranchReferenceName(w.Branch) {
		// Settled without creating the branch, after confirming its parent against this
		// advertisement: the root IS the parent's tip.
		w.recordAbsentBranch(rootBranch.Short(), rootHash.String(), ObservedByPush)
	} else {
		w.recordRemoteObservation(revision, ObservedByPush)
	}
	w.Log.V(1).Info("Remote observed by push",
		"branch", w.Branch, "outcome", string(outcome.Kind), "head", revision)
}

// endPushCycle clears the cycle's root after a push the remote answered. Only a push that left the
// branch absent keeps the parent it would be created from.
func (w *BranchWorker) endPushCycle(kind PushOutcomeKind) {
	w.pushCycleRootBranch = ""
	w.pushCycleRootHash = plumbing.ZeroHash
	if kind != PushNoBranch {
		w.clearNewBranchParent()
	}
}

// parentConfig is one immutable snapshot of the parent configuration. Every operation that
// consults the parent loads it once, at its start, and uses that name and generation throughout:
// nothing re-reads the current name after a network call returns, so an old lookup can never
// report a result about a newly configured parent.
type parentConfig struct {
	// name is spec.parentBranch; empty means the remote's default branch.
	name string
	// gen rises on every change. Zero is the configuration a worker starts with.
	gen uint64
}

// parentSnapshot returns the current parent configuration.
func (w *BranchWorker) parentSnapshot() parentConfig {
	if p := w.parentState.Load(); p != nil {
		return *p
	}
	return parentConfig{}
}

// ParentBranch is the configured parent branch, or empty for the remote's default branch.
func (w *BranchWorker) ParentBranch() string { return w.parentSnapshot().name }

// SetParentBranch configures the parent branch, and reports whether the value changed. A change
// raises the generation, which is what every later decision compares against: trust earned under
// the old one no longer counts, a cycle planned under it is rebuilt before its push is admitted,
// and retained writes are replayed on the new parent. The worker and its clone are kept, and an
// existing write branch, which a parent cannot move, keeps its trust at no cost.
// spec.parentBranch is immutable, so a change reaches a live worker only when the targets on its
// branch were replaced.
func (w *BranchWorker) SetParentBranch(name string) bool {
	current := w.parentSnapshot()
	if current.name == name {
		return false
	}
	w.parentState.Store(&parentConfig{name: name, gen: current.gen + 1})
	w.Log.V(1).Info("Parent branch changed", "branch", w.Branch, "parentBranch", name)
	return true
}

// prepareBranch is PrepareBranch for this worker's branch and parent. A configured parent the
// remote does not carry is recorded as an observation, because it is one: the advertisement said
// so, and the GitTarget reports it as ParentBranchNotFound.
func (w *BranchWorker) prepareBranch(
	ctx context.Context, repoPath string, auth []gitclient.Option,
) (*PullReport, error) {
	parent := w.parentSnapshot()
	report, err := PrepareBranchFrom(ctx, w.repo.URL, repoPath, w.Branch, parent.name, auth)
	w.noteFetchParent(parent, err)
	return report, err
}

// syncToRemote is syncToRemoteFn for this worker's branch and parent; see prepareBranch.
func (w *BranchWorker) syncToRemote(
	ctx context.Context, repo *gogit.Repository, auth []gitclient.Option,
) (*PullReport, error) {
	parent := w.parentSnapshot()
	report, err := syncToRemoteFn(ctx, repo, plumbing.NewBranchReferenceName(w.Branch), parent.name, auth)
	w.noteFetchParent(parent, err)
	return report, err
}

// noteFetchParent records what a fetch under one parent snapshot proved: on success, that the
// checkout now matches that configuration; on a missing parent, the observation, naming the
// parent the fetch actually asked for.
func (w *BranchWorker) noteFetchParent(parent parentConfig, err error) {
	if err == nil {
		w.baseParentGen.Store(parent.gen)
		return
	}
	var unresolved *DefaultBranchUnresolvedError
	switch {
	case errors.As(err, &unresolved):
		w.recordMissingParent(parent.name, unresolved.Head)
	case errors.Is(err, ErrParentBranchNotFound):
		w.recordMissingParent(parent.name, parent.name)
	}
}

// clearNewBranchParent forgets the parent once the remote carries the write branch.
func (w *BranchWorker) clearNewBranchParent() {
	w.newBranchParent = ""
	w.newBranchParentHash = plumbing.ZeroHash
}

// Push cycle outcomes and retry reasons.
//
// A push CYCLE is one call to pushPendingCommits, replay retries included, and it ends exactly
// once: `pushed` when the remote took the work, `failed` when it did not. A retry is not an
// outcome, which is why it is counted separately — a cycle that replays twice and succeeds is one
// `pushed` and two retries, and rate(retries)/rate(pushes) is the contention signal.
const (
	pushOutcomePushed = "pushed"
	pushOutcomeFailed = "failed"

	pushRetryRemoteMoved = "remote_moved"
)

// Fetch reasons. Every call that runs SmartFetch is counted under exactly one of these, and the
// set is deliberately split finer than the call sites: `publication` and `recovery` both come out
// of commitPendingWrites, and keeping them apart is what lets the steady-state claim be asserted
// at zero without the assertion failing the first time a push fails.
const (
	// fetchReasonPublication is the fetch at the head of a publication cycle.
	fetchReasonPublication = "publication"
	// fetchReasonRecovery is a cycle that had to re-establish a base it could not trust.
	//
	// Kept apart from `publication` so the steady-state claim can be asserted at zero without the
	// assertion failing the first time a push fails. A `recovery` series that climbs is a bug
	// report, not a cost.
	fetchReasonRecovery = "recovery"
	// fetchReasonContention is the reset onto the remote tip after a push was rejected because
	// somebody else moved the branch. One confirmed rejection is exactly one of these.
	fetchReasonContention = "contention"
	// fetchReasonRefresh is the periodic top-up of an idle branch's view of the remote: the
	// refresher found the branch somewhere other than the checkout and reset onto it. The
	// advertisement that precedes it is NOT counted here, because this counter's documented
	// meaning is every call that reads the remote through a SmartFetch.
	fetchReasonRefresh = "refresh"
	// fetchReasonPushFailureProbe is the fallback lookup after a push that failed WITHOUT the
	// remote ever saying where the branch is — a dropped connection, an auth failure, a
	// server-side refusal.
	//
	// It is not contention, and counting it as such was wrong: an auth failure would inflate the
	// series an operator reads as "other writers are fighting me over this branch", on a target
	// with no other writers at all. The probe often finds the remote unmoved, in which case no
	// reset follows and this is the only fetch the failure costs.
	fetchReasonPushFailureProbe = "push_failure_probe"
	// fetchReasonForcedRecheck is an operator or controller asking the worker to re-read Git.
	fetchReasonForcedRecheck = "forced_recheck"
)

// Commit-failure reasons. A failure here is a window that was accepted, routed, and then died before
// it could reach the remote: the window is dropped and its events are gone until the next resync
// re-derives them.
const (
	// commitFailureRefused is a plan the acceptance gate or a write-boundary precondition
	// rejected. It is also surfaced as a GitTarget condition, but a condition is not a rate.
	commitFailureRefused = "refused"
	// commitFailureError is any other write fault.
	commitFailureError = "error"
)

// Queue-drop kinds. The set covers every item that can be refused by a full queue.
const (
	queueRefusalWrite   = "write"
	queueRefusalAttach  = "attach"
	queueRefusalResync  = "resync"
	queueRefusalRefresh = "refresh"
)

// commitLabels is the {author_kind, message_source} pair one commit is counted under.
type commitLabels struct {
	authorKind    string
	messageSource string
}

// publishCommitsForPush counts the commits a successful push put on the remote, by
// {author_kind, message_source}.
//
// It reads the pending writes as they stand after the push. executePendingWrites stamps each
// write's CommitSHA in place, so a write a conflict replay rebuilt to nothing has a zero SHA and is
// not counted: this is the commits that reached the remote, not the ones that were created.
func (w *BranchWorker) publishCommitsForPush(pendingWrites []PendingWrite) {
	if telemetry.GitCommitsTotal == nil {
		return
	}
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	counts := map[commitLabels]int64{}
	for _, pendingWrite := range pendingWrites {
		// A write the last replay refused was not pushed: the loop settles it as a refusal.
		if !pendingWrite.createdCommit() || pendingWrite.replayRefusal != nil {
			continue
		}
		counts[commitLabels{
			authorKind:    pendingWrite.authorKind(),
			messageSource: pendingWrite.messageSource(),
		}]++
	}
	for labels, count := range counts {
		// Label by the recording BranchWorker's GitBranch {gitbranch_namespace, gitbranch_name}
		// plus author_kind and message_source. The prefixed key names avoid the reserved
		// Prometheus pod-scrape labels `namespace`/`name`.
		telemetry.GitCommitsTotal.Add(ctx, count, metric.WithAttributes(w.gitBranchAttrs(
			attribute.String("author_kind", labels.authorKind),
			attribute.String("message_source", labels.messageSource),
		)...))
	}
}

// recordPushOutcome closes out one push cycle: its terminal outcome and its wall time.
//
// The duration spans every replay retry on purpose. What an operator needs is how long it took the
// mirror to accept the work, not how fast one attempt was, and a cycle that replayed three times
// took what it took.
func (w *BranchWorker) recordPushOutcome(outcome string, started time.Time) {
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	if telemetry.GitPushesTotal != nil {
		telemetry.GitPushesTotal.Add(ctx, 1,
			metric.WithAttributes(w.gitBranchAttrs(attribute.String("outcome", outcome))...))
	}
	if telemetry.GitPushDurationSeconds != nil {
		telemetry.GitPushDurationSeconds.Record(ctx, time.Since(started).Seconds(),
			metric.WithAttributes(w.gitBranchAttrs()...))
	}
}

// recordPushRetry counts one replay round inside a push cycle.
func (w *BranchWorker) recordPushRetry(reason string) {
	if telemetry.GitPushRetriesTotal == nil {
		return
	}
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	telemetry.GitPushRetriesTotal.Add(ctx, 1,
		metric.WithAttributes(w.gitBranchAttrs(attribute.String("reason", reason))...))
}

// recordFetch counts one call that reaches the remote through SmartFetch.
//
// It is recorded BEFORE the call, not after it: the connection to the Git host is spent whether
// or not the fetch succeeds, and this counter measures what we asked of the remote, not what came
// back. A fetch that fails is exactly the one an operator most wants counted.
func (w *BranchWorker) recordFetch(reason string) {
	if telemetry.GitFetchesTotal == nil {
		return
	}
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	telemetry.GitFetchesTotal.Add(ctx, 1,
		metric.WithAttributes(w.gitBranchAttrs(attribute.String("reason", reason))...))
}

// gitBranchAttrs is the {gitbranch_namespace, gitbranch_name, provider_namespace, provider_name}
// identity every Git-side metric this worker records is labelled by. The GitBranch it writes for
// matches a series to that object's conditions and `kubectl get gitbranch`; its GitProvider groups
// the branches one credential writes, which is how an expired credential shows. The provider adds
// no series: it is immutable on the GitBranch, and a worker serves one GitBranch. spec.branch is
// left off for the same reason and because the GitBranch already names it. The keys are prefixed,
// never bare namespace/name: a Prometheus pod scrape with honor_labels=false renames a bare
// `namespace` to exported_namespace and puts the scraped pod's own in its place, which silently
// breaks every per-branch selector.
func (w *BranchWorker) gitBranchAttrs(extra ...attribute.KeyValue) []attribute.KeyValue {
	return append([]attribute.KeyValue{
		attribute.String("gitbranch_namespace", w.GitBranch.Namespace),
		attribute.String("gitbranch_name", w.GitBranch.Name),
		attribute.String("provider_namespace", w.GitProviderNamespace),
		attribute.String("provider_name", w.GitProviderRef),
	}, extra...)
}

// recordCommitFailure counts one window that died between routing and pushing: its
// events are lost until the next resync re-derives them. An unreachable remote is not one: the
// decided write stays pending for the publication retry.
//
// `refused` and `error` need different people. A refusal is a Git path a human has to fix and will
// not clear on its own; an error may be transient.
func (w *BranchWorker) recordCommitFailure(reason string) {
	if telemetry.GitCommitFailuresTotal == nil {
		return
	}
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	telemetry.GitCommitFailuresTotal.Add(ctx, 1,
		metric.WithAttributes(w.gitBranchAttrs(attribute.String("reason", reason))...))
}

// Commit window timer sources, reported as the commit_windows_total `timer_source` label.
const (
	windowTimerSourceTarget        = "target"
	windowTimerSourceCommitRequest = "commit_request"
)

// metricLabel is the commit_windows_total `close_reason` value for this reason. The log field keeps
// its hyphenated form; a label value follows the snake_case the other labels use.
func (r windowFinalizeReason) metricLabel() string {
	switch r {
	case windowFinalizeReasonIdleTimeout:
		return "idle_timeout"
	case windowFinalizeReasonMaxDuration:
		return "max_duration"
	case windowFinalizeReasonAttachNext:
		return "attach_next"
	case windowFinalizeReasonIdentityChange:
		return "identity_change"
	case windowFinalizeReasonBufferLimit:
		return "buffer_limit"
	case windowFinalizeReasonResyncBeforeApply:
		return "resync_before_apply"
	case windowFinalizeReasonShutdown:
		return "shutdown"
	case windowFinalizeReasonForeignCommit:
		return "foreign_commit"
	case windowFinalizeReasonApplyDue:
		return "apply_due"
	case windowFinalizeReasonUnspecified:
		return "unspecified"
	default:
		return "unspecified"
	}
}

// recordWindowClosed counts one closed commit window and observes how long it collected. The
// labels are the window's GitTarget, because the timers that shape it are that target's (or an
// attached request's), not the branch's.
func (w *BranchWorker) recordWindowClosed(window *openWindow, reason windowFinalizeReason) {
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	target := []attribute.KeyValue{
		attribute.String("gittarget_namespace", window.GitTargetNamespace),
		attribute.String("gittarget_name", window.GitTarget),
	}
	if telemetry.GitCommitWindowsTotal != nil {
		source := windowTimerSourceTarget
		if window.pendingCR != nil {
			source = windowTimerSourceCommitRequest
		}
		telemetry.GitCommitWindowsTotal.Add(ctx, 1, metric.WithAttributes(append(target,
			attribute.String("close_reason", reason.metricLabel()),
			attribute.String("timer_source", source),
		)...))
	}
	if telemetry.GitCommitWindowDurationSeconds != nil {
		telemetry.GitCommitWindowDurationSeconds.Record(ctx, w.now().Sub(window.openedAt).Seconds(),
			metric.WithAttributes(target...))
	}
}

// recordQueueRefusal counts one item refused at enqueue: the queue was full, intake was paused, or the
// worker was stopping.
//
// The producer keeps a refused item and offers it again: the watch delivers a live write again from
// its unadvanced cursor, and the controller re-sends an attach on its next poll.
func (w *BranchWorker) recordQueueRefusal(kind string) {
	if telemetry.GitQueueRefusalsTotal == nil {
		return
	}
	ctx := w.ctx
	if ctx == nil {
		ctx = context.Background()
	}
	telemetry.GitQueueRefusalsTotal.Add(ctx, 1,
		metric.WithAttributes(w.gitBranchAttrs(attribute.String("kind", kind))...))
}

// getGitProvider fetches the GitProvider for this worker.
func (w *BranchWorker) getGitProvider(ctx context.Context) (*configv1alpha3.GitProvider, error) {
	var provider configv1alpha3.GitProvider
	namespacedName := types.NamespacedName{
		Name:      w.GitProviderRef,
		Namespace: w.GitProviderNamespace,
	}

	if err := w.Client.Get(ctx, namespacedName, &provider); err != nil {
		return nil, fmt.Errorf("failed to fetch GitProvider: %w", err)
	}

	// The worker writes to ITS repository and reads this object for the credentials, commit
	// identity and signing to write it with. Those two must describe the same repository, or the
	// worker would present a credential issued for one remote to another: a GitProvider recreated
	// against a different URL is normally replaced at the next reconcile, but a replacement that
	// cannot pass the Validated gate never arrives, and the old worker would otherwise go on
	// mirroring into a repository the operator has abandoned, using the new one's credential.
	//
	// So a mismatch is fatal to the cycle rather than something to work around. The worker stops
	// acting and says why; the reconcile that replaces it is what resolves it. A worker with no
	// identity of its own (the CLI, and tests that never reach a remote) has nothing to compare.
	if !w.repo.IsZero() {
		if found := (RepoIdentity{ProviderUID: provider.UID, URL: provider.Spec.URL}); found != w.repo {
			return nil, fmt.Errorf(
				"GitProvider %s/%s now names %s; this worker is for %s and is waiting to be replaced",
				w.GitProviderNamespace, w.GitProviderRef, found, w.repo)
		}
	}

	return &provider, nil
}

// commitWindowFor returns the commit-window duration for ONE GitTarget.
//
// commitWindowFor resolves a GitTarget's commit window timers.
//
// The window is a GitTarget field but this worker serves every target sharing a (provider,
// branch), so resolving per target is what makes the field mean what it says. Affordable because a
// window is bound to one target already, so this is read once per window, not per event.
//
// There is nothing to parse and nothing to reject: the fields are metav1.Durations behind a
// duration pattern, so a malformed value never reaches storage. An unreadable GitTarget takes the
// fallback, since a missing target is no reason to change how the events in hand are batched.
func (w *BranchWorker) commitWindowFor(
	ctx context.Context,
	targetName, targetNamespace string,
	fallback commitWindowDefaults,
) commitWindowDefaults {
	// No client is legitimate — the CLI and the narrower unit tests run a worker with none — and it
	// means there is no GitTarget to ask, not that the default is wrong.
	if w.Client == nil || targetName == "" || targetNamespace == "" {
		return fallback
	}
	target, err := w.getGitTarget(ctx, targetName, targetNamespace)
	if err != nil {
		w.Log.V(1).Info("Could not read GitTarget for its commit window, using the default",
			"gitTarget", targetNamespace+"/"+targetName, "error", err.Error())
		return fallback
	}
	resolved := fallback
	if target.Spec.Commit == nil || target.Spec.Commit.Window == nil {
		return resolved
	}
	if idle := target.Spec.Commit.Window.IdleTimeout; idle != nil {
		resolved.idleTimeout = idle.Duration
	}
	if maxDuration := target.Spec.Commit.Window.MaxDuration; maxDuration != nil {
		resolved.maxDuration = maxDuration.Duration
	}
	return resolved
}

// LastRemoteObservation returns what the worker last proved about the target branch, and whether
// it has proved anything at all. It reads no repository and opens no connection.
func (w *BranchWorker) LastRemoteObservation() (RemoteObservation, bool) {
	observed := w.lastObservation.Load()
	if observed == nil {
		return RemoteObservation{}, false
	}
	return *observed, true
}

// recordRemoteObservation stores what a confirmed look at the remote learned.
//
// It is called from both producers — a fetch that reset onto the tip, and a push the server
// accepted — because both prove the same kind of fact. An error path must NOT call it: a push
// that died mid-upload or an advertisement that never arrived observed nothing, and recording a
// guess there is how a stale revision reaches status.
//
// An empty revision is a branch the remote does not carry, observed without a parent to name: an
// empty repository, which leaves the first commit unborn.
func (w *BranchWorker) recordRemoteObservation(revision string, by ObservationSource) {
	if revision == "" {
		w.recordAbsentBranch("", "", by)
		return
	}
	w.publishObservation(RemoteObservation{Commit: revision, At: time.Now(), By: by, Repo: w.repo})
}

// recordMissingParent records that the write branch is absent and so is the parent it would be
// created from. requested is spec.parentBranch as the observing operation read it (empty for the
// remote's default branch); branch is the parent that is missing: the configured one, or the
// branch an unresolved HEAD names. Only a look at the remote can prove it, so it is by Fetch.
func (w *BranchWorker) recordMissingParent(requested, branch string) {
	w.publishObservation(RemoteObservation{
		At: time.Now(), By: ObservedByFetch, Repo: w.repo,
		ParentState: ParentMissing, ParentRequested: requested, ParentBranch: branch,
	})
}

// recordAbsentBranch records that the write branch is absent, and where the parent it would be
// created from is. An empty parent is an empty repository, which has none.
func (w *BranchWorker) recordAbsentBranch(parentBranch, parentCommit string, by ObservationSource) {
	state := ParentFound
	if parentBranch == "" {
		state, parentCommit = ParentUnborn, ""
	}
	w.publishObservation(RemoteObservation{
		At: time.Now(), By: by, Repo: w.repo,
		ParentState: state, ParentBranch: parentBranch, ParentCommit: parentCommit,
	})
}

func (w *BranchWorker) publishObservation(observed RemoteObservation) {
	// Stamped with the repository it is about, so that whoever publishes it can tell whether it
	// still describes the repository the GitTarget points at. See RemoteObservation.Repo.
	w.lastObservation.Store(&observed)
	w.inbox.observe(observed)
	// Delivered here, at the one point where an observation is made, rather than by each caller
	// against whichever GitTargets it happened to be holding. The fact is about the BRANCH, so it
	// goes to the branch, and every target on it reads the same answer with the time it was
	// actually proved.
	if w.remoteReporter != nil {
		w.remoteReporter(observed)
	}
}

// syncWithRemote fetches latest changes from remote and resets onto them. It is the
// no-retained-writes half of materialize.
//
// reason is a parameter for the same purpose it is on refreshRemoteAndRebuildPendingWrites: the
// caller knows why it is reading the remote, and a hard-wired constant here would file somebody
// else's fetch under `forced_recheck`.
func (w *BranchWorker) syncWithRemote(ctx context.Context, reason string) error {
	w.repoMu.Lock()
	defer w.repoMu.Unlock()

	provider, err := w.getGitProvider(ctx)
	if err != nil {
		return fmt.Errorf("failed to get GitProvider: %w", err)
	}

	auth, err := getAuthFromSecret(ctx, w.Client, provider, w.sshHostKeys, w.credentialPolicy)
	if err != nil {
		return fmt.Errorf("failed to get auth: %w", err)
	}

	repoPath := w.repoPath()

	// Somebody stated that the remote moved: see refreshRemoteAndRebuildPendingWrites.
	w.invalidateBase(reason)

	// PrepareBranch handles both initial and update cases
	w.recordFetch(reason)
	report, err := w.prepareBranch(ctx, repoPath, auth)
	if err != nil {
		return fmt.Errorf("failed to sync with remote: %w", err)
	}

	w.updateBranchMetadataFromPullReport(report)

	// Log if incoming changes were detected
	if report.IncomingChanges {
		w.Log.Info("Detected remote changes during sync",
			"branch", report.HEAD.ShortName,
			"newSHA", report.HEAD.Sha)
	}

	return nil
}

// updateBranchMetadataFromPullReport updates metadata from a PullReport.
func (w *BranchWorker) updateBranchMetadataFromPullReport(report *PullReport) {
	// A branch the remote does not have is recorded as an observation with no revision, which is
	// the honest answer rather than a missing one: a git branch without a commit does not exist.
	if report.ExistsOnRemote {
		w.recordRemoteObservation(report.HEAD.Sha, ObservedByFetch)
	} else {
		w.recordAbsentBranch(report.ParentBranch, report.HEAD.Sha, ObservedByFetch)
	}

	// This is the ONE place trust can be gained by a fetch, because it is the one place a fetch
	// is known to have been followed by a reset: every call site is a PrepareBranch or a
	// syncToRemote, and both paths inside syncToRemote leave a clean worktree — checkoutAndReset
	// passes Force, and makeHeadUnborn clears the index and the tree.
	//
	// Trust is gained unconditionally, including when the target branch does not exist on the
	// remote (SmartFetch fell back to the default branch) and when it is unborn. An earlier
	// version excluded both, on the reasoning that such a worktree is "not based on the target
	// branch". That reasoning inverts the invariant: the invariant is that the worktree matches
	// the remote's state for this branch, and for a branch the remote does not have, a worktree
	// based on the default branch (or an empty one) IS that state. There is nothing on the remote
	// left to learn, so a fetch per cycle could only ever return the same answer.
	//
	// What makes it safe is the push. For a branch we believe absent, the push is rooted on the
	// parent the branch is created from, and validatePushState refuses it if that parent moved or
	// if somebody has since created the branch; either rejection invalidates the base and replays
	// onto what the advertisement showed. The push then declares Old = zero, so the server also
	// refuses a branch created after the advertisement. So the cost of being wrong is one
	// rejection, not a bad write.
	w.setBaseTrusted(true)
	w.markCheckoutReset()
	if report.ExistsOnRemote {
		w.clearNewBranchParent()
	}

	// Log if this was an unborn branch
	if report.HEAD.Unborn {
		w.Log.Info("Branch is unborn (no commits yet)", "branch", report.HEAD.ShortName)
	}
}

func buildBootstrapOptions(encryptionConfig *ResolvedEncryptionConfig) pathBootstrapOptions {
	if encryptionConfig == nil || len(encryptionConfig.AgeRecipients) == 0 {
		return pathBootstrapOptions{Enabled: true}
	}

	return pathBootstrapOptions{
		Enabled:           true,
		IncludeSOPSConfig: true,
		TemplateData: bootstrapTemplateData{
			AgeRecipients: encryptionConfig.AgeRecipients,
		},
	}
}

// getAuthFromSecret is defined in helpers.go

// SeedRemoteObservationForTest marks the base trusted and records an observation made at `at`. It
// exists so a test in another package can build a worker in a known state; nothing in production
// calls it.
func (w *BranchWorker) SeedRemoteObservationForTest(revision string, at time.Time, by string) {
	w.baseTrustedState.Store(true)
	observed := RemoteObservation{Commit: revision, At: at, By: ObservationSource(by), Repo: w.repo}
	w.lastObservation.Store(&observed)
}

// RecordMissingParentForTest records the observation a fetch makes when neither the write branch
// nor its configured parent exists, for tests in another package.
func (w *BranchWorker) RecordMissingParentForTest(parent string) {
	w.recordMissingParent(parent, parent)
}

// RecordUnresolvedDefaultBranchForTest records the observation a fetch makes when the write branch
// is absent, no parent is configured, and the remote's HEAD names head, which it does not carry
// (empty: no HEAD at all), for tests in another package.
func (w *BranchWorker) RecordUnresolvedDefaultBranchForTest(head string) {
	w.recordMissingParent("", head)
}

// BaseTrustedForTest exposes the flag to tests in another package.
func (w *BranchWorker) BaseTrustedForTest() bool { return w.baseTrusted() }

// releaseInflightItem takes one item off the in-flight count, which the queue-depth gauge reads.
func (w *BranchWorker) releaseInflightItem() {
	w.inflightItems.Add(-1)
}