plan_flush.go

// SPDX-License-Identifier: Apache-2.0

package git

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"math"
	"os"
	"path"
	"path/filepath"
	"sort"
	"strings"

	gogit "github.com/go-git/go-git/v6"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"sigs.k8s.io/controller-runtime/pkg/log"
	sigsyaml "sigs.k8s.io/yaml"

	v1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3"
	"github.com/ConfigButler/gitops-reverser/internal/git/manifestedit"
	"github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer"
	"github.com/ConfigButler/gitops-reverser/internal/manifestreport"
	"github.com/ConfigButler/gitops-reverser/internal/types"
	"github.com/ConfigButler/gitops-reverser/internal/typeset"
)

// planEventsToWorktree is the plan-then-flush write path: model the subtree once, resolve each
// event to a single-identity action, apply to commit-scoped buffers, flush only what changed.
//
// Every event is a single-identity intent and the writer NEVER mark-and-sweeps a batch.
// Whole-folder mark-and-sweep is the resync path, not steady state.
// mapperForCluster returns the GVK->GVR lookup for a source cluster: the per-cluster registry
// when a cluster is named and a cluster resolver is wired, else the default (local) mapper.
// The CLI and tests leave clusterMapper nil, so they always resolve against `mapper`.
func (w *BranchWorker) mapperForCluster(clusterID string) typeset.Lookup {
	if clusterID != "" && w.clusterMapper != nil {
		if lk := w.clusterMapper(clusterID); lk != nil {
			return lk
		}
	}
	return w.mapper
}

// clusterIDForEvents returns the source cluster the events in one base belong to. Events in a
// single flush share a GitTarget (they are grouped by base), so they share a cluster; the
// first non-empty id wins, and an all-empty set is the local cluster.
func clusterIDForEvents(events []Event) string {
	for _, ev := range events {
		if ev.SourceCluster != "" {
			return ev.SourceCluster
		}
	}
	return ""
}

// planEventsToWorktree flushes one GitTarget base's events into the worktree, and reports whether
// the tree changed and, per event and in order, whether that event changed it: the write's actions
// (docs/design/applier-echoes.md, F2).
func (w *BranchWorker) planEventsToWorktree(
	ctx context.Context,
	worktree *gogit.Worktree,
	base string,
	events []Event,
	policy *manifestanalyzer.PlacementPolicy,
	namespaces namespacePolicy,
	pruneMode v1alpha3.PruneMode,
) (bool, []bool, error) {
	root := worktree.Filesystem().Root()
	scoped, err := scanRenderScope(root, base)
	if err != nil {
		return false, nil, err
	}

	// Every event in a base shares one GitTarget (events are grouped by base), so they share
	// one source cluster; resolve this subtree's GVK->GVR against that cluster's registry.
	mapper := w.mapperForCluster(clusterIDForEvents(events))
	batch := newWriteBatch(ctx, w.contentWriter, mapper, scoped.scan, policy, namespaces, scoped.writeSubdir)
	batch.pruneMode = pruneMode
	batch.target = placementTargetForEvents(events)
	if err := batch.refusal(); err != nil {
		return false, nil, err
	}
	// Publish what this folder's shape resolved to before folding anything in. It is a fact about
	// the scan, so it must not depend on the events applying — or on their being applied at all,
	// which is what makes it available to a suspended target. It runs AFTER the acceptance gate
	// above on purpose: a folder the operator has refused to manage is one whose layout it should
	// not be making claims about, and GitPathAccepted=False already says why.
	w.scanLayout(ctx, batch, worktree)
	// The one write-plan precondition that is not about the folder at all, which is why it is
	// raised AFTER the layout is published: the folder is fine and its shape is still worth
	// reporting; what is wrong is the configuration pointed at it.
	if err := batch.sourceNamespaceRefusal(); err != nil {
		return false, nil, err
	}
	// Before the first event is folded in, not as they are reached: an upsert that refuses aborts
	// the batch, and a DELETE queued behind it would never be seen by the eligibility evidence the
	// refusal carries. Gated on the same prune mode applyDelete consults, because a suppressed
	// delete removes nothing and so disqualifies nothing.
	if batch.pruneMode.OrDefault().AppliesEventDeletes() {
		batch.declarePlannedRemovals(countDeleteEvents(events))
	}
	changedBy := make([]bool, len(events))
	for i, event := range events {
		batch.beginEvent()
		if err := batch.applyEvent(ctx, event); err != nil {
			return false, nil, err
		}
		changedBy[i] = batch.eventChanged()
	}
	// The flush is anchored at renderBase — spec.path, or the common ancestor of spec.path
	// and every base it reads. The write jail (writeSubdir) is enforced inside the batch, so
	// a planned write outside spec.path is refused even though the scan reached past it.
	changed, err := batch.flush(ctx, worktree, root, scoped.renderBase)
	return changed, changedBy, err
}

// countDeleteEvents counts the events in a batch that remove a document.
func countDeleteEvents(events []Event) int {
	count := 0
	for _, event := range events {
		if !event.IsFieldPatch() && event.Operation == "DELETE" {
			count++
		}
	}
	return count
}

// writeBatch is the commit-scoped plan-then-flush working set for one GitTarget
// subtree. The store is the byte-free model the batch resolves identities against;
// contentByPath holds the worktree bytes so a touched file is hydrated lazily into
// a fileBuffer; buffers accumulates the mutations the events produce.
type writeBatch struct {
	writer        eventContentWriter
	mapper        typeset.Lookup
	store         *manifestanalyzer.ManifestStore
	docLoc        map[*manifestanalyzer.DocumentModel]manifestanalyzer.RecordRef
	contentByPath map[string][]byte
	buffers       map[string]*fileBuffer
	// intents records what each document this flush writes must render to, so the
	// render precondition can tell a change the flush MEANT from one it merely caused.
	// Anything not named here has to come out of the re-render untouched.
	intents []manifestanalyzer.WriteIntent
	// removedDocuments counts the documents this flush takes out of the subtree, by any route:
	// an explicit DELETE, a mark-and-sweep drop, or a $patch: delete authored into an overlay.
	// Nothing about the write plan reads it — it is evidence for spec.onRefusal, which must never
	// answer a refused removal with a commit that asks the reconciler to put the object back.
	removedDocuments int
	// plannedRemovals counts the removals this flush INTENDS before it starts applying anything.
	//
	// Counting executions alone was not enough, and a review found where: both paths apply their
	// upserts first and abort on the first refusal, so an unplaceable edit raises its issue — and
	// captures its evidence — before the sweep that would have removed a document is ever
	// reached. The flush is all-or-nothing, so what the batch MEANT to do decides eligibility, and
	// it has to be known before the first operation can return early.
	plannedRemovals int
	// putToKustomize records that this flush touched a kustomize render root — it edited a
	// governed document, or placed a new one into a kustomization's resources:. It is what
	// turns the oracle on, and it is deliberately NOT the same question as WriteIntent.Governed:
	// that one additionally ASSERTS the document is rendered, which a new document is not
	// entitled to claim (its resources: entry can legitimately fail to be added — see
	// appendKustomizationResource — leaving the file written but outside every render).
	putToKustomize bool
	// target is the GitTarget this batch writes for, carried only so the placement metrics
	// can name it (see placement_metrics.go). It is set by the caller — the live path reads
	// it off the events, the resync path from the request — and is empty for the CLI and for
	// tests, where the counters are simply unlabelled.
	target placementTarget
	// documents is the write-boundary census for this batch, tallied as documents are decided and
	// published by flush once the bytes are actually on disk. See document_metrics.go for why it
	// is not published at the decision site.
	documents map[documentKey]int64
	// policy is the GitTarget's declared new-file placement policy, consulted
	// only for a resource with no existing document. nil means no declared policy —
	// placement falls through to the folder's one kustomize root and then the built-in path.
	policy *manifestanalyzer.PlacementPolicy
	// namespaces is the GitTarget's declared namespace behavior — spec.serializeNamespace — which
	// decides whether the bytes this batch writes carry metadata.namespace at all. The zero value
	// is "declare nothing", i.e. infer per document, which is what every caller with no GitTarget
	// to read (the CLI, most tests) gets.
	namespaces namespacePolicy
	// pruneMode gates the EXPLICIT delete path only; inferred mark-and-sweep is gated in the
	// planner. Always read through OrDefault: the zero value is unset, not `never`, and the resync
	// batch never sets it — reading it literally would stop that batch mirroring deletes.
	pruneMode v1alpha3.PruneMode
	// writeSubdir is spec.path expressed relative to the render anchor (renderBase) — the
	// write jail. It is "" for a self-contained subtree (renderBase == spec.path), where
	// every scanned path is writable; it is non-empty only when the scan reached past
	// spec.path into a base it renders (render-root scoping), and then a planned write must
	// stay within it. The store and every path in it are keyed relative to renderBase, so a
	// writable path is one under writeSubdir. See internal/git/render_scope.go.
	writeSubdir string
	// createdRoot is the kustomization.yaml this batch WROTE, for a folder that had none and a
	// target that declared spec.placement.useKustomize. It is nil in every other case, including
	// the ordinary one where a root was already there. It exists so the second new document in a
	// batch joins the root the first one created: the store was built before the batch, so nothing
	// in it knows the file exists.
	createdRoot *manifestanalyzer.KustomizationInfo
	// scan is the folder scan the store was built from, kept so a re-entry can rebuild the store
	// over the batch's staged files (see stagedRender).
	scan manifestanalyzer.FolderScan
	// layout is what the scan resolved about this folder's shape. It is published as
	// status.placement, and createNew reads it: a folder covering several render roots has no
	// single one to place a new document into, so placing one is refused rather than guessed.
	layout manifestanalyzer.LayoutResolution
	// coldBundles tracks new resources placed at a path that held nothing before the batch.
	// LocateNew resolves against the PRE-BATCH store, so it cannot see two new resources landing
	// on one path; without this each write would silently discard the last. Members are re-sorted
	// by resource identity so the result does not depend on event order.
	// See docs/layout/new-file-placement-rules.md, "Collision and append behavior".
	coldBundles map[string][]coldBundleMember
	// rendered records what this batch has already done to an inherited object's presence in the
	// render: true once it retired the owned patch hiding the object, false once it authored one.
	// The store was built before the batch, so without this a second event for the same object
	// (ADDED then MODIFIED, or back and gone again) read the pre-batch render and was wrong about
	// it. See rendersInBatch.
	rendered map[renderKey]bool
	// removedGenerated records the generated objects this batch already took out of the folder, so
	// a second removal of one (a live delete and a sweep, or a replay) is a no-op rather than an
	// edit of an entry that is gone. See removeGenerated.
	removedGenerated map[*manifestanalyzer.GeneratedObject]bool
	// touched records, while one event is applied, each file it touched with that file's bytes
	// before the event. Nil when nothing is recorded. See beginEvent.
	touched map[string]touchedFile
}

// touchedFile is a file's bytes as they were before the event being applied touched it, and
// whether the file was there at all.
type touchedFile struct {
	content []byte
	present bool
}

// beginEvent starts recording what the next event changes; eventChanged answers once it is applied.
// A write keeps only the objects whose plan changed its tree (docs/design/applier-echoes.md, F2),
// and the bytes are the one judge of that: every path an event takes, a field patch, an override
// edit or a kustomization entry, reaches its file through buffer.
func (wb *writeBatch) beginEvent() { wb.touched = map[string]touchedFile{} }

