package provision

// Reconcile turns bucket instance records into registry routes, for THIS
// agent's own region and pool set — plus, since the multi-region ingress
// design's Part 3, the serving classes every platform box materializes
// wherever they are homed (see "serving classes" below). Spec:
// docs/superpowers/specs/2026-08-04-instance-provisioning-design.md §4.
//
// DNS seam (spec §10): multi-region deployments write the per-slug CNAME at
// provision time in the console's create handler, not here — the reconcile
// pass stays DNS-free so a box never needs Route53 credentials.

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net"
	"os"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// ReconcileConfig is everything one pass reads its world through — the same
// injection-seam shape domains.Config uses (tests supply fixtures, the host
// agent supplies production values).
type ReconcileConfig struct {
	ClaimEnabled bool
	ClaimInstall func(context.Context, ClaimRecord, func(registry.ClaimOwnership) error) error
	ClaimRetire  func(context.Context, registry.Route) error

	// Region is this agent's region code. "" is the unconstrained region —
	// a box that hasn't been told its place serves only region-less
	// records, never someone else's residency promise.
	Region string
	// Pools are the pools this box serves. Empty means the shared "" pool
	// only — NOT "every pool" (poolServed below).
	Pools []string
	// DataDir feeds DefaultDBPath ("/data" in production).
	DataDir string
	// Now is the clock seam; nil = time.Now.
	Now func() time.Time
	// Inbound is this box's inbound-work declaration (CARLOS_INBOUND, design
	// §8). "" and InboundAny are today's behaviour exactly — an inert
	// default, the apps-domain-style gate this repo already uses — and
	// InboundNone means the box serves sidecar work only: no certs, no
	// inbound ports, possibly a private subnet with egress-only networking.
	// Read by ReconcileSidecars' placement gate (matches a record's own
	// "no-inbound" placement request against this box) AND by Reconcile's
	// mint gate below (refuses ALL routed instance minting outright — an
	// ordinary instance record carries no placement vocabulary to match).
	//
	// Callers building this from a raw env reading MUST pass it through
	// NormalizeInbound first (host/agent.go's provisionTickConfig does) —
	// both gates below compare this field with exact string equality
	// against InboundNone, so an un-folded value would defeat them silently.
	Inbound string
	// Layout answers which layout this box's intent names, "" or
	// release.LayoutModeFlat for flat and release.LayoutModeNested for
	// nested, read from the deploy store the pass already has in hand
	// (isolated-exec design §8.6 step 5, brought forward in plan 5): a row
	// minted on a nested box is minted at the nested paths, never flat
	// and then migrated, so a new tenant never runs flat beside nested
	// siblings, not even for one pass. An error defers every mint on this
	// pass rather than guessing: a box that cannot tell its layout must
	// not mint a flat row the fleet may want nested. Nil is flat (tests,
	// and a caller with no intent to read).
	Layout func(ctx context.Context, s release.Store) (string, error)
	// DryRun reports what the pass would do and writes nothing — not to the
	// registry, not anywhere. The backfill mitigation (design §4.3): a
	// production route is adopted only after someone has read what adopting
	// it would mean, because adoption is what grants the removal phase
	// authority over that row.
	DryRun bool
	// ReleasePub verifies the production flag before a trash marker is
	// allowed to retire an app's routes (#340), the same guard and the same
	// key domains.Config carries for the same decision. nil leaves the
	// retirement UNGUARDED rather than disabling it — domains §1c's posture,
	// carried over deliberately so the two passes cannot reach opposite
	// conclusions about the same app: a keyless deployment has no way to
	// make the guard pass, and holding every trash forever there would break
	// hiding entirely. Nothing destructive to the bucket happens on this
	// path; the auto-purge that does still lives in domains, still refuses
	// without a key.
	ReleasePub ed25519.PublicKey
	// ConfigGate answers whether this box can provide the config of the
	// route a record would mint (amadan #213): nil error means yes, an error
	// means do not mint it this pass. It is handed the route the mint would
	// create (routeFor / sidecarRouteFor), so it reads App, Account,
	// Environment and the shape off the one renderer, and it decides for
	// itself which shapes need config at all. A refused record is retried
	// every pass, so a box that gains the key mints on its next tick. The
	// gate owns its own logging and operator-visible record; this pass only
	// honours the answer. Existing routes are not unminted by a refusal: the
	// wake-time gate (activator.ExecBackend.ConfigGate) is what stops one
	// starting. Nil: no gate, every record mints as before.
	ConfigGate func(ctx context.Context, rt registry.Route) error
	// Hold answers whether a record's row on this box belongs to a running
	// move (moves spec §5.2, §9.6, §19 K9) and, if so, the hold status and
	// the op holding it. When ok, a row minted for rec is minted held
	// (status, HoldOp op) at flat paths, never ready and never nested
	// (Phase 1 lands flat, G11); an unheld row of this pass's own is held
	// with SetHoldIf (Upsert never changes a hold on conflict, G10) once a
	// fresh read of the record, under the host's hold lock
	// (WithHostHoldLock), still asks for it, except under a recovery hold
	// (restoring, releasing), which only mints (holdExisting); a row
	// already held, by this op or another, is left as it is. Nil: no move
	// holds anything, main's pass exactly.
	Hold func(rec Record) (status, op string, ok bool)
	// KeepParticipant answers whether a host's claim names this box as a
	// participant of a running operation (claim spec §8, moves §9.6): the
	// removal phase never removes such a row, whatever the record's region
	// says. The operation removes it at its own finalization. Nil: none is.
	KeepParticipant func(host string) bool
	// HoldWritten is called after every hold this pass writes (a held mint
	// or a SetHoldIf that landed), so the caller can touch the
	// stage-requested marker: root re-judges what it staged under the new
	// hold (registry.SetHold's contract, G10). Nil: nobody is told.
	HoldWritten func(host string)
	// RunSeen is called with every in-scope record of an untrashed app that
	// carries a run, once per pass, before the pass acts on the record (run
	// spec §5.2: pickup rides reconcile's own record reads, so an idle box
	// pays nothing extra). It must not block: the box's runs worker only
	// notes the record and acts after the pass, by which time a row this
	// pass minted is in the registry. Nil: no one is told.
	RunSeen func(rec Record)
	// Seeds is the seeded-instances wiring (seedreconcile.go). Nil: every
	// seeded record is left alone.
	Seeds *SeedHooks

	// beforeHoldWrite is the tests' pause under the host's hold lock, just
	// before holdExisting's conditional write (TestReconcileHoldWrite...).
	// Nil in the agent.
	beforeHoldWrite func(host string)
}

// InboundAny and InboundNone are the closed vocabulary ReconcileConfig.Inbound
// is read against — the BOX's half of design §8's two independent halves (the
// record's half is PlacementAny/PlacementNoInbound). "" reads as InboundAny:
// a box that has not been told is a box that serves, which is every box that
// exists today.
const (
	InboundAny  = "any"
	InboundNone = "none"
)

// NormalizeInbound folds a raw CARLOS_INBOUND reading (whitespace and case)
// and maps it onto the closed vocabulary above. Review fix (Task 10
// finding): the box-side comparisons this feeds are exact string equality
// against InboundNone, so an un-folded "None"/"NONE"/"off" fell through as
// if the variable were unset — a box meant to be no-inbound would silently
// get handed traffic it structurally cannot serve, the same silent outage
// this task exists to prevent, just triggered from the other direction.
//
// An UNRECOGNISED value (anything other than "", "any" or "none" once
// folded) is treated as InboundNone — restrictive, not inert — and logged
// loudly, naming the offending value. The only way to produce an
// unrecognised value is to have typed one: CARLOS_INBOUND is never set by
// anything but an operator's own hand, and an operator only sets this
// variable at all when they mean to restrict. So the restrictive reading
// matches intent in the overwhelmingly likely case, and it is this repo's
// house style for an unparseable declaration already (a provision manifest
// with schema > 1 is refused outright, never guessed at). The failure mode
// of guessing wrong here is loud and bounded — new mints are refused with a
// log line naming host and reason, nothing already running is evacuated —
// where the opposite default's failure mode is a box quietly accepting
// traffic it cannot serve.
func NormalizeInbound(raw string) string {
	v := strings.ToLower(strings.TrimSpace(raw))
	switch v {
	case "", InboundAny, InboundNone:
		return v
	default:
		log.Printf("provision: CARLOS_INBOUND=%q is not a recognised value (want \"\", %q or %q); "+
			"treating this box as CARLOS_INBOUND=%s (no-inbound, restrictive) rather than guessing it means unrestricted",
			raw, InboundAny, InboundNone, InboundNone)
		return InboundNone
	}
}

// acctApp identifies one app within one account — the unit both the record
// scan and the manifest/seed cache below key on, since app names are unique
// only within an account, never globally.
type acctApp struct{ acct, app string }

// trashState is what one app's trash marker means to this pass. Two fields
// and not one bool, because the two questions have different safe answers:
// see the trash cache in Reconcile for which is which.
type trashState struct {
	declineMint bool
	removable   bool
}

// productionClear reports whether a trashed app's routes may be retired —
// i.e. whether we can positively see that it is NOT flagged production.
// Dropping every route of a production app is an outage, which is precisely
// what the flag exists to prevent, so an unreadable flag refuses in the same
// direction as an ON one: the state an attacker with delete-but-not-sign can
// manufacture is exactly the unreadable one.
//
// A nil key is the keyless box, which acts on the marker — domains §1c's
// posture, and the comment on ReconcileConfig.ReleasePub says why.
//
// Both refusals log EVERY pass, deliberately: an operator who wrote that
// marker on purpose has to be told why nothing happened, or they will assume
// deletion is broken.
func productionClear(ctx context.Context, s release.Store, pub ed25519.PublicKey, acct, app string) bool {
	if pub == nil {
		return true
	}
	flag, err := release.GetProduction(ctx, s, pub, acct, app)
	switch {
	case err != nil:
		log.Printf("provision: cannot read the production flag for %s/%s — refusing to act on the trash marker: %v", acct, app, err)
		return false
	case flag != nil && flag.Production:
		log.Printf("provision: trash marker on production app %s/%s — refusing to act on it; turn production off first", acct, app)
		return false
	}
	return true
}

// ---- serving classes -------------------------------------------------------
//
// Which boxes materialize a record. The multi-region ingress design's Part 3
// table, verbatim, and the ONE place this pass states it:
//
//	Everywhere          static (bake-served): the edge resolves the app's
//	                    channel pointer and serves the shipped tar out of the
//	                    deployment store every box already reads, so no box
//	                    needs the home box to serve it.
//	Everywhere, probed  addr routes whose upstreams are ALL publicly dialable:
//	                    each edge runs its own prober and dials the backends
//	                    directly, so per-edge live sets are per-edge by design.
//	Home-boxed          exec, unit-owned, socket-activated, tunnel- and
//	                    fleet-backed: the process exists on exactly ONE box,
//	                    and a request that lands elsewhere backhauls to it
//	                    (internal/edge's hop). Unchanged invariant.
//
// Everything here decides SCOPE only. Nothing about the route a record mints
// changes with the box that mints it — routeFor stays the one renderer, so a
// far box's copy and the home box's copy are the same route, column for
// column, including the record's own Region.
//
// Single-box inertness: with one box there is no far box, every record is in
// scope for the same reason it was before, and the widened half of the gate
// never fires.