// eventChanged reports whether the event applied since beginEvent changed any file's bytes, and
// stops recording.
func (wb *writeBatch) eventChanged() bool {
	touched := wb.touched
	wb.touched = nil
	for rel, before := range touched {
		after := wb.buffers[rel]
		if before.present != (after.current != nil) || !bytes.Equal(before.content, after.current) {
			return true
		}
	}
	return false
}

// renderKey names one document of the store by where it lives and what it is.
type renderKey struct {
	path, kind, name string
}

// rendersInBatch reports whether the document at path renders once the batch's edits so far are
// applied: what the batch itself did to it when it did anything, and the pre-batch render when
// it did not.
func (wb *writeBatch) rendersInBatch(path, kind, name string) bool {
	if r, ok := wb.rendered[renderKey{path: path, kind: kind, name: name}]; ok {
		return r
	}
	return wb.store.Renders(path, kind, name)
}

// setRenderedInBatch records that the batch took the document at path in or out of the render.
func (wb *writeBatch) setRenderedInBatch(path, kind, name string, rendered bool) {
	if wb.rendered == nil {
		wb.rendered = map[renderKey]bool{}
	}
	wb.rendered[renderKey{path: path, kind: kind, name: name}] = rendered
}

// coldBundleMember is one new document contributing to a brand-new shared bundle
// file within this batch. Retained (rather than re-parsed from buf.current) so a
// later collision on the same path can re-sort and rebuild the whole file from
// scratch, independent of which new resource's event the writer processed first.
type coldBundleMember struct {
	identifier types.ResourceIdentifier
	content    []byte
	// sensitive records whether this member is an encrypted (sensitive) resource, so
	// createNew can refuse to co-mingle sensitive and plaintext documents in one
	// brand-new file regardless of the order their events arrived (Option B2's
	// write-safety guard — see createNew).
	sensitive bool
}

func newWriteBatch(
	ctx context.Context,
	writer eventContentWriter,
	mapper typeset.Lookup,
	scan manifestanalyzer.FolderScan,
	policy *manifestanalyzer.PlacementPolicy,
	namespaces namespacePolicy,
	writeSubdir string,
) *writeBatch {
	// The writer allowlist retains build directives (kustomization.yaml) and the operator's
	// own .sops.yaml bootstrap config outside the managed model — these are auxiliary input,
	// not documents to materialise or to mis-refuse as standalone non-KRM. Every other KRM
	// document is still materialised: the live writer indexes the whole subtree for
	// placement. The scan also carries the foreign-content view and the active
	// .gittargetignore, so the structure-only acceptance gate (run by writeBatch.refusal) and
	// the write-plan precondition (run by writeBatch.flush) read both from the store.
	store := manifestanalyzer.BuildStoreFromScan(ctx, scan, mapper, manifestanalyzer.WriterAllowlist(),
		manifestanalyzer.WithDeclaredNamespace(namespaces.declaredNamespace()))
	// Surface the store's build-time warnings (ambiguous namespace or override
	// context, scope mismatches) once per batch: these drive silent fallbacks —
	// e.g. an ambiguous override chain falls back to write-through — and without
	// this line the live path would leave no trace of why. The analyzer CLI and
	// scan mode show the same diagnostics offline.
	logStoreDiagnostics(ctx, store.Diagnostics)
	// The generator inputs are hydrated beside the YAML files: a generated ConfigMap's edit is
	// written into one of them, and the oracle renders both.
	contentByPath := make(map[string][]byte, len(scan.YAMLFiles)+len(scan.InputFiles))
	for _, f := range scan.RenderFiles() {
		contentByPath[f.Path] = f.Content
	}
	batch := &writeBatch{
		writer:        writer,
		mapper:        mapper,
		store:         store,
		docLoc:        store.DocumentLocations(),
		contentByPath: contentByPath,
		buffers:       map[string]*fileBuffer{},
		policy:        policy,
		namespaces:    namespaces,
		writeSubdir:   writeSubdir,
		scan:          scan,
	}
	// Resolved with the store rather than by each caller, so no write path can reach createNew
	// with an unresolved layout and place a new document into a folder that has no single root
	// to place it in. Publishing it is a separate step (scanLayout), because only the paths that
	// know which GitTarget they serve can publish.
	batch.layout = manifestanalyzer.ResolveLayout(store, writeSubdir)
	return batch
}

// refusal runs the acceptance gate over the batch's store; a refusal aborts the commit before any
// file is touched.
//
// Structure-only on purpose: refusing on a discovery-derived fact (unwatched / out-of-scope) would
// turn a discovery wobble into a stuck, unwritable GitTarget.
func (wb *writeBatch) refusal() error {
	return manifestanalyzer.RefusalError(manifestanalyzer.AcceptStructureOnly(wb.store))
}

// sourceNamespaceRefusal is the write-plan precondition for the one-source-namespace rule.
//
// This is the CORRECTNESS layer and it holds whatever admission did: the WatchRule webhook is
// one-shot, cannot see a serializeNamespace flipped afterwards, and fails open.
// See docs/spec/where-validation-lives.md.
func (wb *writeBatch) sourceNamespaceRefusal() error {
	issues := manifestanalyzer.MultipleSourceNamespacesRefusal(
		wb.namespaces.declaresNamespaceFree(),
		wb.namespaces.SourceNamespaces,
		wb.namespaces.SourceNamespaceWildcard,
		wb.writeSubdir,
	)
	if len(issues) == 0 {
		return nil
	}
	return manifestanalyzer.RefusalError(manifestanalyzer.Acceptance{Accepted: false, Issues: issues})
}

// fileBuffer is the commit-scoped, hydrated working copy of one file under the
// GitTarget base path. original is the worktree bytes (nil for a file the batch
// creates); current is the bytes after applying actions (nil means the file should
// be removed). Dirty/Deleted are derived exactly as the design's FileModel — two
// byte slices are the whole state machine, so there is no flag to forget to flip.
type fileBuffer struct {
	rel      string
	original []byte
	current  []byte
	// removedDocument records that this flush took a document OUT of this file. The two byte
	// slices cannot say so: removing one document from a multi-document file leaves the file
	// present and its bytes non-nil, which is byte-for-byte the shape of an ordinary edit. It is
	// set by every path that removes a document, and it is only ever read as evidence for
	// spec.onRefusal (see writeBatch.holdsExistingDocument).
	removedDocument bool
}

func (b *fileBuffer) dirty() bool   { return b.current != nil && !bytes.Equal(b.current, b.original) }
func (b *fileBuffer) deleted() bool { return b.current == nil && b.original != nil }

// holdsExistingDocument reports that the folder ALREADY held the DOCUMENT this write edits, and
// that the flush removes no document anywhere. It is the evidence AcceptanceIssue.ExistingDocument
// carries, and it is a method on the batch so every raise site asks the same question the same way.
//
// **File survival is not the question, and a review found the hole where it was.** The first
// version asked only whether the file existed and survived, so deleting one document out of a
// two-document file answered yes — another document kept the file alive — and a refused deletion
// qualified for a reapply. A reapply asks the applier to apply what Git holds, and Git still holds
// the document the refusal failed to remove, so it would RECREATE an object somebody deliberately
// deleted from the cluster.
//
// The batch-wide arm is deliberate rather than per-file: the flush is all-or-nothing, so a refusal
// aborts the removal along with everything else, and the reapply it would earn covers the whole
// folder rather than one file. A batch that meant to remove anything therefore earns no reapply.
func (wb *writeBatch) holdsExistingDocument(b *fileBuffer) bool {
	if b.original == nil || b.deleted() || b.removedDocument {
		return false
	}
	return wb.removedDocuments == 0 && wb.plannedRemovals == 0
}

// declarePlannedRemovals records how many documents this flush means to remove, before it applies
// anything. Both callers know the answer up front — the resync from its plan, the live path from
// the DELETE events in its batch — and must say so there, because an operation that refuses early
// captures its evidence before any removal has run. See plannedRemovals.
func (wb *writeBatch) declarePlannedRemovals(count int) { wb.plannedRemovals += count }

// buffer returns the hydrated working copy for a base-relative path, reading the
// worktree bytes into Original/Current on first touch. A path with no worktree
// bytes is a new file (Original nil).
func (wb *writeBatch) buffer(rel string) *fileBuffer {
	b, ok := wb.buffers[rel]
	if !ok {
		b = &fileBuffer{rel: rel}
		if orig, found := wb.contentByPath[rel]; found {
			b.original = orig
			b.current = orig
		}
		wb.buffers[rel] = b
	}
	if _, seen := wb.touched[rel]; wb.touched != nil && !seen {
		wb.touched[rel] = touchedFile{content: bytes.Clone(b.current), present: b.current != nil}
	}
	return b
}

// upsertOutcome is what an upsert actually did to the worktree bytes, so a caller can
// count create/update accurately from the apply rather than from a separate plan
// estimate (which mislabels a re-encrypted sensitive resource as skipped).
type upsertOutcome int

const (
	upsertNoChange upsertOutcome = iota
	upsertCreated
	upsertUpdated
	// upsertSkippedUnsafe is a deliberate, fail-safe refusal to write a resource:
	// its placement could not be resolved safely, or writing would co-mingle a
	// sensitive and a plaintext document, or would overwrite a multi-document file.
	// It is distinct from upsertNoChange (a genuine no-op) so the resync path can
	// count it and surface it, rather than have a not-mirrored resource vanish with
	// no signal (placement Option B2's fail-safe skips — see createNew/writeWholeFile).
	upsertSkippedUnsafe
)

// applyEvent folds one event into the batch: a field patch sets bounded fields on an
// existing parent, a DELETE removes a document, anything else is an upsert (the
// object-bearing event the stream guarantees for non-deletes). The steady-state
// writer does not need the upsert outcome (it flushes by byte state), so it is
// discarded here; the resync planner consumes it for stats.
func (wb *writeBatch) applyEvent(ctx context.Context, event Event) error {
	switch {
	case event.IsFieldPatch():
		return wb.applyFieldPatch(ctx, event)
	case event.Operation == "DELETE":
		return wb.applyDelete(ctx, event)
	default:
		_, err := wb.applyUpsert(ctx, event)
		return err
	}
}

// applyUpsert resolves an object-bearing event against the subtree. When a managed
// document for its identity already lives there — even moved off the built-in path —
// the resource is edited where it lives: a non-sensitive document is patched in place;
// a sensitive document is re-encrypted wholesale AT ITS EXISTING PATH (never patched in
// place — that would drop the SOPS metadata and write the secret back in cleartext, and
// never at the built-in path, which would orphan the moved copy). A resource with no
// existing document is placed by createNew. It returns what it did to the bytes
// (created / updated / no change).
func (wb *writeBatch) applyUpsert(ctx context.Context, event Event) (upsertOutcome, error) {
	// An inherited object back in the cluster first loses the owned delete patch that hid it, so
	// the upsert below edits a base document the render holds again.
	basePath, retired, err := wb.retireInheritedDelete(ctx, event)
	if err != nil {
		return upsertNoChange, err
	}
	if basePath != "" {
		// The store was built while the patch still hid the object, so it knows nothing of how the
		// overlay renders it (its labels, its namespace, what the build supplies). Planning the
		// edit against that store would write the overlay's values into the read-only base.
		defer wb.stagedRender(ctx)()
	}
	intentsBefore := len(wb.intents)
	outcome, err := wb.upsert(ctx, event)
	if err == nil && retired {
		// The object renders again, and the oracle must prove it renders as the live object: when
		// the base already matches, the upsert wrote nothing and declared nothing, so the presence
		// is declared here. For the mirror it is a created document either way.
		if !intendsObject(wb.intents[intentsBefore:], event.Object) {
			wb.intend(intentFor(event.Object, basePath, false))
		}
		if outcome == upsertNoChange {
			outcome = upsertCreated
		}
	}
	if err == nil {
		// Tallied here rather than at either caller: the live path reaches this through applyEvent
		// and the resync path calls it directly, so this is the one place both are covered exactly
		// once. Published by flush, never here — see document_metrics.go.
		wb.tallyDocument(event.Identifier, documentOutcomeForUpsert(outcome))
	}
	return outcome, err
}

// stagedRender rebuilds the store over the batch's own staged files, the tree this batch is about
// to commit, and puts it in place until the returned restore is called.
//
// It is scoped to one upsert on purpose. Placement reasons about the PRE-batch tree (a path that
// held nothing before the batch is a new bundle, not an existing file), so the rest of the batch
// keeps the store it started with; only the re-entering object is planned against the render it
// is coming back into.
func (wb *writeBatch) stagedRender(ctx context.Context) func() {
	scan := wb.scan
	scan.YAMLFiles, scan.InputFiles = wb.stagedFiles()
	staged := manifestanalyzer.BuildStoreFromScan(ctx, scan, wb.mapper, manifestanalyzer.WriterAllowlist(),
		manifestanalyzer.WithDeclaredNamespace(wb.namespaces.declaredNamespace()))
	store, docLoc := wb.store, wb.docLoc
	wb.store, wb.docLoc = staged, staged.DocumentLocations()
	return func() { wb.store, wb.docLoc = store, docLoc }
}

// stagedFiles splits the batch's tree into the scan's two views: the YAML files the store models,
// and the generator inputs it only renders.
func (wb *writeBatch) stagedFiles() ([]manifestedit.FileContent, []manifestedit.FileContent) {
	inputs := make(map[string]struct{}, len(wb.scan.InputFiles))
	for _, f := range wb.scan.InputFiles {
		inputs[f.Path] = struct{}{}
	}
	var yamlFiles, inputFiles []manifestedit.FileContent
	for _, f := range wb.files() {
		if _, input := inputs[f.Path]; input {
			inputFiles = append(inputFiles, f)
			continue
		}
		yamlFiles = append(yamlFiles, f)
	}
	return yamlFiles, inputFiles
}

// intendsObject reports whether one of intents is about obj.
func intendsObject(intents []manifestanalyzer.WriteIntent, obj *unstructured.Unstructured) bool {
	for _, in := range intents {
		if in.Kind == obj.GetKind() && in.Name == obj.GetName() {
			return true
		}
	}
	return false
}

// upsert is applyUpsert's body, split out so the census above wraps every return path.
func (wb *writeBatch) upsert(ctx context.Context, event Event) (upsertOutcome, error) {
	id, ok := manifestIdentity(event.Object)
	if !ok {
		return wb.createNew(ctx, event)
	}
	dm := wb.store.ByManifestIdentity[id]
	if dm == nil {
		// No document holds it, but a configMapGenerator may render it: then the edit belongs
		// in the generator's inputs, and a new document beside it would collide in the build.
		if generated := wb.store.GeneratedObjectsFor(event.Object); len(generated) > 0 {
			return wb.writeGenerated(ctx, event, generated)
		}
		return wb.createNew(ctx, event)
	}
	filePath := wb.docLoc[dm].FilePath
	if !wb.writer.isSensitiveIdentifier(event.Identifier) {
		return wb.patchExisting(ctx, event, filePath, id, dm)
	}
	return wb.rewriteSensitive(ctx, event, filePath)
}

// rewriteSensitive re-encrypts a sensitive document wholesale at its existing path.
//
// Its intent is UNCHECKED: the file is SOPS ciphertext, so kustomize renders the encrypted
// blob and no plaintext live object can ever equal it. The oracle is told to expect this
// object to move without being able to say what to — while still holding the write to
// disturbing nothing else, which is the half that protects other environments.
func (wb *writeBatch) rewriteSensitive(ctx context.Context, event Event, filePath string) (upsertOutcome, error) {
	outcome, err := wb.writeWholeFile(ctx, event, filePath)
	if err == nil && wroteBytes(outcome) {
		wb.intend(markUnchecked(intentFor(event.Object, filePath, false), true))
	}
	return outcome, err
}

// wroteBytes reports whether an upsert actually changed the worktree, which is the only
// case that must give the oracle an intent.
func wroteBytes(o upsertOutcome) bool {
	return o == upsertCreated || o == upsertUpdated
}

// createNew places a resource with no existing document, per
// docs/layout/new-file-placement-rules.md, and adds any kustomize resources: entry it requires.
// A placement that cannot be honoured safely (a sensitive resource colliding with an existing
// file) is logged and left unwritten rather than mis-written; the next event or resync retries.
func (wb *writeBatch) createNew(ctx context.Context, event Event) (upsertOutcome, error) {
	kind := ""
	// Labels come off the event's object — the SANITIZED one (see Event.Object), and the only
	// place they exist at this point: placement runs precisely when there is no document in Git
	// to read them from. Sanitized is also what the "{label:key}" variable must see, and what the
	// GitTarget's Validated gate assumes: a label the writer strips is already absent here, so a
	// template naming one is refused at the gate rather than quietly bucketing every resource of
	// its type together. An object-less event cannot reach this function (a field patch branches
	// off in applyEvent, and a non-delete always carries an object), but if one ever did, nil
	// labels place it at the unlabeled sentinel rather than guess.
	var labels map[string]string
	if event.Object != nil {
		kind = event.Object.GetKind()
		labels = event.Object.GetLabels()
	}
	// A folder covering several render roots has no single one for a NEW document, and picking
	// one would hand it to an environment nobody named. An existing document is unaffected: it
	// is edited where it lives, and this branch is not reached for it.
	if issues := manifestanalyzer.AmbiguousLayoutRefusal(wb.layout, wb.writeSubdir); len(issues) > 0 {
		return upsertNoChange, manifestanalyzer.RefusalError(
			manifestanalyzer.Acceptance{Accepted: false, Issues: issues})
	}
	sensitive := wb.writer.isSensitiveIdentifier(event.Identifier)
	// WriteScope tells placement the write jail: when render-root scoping re-rooted the scan
	// past spec.path, a declared/built-in path is rebased under the jail rather than escaping
	// it (see finishPlacement). It is "" for a self-contained subtree, where placement already
	// resolves relative to spec.path.
	placement, err := manifestanalyzer.LocateNew(wb.store, wb.policy, manifestanalyzer.PlacementRequest{
		Identifier: event.Identifier,
		Kind:       kind,
		Sensitive:  sensitive,
		WriteScope: wb.writeSubdir,
		Labels:     labels,
	})
	if err != nil {
		refusal := placementRefusalReason(err)
		log.FromContext(ctx).Info("Skipping new resource: placement could not be resolved safely",
			"resource", event.Identifier.String(), "refusal", string(refusal), "reason", err.Error())
		recordPlacementRefusal(ctx, wb.target, event.Identifier, refusal)
		return upsertSkippedUnsafe, nil
	}

	// The LIVE object, kept before the namespace strip below rewrites it. The bytes we write
	// and the object the render must produce are not the same thing, and only this scope
	// still holds both — see intentFor.
	live := event.Object

	// A destination that infers its namespace from build context (a kustomization's
	// namespace: transformer) must keep metadata.namespace out of the written bytes,
	// exactly as patchExisting already does for an in-place edit of an existing
	// document in the same context — otherwise the new document would silently break
	// the convention every sibling in that directory follows. An explicit
	// spec.serializeNamespace overrides that inference in both directions.
	if wb.namespaces.omitNamespace(placement.NamespaceInherited) && event.Object != nil {
		event.Object = event.Object.DeepCopy()
		event.Object.SetNamespace("")
	}

	outcome, refusal, err := wb.placeNewDocument(ctx, event, placement, sensitive)
	if err != nil || !wroteBytes(outcome) {
		// A skipped write is a resource the mirror does not hold. Count it with the refusals
		// LocateNew raised, from the same closed reason set, so "resources we declined to
		// place" is one series rather than a log line here and a metric there.
		if outcome == upsertSkippedUnsafe {
			recordPlacementRefusal(ctx, wb.target, event.Identifier, refusal)
		}
		return outcome, err
	}
	// Recorded here rather than at resolution: this is the point at which the document is
	// really in the mirror at this path, so placements_total and placement_refusals_total
	// partition every new resource instead of double-counting the ones that resolved and
	// then could not be written.
	recordPlacement(ctx, wb.target, event.Identifier, placement.Source, placement.Append)

	// AFTER the write, for the same reason the placement is counted here. placeNewDocument can
	// still decline — a multi-document target it will not overwrite, or a new file that would
	// mix sensitive and plaintext documents — and registering the entry first meant a resource we
	// REFUSED still put its file into the folder's render. For the multi-document case that is
	// foreign content we declined to own, added to resources: on our say-so; and either way it
	// counted as outcome="added", the value that is supposed to mean "the file we just wrote will
	// build". Pinned by TestPlacementMetrics_RefusedPlacementLeavesTheKustomizationAlone.
	if err := wb.resolveKustomizeRootForNew(ctx, &placement); err != nil {
		return upsertNoChange, err
	}
	if placement.Kustomization != nil {
		wb.appendKustomizationResource(ctx, event, placement)
	}

	// A new document inside a render root is governed by the folder's images:/replicas: entries
	// immediately, but has no override chain to route onto, so an entry that overrides it makes
	// the resource never converge. Declaring it governed puts it in front of the oracle, turning a
	// silent non-converging commit into a refusal naming the file and object.
	wb.putToKustomize = wb.putToKustomize || placement.Kustomization != nil || wb.createdRoot != nil
	wb.intend(markUnchecked(intentFor(live, placement.Path, false), sensitive))
	return outcome, nil
}

// resolveKustomizeRootForNew settles what renders a new document, for a target that declared
// spec.placement.useKustomize. It creates the folder's root when there is none — with this document
// already registered in the bytes it writes, so a LATER document in the same batch joins it through
// the ordinary resources: append — and otherwise refuses a placement no kustomization would render.
//
// Both halves are here rather than in createNew because they answer one question between them: is
// this document going to be rendered at all?
func (wb *writeBatch) resolveKustomizeRootForNew(
	ctx context.Context,
	placement *manifestanalyzer.PlacementResult,
) error {
	if created := wb.bootstrapKustomization(ctx, *placement); created != nil {
		placement.Kustomization = created
	}
	issues := wb.unrenderedPlacementRefusal(*placement)
	if len(issues) == 0 {
		return nil
	}
	return manifestanalyzer.RefusalError(manifestanalyzer.Acceptance{Accepted: false, Issues: issues})
}

// unrenderedPlacementRefusal reports the refusal for a placement under spec.placement.useKustomize
// that no kustomization would render. A document is rendered when a kustomization governs its path,
// and there are three ways for that to be true: one already governed it and needs the entry
// (LocateNew set Kustomization), one already governed it and already lists the path, or this batch
// created the root — which registers the first document in the bytes it writes, so it reports no
// Kustomization to append to.
func (wb *writeBatch) unrenderedPlacementRefusal(
	placement manifestanalyzer.PlacementResult,
) []manifestanalyzer.AcceptanceIssue {
	governed := placement.Kustomization != nil ||
		wb.createdRoot != nil ||
		manifestanalyzer.GoverningKustomization(wb.store, wb.writeSubdir, placement.Path) != nil
	useKustomize := wb.policy != nil && wb.policy.UseKustomize
	return manifestanalyzer.UnrenderedPlacementRefusal(
		useKustomize, governed, placement.Path, wb.layout.RenderRoot)
}