// everywhereClass reports whether rec's serving CLASS is one every platform
// box materializes. It answers the class question only — the tenancy
// questions (pool, fleet placement) are inScope's, below.
//
// Conservative by construction: anything this function cannot positively
// classify as servable from a box that holds nothing but the bucket is
// home-boxed. That is the direction a wrong answer must fail in — a route that
// stays home-boxed is served exactly as it is today (and backhauled from
// elsewhere), while one wrongly promoted to everywhere is a box proxying to an
// upstream it cannot reach.
func everywhereClass(rec Record) bool {
	// A replica is the residence rule made structural (#162): it belongs to
	// the box its placement names and nowhere else, and this widening is the
	// mint gate for every non-fleet box. ValidateShape refuses a replica
	// with an upstream, so the addr arm below could never fire on one — but
	// "could never through THIS door" is not "cannot", and a replica of a
	// static kind would pass the kind arm above if the kind rail ever
	// moved. The role check is FIRST and absolute.
	if rec.Role == "replica" {
		return false
	}
	// Unit names the systemd unit that owns this route's process (issue
	// #110); adoption CYCLES it. That unit exists on the home box, so a far
	// box adopting the row would be stamping authority over a unit name that
	// means something else there, or nothing at all.
	if rec.Unit != "" {
		return false
	}
	// Belt and braces: ValidateShape already refuses hibernation on both
	// classes below (a static route has no process to wake, an addr-backed one
	// has no socket), and this pass validates every record before the gate. A
	// hibernating route needs the activator and the socket only its home box
	// has, so if that door ever moved, this must not be the hole it opens.
	if rec.HibernateOrDefault() {
		return false
	}
	if rec.KindOrDefault() == "static" {
		return true
	}
	ups := rec.AddrsOrDefault()
	if len(ups) == 0 {
		return false // socket-backed: the exec child lives on one box
	}
	for _, a := range ups {
		if !publiclyDialable(a) {
			return false
		}
	}
	return true
}

// everywhereRoute is everywhereClass read off a registry ROW instead of a
// record — the removal phase's half of the same question, asked when the
// record is gone and the row is all that is left to ask.
//
// The two must agree, or the widened mint gate leaves immortal routes: a far
// box that mints an everywhere-class route and then cannot see it as one is a
// box serving a retired site forever. They agree by construction today because
// routeFor is the one renderer between them, and
// TestEverywhereClassAgreesAcrossRecordAndRow locks that: a class rule added to
// one of these functions alone recreates exactly the immortality bug this pair
// was written to close, with every other test still green.
//
// They are separate functions rather than one over a shared shape because the
// row carries columns no record can state (Backing, Tunnel, Caps), and a row
// that has GROWN one of those since it was minted is deliberately read as
// home-boxed here. State that cost honestly, because it is bigger than a stale
// row: an operator who runs `carlos route -grant` (or -backing/-tunnel) against
// the FAR box's copy takes that copy out of removal scope permanently. Deleting
// the record then retires the route on its home box and leaves the far box
// still SERVING the retired site, grants intact, with no record left anywhere
// to explain it — so `carlos remove` stops being fleet-wide the moment anyone
// touches a far copy. The alternative (removing rows carrying operator columns)
// hands an unattended bucket-record deletion the grant purge registry.Delete
// documents as human-gated, which is worse. The conservative gate stands; the
// gap is tracked in the bookkeeping issue, not papered over here.
func everywhereRoute(rt registry.Route) bool {
	// The replica mark is everywhereClass's role check, read off the row
	// (the record whose class it was may be exactly what is gone) — the
	// same FIRST-and-absolute position, for the same reason.
	if rt.Replica {
		return false
	}
	if rt.Unit != "" || rt.Hibernate || rt.Backing != "" || rt.Tunnel != "" || rt.Caps != "" {
		return false
	}
	if rt.Kind == "static" {
		return true
	}
	ups := rt.AddrList()
	if len(ups) == 0 {
		return false
	}
	for _, a := range ups {
		if !publiclyDialable(a) {
			return false
		}
	}
	return true
}

// publiclyDialable reports whether one upstream is an address every edge in
// the deployment can reach — the "publicly-reachable upstreams" half of the
// spec's table.
//
// A NAME is not dialable for this purpose even if it resolves to a public
// address today: validAddrsElement's doc comment already records why a name
// cannot be judged at all here (it is free to move after the check), and the
// legacy Addr column is full of names pointing at box-local backends. It stays
// home-boxed, which is where those routes are serving from today.
//
// Loopback and link-local are the spec's own examples (another box's
// 127.0.0.1 is not this box's). RFC1918 and ULA are refused for exactly the
// same reason one layer out: a private address is private to ONE network, and
// two boxes in two regions are two networks. IsGlobalUnicast covers the rest
// (unspecified, multicast, loopback, link-local).
func publiclyDialable(addr string) bool {
	h, _, err := net.SplitHostPort(addr)
	if err != nil {
		return false
	}
	ip := net.ParseIP(h)
	if ip == nil {
		return false
	}
	return ip.IsGlobalUnicast() && !ip.IsPrivate()
}

// inScope answers "is this region/pool pair this pass's business" — the one
// gate the record scan and the removal phase both ask, so a record this box
// mints is a record this box can retire.
//
// Pool is compared FIRST and never widened. It is a TENANCY boundary, not a
// geography: a pool names the boxes a tenant's routes are allowed to land on
// (customer fleets, dedicated boxes), and "everywhere" in the spec's table
// means every box that already serves this tenant, not every box in the world.
//
// Region equality is the unchanged home-box test. It is widened for the
// everywhere classes, and only between PLATFORM regions: a fleet placement
// ("fleet/<name>", registry.FleetPlacement) is the other tenancy boundary and
// is refused from both sides — the platform's boxes do not pick up a
// customer fleet's routes, and a customer's box does not pick up the
// platform's.
func (cfg ReconcileConfig) inScope(region, pool string, everywhere bool) bool {
	if !poolServed(cfg.Pools, pool) {
		return false
	}
	if region == cfg.Region {
		return true
	}
	if !everywhere {
		return false
	}
	_, recordOnFleet := registry.FleetPlacement(region)
	_, boxOnFleet := registry.FleetPlacement(cfg.Region)
	return !recordOnFleet && !boxOnFleet
}

// RouteFor is the one answer to "what route does this record describe". The
// mint path builds from it and the adoption match test (design §4.2)
// compares against it — one function, so a shape the reconciler would mint
// and a shape it will adopt can never drift apart.
//
// Socket and DBPath are DERIVED, never declared: DefaultSocket and
// DefaultDBPath exist so a hand-added route and a provisioned one land on
// identical path shapes, and their doc comment records what two names for
// one database cost. Deriving DBPath EXPLICITLY here — rather than leaving it
// empty on a route that has one — is also what keeps ops.FleetDBs'
// hibernating-exclusion correct: internal/ops/litestream.go builds that
// exclusion set from `r.Hibernate && r.DBPath != ""`, so a hibernating route
// whose row carried no path would be swept up by the dataDir glob and
// replicated box-wide as if it never slept.
//
// nested mints the derived paths at the nested layout's helpers instead
// (InstanceDir, NestedSocket): the shape the layout pass would otherwise
// move a flat row to. Only the mint branch passes true, from
// ReconcileConfig.Layout; the drift and adoption comparisons derive flat
// and judge a nested row by NestedPaths.
func RouteFor(rec Record, dataDir, seed string, nested bool, policy *UpdatePolicy) registry.Route {
	rt := registry.Route{
		Host: rec.Host, Slug: rec.Slug, Kind: rec.KindOrDefault(), Unit: rec.Unit,
		App: rec.App, Account: rec.Account, Pool: rec.Pool, Region: rec.Region,
		Channel: rec.Channel, Environment: rec.Environment,
		Hibernate: rec.HibernateOrDefault(), StatusNamed: rec.StatusNamedOrDefault(),
		Steering:    rec.Steering,
		Replicas:    replicaSet(rec.Replicas),
		Replica:     rec.Role == "replica",
		Gate:        rec.Gate,
		Provisioned: true, Seed: seed, Status: "ready",
		// The record's own settings over the app's policy (settings spec
		// §3.2); "" for default settings, so an untouched fleet mints the
		// row it always did.
		UpdateSettings: EffectiveUpdates(rec, policy).Column(),
		SeedSource:     SeedColumn(rec),
	}
	if rec.DerivesPaths() {
		rt.Socket = DefaultSocket(rec.Host)
		rt.DBPath = DefaultDBPath(dataDir, rec.Host)
		// Only an activator-managed row has a nested shape: an unflagged
		// socket route is another supervisor's process, whose paths stay
		// flat in every release (design §8.2 step 1).
		if nested && rt.Hibernate {
			rt.Socket = NestedSocket(rec.Host)
			rt.DBPath = NestedDBPath(dataDir, rec.Host)
		}
	}
	rt.Addr, rt.Addrs, rt.Health = upstreamCols(rec)
	return rt
}

// mintLayoutReader answers "mint nested?" for one pass, asking cfg.Layout
// at most once and only when a mint needs the answer. A nil Layout is
// flat. The answer is cached error and all: a store that failed once this
// pass is not asked again per record.
func mintLayoutReader(ctx context.Context, s release.Store, cfg ReconcileConfig) func() (bool, error) {
	var (
		asked  bool
		nested bool
		err    error
	)
	return func() (bool, error) {
		if cfg.Layout == nil {
			return false, nil
		}
		if !asked {
			asked = true
			var mode string
			mode, err = cfg.Layout(ctx, s)
			nested = err == nil && mode == release.LayoutModeNested
		}
		return nested, err
	}
}

// upstreamCols renders a record's THREE upstream columns exactly as a minted
// route carries them: the legacy single addr, the comma-joined addr set, and
// the probe-config JSON.
//
// It exists as its own function because TWO paths now write these columns —
// the mint (routeFor, above) and the owned-row converge (convergeOwned, which
// calls registry.SetUpstreams) — and a second, independent derivation is how
// a route that was minted correctly comes to be "converged" onto a different
// value every pass, forever. One derivation, two callers.
//
// All three move together because they are one fact seen three ways: addr and
// addrs are two spellings of "where this route proxies" (registry's AddrList
// folds addr in whenever addrs is empty), and health decides which of those
// upstreams the edge will use. A record that migrates from the legacy addr to
// an addrs set — which is exactly what the console's PATCH door does to a
// pre-existing single-upstream tenant — must clear the row's addr in the same
// write, or the row keeps a decommissioned upstream that resurfaces through
// the fold the moment the set is cleared again.
//
// Addrs is read from the record's EXPLICIT field only, never
// AddrsOrDefault's Addr-folding — a legacy Addr-only record must mint (and
// converge to) a byte-identical route to before these fields existed
// (registry.Route.Addrs documents the fold-back that keeps single-upstream
// callers working off Addr alone). Health, by contrast, goes through
// HealthOrDefault: a multi-upstream route with no stated probe MUST be
// probed, or the edge has no way to tell which of its upstreams is dark.
func upstreamCols(rec Record) (addr, addrs, health string) {
	addr = rec.Addr
	if len(rec.Addrs) > 0 {
		addrs = strings.Join(rec.Addrs, ",")
	}
	// WithDefaults, not the raw pointer: a record with explicit-but-partial
	// probe config (e.g. only Path set) must still mint a route whose
	// Health JSON carries the platform's fully-materialized defaults for
	// every field it left at zero. An un-defaulted interval_ms of 0
	// reaching the edge would mean a tight-loop probe against a customer
	// upstream.
	if h := rec.HealthOrDefault(); h != nil {
		withDefaults := h.WithDefaults()
		// Health is a struct of strings and ints — json.Marshal cannot fail
		// on it (no channels, funcs, cyclic pointers, unsupported types).
		// The err == nil guard exists to fail CLOSED on the impossible case
		// rather than panic: leaving health == "" would mean the route
		// mints unprobed, not that a bad value reaches the edge.
		if b, err := json.Marshal(withDefaults); err == nil {
			health = string(b)
		}
	}
	return addr, addrs, health
}

// shapeMatches reports whether an existing route is the same ROUTE the
// record describes — the gate on adoption (design §4.2), and so the gate on
// this pass acquiring removal authority over a row it did not mint.
//
// Channel, Environment, Hibernate and Steering are deliberately absent: those
// are the columns the record is taking ownership OF, and a hand-added route
// on the wrong channel is the whole reason to adopt one. Requiring them to
// agree would make adoption impossible in exactly the case it exists for —
// Steering named explicitly here because a record carrying
// steering:"latency" over an unprovisioned row is a real deployment shape
// (an operator hand-added the route before the console's steering PATCH
// door existed to touch it), and refusing that adoption forever would buy
// nothing: convergeOwned stamps Steering onto the row unconditionally the
// moment adoption succeeds (ruled 2026-08-23, plan amendment b485ec5,
// superseding this task's own first pass, which compared it here).
//
// Backing, Tunnel and Caps are deliberately required to be empty (unset) on
// the existing row — registry.Route's own doc comments name these three as
// the box-scoped operator columns, and a record has a field for none of
// them, so a record can never fully describe a row that carries one:
//   - Backing "unit" names a carlos-app@<host> systemd tenant; adopting
//     across it would claim a process shape the record says nothing about.
//   - Tunnel is routing-decisive and OUTRANKS the Socket/Addr a record can
//     state: internal/edge picks the tunnel proxy before it ever looks at
//     Socket, so a tunnel-backed route can agree with a record on every
//     other compared column while every byte it actually serves goes down
//     a channel the record cannot see or reclaim.
//   - Caps is the operator-granted, box-local capability grant column
//     (port:25, host-role, ...). registry.Delete's purge of a route's
//     grants is already treated as a human-gated, last-resort action
//     elsewhere in this file (the shape-drift log below); adoption must not
//     hand that purge to an unattended bucket-record deletion.
//
// A row this pass cannot state the shape of is a row this pass must not
// claim to describe, no matter how well everything else lines up.
//
// These three refusals guard the adoption INSTANT, not the row forever after:
// `carlos route -backing/-unit/-tunnel/-grant` still work on a Provisioned
// row, so an already-owned route can acquire one of them later. See the
// removal phase's comment in Reconcile for what that costs.
//
// Platform and Canary are likewise required false: those flags mark a row
// domains.Reconcile minted and re-mints/removes on its own authority
// (mintedRoute leaves Slug/Pool/Region/DBPath at their zero values, so a
// platform apex on an unconstrained-region box can otherwise agree with a
// record on every other compared column). Adoption must not hand this
// pass's removal authority to a row another pass already owns and
// periodically rebuilds — nor let it write Channel/Environment onto a row
// the CLI itself refuses (platformCanaryOwnedRow, cli.go) for the same
// reason.
//
// Everything else that identifies the route is compared, including the
// derived Socket and DBPath — two rows agreeing on kind and app but not on
// where the database lives are not the same route, and adopting across that
// difference would point a running process at a second copy of its own data.
func shapeMatches(existing, want registry.Route) bool {
	return shapeMismatch(existing, want) == ""
}

// shapeMismatch names the first column on which existing disagrees with
// want — including the box-scoped and pass-ownership guards shapeMatches
// documents — or "" when every compared column agrees. It backs both
// shapeMatches and the adoption refusal log: printing only a couple of
// columns there (the original message did app/kind/addr) leaves every
// refusal over any OTHER column looking identical and undiagnosable to the
// operator the refusal message exists for.
//
// `replication` is box-scoped and inexpressible in a record exactly like
// Backing/Tunnel/Caps, and is DELIBERATELY not listed here — see the
// asymmetry note above the removal loop below for why refusing adoption over
// it would cost the fleet the backfill and buy nothing.
func shapeMismatch(existing, want registry.Route) string {
	switch {
	case existing.Backing != "":
		return fmt.Sprintf("backing %q (box-scoped, not expressible in a record)", existing.Backing)
	case existing.Tunnel != "":
		return fmt.Sprintf("tunnel %q (box-scoped, not expressible in a record)", existing.Tunnel)
	case existing.Caps != "":
		return fmt.Sprintf("caps %q (box-scoped, not expressible in a record)", existing.Caps)
	case existing.Platform:
		return "platform (owned by the domains reconcile pass)"
	case existing.Canary:
		return "canary (owned by the domains reconcile pass)"
	// Replicas is deliberately absent, replication's twin: the column is
	// CONVERGED in the owned arm (SetReplicas — the record is authoritative
	// in both directions), so a divergence here is repaired, never a reason
	// to refuse adoption. Listing it would opt every route with a settled
	// replica set out of the backfill permanently and invisibly, the exact
	// trade the replication paragraph below refuses — and a replica set
	// grants no authority over the route's process; it names where the
	// account's mirrors live.
	case existing.App != want.App:
		return fmt.Sprintf("app %q/%q", existing.App, want.App)
	case existing.Account != want.Account:
		return fmt.Sprintf("account %q/%q", existing.Account, want.Account)
	case existing.Kind != want.Kind:
		return fmt.Sprintf("kind %q/%q", existing.Kind, want.Kind)
	case existing.Addr != want.Addr:
		return fmt.Sprintf("addr %q/%q", existing.Addr, want.Addr)
	case existing.Addrs != want.Addrs:
		return fmt.Sprintf("addrs %q/%q", existing.Addrs, want.Addrs)
	case existing.Health != want.Health:
		return fmt.Sprintf("health %q/%q", existing.Health, want.Health)
	case existing.Unit != want.Unit:
		return fmt.Sprintf("unit %q/%q", existing.Unit, want.Unit)
	case existing.Slug != want.Slug:
		return fmt.Sprintf("slug %q/%q", existing.Slug, want.Slug)
	case existing.Pool != want.Pool:
		return fmt.Sprintf("pool %q/%q", existing.Pool, want.Pool)
	case existing.Region != want.Region:
		return fmt.Sprintf("region %q/%q", existing.Region, want.Region)
	case existing.Socket != want.Socket:
		return fmt.Sprintf("socket %q/%q", existing.Socket, want.Socket)
	case existing.DBPath != want.DBPath:
		return fmt.Sprintf("db %q/%q", existing.DBPath, want.DBPath)
	default:
		return ""
	}
}

// convergeOwned applies the record's Channel, Environment, StatusNamed,
// UPSTREAM SET and Steering onto a route it owns — the columns the record is
// authoritative for once it owns the row, whether ownership was just granted
// by adoption or was already held by an earlier pass. Both the owned-route
// converge branch and the adoption branch call this, so there is exactly one
// place these columns get written from a record, never two copies drifting
// apart.
//
// Addrs/Health converge UNCONDITIONALLY on divergence — no "only when the
// record states one" guard of the kind Unit carries — and that difference is
// deliberate. Unit has a supported out-of-band writer (`carlos route -unit`,
// issue #110) whose stamp a record that never mentioned the field must not
// treat as drift. The upstream set has no such writer: the record IS the
// desired state, the console's PATCH door is how it changes, and clearing the
// set (PATCH with an empty addrs) has to reach the row or that door would move
// the bucket and nothing else. Both values come from upstreamCols, the same
// derivation the mint uses, so a route minted from a record cannot be
// "converged" onto anything but what it already carries.
//
// Steering converges the same way, for the same reason: it has no supported
// out-of-band writer either (Task 3's console door and CLI both write the
// RECORD, never the row directly), so "" is as much an answer as "latency" —
// a record that clears its steering opt-in must reach the row, or the
// console's PATCH door would move the bucket and nothing else, the exact
// silent-stomp shape the upstream-set paragraph above already documents.
//
// StatusNamed converges only when the record STATES it (design §3, the same
// tri-state Hibernate carries): absent means "this record does not speak to
// the status page", and stamping a derived false over an operator's `carlos
// add -status-named` would be the silent stomp this design exists to end.
// Unlike Hibernate it is converged at the adoption instant too — it decides
// what a public page prints, not whether a live process is allowed to sleep.
//
// Every write logs on success as well as on failure. These are the writes
// that REVERT an operator's own `carlos route -environment` or `-hibernate`;
// a row that moved with nothing saying so is the exact failure class of the
// production incident behind #130, seen from the outside.
//
// dryRun is a parameter here, not left to callers, deliberately: this
// function is reached from two branches, and a dry-run guard living in only
// one of the two callers is exactly the single-unguarded-write failure mode
// the pass exists to prevent (design §4.3). Guarding here means both call
// sites are safe by construction, not by each one remembering to check.
func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, policy *UpdatePolicy, policyKnown bool, dryRun bool) {
	if rec.Channel != "" && existing.Channel != rec.Channel {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s channel %s -> %s", rec.Host, existing.Channel, rec.Channel)
		} else if err := reg.SetChannel(rec.Host, rec.Channel); err != nil {
			log.Printf("provision: converge channel on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s channel %s -> %s", rec.Host, existing.Channel, rec.Channel)
		}
	}
	if existing.Environment != rec.Environment {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s environment %q -> %q", rec.Host, existing.Environment, rec.Environment)
		} else if err := reg.SetEnvironment(rec.Host, rec.Environment); err != nil {
			log.Printf("provision: converge environment on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s environment %q -> %q", rec.Host, existing.Environment, rec.Environment)
		}
	}
	if rec.StatusNamed != nil && existing.StatusNamed != *rec.StatusNamed {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s status-named %v -> %v", rec.Host, existing.StatusNamed, *rec.StatusNamed)
		} else if err := reg.SetStatusNamed(rec.Host, *rec.StatusNamed); err != nil {
			log.Printf("provision: converge status-named on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s status-named %v -> %v", rec.Host, existing.StatusNamed, *rec.StatusNamed)
		}
	}
	// One UPDATE for all three columns, because they are one fact: where the
	// route proxies (addr, addrs — two spellings of the same thing, which
	// registry.AddrList folds together) and the probe that decides which of
	// those upstreams the edge will send traffic to. Writing them separately
	// would leave a window in which a route's new upstreams are live under the
	// OLD probe's path/expected-status — a probe that can mark every new
	// upstream dark — and leaving addr out entirely would leave the row saying
	// two contradictory things forever (see upstreamCols).
	if addr, addrs, health := upstreamCols(rec); existing.Addr != addr ||
		existing.Addrs != addrs || existing.Health != health {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s upstreams addr %q -> %q, addrs %q -> %q (probe %q -> %q)",
				rec.Host, existing.Addr, addr, existing.Addrs, addrs, existing.Health, health)
		} else if err := reg.SetUpstreams(rec.Host, addr, addrs, health); err != nil {
			log.Printf("provision: converge upstreams on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s upstreams addr %q -> %q, addrs %q -> %q (probe %q -> %q)",
				rec.Host, existing.Addr, addr, existing.Addrs, addrs, existing.Health, health)
		}
	}
	if existing.Steering != rec.Steering {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s steering %q -> %q", rec.Host, existing.Steering, rec.Steering)
		} else if err := reg.SetSteering(rec.Host, rec.Steering); err != nil {
			log.Printf("provision: converge steering on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s steering %q -> %q", rec.Host, existing.Steering, rec.Steering)
		}
	}
	// Replicas CONVERGES, where steering merely repoints: the bucket's value
	// is authoritative in BOTH directions, including down to none — a set
	// that shrank must actually shrink on the row, or the edges file goes on
	// publishing a replica placement the account no longer has. This is the
	// one deliberate widening the owned arm permits, and the reason is in
	// Record.Replicas' doc comment: replica placement is the bucket's
	// decision to own, not an operator column to preserve.
	if want := replicaSet(rec.Replicas); existing.Replicas != want {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s replicas %q -> %q", rec.Host, existing.Replicas, want)
		} else if err := reg.SetReplicas(rec.Host, want); err != nil {
			log.Printf("provision: converge replicas on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s replicas %q -> %q", rec.Host, existing.Replicas, want)
		}
	}
	// The replica MARK converges the same both-ways way, from the same
	// record: it decides how the edges file publishes the row (replica
	// mark vs plain home), which is exactly the fact steering's fold sorts
	// hosts by. Upsert would preserve it wrong (a re-mint carrying no role
	// must DEMOTE a row that carried one), so the narrow setter owns it.
	if want := rec.Role == "replica"; existing.Replica != want {
		if dryRun {
			log.Printf("provision: dry-run: would converge %s replica mark %v -> %v", rec.Host, existing.Replica, want)
		} else if err := reg.SetReplica(rec.Host, want); err != nil {
			log.Printf("provision: converge replica mark on %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged %s replica mark %v -> %v", rec.Host, existing.Replica, want)
		}
	}
	// Update settings (settings spec §4): the record over the app policy,
	// both directions, including back to "". Left alone when the policy
	// could not be read this pass.
	if policyKnown {
		if want := EffectiveUpdates(rec, policy).Column(); existing.UpdateSettings != want {
			if dryRun {
				log.Printf("provision: dry-run: would converge %s update settings %q -> %q", rec.Host, existing.UpdateSettings, want)
			} else if err := reg.SetUpdateSettings(rec.Host, want); err != nil {
				log.Printf("provision: converge update settings on %s: %v", rec.Host, err)
			} else {
				log.Printf("provision: converged %s update settings %q -> %q", rec.Host, existing.UpdateSettings, want)
			}
		}
	}
}