// placeNewDocument writes the new document at its resolved placement: appended to an existing
// accepted bundle, folded into a same-batch cold bundle, or as a file of its own.
//
// It returns the refusal reason alongside the outcome, and only for upsertSkippedUnsafe, so
// the caller can count WHY a new resource was left out of the mirror without inspecting a log
// message. The multi-document refusal is attributed here rather than inside writeWholeFile
// because that function also serves in-place updates, where the same skip is not a placement
// decision at all.
func (wb *writeBatch) placeNewDocument(
	ctx context.Context,
	event Event,
	placement manifestanalyzer.PlacementResult,
	sensitive bool,
) (upsertOutcome, manifestanalyzer.PlacementRefusalReason, error) {
	if placement.Append {
		outcome, err := wb.appendNewDocument(ctx, event, placement.Path)
		return outcome, "", err
	}

	buf := wb.buffer(placement.Path)
	if buf.original == nil {
		// Nothing occupied this path pre-batch, so route through the cold-bundle path: a collision
		// LocateNew could not see must form a deterministic multi-document file rather than one
		// write discarding another.
		//
		// A sensitive resource never shares a file, and plaintext never joins a bundle holding a
		// sensitive member. Skip rather than co-mingle; the next event or resync retries.
		if buf.current != nil && (sensitive || wb.coldBundleHasSensitive(placement.Path)) {
			log.FromContext(ctx).Info(
				"Skipping new resource: sensitive and plaintext resources must not share a new file",
				"resource", event.Identifier.String(), "file", placement.Path, "sensitive", sensitive)
			return upsertSkippedUnsafe, manifestanalyzer.PlacementRefusedMixedSensitivityNewFile, nil
		}
		outcome, err := wb.writeColdBundleMember(ctx, event, placement.Path, sensitive)
		return outcome, "", err
	}
	outcome, err := wb.writeWholeFile(ctx, event, placement.Path)
	if outcome == upsertSkippedUnsafe {
		return outcome, manifestanalyzer.PlacementRefusedMultiDocumentTarget, err
	}
	return outcome, "", err
}

// writeColdBundleMember writes a resource to a path nothing occupied before this batch. Every
// member seen at rel is re-sorted by resource identity and the file rebuilt, so the result does
// not depend on which event was processed first. The single-member case is byte-identical to a
// plain write.
func (wb *writeBatch) writeColdBundleMember(
	ctx context.Context,
	event Event,
	rel string,
	sensitive bool,
) (upsertOutcome, error) {
	content, err := wb.writer.buildContentForWrite(ctx, wb.target, event)
	if err != nil {
		return upsertNoChange, err
	}
	if wb.coldBundles == nil {
		wb.coldBundles = map[string][]coldBundleMember{}
	}
	wb.coldBundles[rel] = append(
		wb.coldBundles[rel],
		coldBundleMember{identifier: event.Identifier, content: content, sensitive: sensitive},
	)
	members := wb.coldBundles[rel]
	sort.Slice(members, func(i, j int) bool {
		return members[i].identifier.Key() < members[j].identifier.Key()
	})

	var rebuilt []byte
	for _, m := range members {
		rebuilt = appendYAMLDocument(rebuilt, m.content)
	}
	wb.buffer(rel).current = rebuilt
	return upsertCreated, nil
}

// coldBundleHasSensitive reports whether any member already staged for the
// brand-new file at rel is an encrypted (sensitive) resource, so createNew can
// refuse to add a plaintext member that would co-mingle with it.
func (wb *writeBatch) coldBundleHasSensitive(rel string) bool {
	for _, m := range wb.coldBundles[rel] {
		if m.sensitive {
			return true
		}
	}
	return false
}

// appendNewDocument adds a resource with no existing document as an additional
// document in an existing accepted plaintext file (a "bundle" placement). Unlike
// writeWholeFile it never replaces the file's existing bytes — every prior document
// in the buffer survives untouched, byte for byte; LocateNew never returns an
// Append placement for a sensitive resource (see its doc comment), so this path is
// plaintext-only.
func (wb *writeBatch) appendNewDocument(ctx context.Context, event Event, rel string) (upsertOutcome, error) {
	content, err := wb.writer.buildContentForWrite(ctx, wb.target, event)
	if err != nil {
		return upsertNoChange, err
	}
	buf := wb.buffer(rel)
	buf.current = appendYAMLDocument(buf.current, content)
	return upsertCreated, nil
}

// appendYAMLDocument appends newDoc as an additional "---\n"-separated document
// after existing. existing is assumed to already be valid, accepted YAML (single- or
// multi-document); newDoc is assumed to be exactly one well-formed document
// (sanitize.MarshalToOrderedYAML's output, which always ends in a newline).
func appendYAMLDocument(existing, newDoc []byte) []byte {
	if len(existing) == 0 {
		return newDoc
	}
	const separator = "---\n"
	out := make([]byte, 0, len(existing)+len(separator)+len(newDoc))
	out = append(out, existing...)
	if out[len(out)-1] != '\n' {
		out = append(out, '\n')
	}
	out = append(out, separator...)
	out = append(out, newDoc...)
	return out
}

// appendKustomizationResource adds the new document's path to its resources:
// sequence as part of the same commit, so kustomize picks up the file createNew just
// placed inside the kustomization's directory — the "add to the right kustomize
// file." The entry is rendered relative to the kustomization's own directory
// (resources: entries are relative to the kustomization file, not the repo root).
// A failure here only drops the resources: entry (logged as a diagnostic); the
// resource's own file is still written, since a human can add the missing entry by
// hand and the next placement for that directory re-detects the gap.
func (wb *writeBatch) appendKustomizationResource(
	ctx context.Context,
	event Event,
	placement manifestanalyzer.PlacementResult,
) {
	k := placement.Kustomization
	entry := placement.Path
	if dir := path.Dir(k.Path); dir != "." {
		if rel, err := filepath.Rel(dir, placement.Path); err == nil {
			entry = filepath.ToSlash(rel)
		}
	}

	buf := wb.buffer(k.Path)
	if buf.current == nil {
		// The kustomization vanished within this batch; nothing to edit — and the file it
		// would have registered is now outside every render, which is the same user-visible
		// outcome as a failed edit, so it is counted as one.
		recordKustomizationEntry(ctx, wb.target, kustomizationEntryFailed)
		return
	}
	res, diags := manifestedit.AppendKustomizationResource(k.Path, buf.current, entry)
	switch res.Mode {
	case manifestedit.EditPatched:
		buf.current = res.Content
		recordKustomizationEntry(ctx, wb.target, kustomizationEntryAdded)
		log.FromContext(ctx).Info("Added resources: entry for new file",
			"kustomization", k.Path, "entry", entry, "resource", event.Identifier.String())
	case manifestedit.EditNoChange:
		recordKustomizationEntry(ctx, wb.target, kustomizationEntryNoChange)
	case manifestedit.EditSkipped, manifestedit.EditDeleted, manifestedit.EditWholeReplace:
		// The document is committed and its resources: entry is not, so kustomize will never
		// build the file: it is in Git, it looks mirrored, and nothing applies it. The counter
		// is the only signal that is not a log line.
		recordKustomizationEntry(ctx, wb.target, kustomizationEntryFailed)
		log.FromContext(ctx).Info("Could not add resources: entry for new file",
			"kustomization", k.Path, "entry", entry, "resource", event.Identifier.String())
		logManifestDiagnostics(ctx, diags)
	}
}

// applyFieldPatch sets only the patch's declared field paths on the existing parent document.
//
// Two refusals make a partial intent safe:
//   - NO creation path: a patch whose parent is absent is dropped rather than fabricated.
//   - A document that cannot be patched field-by-field is SKIPPED, never whole-replaced, which
//     would delete every field the subresource did not mention.
//
// The document index is re-derived from CURRENT bytes so an earlier event in the batch that
// shifted a multi-document file does not misdirect the edit.
func (wb *writeBatch) applyFieldPatch(ctx context.Context, event Event) error {
	filePath, id, ok := wb.resolveFieldPatchTarget(event)
	if !ok {
		log.FromContext(ctx).Info("Dropping field patch: parent manifest not present in Git",
			"resource", event.Identifier.String(), "source", event.FieldPatch.Source,
			"reason", "subresource_patch_no_parent")
		return nil
	}

	assignments := event.FieldPatch.Assignments
	dm := wb.store.ByManifestIdentity[id]
	governed := dm != nil && dm.Overrides != nil
	if governed {
		assignments = wb.routeGovernedFieldAssignments(ctx, event, dm, assignments)
		// A routed scale changes only the kustomization entry, which still moves what this
		// document renders to — so it must be declared, or the oracle would read its own
		// intended write as collateral damage. It is UNCHECKED because a field patch carries
		// a few audited assignments, never a whole object to compare the render against: the
		// oracle can still prove the write disturbs nothing else, but not that it landed.
		wb.putToKustomize = true
		wb.intend(fieldPatchIntent(filePath, id, governed))
		if len(assignments) == 0 {
			return nil
		}
	}

	buf := wb.buffer(filePath)
	idx, found := currentDocIndex(filePath, buf.current, id)
	if !found {
		// An earlier event in this batch already removed the document; nothing to patch.
		return nil
	}

	res, diags := manifestedit.PatchFields(
		buf.current, idx, id, assignments, manifestedit.EditOptions{},
	)
	switch res.Mode {
	case manifestedit.EditPatched:
		buf.current = res.Content
		if !governed {
			wb.intend(fieldPatchIntent(filePath, id, false))
		}
	case manifestedit.EditNoChange, manifestedit.EditDeleted:
		// No-op: the audited value already matched (or, impossible here, a delete).
	case manifestedit.EditSkipped, manifestedit.EditWholeReplace:
		// EditSkipped (encrypted, non-editable, or snapshot drift), or a defensive
		// EditWholeReplace we must never apply from a partial desired.
		log.FromContext(ctx).Info("Field patch not applied: parent is encrypted or not field-patchable",
			"resource", event.Identifier.String(), "source", event.FieldPatch.Source,
			"reason", "subresource_patch_unsafe")
		logManifestDiagnostics(ctx, diags)
	}
	return nil
}

// routeGovernedFieldAssignments diverts a spec.replicas assignment whose value a
// replicas override governs to its kustomization entry (the /scale subresource
// case of the images/replicas edit-through) and returns the assignments the file
// patch should still apply. An
// ungoverned assignment — any other path, a non-integer value, no matching
// entry — keeps today's bounded file patch.
func (wb *writeBatch) routeGovernedFieldAssignments(
	ctx context.Context,
	event Event,
	dm *manifestanalyzer.DocumentModel,
	assignments []manifestedit.FieldAssignment,
) []manifestedit.FieldAssignment {
	kept := make([]manifestedit.FieldAssignment, 0, len(assignments))
	for _, a := range assignments {
		if len(a.Path) == 2 && a.Path[0] == "spec" && a.Path[1] == "replicas" {
			if count, isInt := assignmentInt64(a.Value); isInt {
				if edit, governed := manifestanalyzer.ReplicaCountEdit(dm, count); governed {
					wb.applyOverrideEdits(ctx, event, []manifestanalyzer.OverrideEdit{edit})
					continue
				}
			}
		}
		kept = append(kept, a)
	}
	return kept
}

// assignmentInt64 reads a field-assignment value as a whole number (audit JSON
// may deliver it as int64 or float64).
func assignmentInt64(v any) (int64, bool) {
	switch n := v.(type) {
	case int64:
		return n, true
	case int:
		return int64(n), true
	case int32:
		return int64(n), true
	case float64:
		if n == math.Trunc(n) {
			return int64(n), true
		}
	}
	return 0, false
}