// Reconcile materializes bucket instance records into registry routes for
// THIS agent's region and pools: mint on appearance, converge seed, remove
// on record deletion — removal only for routes this pass (or an earlier one
// running the same scope) minted, marked by Provisioned, the same authority
// discipline as domains.Reconcile's canary removal. Per-item failures log
// and skip; a failed account-manifest or record read additionally disables
// the removal phase for the whole pass — a transient read failure must
// never mass-retire a fleet. Minting is additive and always safe, so it
// proceeds regardless.
func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig) error {
	if cfg.Now == nil {
		cfg.Now = time.Now
	}

	// scan is the record-collection phase over st: the in-scope records by
	// host, the hosts whose record is there but cannot be used (protect),
	// and whether anything failed to read (readFailed, which disables
	// removal for the whole pass). The pass scans s; a pass over a record
	// view snapshot that has a removal to decide scans the bucket too
	// (freshStore), so both judge every record by exactly these rules (S3
	// change-index design §4.4).
	type recordScan struct {
		recs       map[string]Record
		protect    map[string]bool
		readFailed bool
	}
	scan := func(st release.Store) (recordScan, error) {
		known, readFailed, err := knownAccounts(ctx, st)
		if err != nil {
			return recordScan{}, err
		}

		// --- collect in-scope records --------------------------------------------
		keys, partial, err := listAppSubtree(ctx, st, known, "instances/")
		if err != nil {
			return recordScan{}, fmt.Errorf("provision: list apps: %w", err)
		}
		if partial {
			readFailed = true // unseen records must not read as retired
		}
		recs := map[string]Record{} // host -> record, in-scope only
		// protect holds hosts whose record KEY parsed (ParseRecordKey succeeded)
		// but whose body could not be trusted — unparseable JSON, or a body that
		// disagrees with its own path. Such a host must NEVER be treated as "the
		// bucket retired it": a non-additive Record encoding change reaching an
		// older binary would make EVERY existing record fail exactly this way,
		// and without this guard one tick would delete every provisioned route
		// in scope. This is narrower than readFailed (which vetoes removal for
		// the WHOLE pass): only the specific host with the unusable body is held
		// back — every other record still converges/removes normally the same
		// pass.
		protect := map[string]bool{}
		var recordKeys []string
		for _, key := range keys {
			if _, _, _, ok := ParseRecordKey(key, known); ok {
				recordKeys = append(recordKeys, key)
			}
		}
		for _, part := range chunks(recordKeys) {
			bodies := getEach(ctx, st, part)
			for _, key := range part {
				acct, app, host, _ := ParseRecordKey(key, known)
				raw, err := bodies[key].body, bodies[key].err
				if err != nil {
					log.Printf("provision: skipping record %s: read failed: %v", key, err)
					// The record's absence from recs below must not read as "the
					// bucket retired it" — a transient Get failure disables removal
					// for the whole pass, same as an unreadable account manifest.
					readFailed = true
					continue
				}
				rec, err := DecodeRecord(raw)
				if err != nil {
					log.Printf("provision: skipping record %s: malformed: %v", key, err)
					protect[host] = true
					continue
				}
				if rec.Host != host || rec.Account != acct || rec.App != app {
					log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)",
						key, rec.Host, rec.Account, rec.App)
					protect[host] = true
					continue
				}
				// The record IS the authorization, and this pass turns its host into
				// a socket path and a database path (DefaultSocket/DefaultDBPath)
				// and a registry primary key. The console's create door enforces
				// this same shape — but a door is not a guarantee about what is
				// already in the bucket: anything with bucket write access, and any
				// record written by a different (older, or future) writer, arrives
				// here unexamined. Both doors hold the invariant, neither trusts
				// the other. Protected like a corrupted body, and for the same
				// reason: a record that cannot be used is not a record that was
				// retired.
				if !ValidHostShape(rec.Host) {
					log.Printf("provision: skipping record %s: host %q is not a hostname", key, rec.Host)
					protect[host] = true
					continue
				}
				// ValidateShape is the console create door's policy check — but a
				// door is not a guarantee about what is already in the bucket (the
				// same reasoning as the host-shape check just above). This pass
				// materializes whatever it finds, so a record whose Kind/Addr/Unit/
				// Environment combination is malformed must be caught here too, or
				// a bad value becomes a live route. Protected like a corrupted
				// body: the record is there, it just cannot be used.
				if err := ValidateShape(rec); err != nil {
					log.Printf("provision: skipping record %s: invalid shape: %v", key, err)
					protect[host] = true
					continue
				}
				// ValidateShape deliberately does not inspect Channel (it validates
				// shape fields, and Channel is not one), but a channel-less record
				// must not become — or converge — a route pinned to no channel:
				// that is issue #130's failure mode arriving through a different
				// door than the one Task 3 closed. Protected like a corrupted body:
				// the record is there, it just cannot be used.
				if rec.Channel == "" {
					log.Printf("provision: skipping record %s: empty channel", key)
					protect[host] = true
					continue
				}
				// Scope. For a home-boxed record this is the region/pool test it
				// always was; for a static or public-addr one the region half widens
				// to every platform box in the deployment (everywhereClass, above).
				// A replica record narrows it: it belongs ONLY to the box whose
				// placement is the record's own Region — the placement the create
				// door validated as armed — so a replica record minted on the wrong
				// region would mint a route with no mirror behind it and no lease
				// speaking to it, a process holding live order data nobody
				// authorized into that region (the residence rule, #162). Every
				// other box skips it exactly like any other out-of-scope record;
				// the replica's own box sees Region == cfg.Region through the
				// ordinary test above.
				if !cfg.inScope(rec.Region, rec.Pool, everywhereClass(rec)) {
					continue // another agent's jurisdiction, not this pass's business
				}
				// Two accounts writing the same host is a collision this pass
				// cannot arbitrate silently: whichever record this listing reaches
				// FIRST wins the host, and the loser is skipped loudly rather than
				// overwriting the winner in the map with no trace of the conflict.
				if prev, dup := recs[rec.Host]; dup {
					log.Printf("provision: duplicate host %s: keeping %s, skipping %s", rec.Host,
						RecordKey(prev.Account, prev.App, prev.Host), RecordKey(rec.Account, rec.App, rec.Host))
					continue
				}
				recs[rec.Host] = rec
			}
		}
		return recordScan{recs: recs, protect: protect, readFailed: readFailed}, nil
	}
	sc, err := scan(s)
	if err != nil {
		return err
	}
	recs, protect, readFailed := sc.recs, sc.protect, sc.readFailed

	// --- manifest seed cache, per (acct, app), lazily -------------------------
	// known=false (a genuine read failure, not "no manifest") means the seed
	// value returned is a mint-only fallback: safe to hand a BRAND NEW route
	// (nothing existing to protect; the next readable pass converges it),
	// never safe to feed the converge branch below — SetSeed-ing an existing
	// route to "" on a transient blip would erase a working seed for a
	// reason that has nothing to do with the manifest actually changing.
	// Each app with records costs two reads below (its manifest and its trash
	// marker), made lazily, one at a time, inside the record loop. Both are
	// read here first, eight at a time, for every app the records name; the
	// lookups below then judge exactly what they judged before, from these
	// results, and read for themselves only for an app not prefetched (a
	// route's app with no record, at removal time).
	// The manifest is kept only as the seed it renders to, never whole: the
	// lookup below has always reduced it to that at once.
	type manifestRead struct {
		seed   string
		policy *UpdatePolicy
		err    error
	}
	type tombRead struct {
		tomb *release.AppTombstone
		err  error
	}
	var recApps []string
	appOf := map[string]acctApp{}
	for _, rec := range recs {
		k := acctApp{rec.Account, rec.App}
		id := rec.Account + "/" + rec.App
		if _, seen := appOf[id]; !seen {
			appOf[id] = k
			recApps = append(recApps, id)
		}
	}
	slices.Sort(recApps)
	manifests := map[acctApp]manifestRead{}
	tombs := map[acctApp]tombRead{}
	var prefetchMu sync.Mutex
	forEach(recApps, func(id string) {
		k := appOf[id]
		// Trash first, as the record loop asks: an app in the trash, or one
		// whose marker will not read, is never minted for, so its manifest
		// was never read and is not read here either.
		tomb, terr := release.ReadAppTombstone(ctx, s, k.acct, k.app)
		prefetchMu.Lock()
		tombs[k] = tombRead{tomb, terr}
		prefetchMu.Unlock()
		if tomb != nil || terr != nil {
			return
		}
		m, merr := GetManifest(ctx, s, k.acct, k.app)
		r := manifestRead{err: merr}
		if merr == nil {
			r.seed = seedJSON(m)
			r.policy = m.Updates
		}
		prefetchMu.Lock()
		manifests[k] = r
		prefetchMu.Unlock()
	})

	seedCache := map[acctApp]string{}
	seedKnown := map[acctApp]bool{}
	seenManifest := map[acctApp]bool{}
	seedFor := func(acct, app string) (seed string, known bool) {
		k := acctApp{acct, app}
		if seenManifest[k] {
			return seedCache[k], seedKnown[k]
		}
		seenManifest[k] = true
		var rendered string
		var err error
		if r, ok := manifests[k]; ok {
			rendered, err = r.seed, r.err
		} else {
			var m *Manifest
			if m, err = GetManifest(ctx, s, acct, app); err == nil {
				rendered = seedJSON(m)
			}
		}
		if err != nil {
			if errors.Is(err, os.ErrNotExist) {
				seedCache[k], seedKnown[k] = "", true // absent manifest: confirmed no seed
				return seedCache[k], seedKnown[k]
			}
			// A record already written is truth regardless — the manifest
			// only governs the API door — but a genuine read failure must
			// still hold back removal: this pass cannot tell a disabled
			// feature from a transient blip, so it treats it as the latter.
			log.Printf("provision: manifest %s/%s: read failed: %v", acct, app, err)
			readFailed = true
			return "", false
		}
		seedCache[k], seedKnown[k] = rendered, true
		return seedCache[k], seedKnown[k]
	}

	// policyFor is the app's update policy for this pass (settings spec
	// §3.2). known is false when the manifest could not be read: the caller
	// then leaves every row's update_settings as it is, because a read
	// failure is not evidence the policy went away. An absent manifest is
	// known: no policy.
	policyFor := func(acct, app string) (*UpdatePolicy, bool) {
		k := acctApp{acct, app}
		if r, ok := manifests[k]; ok {
			if r.err != nil {
				return nil, errors.Is(r.err, os.ErrNotExist)
			}
			return r.policy, true
		}
		m, err := GetManifest(ctx, s, acct, app)
		switch {
		case errors.Is(err, os.ErrNotExist):
			return nil, true
		case err != nil:
			return nil, false
		}
		return m.Updates, true
	}
	// policyUnread says a row was not minted because policyFor could not
	// read the app's policy. A read failure is not "no policy": minting now
	// would mint an instance that inherits manual as auto, and root would
	// adopt the head over its data's build. Absent is known.
	policyUnread := func(host string) {
		log.Printf("provision: %s: not minted this pass: the app's update policy could not be read", host)
	}

	// --- trash cache, per (acct, app), lazily ---------------------------------
	// An app in the trash provisions nothing, and its existing routes are
	// RETIRED — both halves are this pass's job (#340).
	//
	// Retirement used to be domains.Reconcile's alone, on the grounds that it
	// held the production-flag guard. That deferral was a live bug: the
	// domains pass runs only on the box holding the housekeeper lease
	// (agent.go's domainsLoop wraps it in runHousekeeperDuties), and its
	// registry write is local, so on a multi-box deployment every OTHER box
	// kept a trashed app's route forever and went on serving it from its own
	// edge. Single-box deployments were correct by accident. This pass runs
	// on every box and already reads the marker, so the trigger belongs here
	// — and the guard comes with it (cfg.ReleasePub). Both passes now remove
	// and neither mints, so they agree rather than fight.
	//
	// One marker, two DIFFERENT answers, and conflating them is the trap:
	//
	//   declineMint — true also when the marker is UNREADABLE. Not minting
	//     is the reversible direction, and a bucket writer who could plant
	//     garbage here to stall minting can already retire the instance
	//     outright by deleting its record.
	//   removable — true ONLY for a marker positively read and present,
	//     whose app is not flagged production. Removal is not reversible, so
	//     an unreadable marker (or an unreadable flag) must never reach it.
	//
	// trashOver is that lookup over st, answering from pre where pre holds
	// the app: the pass's own over s and its prefetch, and the removal
	// phase's over the bucket when this pass reads a snapshot (below).
	trashOver := func(st release.Store, pre map[acctApp]tombRead) func(acct, app string) trashState {
		trashCache := map[acctApp]trashState{}
		trashSeen := map[acctApp]bool{}
		return func(acct, app string) trashState {
			k := acctApp{acct, app}
			if trashSeen[k] {
				return trashCache[k]
			}
			trashSeen[k] = true
			var tomb *release.AppTombstone
			var err error
			if r, ok := pre[k]; ok {
				tomb, err = r.tomb, r.err
			} else {
				tomb, err = release.ReadAppTombstone(ctx, st, acct, app)
			}
			switch {
			case err != nil:
				log.Printf("provision: trash marker for %s/%s unreadable: %v — not minting for this app", acct, app, err)
				trashCache[k] = trashState{declineMint: true}
			case tomb != nil:
				log.Printf("provision: %s/%s is in the trash — not minting instance routes for it", acct, app)
				trashCache[k] = trashState{declineMint: true, removable: productionClear(ctx, st, cfg.ReleasePub, acct, app)}
			}
			return trashCache[k]
		}
	}
	trashFor := trashOver(s, tombs)
	trashedFor := func(acct, app string) bool { return trashFor(acct, app).declineMint }

	// --- mint / converge / conflict -------------------------------------------
	routes, err := reg.List()
	if err != nil {
		return fmt.Errorf("provision: list routes: %w", err)
	}
	byHost := map[string]registry.Route{}
	for _, rt := range routes {
		byHost[rt.Host] = rt
	}
	mintLayout := mintLayoutReader(ctx, s, cfg)
	// The import journal is read before the records are acted on, every
	// pass (seeded-instances spec §5.7, seedreconcile.go). A host it handled
	// is skipped by the loop below, and every journaled host by the removal
	// phase: the journal owns the row until its row is gone. An unreadable
	// journal could hide any host, so the removal phase does not run.
	journalHandled, journaled, journalOK := journalPass(ctx, s, reg, cfg, recs, readFailed, protect)
	if !journalOK {
		readFailed = true
	}

	for _, rec := range recs {
		if journalHandled[rec.Host] {
			continue
		}
		if trashedFor(rec.Account, rec.App) {
			continue
		}
		if cfg.RunSeen != nil && rec.ActiveRun != nil {
			cfg.RunSeen(rec)
		}
		// A mint-only fallback ("", the record-already-written-is-truth
		// case) is fine here even when the manifest is unreadable this
		// pass — there is no existing seed value a brand new route could
		// clobber, and a later pass converges it once the manifest reads.
		seed, seedIsKnown := seedFor(rec.Account, rec.App)
		existing, ok := byHost[rec.Host]
		// A seed is dispatched before anything else (seeded-instances spec
		// §3.2, codex 13): an aborted one is finished, an unready one is
		// skipped, and a seeded row never carries the manifest's seed
		// command.
		if rec.Seed != nil {
			var row *registry.Route
			if ok {
				row = &existing
			}
			switch seedAction(rec, row, cfg, mintLayout, journalOK) {
			case seedSkip:
				continue
			case seedSwap:
				// The swap mints the import's row, so it waits for the
				// app's update policy as a mint does (settings spec §3.2).
				p, pKnown := policyFor(rec.Account, rec.App)
				if !pKnown {
					policyUnread(rec.Host)
					continue
				}
				swapImport(ctx, s, reg, cfg, rec, row, mintLayout, p)
				continue
			case seedFinishAbort:
				// A --from seed this box minted is still running: its row
				// goes first, and a later pass that finds no row rolls the
				// seed back (seedreconcile.go).
				if ok && rec.Seed.Kind == seeds.KindReplica {
					retireAbortedSeedRow(ctx, reg, cfg, rec, existing)
					continue
				}
				finishAbortedSeed(ctx, s, reg, cfg, rec)
				continue
			}
			seed, seedIsKnown = "", true
		}
		// A move's hold is judged first (moves spec §9.6). A missing row is
		// minted held by mintHeld, which has a ready mint's refusals and
		// the same claim hook (below) around its write, so the hold and the
		// claim both have to allow the mint. An existing row is held in
		// place with a narrow status write, outside the claim hook as the
		// other status writers are, and then converged as usual: the owned
		// branch's narrow setters never touch status or hold_op.
		if cfg.Hold != nil {
			if status, op, held := cfg.Hold(rec); held {
				if !ok {
					p, known := policyFor(rec.Account, rec.App)
					mintHeld(ctx, reg, cfg, rec, seed, status, op, p, known)
					continue
				}
				holdExisting(ctx, s, reg, cfg, rec, existing, status, op)
			}
		}

		// Refusals that need no claim are decided BEFORE cfg.claimInstall: the
		// hook activates the host guard and writes inventory, effects that a
		// record this pass will not act on must never cause. Only work that
		// is going to touch the registry reaches it.
		var prior *registry.Route
		nested := false
		if ok {
			prior = &existing
			nested = NestedPaths(cfg.DataDir, existing)
		} else {
			// A box that declared CARLOS_INBOUND=none (design §8) has told
			// this pass what work it will ACCEPT, not made a claim about its
			// network — fleet boxes are already outbound-only dial in this
			// estate. Ordinary instance records carry no placement
			// vocabulary of their own (that is ReconcileSidecars' gate,
			// Task 9, over SidecarRecord.Placement); every routed instance
			// this pass mints needs inbound service, so a no-inbound box
			// must refuse ALL of them outright, loudly, rather than mint
			// something it can never serve — a misfiled record becomes a
			// log line here, never a silent outage.
			if cfg.Inbound == InboundNone {
				log.Printf("provision: refusing to mint route %s (app %q acct %q): this box declares "+
					"CARLOS_INBOUND=none (no-inbound) and cannot serve routed instance traffic", rec.Host, rec.App, rec.Account)
				continue
			}
			// The box's layout decides the paths a fresh row is minted at.
			// Read once per pass, lazily, so a pass with nothing to mint
			// never reads it; a read that fails defers every mint on this
			// pass (ReconcileConfig.Layout).
			var lerr error
			if nested, lerr = mintLayout(); lerr != nil {
				log.Printf("provision: %s: not minting this pass: the box's layout intent could not be read: %v", rec.Host, lerr)
				continue
			}
		}
		// The route's whole shape now comes from the record (routeFor):
		// Kind, Addr, Unit, Environment, StatusNamed, Hibernate and the
		// path derivation. A record carrying none of them still mints the
		// exact route this branch hardcoded before they existed — that is
		// the additivity rule, pinned by TestRecordWireShapeIsAdditive and
		// TestReconcileMints.
		p, pKnown := policyFor(rec.Account, rec.App)
		if !ok && !pKnown {
			policyUnread(rec.Host)
			continue
		}
		nr := RouteFor(rec, cfg.DataDir, seed, nested, p)
		// A box that cannot provide this instance's config must not
		// admit it: it would start without its secrets, and a keyed
		// sibling would be left with a route that lost the race to one
		// that was never going to work. Refused before the claim hook, the
		// write and the dry-run line, so a dry run reports what a real pass
		// would do.
		if !ok && cfg.ConfigGate != nil {
			if err := cfg.ConfigGate(ctx, nr); err != nil {
				continue
			}
		}
		// An existing row this record neither owns nor may adopt is known to be
		// left alone before any claim work: with claims on, the hook would
		// activate the guard and write inventory only to be refused after.
		// (Flag off, the same decision is made inside the write below.)
		if ok && cfg.ClaimEnabled && !(existing.Provisioned && existing.App == rec.App && existing.Account == rec.Account) {
			if reason := shapeMismatch(existing, nr); reason != "" {
				log.Printf("provision: %s exists but is not the route this record describes (%s) — leaving it alone",
					rec.Host, reason)
				continue
			}
		}
		selected := ClaimRecord{Host: rec.Host, Key: RecordKey(rec.Account, rec.App, rec.Host), Family: "instance", Account: rec.Account, App: rec.App, CreatedAt: rec.CreatedAt, Value: rec, Route: nr, DryRun: cfg.DryRun}
		if err := cfg.claimInstall(ctx, selected, prior, func(owner registry.ClaimOwnership) error {
			switch {
			case !ok:
				nr.ClaimOwnership = owner
				if cfg.DryRun {
					log.Printf("provision: dry-run: would mint %s route %s (app %q acct %q channel %q)",
						nr.Kind, rec.Host, rec.App, rec.Account, rec.Channel)
					return nil
				}
				if err := reg.Upsert(nr); err != nil {
					log.Printf("provision: mint %s: %v", rec.Host, err)
					return nil
				}
				log.Printf("provision: minted %s route %s (app %q acct %q)", nr.Kind, rec.Host, rec.App, rec.Account)
			case existing.Provisioned && existing.App == rec.App && existing.Account == rec.Account:
				// The columns this record owns (design §4.1, issue #130). Narrow
				// UPDATEs, never a re-Upsert: a read-modify-write racing a
				// deploy's SetVersions or the activator's SetNextDue would
				// clobber those, and Upsert cannot express "clear Environment"
				// at all (it preserves a non-empty value by design).
				convergeOwned(reg, existing, rec, p, pKnown, cfg.DryRun)
				// Absent means "this record does not speak to hibernation" —
				// never a derivation stamped over an operator's own setting.
				//
				// Converging hibernation OFF also drops the route's replication
				// rung, inside registry.SetHibernate — a rung is read only on
				// the activator's wake path, so on a route the record has just
				// stopped hibernating it would describe nothing that happens
				// while `carlos routes` went on printing it. A record cannot
				// express the rung (it is box-scoped, like caps and tunnel), so
				// this pass cannot restore one either; the drop is LOGGED with
				// the flip rather than left for an operator to notice, because
				// it is the one operator-typed value a converge tick can
				// silently spend.
				//
				// A nested row keeps its flag whatever the record says: only a
				// row the activator manages is one the layout revert can stop
				// and move back, so clearing the flag would strand its files
				// under the instance tree (design §8.2, downgrade rules). A row
				// with a layout journal is mid-move whatever its paths say, and
				// recovery needs it managed just the same.
				if rec.Hibernate != nil && existing.Hibernate != *rec.Hibernate && !*rec.Hibernate &&
					(NestedPaths(cfg.DataDir, existing) || LayoutJournalExists(cfg.DataDir, rec.Host)) {
					log.Printf("provision: %s: record asks to clear hibernate but the row's paths are nested; revert the layout first", rec.Host)
				} else if rec.Hibernate != nil && existing.Hibernate != *rec.Hibernate {
					dropped := ""
					if !*rec.Hibernate && existing.Replication != "" {
						dropped = fmt.Sprintf(" (replication %q cleared: only a hibernating route's rung is read)", existing.Replication)
					}
					if cfg.DryRun {
						log.Printf("provision: dry-run: would converge %s hibernate %v -> %v%s", rec.Host, existing.Hibernate, *rec.Hibernate, dropped)
					} else if err := reg.SetHibernate(rec.Host, *rec.Hibernate); err != nil {
						log.Printf("provision: converge hibernate on %s: %v", rec.Host, err)
					} else {
						log.Printf("provision: converged %s hibernate %v -> %v%s", rec.Host, existing.Hibernate, *rec.Hibernate, dropped)
					}
				}
				if existing.Gate != rec.Gate {
					if cfg.DryRun {
						log.Printf("provision: dry-run: would converge %s gate %s -> %s", rec.Host, gateWord(existing.Gate), gateWord(rec.Gate))
					} else if err := reg.SetGate(rec.Host, rec.Gate); err != nil {
						log.Printf("provision: converge gate on %s: %v", rec.Host, err)
					} else {
						log.Printf("provision: converged %s gate %s -> %s", rec.Host, gateWord(existing.Gate), gateWord(rec.Gate))
					}
				}
				// The seed source goes first, and the seed command follows only
				// once it has landed (seeded-instances preflight M10). A row
				// whose record has a seed gets its seed column before its seed
				// command is cleared, so between the two writes it is placeable
				// (seededRestore, never fresh). In the other order, or with the
				// source write failed and the command cleared anyway, the row
				// would hold neither, and a wake then could boot it empty.
				sourceConverged := true
				if col := SeedColumn(rec); existing.SeedSource != col {
					if cfg.DryRun {
						log.Printf("provision: dry-run: would converge %s seed source", rec.Host)
					} else if err := reg.SetSeedSource(rec.Host, col); err != nil {
						log.Printf("provision: converge seed source on %s: %v", rec.Host, err)
						sourceConverged = false
					}
				}
				// seedIsKnown guards against stamping a transient manifest-read
				// failure's "" fallback over a working seed — see seedFor.
				if sourceConverged && seedIsKnown && existing.Seed != seed {
					if cfg.DryRun {
						log.Printf("provision: dry-run: would converge %s seed", rec.Host)
					} else if err := reg.SetSeed(rec.Host, seed); err != nil {
						log.Printf("provision: converge seed on %s: %v", rec.Host, err)
					}
				}
				// Unit is compared only when the RECORD states one: `carlos
				// route -unit` (issue #110) is the supported way to stamp the
				// unit owning a route's process out of band, and a record that
				// never spoke to Unit must not treat that stamp as drift — or
				// this would fire on a healthy, correctly-stamped route every
				// pass, forever.
				// Addr is deliberately NOT compared here any more: it is a
				// CONVERGED column now (convergeOwned writes it beside Addrs and
				// Health through SetUpstreams), and a drift log that kept naming a
				// column the same pass had just fixed would fire forever on a
				// healthy row — the exact "undiagnosable refusal" this message was
				// widened to prevent. What remains are the columns nothing
				// converges in place: the route CLASS and the two filesystem paths
				// derived from it.
				//
				// A nested row's paths differ from the record's flat derivation
				// by design (the layout pass moved them), so when that is the
				// only difference the line says so, rather than pointing an
				// operator at `carlos remove`.
				want := RouteFor(rec, cfg.DataDir, seed, false, p)
				unitDrift := rec.Unit != "" && want.Unit != existing.Unit
				if want.Kind != existing.Kind || unitDrift ||
					want.Socket != existing.Socket || want.DBPath != existing.DBPath {
					if want.Kind == existing.Kind && !unitDrift && NestedPaths(cfg.DataDir, existing) {
						log.Printf("provision: %s: row is on the nested layout (db %q socket %q); the record's shared paths are not drift",
							rec.Host, existing.DBPath, existing.Socket)
					} else {
						log.Printf("provision: %s: record describes a different route SHAPE than the row "+
							"(kind %q/%q unit %q/%q socket %q/%q db %q/%q) — not converged in place; "+
							"`carlos remove %s` purges this route's capability grants (registry.Delete's contract, "+
							"not a bug) before the next pass re-mints it, so treat that as a last resort, not a routine fix",
							rec.Host, want.Kind, existing.Kind, want.Unit, existing.Unit,
							want.Socket, existing.Socket, want.DBPath, existing.DBPath, rec.Host)
					}
				}
			default:
				// A route exists that this pass did not mint. If it is the same
				// route the record describes, adopt it — that is what makes
				// backfilling the fleet a bucket-side write rather than a box
				// session (design §4.2). If it is not, leave it exactly as alone
				// as before: adoption grants this pass's removal phase authority
				// over the row, so the match test is the only thing standing
				// between a backfill and a retired production route.
				// A hand-added row on a nested box (`carlos add` there mints
				// nested too) matches the record at the nested paths; the row's
				// own shape says which derivation to compare against.
				want := RouteFor(rec, cfg.DataDir, seed, NestedPaths(cfg.DataDir, existing), p)
				if reason := shapeMismatch(existing, want); reason != "" {
					log.Printf("provision: %s exists but is not the route this record describes (%s) — leaving it alone",
						rec.Host, reason)
					return nil
				}
				if cfg.DryRun {
					log.Printf("provision: dry-run: would adopt existing route %s (app %q acct %q) — this record would own its channel and environment",
						rec.Host, rec.App, rec.Account)
					// convergeOwned reports channel and environment together,
					// and only the columns that actually differ (design §4.1) —
					// a fixed "converging channel X -> Y" line here would both
					// omit environment (the defect this fixes) and claim a
					// convergence that would not happen when the route is
					// already on the record's channel. Every write inside
					// convergeOwned is dry-run guarded there, so this call
					// writes nothing.
					convergeOwned(reg, existing, rec, p, pKnown, true)
					return nil
				}
				if err := reg.SetProvisioned(rec.Host, true); err != nil {
					log.Printf("provision: adopt %s: %v", rec.Host, err)
					return nil
				}
				log.Printf("provision: adopted existing route %s (app %q acct %q) — this record now owns its channel and environment",
					rec.Host, rec.App, rec.Account)
				// Hibernate is deliberately NOT converged here even when
				// explicit: this first pass over a hand-added route should
				// change what it serves and who owns the column, not whether a
				// live process is allowed to sleep. It converges on the next
				// pass through the owned branch above, by which point the
				// adoption is already visible in the log.
				convergeOwned(reg, existing, rec, p, pKnown, cfg.DryRun)
			}
			return nil
		}); err != nil {
			log.Printf("provision: claim install %s: %v", rec.Host, err)
		}
	}

	// --- removal, only when nothing failed to read this pass ------------------
	// Three conditions, each load-bearing: Provisioned (the flag — not the
	// app, not the channel — grants removal authority; a hand-added route
	// never has it), scope (this box's own region/pool jurisdiction — a
	// foreign box's provisioned route is not this pass's business), and a
	// retirement the BUCKET decided rather than one this pass guessed —
	// either the record is gone, or the app carries a trash marker (#340,
	// and see the trash cache above for why the marker's two answers are
	// not the same answer).
	//
	// Scope is asked through the SAME inScope as the record scan above, with
	// the class read off the row (everywhereRoute) because the record whose
	// class it was may be exactly what is gone. Both halves must widen
	// together:
	// a far box that mints a static route under the widened mint gate and then
	// judges it out of scope here would serve a retired site forever.
	//
	// What reaches reg.Delete here may carry Backing, Tunnel or Caps even
	// though shapeMismatch refuses to ADOPT a row carrying any of them: those
	// refusals cover the adoption instant only. `carlos route -backing/-unit/
	// -tunnel/-grant` all still work on a Provisioned row, so a route can
	// acquire an operator column after it was minted or adopted and remain
	// one unattended record-deletion away from registry.Delete's grant purge.
	// That asymmetry predates this pass's adoption step; it is stated here so
	// the adoption guard is not mistaken for a standing guarantee.
	//
	// `replication` (discussion #10) is the same KIND of column — box-scoped,
	// operator-typed, inexpressible in a record — and is deliberately exempt
	// from shapeMismatch rather than added to it. The three that are listed
	// there are listed for a reason that does not hold for this one: each
	// grants or relocates AUTHORITY over a route's process (a capability, a
	// process owner, an ingress path), and adoption is itself an authority
	// transfer, so handing bucket control a row that already carries one is
	// the thing worth stopping to look at. A replication rung grants nothing.
	// It names how often a database is copied off the box, and it survives
	// adoption unchanged: adoption's own write is SetProvisioned, the owned
	// branch converges through narrow setters that never touch this column,
	// and the one path that DOES clear it — registry.SetHibernate, when a
	// record turns hibernation off — is the honest answer rather than a loss,
	// since a non-hibernating route's rung is read by nobody.
	//
	// Listing it would mean an operator who tuned a large sleeping tenant's
	// rung had, by that act, opted the route out of the backfill this pass
	// exists to perform — permanently and invisibly, until someone thought to
	// clear a column to let a record adopt a row. That is the exact opposite
	// trade from the other three: it would strand precisely the hibernating
	// tenants the ladder was built for, in exchange for guarding an authority
	// this column does not carry.
	//
	// retire runs that phase over one set of facts: the records, the
	// protected hosts and the trash lookup. With stopAtFirst it removes
	// nothing and reports whether any row would go. A record view
	// snapshot's absence is not the bucket's (a write that missed its bump
	// is missing from a Complete snapshot until the backstop rebuild), so a
	// pass over a snapshot that would remove anything decides every removal
	// again from the bucket's own walk, below (S3 change-index design §4.4).
	retire := func(recs map[string]Record, protect map[string]bool, trashFor func(acct, app string) trashState, stopAtFirst bool) (wouldRemove bool) {
		for _, rt := range routes {
			// Removal authority is scoped to this pass's OWN kind. A sidecar
			// row is Provisioned too — that flag is what grants a pass removal
			// authority — and it can never appear in the INSTANCE record set,
			// so without this guard the two passes would delete each other's
			// rows on every tick, forever, with no error anywhere and nothing
			// in the logs that reads as a failure. ReconcileSidecars carries
			// the mirror of this guard; both directions are asserted
			// (TestInstanceReconcileNeverRemovesSidecarRows).
			if rt.Kind == SidecarKind {
				continue
			}
			if !rt.Provisioned || !cfg.inScope(rt.Region, rt.Pool, everywhereRoute(rt)) {
				continue
			}
			// Two triggers, asked in this order because the first is the
			// cheap one: the bucket retired the instance (its record is
			// gone), or the bucket retired the whole APP (a trash marker,
			// while the record stays put). Both name the bucket's own
			// decision; neither is a guess this pass made.
			//
			// The trash lookup needs both halves of the app's identity off
			// the row. A Provisioned row always carries them — routeFor
			// copies them from the record — so a row missing either is a
			// shape this pass did not mint, and not one it will retire.
			why := "record gone"
			if _, ok := recs[rt.Host]; ok {
				if rt.Account == "" || rt.App == "" || !trashFor(rt.Account, rt.App).removable {
					continue
				}
				why = "app in the trash"
			}
			if journaled[rt.Host] {
				continue // the import journal owns the row (seedreconcile.go)
			}
			if protect[rt.Host] {
				log.Printf("provision: not removing %s: its record exists but could not be read cleanly this pass", rt.Host)
				continue
			}
			// A participant of a running operation keeps its row whatever
			// the record's region says (moves spec §9.6, claim spec §8): a
			// move's source after the flip, its target during a rollback.
			// The operation deletes the row at its own finalization.
			if cfg.KeepParticipant != nil && cfg.KeepParticipant(rt.Host) {
				log.Printf("provision: not removing %s (%s): participant of a running operation", rt.Host, why)
				continue
			}
			if stopAtFirst {
				return true
			}
			if cfg.DryRun {
				log.Printf("provision: dry-run: would remove retired %s route %s (%s)", rt.Kind, rt.Host, why)
				continue
			}
			if err := cfg.claimRetire(ctx, reg, rt); err != nil {
				log.Printf("provision: remove retired %s route %s: %v", rt.Kind, rt.Host, err)
				continue
			}
			log.Printf("provision: removed retired %s route %s (%s)", rt.Kind, rt.Host, why)
		}
		return false
	}
	if readFailed || !retire(recs, protect, trashFor, onSnapshot(s)) {
		return nil
	}
	// A snapshot pass with a removal to make: the record walk again, over
	// the bucket, and the removal phase over what it read, by the walk's
	// own rules (invalid shapes and empty channels protected before scope,
	// hosts matched across every key, the bucket's trash marker). Anything
	// it cannot read keeps every row this pass, as in the walk.
	live := freshStore(s)
	lsc, err := scan(live)
	if err != nil || lsc.readFailed {
		log.Printf("provision: not removing any route this pass: the record view lacks a record or holds a trash marker, and the bucket's own walk could not be read cleanly%s", walkErrSuffix(err))
		return nil
	}
	retire(lsc.recs, lsc.protect, trashOver(live, nil), false)
	return nil
}

// mintHeld mints a move participant's row held (moves spec §9.6 "mint"): the
// record's own route at flat paths whatever the box's layout intent (Phase 1
// lands flat, G11), with status and hold_op set by the INSERT, so it is never
// ready for even one pass. The refusals a ready mint has (no-inbound box,
// the config gate, a dry run) apply unchanged, and so does its claim hook
// (claim Plan 6b, cfg.claimInstall): the write runs inside it, after every
// refusal that needs no claim, so with claims on the row carries the owner
// the hook grants and the hook's guard and inventory effects happen only for
// a mint that is going to land. With claims off the hook is a pass-through.
// A status that is not a hold, or an empty op, mints nothing: a
// participant's row is never minted ready.
func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record, seed, status, op string, policy *UpdatePolicy, policyKnown bool) {
	if !registry.IsHold(status) || op == "" {
		log.Printf("provision: %s: not minting: the move's hold %q by %q is not a hold", rec.Host, status, op)
		return
	}
	if cfg.Inbound == InboundNone {
		log.Printf("provision: refusing to mint held route %s (app %q acct %q): this box declares "+
			"CARLOS_INBOUND=none (no-inbound) and cannot serve routed instance traffic", rec.Host, rec.App, rec.Account)
		return
	}
	if !policyKnown {
		// The same wait as the ordinary mint: an instance that inherits
		// manual from its app must not land held as auto.
		log.Printf("provision: %s: not minted this pass: the app's update policy could not be read", rec.Host)
		return
	}
	nr := RouteFor(rec, cfg.DataDir, seed, false, policy)
	nr.Status, nr.HoldOp = status, op
	if cfg.ConfigGate != nil {
		if err := cfg.ConfigGate(ctx, nr); err != nil {
			return
		}
	}
	written := false
	selected := ClaimRecord{Host: rec.Host, Key: RecordKey(rec.Account, rec.App, rec.Host), Family: "instance", Account: rec.Account, App: rec.App, CreatedAt: rec.CreatedAt, Value: rec, Route: nr, DryRun: cfg.DryRun}
	if err := cfg.claimInstall(ctx, selected, nil, func(owner registry.ClaimOwnership) error {
		nr.ClaimOwnership = owner
		if cfg.DryRun {
			log.Printf("provision: dry-run: would mint %s route %s held %s by %s (app %q acct %q channel %q)",
				nr.Kind, rec.Host, status, op, rec.App, rec.Account, rec.Channel)
			return nil
		}
		if err := reg.Upsert(nr); err != nil {
			log.Printf("provision: mint held %s: %v", rec.Host, err)
			return nil
		}
		log.Printf("provision: minted %s route %s held %s by %s (app %q acct %q)", nr.Kind, rec.Host, status, op, rec.App, rec.Account)
		written = true
		return nil
	}); err != nil {
		log.Printf("provision: claim install held %s: %v", rec.Host, err)
	}
	if written && cfg.HoldWritten != nil {
		cfg.HoldWritten(rec.Host)
	}
}