// resolveFieldPatchTarget locates the parent manifest a field-patch event targets.
// The parent is resolved from its objectRef GVR through the same resource-identity
// inventory the GVR-only delete uses (PlanDelete), which the live-catalog mapper
// populates while scanning the GitTarget folder. The returned identity is the parent
// document's own manifest identity (full GVK from the committed YAML), so the patch
// is applied with the parent's real Kind, never one guessed from the subresource body.
//
// found is false when Git holds no managed document for the parent identity.
func (wb *writeBatch) resolveFieldPatchTarget(event Event) (string, manifestedit.Identity, bool) {
	if action, emitted := manifestanalyzer.PlanDelete(wb.store, event.Identifier); emitted {
		return action.Ref.FilePath, action.Identity, true
	}
	return "", manifestedit.Identity{}, false
}

// patchExisting edits the managed document in place, preserving sibling bytes and hand-authored
// formatting. The no-op / patch / whole-replace / skip choice is a plan decision, not a per-event
// heuristic, and the position is re-derived from CURRENT bytes so an earlier event in the batch
// does not misdirect it.
//
// Under a kustomize images/replicas override chain the desired projection is split first: values
// the chain produces are restored to source form and the divergence routed to the entries.
func (wb *writeBatch) patchExisting(
	ctx context.Context,
	event Event,
	filePath string,
	id manifestedit.Identity,
	dm *manifestanalyzer.DocumentModel,
) (upsertOutcome, error) {
	buf := wb.buffer(filePath)
	idx, ok := currentDocIndex(filePath, buf.current, rawManifestIDForCurrentBytes(id, dm))
	if !ok {
		return upsertNoChange, nil
	}
	gitDoc, _ := manifestedit.NewDocumentAt(filePath, buf.current, idx)
	desired := event.Object
	if wb.namespaces.omitNamespace(dm.NamespaceAbsentFromFile()) && desired != nil {
		desired = desired.DeepCopy()
		desired.SetNamespace("")
	}
	projected, overrideEdits, err := projectThroughKustomize(
		manifestreport.Project(desired), buf.current, idx, dm, wb.overlayAuthorKustomization(filePath))
	if err != nil {
		var fidelity *renderFidelityRefusedError
		if errors.As(err, &fidelity) {
			return upsertNoChange, renderFidelityRefusal(filePath, id, fidelity)
		}
		// The projection could not place the edit. Refusing the whole flush is the point: the
		// alternative is to write the live object through and silently absorb the build's own
		// output into the file that feeds it.
		return upsertNoChange, sourceFormRefusal(filePath, id, wb.holdsExistingDocument(buf), err)
	}
	c := manifestedit.Comparison{
		Git:     gitDoc,
		Desired: projected,
		Options: manifestreport.EditOptions(),
	}
	res, diags := manifestedit.Apply(c, manifestedit.Decide(c))
	outcome := upsertNoChange
	switch res.Mode {
	case manifestedit.EditPatched, manifestedit.EditWholeReplace:
		buf.current = res.Content
		outcome = upsertUpdated
	case manifestedit.EditNoChange, manifestedit.EditSkipped, manifestedit.EditDeleted:
		// No-op, an unsafe edit left untouched, or (impossible here) a delete: leave
		// the bytes as they are. Surface a skip so an operator can see a document Git
		// holds but the editor refused.
		if res.Mode == manifestedit.EditSkipped {
			logManifestDiagnostics(ctx, diags)
		}
	}
	if wb.applyOverrideEdits(ctx, event, overrideEdits) {
		outcome = upsertUpdated
	}

	// Declare what this document must render to. Attribution decided WHERE the edit goes and may
	// be wrong; renderPrecondition adjudicates once the whole plan is known.
	//
	// A GOVERNED document declares intent even when its own bytes did not change. images: entries
	// are shared: bumping two Deployments on one image means the first event's edit already moved
	// what the second renders to, leaving nothing to write. Its render still moves, onto its own
	// live state, and only the declared intent says that is convergence rather than damage.
	//
	// The oracle is armed for ANY document a render root produces, not only one an override chain
	// governs: where live and render disagree the user changed something, and if a transformer
	// owns that field the write never converges. Only the re-render can see that.
	if dm.Rendered != nil {
		wb.putToKustomize = true
	}
	if outcome == upsertUpdated || dm.Overrides != nil {
		wb.intend(intentFor(event.Object, filePath, dm.Overrides != nil))
	}
	return outcome, nil
}

// projectThroughKustomize turns the live projection into the SOURCE FORM of it: the object the
// file should hold once everything the build supplies is left to the build, plus the entry edits
// for the values an images:/replicas: entry supplies.
//
// A plain document uses its parsed Git object as its render. A kustomize document uses the
// DocumentModel's local render. In both cases, a rendered ${...} value that differs in live is
// refused before source-form projection can write the live expansion back into Git.
func projectThroughKustomize(
	projected *unstructured.Unstructured,
	content []byte,
	idx int,
	dm *manifestanalyzer.DocumentModel,
	authorInto string,
) (*unstructured.Unstructured, []manifestanalyzer.OverrideEdit, error) {
	gitRaw, parsed := gitDocRawObject(content, idx)
	if !parsed {
		return projected, nil, nil
	}
	rendered := gitRaw
	if dm.Rendered != nil {
		rendered = dm.Rendered.Object
	}
	if divergences := manifestanalyzer.RenderTokenDivergences(rendered, projected.Object); len(divergences) > 0 {
		return nil, nil, &renderFidelityRefusedError{Divergences: divergences}
	}
	if dm.Rendered == nil {
		return projected, nil, nil
	}
	return manifestanalyzer.SplitDesiredForOverrides(gitRaw, projected, dm.Rendered, authorInto)
}

// overlayAuthorKustomization is the kustomization the writer may author a NEW images:/replicas:
// entry into for an edit to filePath. It is set only when render-root scoping put filePath OUT of
// the write jail — a base document an overlay reads read-only — and the overlay has a supported
// render root of its own: then a value the base supplies can be overridden by authoring an entry
// in the overlay instead of refusing the base write. It is "" for a self-contained subtree and
// for an in-jail document, where the source file itself is writable.
func (wb *writeBatch) overlayAuthorKustomization(filePath string) string {
	if wb.writeSubdir == "" || pathWithin(filePath, wb.writeSubdir) {
		return ""
	}
	if k := wb.store.Kustomizations[wb.writeSubdir]; k != nil && !k.Unsupported {
		return k.Path
	}
	return ""
}

// renderFidelityRefusedError travels from the projection seam to patchExisting, where the file
// and object identity are available to make a normal write-boundary refusal.
type renderFidelityRefusedError struct {
	Divergences []manifestanalyzer.RenderDivergence
}

func (e *renderFidelityRefusedError) Error() string {
	return "rendered token does not match live"
}

// sourceFormRefusal turns a projection that could not place an edit into the same reported
// refusal every other write-boundary violation surfaces as: GitPathAccepted=False / Stalled=True,
// naming the file and the object. It is not an internal error — the folder is fine and the
// operator is fine; the EDIT had nowhere honest to land, and saying so is the whole contract.
func sourceFormRefusal(
	filePath string, id manifestedit.Identity, existingDocument bool, err error,
) error {
	return &manifestanalyzer.AcceptanceRefusedError{
		Issues: []manifestanalyzer.AcceptanceIssue{{
			Kind: manifestanalyzer.IssueUnplaceableEdit,
			Path: filePath,
			// Evidence for spec.onRefusal, threaded from the buffer rather than assumed from the
			// name: an unplaceable EDIT usually does have a document behind it, but the caller is
			// the only place that knows, and guessing here is what the field exists to avoid.
			ExistingDocument: existingDocument,
			// Not solvable, and deliberately so: the alternative to refusing is aligning
			// two lists by position, which is measurably wrong rather than merely risky
			// (see the IssueUnplaceableEdit comment). Nobody can act on it.
			Solvable: false,
			Message: fmt.Sprintf("%s/%s in %s: %v",
				id.Kind, id.Name, filePath, err),
		}},
	}
}

func renderFidelityRefusal(
	filePath string,
	id manifestedit.Identity,
	fidelity *renderFidelityRefusedError,
) error {
	issues := make([]manifestanalyzer.AcceptanceIssue, 0, len(fidelity.Divergences))
	for _, divergence := range fidelity.Divergences {
		issues = append(issues, manifestanalyzer.AcceptanceIssue{
			Kind: manifestanalyzer.IssueRenderDoesNotMatchLive,
			Path: filePath,
			// A live value that diverges from what the folder renders is out-of-band
			// substitution, not a render artifact, so whoever owns the deployment
			// pipeline can reconcile the two.
			Solvable: true,
			Actor:    manifestanalyzer.ActorPlatformOperator,
			Field:    divergence.Field,
			Token:    divergence.Token,
			Message: fmt.Sprintf("%s/%s in %s: rendered token %q at %s does not match live",
				id.Kind, id.Name, filePath, divergence.Token, divergence.Field),
		})
	}
	return &manifestanalyzer.AcceptanceRefusedError{Issues: issues}
}

// renderPrecondition is the oracle: it runs once the whole plan is known and before a byte is
// touched, so a refusal aborts the flush and commits nothing. It only runs when the flush routed
// something through a kustomization, so a repo with no override chain pays nothing.
//
// A refusal is an AcceptanceRefusedError, surfacing as GitPathAccepted=False / Stalled=True with
// the file and object named. Silently not mirroring a resource is the failure this path exists to
// prevent, so it must not be the failure it introduces.
func (wb *writeBatch) renderPrecondition() error {
	if !wb.putToKustomize {
		return nil
	}

	before := make([]manifestedit.FileContent, 0, len(wb.contentByPath))
	for _, path := range sortedContentKeys(wb.contentByPath) {
		before = append(before, manifestedit.FileContent{Path: path, Content: wb.contentByPath[path]})
	}

	var refused *manifestanalyzer.RenderRefusedError
	verifyOptions := manifestanalyzer.VerifyOptions{
		NamespaceSuppliedDownstream: wb.namespaces.declaresNamespaceFree(),
	}
	if err := manifestanalyzer.VerifyBatchRenders(before, wb.files(), wb.intents, verifyOptions); err != nil {
		if errors.As(err, &refused) {
			issues := make([]manifestanalyzer.AcceptanceIssue, 0, len(refused.Reasons))
			for _, reason := range refused.Reasons {
				issues = append(issues, manifestanalyzer.AcceptanceIssue{
					Kind:    manifestanalyzer.IssueRenderRefused,
					Message: reason,
					// This refuses a WRITE, not a folder. Nobody can solve it from the
					// repository or the GitTarget: the oracle refuses a write it cannot
					// vouch for, and neither side can make it vouch.
					Solvable: false,
				})
			}
			return &manifestanalyzer.AcceptanceRefusedError{Issues: issues}
		}
		return err
	}
	return nil
}

// intend records what one document of this flush must render to, so the oracle can tell a
// change the flush MEANT from a change it merely caused. Everything not intended has to
// come out of the render untouched.
// recordDocumentRemoval marks one document leaving the subtree: on the file it left, and on the
// batch. Both are evidence for spec.onRefusal and nothing else — see holdsExistingDocument.
func (wb *writeBatch) recordDocumentRemoval(buf *fileBuffer) {
	if buf != nil {
		buf.removedDocument = true
	}
	wb.removedDocuments++
}