// hostHoldLocks is one in-process mutex per host for the hold decision on
// an existing row: see WithHostHoldLock.
var hostHoldLocks struct {
	mu    sync.Mutex
	hosts map[string]*sync.Mutex
}

// WithHostHoldLock runs fn holding host's hold lock, exclusively within this
// process. Reconcile's hold decision for an existing row (holdExisting: the
// record re-read, the hold asked again of it, the conditional write) runs
// under it, so nothing that ends the hold can run between that read and
// that write. The landing's open and its finalisation (moves Task 10,
// movesland.go) MUST take this same lock around their record read and their
// ReleaseHold or DeleteIfHold, and so must any later writer that ends a hold
// reconcile could re-assert from the record: without it, an open that lands
// in the window is re-held on the next line. Writers in other processes
// (root) are met by SetHoldIf's own condition, not by this lock.
//
// A leaf: its holder reads one record, writes one row and takes no other
// lock. Nothing takes it while holding the held-stage lock (G30), and
// nothing holds it across a wait on another host.
func WithHostHoldLock(host string, fn func()) {
	hostHoldLocks.mu.Lock()
	if hostHoldLocks.hosts == nil {
		hostHoldLocks.hosts = map[string]*sync.Mutex{}
	}
	l := hostHoldLocks.hosts[host]
	if l == nil {
		l = &sync.Mutex{}
		hostHoldLocks.hosts[host] = l
	}
	hostHoldLocks.mu.Unlock()
	l.Lock()
	defer l.Unlock()
	fn()
}

// holdExisting holds an existing row for a move (G10): Upsert never changes
// a hold on conflict, so an unheld row of this pass's own is held with
// SetHoldIf. A row already held by op is left as is; a row held by another
// op is that op's, and is left alone; a row this pass did not mint (not
// Provisioned, or another app's) is not taken into a move.
//
// Three rules keep a hold write from closing a row an operation has opened,
// or from landing over a hold written since the pass's listing (fix round 1):
//
//   - a recovery hold (restoring, releasing) is never written over an
//     unheld row. The recovering operation holds its own row as it enters
//     the recovery, and opens it while the record still names the
//     recovery (moves spec §7.4 reverting `probed`, §7.8 `opened`), so an
//     unheld row there is one it opened. Only a missing row is minted.
//   - the record is read again just before the write, and the hold is
//     asked again of the fresh record, all under the host's hold lock
//     (WithHostHoldLock), which the open and the finalisation take too:
//     this pass listed the records before the routes, and a target opens
//     its row only after its commit, so a fresh record still at landing
//     means the row has not been opened, and under the lock it cannot be
//     opened before the write lands.
//   - the write is SetHoldIf expecting the row unheld: a hold another op
//     wrote since the listing (or since this pass's own route read) is
//     never overwritten; the row is left to it and the next pass judges
//     it afresh.
//
// SetHoldIf can wait out root's activation guard and fail (G30): logged,
// and the next pass tries again.
func holdExisting(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, rec Record, existing registry.Route, status, op string) {
	switch {
	case !registry.IsHold(status) || op == "":
		log.Printf("provision: %s: not holding: the move's hold %q by %q is not a hold", rec.Host, status, op)
		return
	case existing.HoldOp == op:
		return
	case existing.HoldOp != "":
		log.Printf("provision: %s: held %s by %s, not by %s; leaving it", rec.Host, existing.Status, existing.HoldOp, op)
		return
	case status == "restoring" || status == "releasing":
		return
	case !existing.Provisioned || existing.App != rec.App || existing.Account != rec.Account:
		log.Printf("provision: %s: the row is not this pass's (app %q acct %q); not holding it for %s", rec.Host, existing.App, existing.Account, op)
		return
	case cfg.DryRun:
		log.Printf("provision: dry-run: would hold %s %s by %s", rec.Host, status, op)
		return
	}
	WithHostHoldLock(rec.Host, func() {
		// From the bucket, not a record view snapshot: the listing may be one.
		fresh, err := GetRecord(ctx, freshStore(s), rec.Account, rec.App, rec.Host)
		if err != nil {
			log.Printf("provision: %s: not holding this pass: re-read the record: %v", rec.Host, err)
			return
		}
		if st, o, ok := cfg.Hold(*fresh); !ok || st != status || o != op {
			return // the move moved on since the listing; the next pass judges it afresh
		}
		if cfg.beforeHoldWrite != nil {
			cfg.beforeHoldWrite(rec.Host)
		}
		held, err := reg.SetHoldIf(rec.Host, status, op, "")
		if err != nil {
			log.Printf("provision: hold %s %s by %s: %v", rec.Host, status, op, err)
			return
		}
		if !held {
			log.Printf("provision: %s: not held %s by %s: the row's hold changed under this pass; the next pass judges it afresh", rec.Host, status, op)
			return
		}
		log.Printf("provision: held %s %s by %s (was %s)", rec.Host, status, op, existing.Status)
		if cfg.HoldWritten != nil {
			cfg.HoldWritten(rec.Host)
		}
	})
}

// gateWord renders a gate column for a log line: "open" or "set", never the
// hash — it has no business in a journal.
func gateWord(hash string) string {
	if hash == "" {
		return "open"
	}
	return "set"
}

// knownAccounts lists accounts/ and materializes the sqid set this pass
// trusts for ParseRecordKey's "known account" guard — the same three-way
// sqid agreement domains.Reconcile enforces (reconcile.go:106): the key
// path's sqid, the manifest body's own sqid, and the frozen encoding of the
// body's id must all agree, or the manifest is a lie about identity and is
// skipped, loudly. A tombstoned manifest (m.Deleted()) is skipped too —
// retired accounts own no in-scope records.
//
// Listing failure returns an error (nothing here can proceed without it); a
// per-manifest read or parse failure instead sets readFailed and is
// otherwise skipped — never fatal to the pass, but disabling enough that the
// removal phase in Reconcile refuses to run on it.
func knownAccounts(ctx context.Context, s release.Store) (known map[string]bool, readFailed bool, err error) {
	known = map[string]bool{domains.OpsSqid: true}
	keys, err := s.List(ctx, "accounts/")
	if err != nil {
		return nil, false, fmt.Errorf("provision: list accounts: %w", err)
	}
	var manifestKeys []string
	for _, key := range keys {
		if _, ok := domains.AccountManifestSqid(key); ok {
			manifestKeys = append(manifestKeys, key)
		}
	}
	for _, part := range chunks(manifestKeys) {
		bodies := getEach(ctx, s, part)
		for _, key := range part {
			sqid, _ := domains.AccountManifestSqid(key)
			body, err := bodies[key].body, bodies[key].err
			if err != nil {
				log.Printf("provision: skipping %s: read failed: %v", key, err)
				readFailed = true
				continue
			}
			var m domains.AccountManifest
			if err := json.Unmarshal(body, &m); err != nil || m.ID <= 0 || m.Sqid == "" {
				log.Printf("provision: skipping %s: malformed account manifest: %v", key, err)
				readFailed = true
				continue
			}
			if m.Sqid != sqid || domains.Sqid(m.ID) != sqid {
				log.Printf("provision: skipping %s: sqid %q/%q is not Sqid(%d)=%q", key, sqid, m.Sqid, m.ID, domains.Sqid(m.ID))
				readFailed = true
				continue
			}
			if m.Deleted() {
				continue
			}
			known[sqid] = true
		}
	}
	return known, readFailed, nil
}

// trashGate builds the per-(acct, app) "is this app in the trash" predicate
// the SIDECAR pass gates MINTING on, caching each answer for the pass. what
// names the route family in the log line and is all that ever varied between
// callers.
//
// The instance pass had the same shape until #340 gave it a SECOND answer —
// whether a trashed app's routes may also be RETIRED, which needs the
// production-flag guard (trashFor/trashState above). Retirement is not a
// question this predicate can answer, so the sidecar pass keeps the
// mint-only one rather than inheriting an authority its own removal phase
// was never designed around. A trashed app's sidecar rows therefore linger
// until their records go, which is the same behaviour every pass had before
// #340 and the safe direction — but it is a gap, recorded here and in the
// design's follow-ups, not a decision this branch made.
//
// The predicate answers TRUE for an unreadable or malformed marker as well as
// a valid one: not minting is the reversible direction, and a bucket writer
// who could plant a marker to stall minting can already retire the record
// outright.
func trashGate(ctx context.Context, s release.Store, what string) func(acct, app string) bool {
	cache := map[acctApp]bool{}
	seen := map[acctApp]bool{}
	return func(acct, app string) bool {
		k := acctApp{acct, app}
		if seen[k] {
			return cache[k]
		}
		seen[k] = true
		tomb, err := release.ReadAppTombstone(ctx, s, acct, app)
		switch {
		case err != nil:
			log.Printf("provision: trash marker for %s/%s unreadable: %v — not minting for this app", acct, app, err)
			cache[k] = true
		case tomb != nil:
			log.Printf("provision: %s/%s is in the trash — not minting %s routes for it", acct, app, what)
			cache[k] = true
		}
		return cache[k]
	}
}

// ListAllRecords enumerates EVERY instance record in the deployment — the
// same account-qualified walk Reconcile does (list accounts/ for the trusted
// sqid set, list apps/, ParseRecordKey, Get, decode), exported for callers
// that need deployment-GLOBAL route facts rather than one app's.
//
// It exists because the registry is not deployment-global truth. Routes are
// provisioned per box: an everywhereClass record (exec/unit/socket kinds) is
// homed to ONE box, so that host is simply absent from every other box's
// registry. Any deployment-wide decision derived from one box's registry
// therefore changes meaning when the work moves to a different box — which,
// for DNS steering (internal/steering), meant a housekeeper lease migration
// could delete a live steered host's records and NXDOMAIN it until the lease
// came back. The records ARE the deployment's truth; this is how to read all
// of them.
//
// FAIL-CLOSED, and that is the deliberate difference from Reconcile's own
// walk. Reconcile tolerates a partial read (it logs, sets readFailed, and
// vetoes only its removal phase) because it must still MINT on a degraded
// pass. A caller deriving global state has no such split: a short list is
// indistinguishable from records having been deleted, so every read failure
// is an error and no records are returned —
//
//   - accounts/ listing fails, or ANY account manifest is unreadable,
//     malformed, or disagrees with its own sqid: that account's records all
//     fail ParseRecordKey's known-account guard and would vanish silently
//     from the result. Fatal.
//   - apps/ listing fails, or ANY record object will not Get. Fatal.
//
// A record that READS but does not decode — or whose body disagrees with its
// own path — is the other class: a content problem in one object, which is
// skipped and logged (Reconcile's `protect` treatment, minus the registry
// bookkeeping). One malformed record must not blind a caller to every other.
func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) {
	recs, _, _, err := readAllRecords(ctx, s, true)
	return recs, err
}