func (wb *writeBatch) intend(in manifestanalyzer.WriteIntent) {
	if in.Kind == "" || in.Name == "" {
		return // nothing addressable to check; the render comparison keys on kind+name
	}
	wb.intents = append(wb.intents, in)
}

// intentFor builds the intent for an object-bearing write: the document must render to exactly
// the live object.
//
// It takes the LIVE object, not the event. createNew strips metadata.namespace when the
// destination inherits it from a transformer, but the RENDER then has it, so an intent built from
// the stripped object would demand a render without one and the oracle would refuse a perfectly
// planned flush. The bytes and the intent are different objects.
func intentFor(live *unstructured.Unstructured, filePath string, governed bool) manifestanalyzer.WriteIntent {
	desired := manifestreport.Project(live)
	return manifestanalyzer.WriteIntent{
		SourcePath: filePath,
		Kind:       desired.GetKind(),
		Name:       desired.GetName(),
		Desired:    desired,
		Governed:   governed,
	}
}

// unchecked marks a write whose rendered form cannot be predicted, so the oracle lets the
// object move without comparing it (but still holds the write to disturbing nothing else).
func markUnchecked(in manifestanalyzer.WriteIntent, unchecked bool) manifestanalyzer.WriteIntent {
	if unchecked {
		in.Unchecked = true
		in.Desired = nil
	}
	return in
}

// fieldPatchIntent declares a bounded field patch. It is always unchecked: the event carries
// a handful of audited assignments, not a whole object, so there is nothing to require the
// render to equal.
func fieldPatchIntent(filePath string, id manifestedit.Identity, governed bool) manifestanalyzer.WriteIntent {
	return manifestanalyzer.WriteIntent{
		SourcePath: filePath,
		Kind:       id.Kind,
		Name:       id.Name,
		Unchecked:  true,
		Governed:   governed,
	}
}

// files is the batch's complete tree as the flush would leave it: the worktree bytes with
// every buffer folded over them, and deleted files removed. Sorted, so the render is
// reproducible.
func (wb *writeBatch) files() []manifestedit.FileContent {
	byPath := make(map[string][]byte, len(wb.contentByPath)+len(wb.buffers))
	for path, content := range wb.contentByPath {
		byPath[path] = content
	}
	for path, b := range wb.buffers {
		if b.current == nil {
			delete(byPath, path) // the flush deletes this file
			continue
		}
		byPath[path] = b.current
	}
	paths := make([]string, 0, len(byPath))
	for path := range byPath {
		paths = append(paths, path)
	}
	sort.Strings(paths)
	out := make([]manifestedit.FileContent, 0, len(paths))
	for _, path := range paths {
		out = append(out, manifestedit.FileContent{Path: path, Content: byPath[path]})
	}
	return out
}

func sortedContentKeys(m map[string][]byte) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// applyOverrideEdits folds routed override edits into their kustomization file
// buffers, so they flush (and hit the .gittargetignore shadow precondition)
// exactly like any other planned write. It reports whether any buffer changed.
// A skipped edit (drifted or missing entry) is logged and dropped — the source
// file was deliberately left in its source form, so the next event or resync
// re-decides against the changed kustomization rather than guessing now.
func (wb *writeBatch) applyOverrideEdits(
	ctx context.Context,
	event Event,
	edits []manifestanalyzer.OverrideEdit,
) bool {
	if len(edits) == 0 {
		return false
	}
	byPath := map[string][]manifestanalyzer.OverrideEdit{}
	paths := make([]string, 0, len(edits))
	for _, e := range edits {
		if _, seen := byPath[e.KustomizationPath]; !seen {
			paths = append(paths, e.KustomizationPath)
		}
		byPath[e.KustomizationPath] = append(byPath[e.KustomizationPath], e)
	}
	sort.Strings(paths)

	changed := false
	for _, p := range paths {
		buf := wb.buffer(p)
		if buf.current == nil {
			continue // the kustomization vanished within this batch; nothing to edit
		}
		// A skipped or no-change edit leaves buf.current as it is; a patch updates it. Folded
		// here so PatchKustomization (edit an existing entry) and AppendKustomizationOverride
		// (author a new one) share one place that advances the buffer and logs.
		fold := func(res manifestedit.EditResult, diags []manifestedit.Diagnostic, n int, msg string) {
			switch res.Mode {
			case manifestedit.EditPatched:
				buf.current = res.Content
				changed = true
				log.FromContext(ctx).Info(msg,
					"kustomization", p, "resource", event.Identifier.String(), "edits", n)
			case manifestedit.EditNoChange:
				// Another event in this batch already landed the same value.
			case manifestedit.EditSkipped, manifestedit.EditWholeReplace, manifestedit.EditDeleted:
				logManifestDiagnostics(ctx, diags)
			}
		}

		var patches []manifestedit.KustomizationEdit
		var creates []manifestanalyzer.OverrideEdit
		for _, e := range byPath[p] {
			if e.Create {
				creates = append(creates, e)
			} else {
				patches = append(patches, e.Edit)
			}
		}
		if len(patches) > 0 {
			res, diags := manifestedit.PatchKustomization(p, buf.current, patches)
			fold(res, diags, len(patches), "Routed live change to kustomization override")
		}
		// Each authored entry is applied on its own (the writer creates one image component or
		// replica count at a time), reading buf.current so several land in the same file.
		for _, c := range creates {
			res, diags := manifestedit.AppendKustomizationOverride(
				p, buf.current, c.Edit.Section, c.Edit.EntryName, c.Edit.Field, c.Edit.Value)
			fold(res, diags, 1, "Authored kustomization override for a base-supplied value")
		}
	}
	return changed
}

// gitDocRawObject parses one document of a managed file into JSON-typed maps for
// the override projection. parsed is false for an out-of-range index or a body
// that is not a YAML mapping — the projection then simply does not run.
func gitDocRawObject(content []byte, idx int) (map[string]interface{}, bool) {
	body, ok := manifestedit.DocumentBody(content, idx)
	if !ok {
		return nil, false
	}
	raw := map[string]interface{}{}
	if err := sigsyaml.Unmarshal(body, &raw); err != nil {
		return nil, false
	}
	return raw, true
}

// writeWholeFile renders the event's clean content (sanitized, or SOPS-encrypted for a
// sensitive resource) and writes it wholesale at rel: the built-in placement path for a
// new resource, or the existing file path for a located sensitive resource. It keeps the
// two per-event-writer safety rules: it never overwrites a multi-document file (which
// would drop siblings — splicing a single rendered/encrypted document into a multi-doc
// file is unsupported, so it is refused), and a write that matches the current bytes is a
// no-op (the byte state machine, with the semantic-equality guard for comment-only diffs).
func (wb *writeBatch) writeWholeFile(ctx context.Context, event Event, rel string) (upsertOutcome, error) {
	content, err := wb.writer.buildContentForWrite(ctx, wb.target, event)
	if err != nil {
		if wb.writer.isSensitiveIdentifier(event.Identifier) {
			log.FromContext(ctx).Info(
				"Sensitive resource write skipped because encryption failed",
				"resource", event.Identifier.String(),
				"error", err.Error(),
			)
		}
		return upsertNoChange, err
	}

	buf := wb.buffer(rel)
	isNew := buf.current == nil
	if buf.current != nil {
		if manifestedit.DocumentCount(buf.current) > 1 {
			log.FromContext(ctx).Info(
				"Skipping wholesale write: target holds a multi-document file",
				"file", rel,
				"resource", event.Identifier.String(),
			)
			return upsertSkippedUnsafe, nil
		}
		if bytes.Equal(buf.current, content) || manifestsAreSemanticallyEqual(buf.current, content) {
			return upsertNoChange, nil
		}
	}
	buf.current = content
	if isNew {
		return upsertCreated, nil
	}
	return upsertUpdated, nil
}

// applyDelete mirrors an observed removal: the live DELETE path, located by content so a manifest
// moved off its built-in path is still found. The removal itself is the shared removeDocument, so a
// live delete and a snapshot sweep remove a document the same way.
//
// spec.prune.mode gates the whole path, and the check is FIRST, before the document is located: a
// suppressed delete must be indistinguishable from the event never having arrived, so it touches
// no buffer and cannot turn the kustomize oracle on.
func (wb *writeBatch) applyDelete(ctx context.Context, event Event) error {
	if !wb.pruneMode.OrDefault().AppliesEventDeletes() {
		log.FromContext(ctx).V(1).Info("source DELETE not mirrored (spec.prune.mode)",
			"pruneMode", string(wb.pruneMode.OrDefault()), "resource", event.Identifier.Key())
		return nil
	}
	target, found := wb.resolveDelete(event)
	if !found {
		// No document holds it, but a configMapGenerator may render it.
		return wb.deleteGenerated(ctx, event)
	}
	_, err := wb.removeDocument(ctx, event.Identifier, target, documentDeletedLive)
	return err
}

// removeDocument takes one managed document out of what the target renders. It is the ONE removal
// both the live DELETE and the snapshot sweep use; each keeps its own prune gate and its own
// counter (outcome), and nothing else differs.
//
//   - An object the overlay INHERITS from its read-only base gets an owned `$patch: delete` in
//     the overlay; see inherited_delete.go.
//   - Anything else is removed from its own file, preserving sibling documents. When the file
//     empties, every resources: entry naming it goes too, so the root still builds.
//
// Every removal inside a render root declares a Removed intent and goes to the re-render oracle,
// which proves the object left the render and that no unrelated rendered object changed. It
// reports whether a document was removed; one already absent from the file or from the render is
// a no-op.
func (wb *writeBatch) removeDocument(
	ctx context.Context,
	identifier types.ResourceIdentifier,
	target deleteTarget,
	outcome string,
) (bool, error) {
	if !wb.rendersInBatch(target.filePath, target.id.Kind, target.id.Name) {
		// Already out of the render, whoever's $patch: delete did it. There is nothing to remove.
		return false, nil
	}
	if d, inherited := wb.inheritedDeleteFor(target.filePath, target.id); inherited {
		removed, err := wb.authorInheritedDelete(ctx, identifier, target, d)
		if removed {
			wb.tallyDocument(identifier, outcome)
		}
		return removed, err
	}
	buf := wb.buffer(target.filePath)
	if buf.current == nil {
		return false, nil
	}
	idx, ok := currentDocIndex(target.filePath, buf.current, target.id)
	if !ok {
		return false, nil
	}
	wb.intend(manifestanalyzer.WriteIntent{
		SourcePath: target.filePath,
		Kind:       target.id.Kind,
		Name:       target.id.Name,
		Removed:    true,
	})
	// Any delete inside a render root changes what that root renders, so it goes to the
	// oracle — whether or not the file itself goes with it. Gating this on "the file was
	// emptied" would be a rule about our own bookkeeping rather than about the render, and
	// the whole point of the oracle is that it does not take our word for anything.
	if len(wb.kustomizationsListing(target.filePath)) > 0 {
		wb.putToKustomize = true
	}
	wb.tallyDocument(identifier, outcome)
	wb.recordDocumentRemoval(buf)
	res, _ := manifestedit.DeleteDocument(buf.current, idx)
	if !res.FileEmpty {
		buf.current = res.Content
		return true, nil
	}
	// The file is gone. Anything still naming it in a resources: list now names a file that
	// does not exist, and kustomize refuses to build over that — so deleting the manifest is
	// only half the delete.
	buf.current = nil
	wb.dropKustomizationResource(ctx, identifier, target.filePath)
	return true, nil
}