// ListAllRecordsStrict is ListAllRecords with one more answer: the hosts it
// skipped. ListAllRecords is already fail-closed on every READ failure (a
// listing that fails, a manifest or a record that will not Get is an error
// and no records come back), but it logs and drops a record that reads and
// then does not decode, or whose body disagrees with its own path, so the
// set it returns is silently short of those records. A caller that draws a
// conclusion from a record's absence (the seeds sweep deciding no record
// names a snapshot) cannot tell that gap from a record that is not there.
// ListAllRecordsStrict names the gap: unreadable holds the host of each
// record key whose body did not decode or disagreed with its path (keyed by
// the host in the path, never the body's). unreadable is non-nil on
// success. The records returned are exactly ListAllRecords' set, and every
// error is ListAllRecords' error; a caller deleting on absence must keep
// everything that belongs to a host in unreadable.
//
// One more blind spot, shared with ListAllRecords and Reconcile: records
// under an account with no manifest, or whose manifest is soft-deleted, are
// not in recs and not in unreadable. They are invisible, not unreadable.
func ListAllRecordsStrict(ctx context.Context, s release.Store) (recs []Record, unreadable map[string]bool, err error) {
	recs, _, unreadable, err = readAllRecords(ctx, s, true)
	if err != nil {
		return nil, nil, err
	}
	return recs, unreadable, nil
}

// ListAllRecordsTolerant is ListAllRecords for a caller that acts on each
// record it reads, one at a time, and draws nothing from a record's absence
// (the run sweep's expiry and lost-run phases). An account manifest that
// will not read cleanly, an app whose records will not list, or a record
// that will not Get is logged and skipped, and complete comes back false;
// every other record is returned. err is only the top-level listing failing,
// when nothing could be read. A caller that also draws conclusions from
// absence must act on that only when complete is true.
func ListAllRecordsTolerant(ctx context.Context, s release.Store) (recs []Record, complete bool, err error) {
	recs, complete, _, err = readAllRecords(ctx, s, false)
	return recs, complete, err
}

// readAllRecords is both: strict fails on any read failure, as
// ListAllRecords documents; tolerant logs it, skips what it could not read
// and reports complete false. In both modes unreadable holds the path host
// of every record that read but did not decode or disagreed with its path
// (skipped, logged); it is non-nil whenever err is nil.
func readAllRecords(ctx context.Context, s release.Store, strict bool) (recs []Record, complete bool, unreadable map[string]bool, err error) {
	complete = true
	unreadable = map[string]bool{}
	known, readFailed, err := knownAccounts(ctx, s)
	if err != nil {
		return nil, false, nil, err
	}
	if readFailed {
		if strict {
			return nil, false, nil, errors.New("provision: an account manifest could not be read cleanly — refusing to report a partial record set")
		}
		log.Printf("provision: an account manifest could not be read cleanly: that account's records are skipped")
		complete = false
	}
	// Sorted (listAppSubtree sorts) so the result — and anything folded
	// from it — is stable call to call, whatever order the store lists in.
	keys, partial, err := listAppSubtree(ctx, s, known, "instances/")
	if err != nil {
		return nil, false, nil, fmt.Errorf("provision: list apps: %w", err)
	}
	if partial {
		if strict {
			return nil, false, nil, errors.New("provision: an app's records could not be listed — refusing to report a partial record set")
		}
		log.Printf("provision: an app's records could not be listed: they are skipped")
		complete = false
	}
	var recordKeys []string
	for _, key := range keys {
		if _, _, _, ok := ParseRecordKey(key, known); ok {
			recordKeys = append(recordKeys, key)
		}
	}
	for _, part := range chunks(recordKeys) {
		bodies := getEach(ctx, s, part)
		for _, key := range part {
			acct, app, host, _ := ParseRecordKey(key, known)
			raw, err := bodies[key].body, bodies[key].err
			if err != nil {
				if strict {
					return nil, false, nil, fmt.Errorf("provision: read record %s: %w", key, err)
				}
				log.Printf("provision: skipping record %s: unreadable: %v", key, err)
				complete = false
				continue
			}
			rec, err := DecodeRecord(raw)
			if err != nil {
				log.Printf("provision: skipping record %s: malformed: %v", key, err)
				unreadable[host] = true
				continue
			}
			if rec.Host != host || rec.Account != acct || rec.App != app {
				log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)",
					key, rec.Host, rec.Account, rec.App)
				unreadable[host] = true
				continue
			}
			recs = append(recs, rec)
		}
	}
	return recs, complete, unreadable, nil
}

// listAppSubtree returns the keys a record reader needs to see: every key
// under apps/<acct>/<app>/<sub> for each known account and each of its apps,
// sorted — or, on a store that cannot be walked that way, every key under
// apps/, as the readers always listed. The callers parse every key exactly as
// before (ParseRecordKey, ParseSidecarKey); this changes only which keys they
// are shown.
//
// It exists because List(ctx, "apps/") returns every object under every app —
// every release artifact and every log chunk — to find the records among
// them. On the flagship on 2026-10-05 apps/ held 59,288 objects, 50,695 of
// them logs, for 152 instance records: sixty list pages per reader, three
// readers on every serial bump, and most of why a box took 16 to 43 seconds
// to notice a promote (#231). The walk is one ListDirs of apps/, one per
// known account, and one small List per app.
//
// It finds the same records the full listing did. A record's account is
// always one of known (the parsers refuse any other), and its app is a
// directory directly under the account, so no record lives anywhere the walk
// does not look — PROVIDED ListDirs answers in the bucket's shape. Only a
// store that says so (release.HasBucketListDirs: the S3 store, and a prefix
// store over one) is walked; the fleet door, whose ListDirs("apps/") is its
// own account's app directories, and DirStore, which lists by walking its
// whole root, are listed the old way. Assuming the shape on the fleet door
// found no records at all, and the reconcile pass would have read that as
// every route retired.
//
// Failure comes in two sizes. The top-level listing failing is an error, as
// the single List failing was. One account's or one app's listing failing is
// PARTIAL: the keys that were listed come back, and partial reports that some
// were not. The callers treat that as they treat a record they could not
// read — removal is vetoed for the whole pass, everything that was seen still
// converges — because a partial record set must never read as "the bucket
// retired the rest", and one app's bad minute must not stop every other app's
// routes from being minted.
func listAppSubtree(ctx context.Context, s release.Store, known map[string]bool, sub string) (keys []string, partial bool, err error) {
	if !release.HasBucketListDirs(s) {
		keys, err := s.List(ctx, "apps/")
		slices.Sort(keys) // Store.List promises no order; the walk's is sorted
		return keys, false, err
	}
	accts, err := s.ListDirs(ctx, "apps/")
	if err != nil {
		return nil, false, err
	}
	var acctPrefixes []string
	for _, ap := range accts {
		if known[strings.TrimSuffix(strings.TrimPrefix(ap, "apps/"), "/")] {
			acctPrefixes = append(acctPrefixes, ap)
		}
	}

	var (
		mu       sync.Mutex
		prefixes []string
		out      []string
	)
	fail := func(what string, err error) {
		log.Printf("provision: listing %s: %v — its records are unseen this pass", what, err)
		partial = true
	}
	// A handful at a time, accounts and apps alike: each is one small round
	// trip, and a box walks a few dozen accounts and a few hundred apps.
	forEach(acctPrefixes, func(ap string) {
		apps, err := s.ListDirs(ctx, ap)
		mu.Lock()
		defer mu.Unlock()
		if err != nil {
			fail(ap, err)
			return
		}
		for _, p := range apps {
			prefixes = append(prefixes, p+sub)
		}
	})
	forEach(prefixes, func(p string) {
		keys, err := s.List(ctx, p)
		mu.Lock()
		defer mu.Unlock()
		if err != nil {
			fail(p, err)
			return
		}
		out = append(out, keys...)
	})
	slices.Sort(out)
	return out, partial, nil
}

// readChunk is how many keys a judging loop prefetches at once: enough to keep
// listWalkers busy, few enough that a large record set is never held in
// memory whole, and that a loop stopping at its first bad read has read only
// a chunk ahead of it.
const readChunk = 64

// chunks splits keys into runs of at most readChunk, in order.
func chunks(keys []string) [][]string {
	var out [][]string
	for len(keys) > readChunk {
		out = append(out, keys[:readChunk])
		keys = keys[readChunk:]
	}
	if len(keys) > 0 {
		out = append(out, keys)
	}
	return out
}

// fetched is one prefetched object: its body, or why it could not be read.
type fetched struct {
	body []byte
	err  error
}

// getEach reads every key in keys, listWalkers at a time, and returns each
// one's body or error by key.
//
// The record readers used to Get each record in turn inside the loop that
// judged it. With 152 records on the flagship that was the bulk of a 12.5s
// reconcile and a 6.3s moves pass on every serial bump (#231, measured on the
// box with the fast loop's stage timing). The loops are unchanged: they still
// visit keys in the same order and judge each one exactly as before, only
// reading the body from here rather than from the store. A key's read error
// is that key's, as it was.
//
// Callers prefetch one chunk (chunks) at a time and judge it before the next.
// A context already cancelled reads nothing: every key reports ctx.Err(), so
// a judging loop sees a read failure for it, never a missing body.
func getEach(ctx context.Context, s release.Store, keys []string) map[string]fetched {
	out := make(map[string]fetched, len(keys))
	if err := ctx.Err(); err != nil {
		for _, k := range keys {
			out[k] = fetched{err: err}
		}
		return out
	}
	var mu sync.Mutex
	forEach(keys, func(k string) {
		body, _, err := s.Get(ctx, k)
		mu.Lock()
		out[k] = fetched{body: body, err: err}
		mu.Unlock()
	})
	return out
}

// forEach runs f over items, listWalkers at a time, and returns when all
// have finished.
func forEach(items []string, f func(string)) {
	const listWalkers = 8
	var wg sync.WaitGroup
	sem := make(chan struct{}, listWalkers)
	for _, it := range items {
		wg.Add(1)
		sem <- struct{}{}
		go func(it string) {
			defer wg.Done()
			defer func() { <-sem }()
			f(it)
		}(it)
	}
	wg.Wait()
}

// poolServed reports whether pool is one this box serves. An empty
// cfg.Pools means the shared "" pool ONLY, never "every pool" — the same
// convention domains.Config's AppsDomain=="" carries: unconfigured means the
// default, not "all".
func poolServed(pools []string, pool string) bool {
	if len(pools) == 0 {
		return pool == ""
	}
	return slices.Contains(pools, pool)
}

// SharedInstanceEligibility answers, from this box's own reconcile config,
// whether it mints an ORDINARY instance record: one in the shared ""
// pool, needing inbound service. The edges file publishes both answers
// (serves_shared_pool, accepts_inbound) so the console never offers an
// automatic choice of a region whose boxes would refuse the record
// (nearest-region spec §6.1). It is the reconciler's own predicates —
// poolServed and the CARLOS_INBOUND=none refusal in the reconcile pass — so
// publication and minting cannot disagree.
func (cfg ReconcileConfig) SharedInstanceEligibility() (sharedPool, inbound bool) {
	return poolServed(cfg.Pools, ""), cfg.Inbound != InboundNone
}

// seedJSON renders a manifest's seed argv as the wire text SetSeed and a
// minted route's Seed field carry: "" for nil/empty (no seed), else the
// JSON array of the argv strings.
func seedJSON(m *Manifest) string {
	if m == nil || len(m.Seed) == 0 {
		return ""
	}
	b, err := json.Marshal(m.Seed) // []string always marshals; err is unreachable
	if err != nil {
		return ""
	}
	return string(b)
}