// dropKustomizationResource removes the resources: entry naming a file this flush deleted,
// from every supported kustomization that lists it.
//
// It is the counterpart of appendKustomizationResource, and it fails the same way: a
// kustomization it cannot edit only loses its entry (logged), it does not abort the delete —
// the render precondition is what decides whether the resulting tree is committable.
func (wb *writeBatch) dropKustomizationResource(
	ctx context.Context, identifier types.ResourceIdentifier, filePath string,
) {
	for _, listing := range wb.kustomizationsListing(filePath) {
		buf := wb.buffer(listing.kustomization)
		if buf.current == nil {
			continue // the kustomization is itself being deleted in this batch
		}
		res, diags := manifestedit.RemoveKustomizationResource(listing.kustomization, buf.current, listing.entry)
		switch res.Mode {
		case manifestedit.EditPatched:
			buf.current = res.Content
			log.FromContext(ctx).Info("Removed resources: entry for deleted file",
				"kustomization", listing.kustomization, "entry", listing.entry,
				"resource", identifier.String())
		case manifestedit.EditNoChange:
		case manifestedit.EditSkipped, manifestedit.EditDeleted, manifestedit.EditWholeReplace:
			// The entry stays, and it now names a file that does not exist. We do not have to
			// decide how bad that is: the render precondition rebuilds the tree and refuses
			// the flush, because kustomize will not build over a missing resource.
			log.FromContext(ctx).Info("Could not remove resources: entry for deleted file",
				"kustomization", listing.kustomization, "entry", listing.entry,
				"resource", identifier.String())
			logManifestDiagnostics(ctx, diags)
		}
	}
}

// resourceListing is one kustomization's resources: entry naming a given file.
type resourceListing struct {
	kustomization string // the kustomization.yaml's own path
	entry         string // the entry text, relative to the kustomization's directory
}

// kustomizationsListing returns every supported kustomization whose resources: names filePath,
// in deterministic order. It is the one definition of "this file is inside a render root",
// shared by the oracle's trigger and by the entry removal, so the two cannot drift apart.
func (wb *writeBatch) kustomizationsListing(filePath string) []resourceListing {
	dirs := make([]string, 0, len(wb.store.Kustomizations))
	for dir := range wb.store.Kustomizations {
		dirs = append(dirs, dir)
	}
	sort.Strings(dirs)

	var out []resourceListing
	for _, dir := range dirs {
		info := wb.store.Kustomizations[dir]
		if info.Unsupported {
			continue // never edit a kustomization we do not model
		}
		base := path.Dir(info.Path)
		for _, entry := range info.Resources {
			if path.Clean(path.Join(base, entry)) == filePath {
				out = append(out, resourceListing{kustomization: info.Path, entry: entry})
			}
		}
	}
	return out
}

// deleteTarget names the file and manifest identity a delete targets. The document
// position is re-derived from the live bytes at apply time because a multi-document
// file's indices can shift within a batch.
type deleteTarget struct {
	filePath string
	id       manifestedit.Identity
}

// resolveDelete locates the managed document a DELETE event targets, content-first:
//
//  1. A delete event that still carries its object is matched by manifest identity,
//     so it follows a moved manifest (the placement guarantee the per-event writer had).
//  2. A GVR-only delete is resolved through PlanDelete's resource-identity inventory.
//
// found is false when Git holds no managed document for the resource.
func (wb *writeBatch) resolveDelete(event Event) (deleteTarget, bool) {
	if id, ok := manifestIdentity(event.Object); ok {
		if dm := wb.store.ByManifestIdentity[id]; dm != nil {
			return deleteTarget{filePath: wb.docLoc[dm].FilePath, id: rawManifestIDForCurrentBytes(id, dm)}, true
		}
	}
	action, emitted := manifestanalyzer.PlanDelete(wb.store, event.Identifier)
	if emitted {
		id := action.Identity
		if dm := wb.store.ByManifestIdentity[id]; dm != nil {
			id = rawManifestIDForCurrentBytes(id, dm)
		}
		return deleteTarget{filePath: action.Ref.FilePath, id: id}, true
	}
	return deleteTarget{}, false
}

// rawManifestIDForCurrentBytes maps an effective manifest identity back to the raw identity as
// written: where the namespace came from anywhere but the file, the document is located by a
// namespace-less identity.
//
// It reads the DOCUMENT, never spec.serializeNamespace: the setting says what the NEXT write will
// contain, and a folder written before it changed still has to be found.
func rawManifestIDForCurrentBytes(
	id manifestedit.Identity,
	dm *manifestanalyzer.DocumentModel,
) manifestedit.Identity {
	if dm != nil && dm.NamespaceAbsentFromFile() {
		id.Namespace = ""
	}
	return id
}

// currentDocIndex re-derives the position of the managed document for id within the
// file's live bytes. The pre-batch store index can go stale when an earlier event in the
// same batch shifts a multi-document file (a delete drops a document, renumbering its
// successors), so any edit/delete recomputes the position against the current bytes
// rather than trusting the index captured at scan time. ok is false when no document of
// that identity is present in the bytes (e.g. already removed earlier in the batch).
func currentDocIndex(filePath string, content []byte, id manifestedit.Identity) (int, bool) {
	inv, _ := manifestedit.IndexFile(filePath, content)
	loc, ok := inv.Location(id)
	return loc.DocumentIndex, ok
}

// flush writes every dirty buffer and removes every deleted one, staging each change.
//
// Before touching a byte it enforces the write-plan precondition: no path about to be written may
// be shadowed by the active .gittargetignore. A precondition rather than a detector, so the
// unrecoverable state (an ignored file the operator can no longer see) is never reached.
// See docs/spec/gitpath-foreign-content-stringency.md §4.3.
func (wb *writeBatch) flush(ctx context.Context, worktree *gogit.Worktree, root, base string) (bool, error) {
	// Write-plan preconditions run before any byte is touched, so a violation aborts the
	// whole flush and commits nothing (each reuses the existing "refusal aborts before a file
	// is written" seam). They enforce, at the one moment the planned paths are known, the two
	// write-boundary invariants the operator must never break: the .gittargetignore shadow
	// guard (§4.3), the L1 write-scope jail (writes stay inside spec.path), and the L2
	// write-fan-in = 1 rule (never write a live change through into context shared by more
	// than one render root). See
	// docs/design/support-boundary/gittarget-granularity-and-cross-environment-edits.md §1.
	if err := wb.ignoreShadowPrecondition(); err != nil {
		return false, err
	}
	if err := wb.pathScopePrecondition(); err != nil {
		return false, err
	}
	if err := wb.fanInPrecondition(); err != nil {
		return false, err
	}
	if err := wb.renderPrecondition(); err != nil {
		return false, err
	}
	logger := log.FromContext(ctx)
	changed := false
	for _, rel := range sortedBufferKeys(wb.buffers) {
		buf := wb.buffers[rel]
		worktreePath := path.Join(base, rel)
		fullPath := filepath.Join(root, base, rel)
		switch {
		case buf.deleted():
			if _, err := removeFileFromWorktree(logger, worktreePath, fullPath, worktree); err != nil {
				return changed, err
			}
			changed = true
		case buf.dirty():
			if err := writeAndStageFile(worktree, worktreePath, fullPath, buf.current); err != nil {
				return changed, err
			}
			changed = true
		}
	}
	// The bytes are on disk. Only now is the census true: a batch that aborted on a precondition
	// above, or failed mid-write, publishes nothing.
	wb.publishDocumentTally(ctx)
	return changed, nil
}

// ignoreShadowPrecondition tests every planned write in the batch — a created or edited
// file (dirty) and a removed file (deleted) — against the active .gittargetignore matcher.
// On a match it returns an *AcceptanceRefusedError carrying one IssueIgnoreShadowsManaged
// per shadowed path, naming both the path and the matching pattern, so the whole flush is
// aborted and nothing is committed. The write path is unknowable ahead of time (it is
// dynamic and, with configurable placement, templated) but perfectly known here, and the
// matcher is already loaded — so this O(touched files) check is the airtight guarantee that
// static analysis cannot give. A path with no active matcher can never be shadowed.
func (wb *writeBatch) ignoreShadowPrecondition() error {
	if wb.store.Ignore == nil {
		return nil
	}
	var issues []manifestanalyzer.AcceptanceIssue
	for _, rel := range sortedBufferKeys(wb.buffers) {
		buf := wb.buffers[rel]
		if !buf.dirty() && !buf.deleted() {
			continue
		}
		// The .gittargetignore matcher is scoped to spec.path and matches spec.path-relative
		// paths, but a buffer is keyed relative to the render anchor. Translate it back to the
		// write scope before matching; a buffer outside the write jail is refused by the
		// path-scope precondition and never reaches a legitimate ignore match here.
		specRel := relUnder(wb.writeSubdir, rel)
		if pattern := wb.store.Ignore.MatchingPattern(specRel, false); pattern != "" {
			issues = append(issues, manifestanalyzer.AcceptanceIssue{
				Kind: manifestanalyzer.IssueIgnoreShadowsManaged,
				Path: rel,
				// The same answer the parse-time denylist gives, and it has to be given
				// again here: a pattern narrow enough to pass the initial scan can still
				// match a write planned later, and that refusal reaches the same reader.
				// Narrowing the pattern is the author's to do.
				Solvable: true,
				Actor:    manifestanalyzer.ActorRepositoryAuthor,
				Message: fmt.Sprintf(
					"%s pattern %q shadows the managed write path %s; the operator would be blind to its own "+
						"file. Remove the pattern or move the resource out of its match",
					manifestanalyzer.GitTargetIgnoreFileName, pattern, rel),
			})
		}
	}
	if len(issues) == 0 {
		return nil
	}
	return &manifestanalyzer.AcceptanceRefusedError{Issues: issues}
}

// pathScopePrecondition enforces the L1 write-boundary invariant: every planned write stays
// inside the GitTarget write scope (spec.path). It tests each created/edited (dirty) or
// removed (deleted) buffer path and refuses the whole flush — one IssueWriteEscapesScope per
// offender — if the base-relative path is absolute or escapes the subtree via "..". Reading
// shared context outside the scope is legitimate (the analyzer follows ../../base); writing
// outside it never is. Planned paths are base-relative by construction today (scan Rel plus
// ".."-free placement validation), so this is defense-in-depth made explicit and tested,
// symmetric to ignoreShadowPrecondition: it never write-then-detects.
func (wb *writeBatch) pathScopePrecondition() error {
	var issues []manifestanalyzer.AcceptanceIssue
	for _, rel := range sortedBufferKeys(wb.buffers) {
		buf := wb.buffers[rel]
		if !buf.dirty() && !buf.deleted() {
			continue
		}
		if wb.writePathEscapesScope(rel) {
			issues = append(issues, manifestanalyzer.AcceptanceIssue{
				Kind: manifestanalyzer.IssueWriteEscapesScope,
				Path: rel,
				// Evidence for spec.onRefusal: the folder already holds this document and this
				// write is not removing it, so re-applying corrects rather than prunes.
				ExistingDocument: wb.holdsExistingDocument(buf),
				// Widening spec.path, or re-placing the write, is the GitTarget owner's call.
				Solvable: true,
				Actor:    manifestanalyzer.ActorPlatformOperator,
				Message: fmt.Sprintf(
					"planned write path %q escapes the GitTarget write scope: the operator only ever writes "+
						"inside spec.path (reads may reach shared context such as ../../base, writes never leave it)",
					rel),
			})
		}
	}
	if len(issues) == 0 {
		return nil
	}
	return &manifestanalyzer.AcceptanceRefusedError{Issues: issues}
}

// writePathEscapesScope reports whether a render-anchor-relative planned write path would
// land outside the GitTarget write scope — an empty path (no destination), an absolute path,
// one whose cleaned form climbs above the anchor with "..", or (when render-root scoping
// re-rooted the scan past spec.path) one that is not within the write jail writeSubdir. A
// read-only base the overlay renders is scanned but never written: its path is outside
// writeSubdir, so a planned write to it is refused here rather than corrupting shared context.
func (wb *writeBatch) writePathEscapesScope(rel string) bool {
	if rel == "" || path.IsAbs(rel) {
		return true
	}
	clean := path.Clean(rel)
	if clean == ".." || strings.HasPrefix(clean, "../") {
		return true
	}
	return wb.writeSubdir != "" && !pathWithin(clean, wb.writeSubdir)
}

// fanInPrecondition enforces the L2 write-boundary invariant: never write through into a source
// file that more than one kustomize render root reaches. Any file two roots read is refused for
// in-place editing, whether or not an images/replicas entry is at stake. It fires only on an
// actual planned write, so a base reached by a single overlay is edited through normally.
func (wb *writeBatch) fanInPrecondition() error {
	var issues []manifestanalyzer.AcceptanceIssue
	for _, rel := range sortedBufferKeys(wb.buffers) {
		buf := wb.buffers[rel]
		if !buf.dirty() && !buf.deleted() {
			continue
		}
		if wb.store.OverridesAmbiguousAt(rel) || wb.store.ReachedByMultipleRenderRoots(rel) {
			issues = append(issues, manifestanalyzer.AcceptanceIssue{
				Kind: manifestanalyzer.IssueWriteFanIn,
				Path: rel,
				// See pathScopePrecondition: evidence, not a description.
				ExistingDocument: wb.holdsExistingDocument(buf),
				// Nobody can solve this from the repository or the GitTarget: the edit
				// has nowhere safe to land while two render roots share the file.
				Solvable: false,
				Message: fmt.Sprintf(
					"planned write to %q would edit in place a source file that more than one kustomize render "+
						"root reaches (write-fan-in must be 1); refusing rather than "+
						"writing the change through into context shared by multiple render roots",
					rel),
			})
		}
	}
	if len(issues) == 0 {
		return nil
	}
	return &manifestanalyzer.AcceptanceRefusedError{Issues: issues}
}

// writeAndStageFile writes a file's bytes to disk (creating parent directories) and
// stages it in the worktree.
//
// A write that fails removes the directories it had just created for it. Nothing downstream can:
// Git does not track empty directories, so they produce no worktree status entry, and
// discardWorktreeLeftovers therefore never learns they exist — a later reset would leave the
// folder sitting there while the state machine reported the worktree clean. It is cheaper to undo
// it here, where the set of directories this write created is still known, than to carry that set
// across the reset boundary.
func writeAndStageFile(worktree *gogit.Worktree, worktreePath, fullPath string, content []byte) error {
	created, err := mkdirAllTrackingCreated(filepath.Dir(fullPath))
	if err != nil {
		return wrapPathErr("create directory for", worktreePath, err)
	}
	// fullPath is an internally derived repo path: the GitTarget segment is run
	// through sanitizePath and the rest comes from the resource's API identity or a
	// content-indexed worktree file, joined under the worktree root — not external input.
	if err := os.WriteFile(fullPath, content, 0o600); err != nil {
		removeCreatedDirs(created)
		return wrapPathErr("write file", worktreePath, err)
	}
	if _, err := worktree.Add(worktreePath); err != nil {
		return wrapPathErr("add file", worktreePath, err)
	}
	return nil
}

// mkdirAllTrackingCreated is os.MkdirAll that reports which directories it actually had to make,
// deepest first, so a failed write can undo exactly its own additions and nothing else.
func mkdirAllTrackingCreated(dir string) ([]string, error) {
	var missing []string
	for probe := dir; ; probe = filepath.Dir(probe) {
		if _, err := os.Stat(probe); err == nil {
			break
		} else if !os.IsNotExist(err) {
			return nil, err
		}
		missing = append(missing, probe)
		if parent := filepath.Dir(probe); parent == probe {
			break
		}
	}
	if err := os.MkdirAll(dir, 0o750); err != nil {
		// MkdirAll creates ancestors before the path it was asked for, so a failure deeper down
		// can still have left some of `missing` behind. Undo whatever it managed before giving up:
		// the caller only learns about the error, and an empty directory produces no worktree
		// status entry for anything later to find.
		removeCreatedDirs(missing)
		return nil, err
	}
	return missing, nil
}

// removeCreatedDirs deletes the directories a failed write created, deepest first and only while
// they are empty. A directory that something else has since put a file in is left alone.
func removeCreatedDirs(created []string) {
	for _, dir := range created {
		entries, err := os.ReadDir(dir)
		if err != nil {
			// Not there, not readable, or not a name this filesystem can hold. Every one of those
			// says there is nothing HERE to remove and says nothing about the ancestors above it,
			// so keep walking rather than stopping.
			//
			// Stopping was a real bug: the deepest entry is the one MkdirAll failed on, and when
			// it failed because the name was too long, ReadDir answers ENAMETOOLONG rather than
			// ENOENT. Treating that as "something else is in there" abandoned every ancestor the
			// call had just created, which is precisely the state this function exists to undo.
			//
			// Walking past it is safe because os.Remove is the real guard: it refuses a directory
			// that is not empty, so a directory somebody else filled survives whatever this loop
			// believes about it.
			continue
		}
		if len(entries) > 0 {
			// Something else is in there, so neither this directory nor any ancestor of it is
			// empty, and none of them are ours to remove.
			return
		}
		_ = os.Remove(dir)
	}
}

// scanWorktreeSubtree walks the GitTarget subtree into a FolderScan, applying the SAME
// ClassifyEntry policy the analyzer uses so the live writer and a dry-run scan agree.
//
// A missing base directory yields an empty scan. Unlike the analyzer scan, a mid-walk read error
// is FATAL: planning against a partial view would re-create an unreadable managed file it skipped
// and churn the mirror. Symlinks are never followed.
func scanWorktreeSubtree(absBase string) (manifestanalyzer.FolderScan, error) {
	ignore, ignoreIssues := loadWorktreeGitTargetIgnore(absBase)
	scan := manifestanalyzer.FolderScan{Ignore: ignore, IgnoreIssues: ignoreIssues}

	walkErr := filepath.WalkDir(absBase, func(p string, d os.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if p == absBase {
			return nil
		}
		rel, relErr := filepath.Rel(absBase, p)
		if relErr != nil {
			return relErr
		}
		rel = filepath.ToSlash(rel)
		switch manifestanalyzer.ClassifyEntry(rel, d, ignore) {
		case manifestanalyzer.RoleSkipDir:
			return filepath.SkipDir
		case manifestanalyzer.RoleManagedYAML:
			content, readErr := os.ReadFile(p) //nolint:gosec // scanning the GitTarget worktree subtree is the feature
			if readErr != nil {
				return readErr
			}
			scan.YAMLFiles = append(scan.YAMLFiles, manifestedit.FileContent{Path: rel, Content: content})
		case manifestanalyzer.RoleOperatorArtifact, manifestanalyzer.RoleBenignPassenger:
			scan.NonYAML = append(scan.NonYAML, rel)
		case manifestanalyzer.RoleForeignFile:
			scan.NonYAML = append(scan.NonYAML, rel)
			scan.Foreign = append(scan.Foreign, manifestanalyzer.ForeignEntry{
				Path: rel, Kind: manifestanalyzer.ForeignFile,
			})
		case manifestanalyzer.RoleForeignSymlink:
			scan.Foreign = append(scan.Foreign, manifestanalyzer.ForeignEntry{
				Path: rel, Kind: manifestanalyzer.ForeignSymlink,
			})
		case manifestanalyzer.RoleIgnored, manifestanalyzer.RoleDescend:
			// Ignored content is never read; a normal directory is simply descended.
		}
		return nil
	})
	if walkErr != nil && !os.IsNotExist(walkErr) {
		return manifestanalyzer.FolderScan{}, walkErr
	}
	sort.Slice(scan.YAMLFiles, func(i, j int) bool { return scan.YAMLFiles[i].Path < scan.YAMLFiles[j].Path })
	sort.Strings(scan.NonYAML)
	sort.Slice(scan.Foreign, func(i, j int) bool { return scan.Foreign[i].Path < scan.Foreign[j].Path })
	return scan, nil
}

// loadWorktreeGitTargetIgnore reads and parses the one honoured .gittargetignore at the
// subtree root. A missing file is the common case and yields a nil matcher with no issues.
func loadWorktreeGitTargetIgnore(absBase string) (*manifestanalyzer.IgnoreMatcher, []manifestanalyzer.AcceptanceIssue) {
	content, err := os.ReadFile(filepath.Join(absBase, manifestanalyzer.GitTargetIgnoreFileName))
	if err != nil {
		return nil, nil
	}
	return manifestanalyzer.LoadGitTargetIgnore(content)
}

// groupEventsByBaseWithIndex buckets events by their sanitized GitTarget base path, preserving
// arrival order within each bucket, and returns, per base, each event's index in events, so a
// result per event can be put back in order. A grouped commit window is single-target (one base) by
// construction; the grouping stays correct for any future multi-target batch.
func groupEventsByBaseWithIndex(events []Event) (map[string][]Event, map[string][]int) {
	byBase := map[string][]Event{}
	indexes := map[string][]int{}
	for i, event := range events {
		base := sanitizePath(event.Path)
		byBase[base] = append(byBase[base], event)
		indexes[base] = append(indexes[base], i)
	}
	return byBase, indexes
}

// sortedBufferKeys returns the buffer paths in lexicographic order so flushing is
// deterministic regardless of map iteration order.
func sortedBufferKeys(buffers map[string]*fileBuffer) []string {
	keys := make([]string, 0, len(buffers))
	for k := range buffers {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

// sortedBaseKeys returns the base paths in lexicographic order so subtrees are
// flushed deterministically.
func sortedBaseKeys(byBase map[string][]Event) []string {
	keys := make([]string, 0, len(byBase))
	for k := range byBase {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

// logStoreDiagnostics surfaces store build-time diagnostics of warning level or
// above at low verbosity — the trace for decisions the writer makes silently
// (ambiguity fallbacks, scope mismatches). Info-level index chatter is dropped.
func logStoreDiagnostics(ctx context.Context, diags []manifestedit.Diagnostic) {
	logger := log.FromContext(ctx)
	for _, d := range diags {
		if d.Level == manifestedit.DiagInfo {
			continue
		}
		logger.V(1).Info("manifest store diagnostic",
			"level", d.Level, "reason", d.Reason, "file", d.Path,
			"documentIndex", d.DocumentIndex, "message", d.Message)
	}
}

// logManifestDiagnostics surfaces manifestedit diagnostics at low verbosity so a
// skipped edit is observable without noise on the happy path.
func logManifestDiagnostics(ctx context.Context, diags []manifestedit.Diagnostic) {
	logger := log.FromContext(ctx)
	for _, d := range diags {
		logger.V(1).Info("manifest edit diagnostic",
			"level", d.Level, "file", d.Path, "documentIndex", d.DocumentIndex, "message", d.Message)
	}
}

// wrapPathErr wraps a worktree file operation error with the action and path.
func wrapPathErr(action, p string, err error) error {
	return &pathOpError{action: action, path: p, err: err}
}

type pathOpError struct {
	action string
	path   string
	err    error
}

func (e *pathOpError) Error() string {
	return "failed to " + e.action + " " + e.path + ": " + e.err.Error()
}
func (e *pathOpError) Unwrap() error { return e.err }