package host

import (
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/controlsock"
	"github.com/carlosframework/platform/internal/dbverify"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/stopevidence"
)

// The residence-move worker (issue #331, instance-move spec): everything a
// move does to DATA happens here, on the SOURCE box — the box whose
// CARLOS_REGION matches the record's region for as long as the move is in
// flight. The console only writes requests; the target box only restores.
//
// One pass = one delta round per in-flight move. The caller runs it on the
// same trigger as the reconcile pass (serial bumps land in seconds via
// carlos-adopt.path), and prepare's convergence is the round-over-round
// shrinking of the delta, not a resident goroutine — restart-safe because
// every step derives from the record and the buckets. "Converged" is a
// round that moved (copied or deferred) at most moveCfg.ReadyDelta bytes,
// 16 MiB by default, not a round that copied nothing: an awake instance
// never produces the latter (#175). Ready therefore means "small gap", and
// the cut's final round below is what makes staging exact. "Unchanged" is
// exact: the copy receipts (lease.Receipts, kept in this box's own bucket
// at activator.MoveReceiptsKey) are what let a multipart source, whose
// copy never shares its ETag, count as unchanged at all.
//
// The cut ordering is the never-two-writers argument, mechanically:
//
//	route Status "moving"   (edge stops serving; RetireExcept stops the child)
//	→ wake fence            (no tick or sweep can wake it again — G12)
//	→ child quiesced        (Quiesced(host) — retries next pass until true)
//	→ final mirror round    (own replica is byte-final; staging matches it)
//	→ local purge           (the warm-cache trap: a stale local file must
//	                         never resurrect a moved-away history)
//	→ CAS flip+clear        (Region=To, Move=nil, ONE write — the last act)
//
// Every step is idempotent; a box that dies mid-cut resumes at the top and
// falls through the already-done steps.

// movesWorker is the one home for the worker's cross-pass state: the wired
// config, the memoized per-region mirror clients (built once per target
// region, not per pass — a client build is a credential-chain resolution),
// the once-per-host status nudge, and the per-host trouble throttle.
// RunAgent writes the pointer once; both converge loops read it.
type movesWorker struct {
	cfg moveCfg
	// pass runs one movesPass at a time. The fast converge and the
	// instances loop both call runMovesPass, and two concurrent passes
	// could drive one cut twice, or lift a fence from a record list read
	// before the cut that installed it.
	pass    sync.Mutex
	mu      sync.Mutex
	mirrors map[string]lease.MirrorStore // dstRegion → client pair
	nudged  sync.Map                     // host → true: status→quiesce hop nudged
	trouble sync.Map                     // host → time.Time of last logged error
	fenced  sync.Map                     // host → uint64: the cut's wake-fence generation

	// Phase 1 (CARLOS_MOVES_PHASE1). stagers memoizes the staging writer per
	// target bucket region; archives keeps T5's verdict per move and exact
	// archive version; measured is this box's last T11 measurements;
	// notices rate-limits the worker's standing log lines; strict is the
	// set of hosts with a Phase 1 move in flight here (ExecBackend.Strict).
	stagers  map[string]lease.StagingWriter
	archives map[string]verifiedArchive
	measured moves.Measurements
	notices  sync.Map // key → time.Time of the last line
	// logged is the move log identities this worker has handed off
	// (movesevents.go): host/op/identity, bounded.
	logged moveLogMemo
	// pageWarned is the host/op/rule pages this worker has warned are not
	// sent because no pager is wired (movespaging.go).
	pageWarned sync.Map
	strict     atomic.Pointer[map[string]bool]
	// holds is what reconcile and the schedule halves ask about this
	// box's running moves (moveshold.go), rebuilt every pass. Nil with the
	// switch off.
	holds *movesHoldView
	// landLeases is the landing's own-location lease per host, held from
	// building through published (movesland.go); guarded by mu.
	landLeases map[string]park.HeldLease
	// lost names the hosts whose claim lists this box in Lost, as this pass
	// read them (lostHere); rebuilt every pass under pass.
	lost map[string]bool
	// walkedIdle says the last pass walked the bucket, found no record with
	// a move (and none it could not decode), and left this worker holding
	// no move state of its own (localMoveState). Only then may a pass skip
	// its walk on the record view's word (movesIdle); guarded by pass.
	walkedIdle bool
}

// staging is the memoized staging writer for a target bucket region.
func (w *movesWorker) staging(ctx context.Context, dstRegion string) (lease.StagingWriter, error) {
	if w.cfg.Staging == nil {
		return nil, errors.New("no staging writer on this box")
	}
	w.mu.Lock()
	m := w.stagers[dstRegion]
	w.mu.Unlock()
	if m != nil {
		return m, nil
	}
	m, err := w.cfg.Staging(ctx, dstRegion)
	if err != nil {
		return nil, err
	}
	w.mu.Lock()
	if w.stagers == nil {
		w.stagers = map[string]lease.StagingWriter{}
	}
	w.stagers[dstRegion] = m
	w.mu.Unlock()
	return m, nil
}

// notice logs a standing condition at most once a minute per key.
func (w *movesWorker) notice(key, format string, args ...any) {
	now := time.Now()
	if last, ok := w.notices.Load(key); ok && now.Sub(last.(time.Time)) < time.Minute {
		return
	}
	w.notices.Store(key, now)
	log.Printf(format, args...)
}

// strictHost is ExecBackend.Strict with the switch on (ruling F9): a host
// is strict while its record has a Phase 1 move in flight that names this
// box's region, as the last pass read it. It is the one strict setter.
func (w *movesWorker) strictHost(host string) bool {
	m := w.strict.Load()
	return m != nil && (*m)[host]
}

// setStrict names the hosts whose spawns on this box must be strict: every
// host with a Phase 1 move in flight of which this box is a participant.
// The record's region is the source until the flip and the target after
// it, so the source is also matched by Move.From: it stays strict after
// the flip for as long as it is a participant (spec §5.3), and the close
// CAS that clears the move ends it.
func (w *movesWorker) setStrict(recs []provision.Record) {
	m := map[string]bool{}
	for _, rec := range recs {
		if movesNamesBox(w.cfg, rec) {
			m[rec.Host] = true
		}
	}
	w.strict.Store(&m)
}

// movesNamesBox reports whether rec carries a Phase 1 move naming this
// box's region: its record's, its source's or its target's. Each record
// code is compared with this box's engine region by the shared placement
// rule (park.PlacementMatches under the verified home binding,
// ParkHomeRegion), so the home's roster name and its "" code are one.
func movesNamesBox(cfg moveCfg, rec provision.Record) bool {
	if rec.Move == nil || rec.Move.Legacy() || rec.Move.Offline() {
		return false
	}
	here := cfg.parkRegion()
	for _, code := range []string{rec.Region, rec.Move.To, rec.Move.From} {
		if park.PlacementMatches(cfg.ParkHomeRegion, code, here) {
			return true
		}
	}
	return false
}

// lostHere is the lost-box gate (moves Q9, Plan 4 Task 10, delta M15) for
// a record's round: a record that names this box's region is gated by
// hostBlocked. A record that names neither this box's region nor a move is
// not read.
func (w *movesWorker) lostHere(ctx context.Context, s release.Store, rec provision.Record) bool {
	if !movesNamesBox(w.cfg, rec) {
		return false
	}
	return w.hostBlocked(ctx, s, rec.Host)
}

// hostBlocked decides, once per pass, whether this box runs no move step
// for host: the host's claim lists this box in Lost (until the claim
// readmits it), or the claim does not read. The quarantine's local
// retirement is the claim's (readmission requires its retirement receipt);
// moves adds only this gate. The decision holds for the whole pass (w.lost):
// the record's round, the target's report pass and the op journal sweep
// all skip a blocked host, and one standing notice is logged.
func (w *movesWorker) hostBlocked(ctx context.Context, s release.Store, host string) bool {
	if w.cfg.Box == "" {
		return false
	}
	if w.lost == nil {
		w.lost = map[string]bool{}
	}
	if blocked, decided := w.lost[host]; decided {
		return blocked
	}
	blocked := false
	obs, err := hostclaim.Read(ctx, s, host)
	switch {
	case err != nil:
		blocked = true
		w.notice(host+"#lost", "moves: %s: the host claim does not read: this box runs no move step for it this pass: %v", host, err)
	case !obs.Absent && obs.Claim != nil:
		for _, box := range obs.Claim.Lost {
			if box == w.cfg.Box {
				blocked = true
				w.notice(host+"#lost", "moves: %s: this box (%s) is lost on this host's claim: it runs no move step until the claim readmits it", host, box)
				break
			}
		}
	}
	w.lost[host] = blocked
	return blocked
}

var armedMovesWorker atomic.Pointer[movesWorker]

func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.MirrorStore, error) {
	w.mu.Lock()
	m := w.mirrors[dstRegion]
	w.mu.Unlock()
	if m != nil {
		return m, nil
	}
	m, err := w.cfg.Mirror(ctx, dstRegion)
	if err != nil {
		return nil, err
	}
	w.mu.Lock()
	w.mirrors[dstRegion] = m
	w.mu.Unlock()
	return m, nil
}

// armMovesPass wires the worker with what only RunAgent has in hand: the
// activator's hibernate coordinates (the own-bucket side of every mirror),
// the data dir the cut purges, the registry the cut quiesces through, the
// service that answers whether a child is down, and the exec backend (the
// Phase 1 launch environment and S2). Called once, before svc.Start: with
// the Phase 1 switch on it sets eb.Strict, which the spawn path reads; with
// it off eb.Strict stays nil, so every spawn is main's. providers is the
// claim wiring's (newClaimWiring builds it before this): with both switches
// on, the park arms take their engine from its MoveParks (armMovesParks,
// Plan 5 Task 9a), behind the wake-source gate (movesWakeSourceRegistered,
// Plan 5 Task 9b), which the moves wake source's registration opens, before
// this runs, in the same agent start. events is the agent's
// rollout sink, which the move log tells the app's members through (Plan 6
// Task 3).
func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, providers ClaimProviders, events activator.EventSink, carry func(ctx context.Context, host, bucket, prefix string) error) {
	dataDir = movesDataDir(dataDir)
	readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv)
	if !ok {
		log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta)
	}
	cfg := moveCfg{
		registerLanding: providers.registerLanding,
		Region:          strings.TrimSpace(os.Getenv("CARLOS_REGION")),
		OwnBucket:       hib.Bucket,
		OwnBucketRegion: hib.Region,
		OwnPrefix:       hib.Prefix,
		DataDir:         dataDir,
		Mirror: func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) {
			return lease.NewS3Mirror(ctx, hib.Region, dstRegion)
		},
		Quiesced: svc.Quiesced,
		Fence: func(ctx context.Context, host string, generation uint64) error {
			// A short deadline: FenceAndStop waits for an awake child to
			// stop, and one slow stop must not hold up every other move in
			// the pass. ErrStopUnproven still means the fence is installed,
			// which is all the cut asks of it; Quiesced answers the rest.
			fctx, cancel := context.WithTimeout(ctx, 2*time.Second)
			defer cancel()
			if _, err := svc.FenceAndStop(fctx, host, generation); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
				return err
			}
			return nil
		},
		LiftFence:    svc.LiftFence,
		SetStatus:    reg.SetStatus,
		CarryRequest: carry,
		OwnsRoute: func(host string) bool {
			rt, err := reg.Get(host)
			return err == nil && rt != nil && rt.Provisioned
		},
		Route: reg.Get,
		Checkpoint: func(ctx context.Context, dbPath string) error {
			host := strings.TrimSuffix(filepath.Base(dbPath), ".db")
			rep, err := lease.ReplicateWithLease(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, host), hib.Region)
			if err != nil {
				return err
			}
			defer rep.Close(ctx)
			return rep.SyncAndWait(ctx)
		},
		Now:          time.Now,
		ReadyDelta:   readyDelta,
		Enabled:      moves.Enabled(os.Getenv),
		claimRecords: writes.Enabled,
		Page:         PageOperator,
		// The move log and the member stream (Plan 6 Task 3, preflight L9).
		Events:   events,
		LogEvent: moveLogWriter(events),
	}
	// The edges box label and the version store, whatever the switch says:
	// the offline move reports under the label (movesoffline.go). armPhase1
	// sets the same two values.
	cfg.Box = metricsBoxLabel(os.Getenv, readMachineID())
	cfg.OptDir = release.OptDir()
	w := newMovesWorker(cfg)
	if cfg.Enabled {
		armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, writes, os.Getenv)
		if writes.Enabled && providers.guard.FenceView != nil {
			cfg.Guard = providers.guard
			cfg.Local.Guard = providers.guard
		}
		w.holds = newMovesHoldView(cfg.Box, cfg.Region, hostclaim.Read, holdStageRequester(dataDir))
		cfg.TimersDrained = w.holds.drainTimers
		if cfg.claimRecords {
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
			if err := armMovesParks(ctx, &cfg, providers); err != nil {
				log.Printf("agent: moves: the park arms stay unarmed: %v", err)
			}
			cancel()
		}
	}
	armOffline(&cfg, dataDir, reg, svc, eb, writes, os.Getenv)
	w.cfg = cfg
	armedMovesWorker.Store(w)
}

// armOffline sets every moveCfg field the offline operator move reads
// (movesoffline.go and Tasks 4 and 5 of docs/superpowers/plans/2026-10-20-
// offline-operator-move.md), whatever the Phase 1 switch says, after
// armPhase1 so a hook it already set is kept. Nothing here acts: the fields
// are read only by offline steps, which run only on a capable box and only
// for a record with an offline move.
func armOffline(cfg *moveCfg, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, getenv func(string) string) {
	cfg.OfflineCapable = offlineCapable(getenv, hibernationConfigured(getenv))
	cfg.Offline = newOfflineView(*cfg, holdStageRequester(dataDir))
	if svc != nil {
		cfg.FenceAs = fenceAsWith(svc)
		cfg.LiftFenceAs = svc.LiftFenceAs
	}
	if reg != nil {
		cfg.Registry = reg
		// The caller holds provision.WithHostHoldLock; a release that
		// landed touches the stage-requested marker, so root re-judges the
		// now-ready row (registry.SetHold's contract, G10).
		cfg.ReleaseHold = releaseHeld(reg, holdStageRequester(dataDir))
		// The row and the timers the move landed on it, in one
		// transaction, only while op holds the row.
		cfg.DeleteIfHold = reg.DeleteIfHoldWithTimers
	}
	if eb != nil {
		cfg.LaunchLayers = eb.LaunchLayers
		cgRoot := eb.CgroupRoot
		if cgRoot == "" {
			cgRoot = "/sys/fs/cgroup"
		}
		execRoot := eb.Cgroups.Root()
		cfg.Procs = func(host string) (int, error) {
			return cgroup.HostProcs(execRoot, host, hostSliceCgroup(cgRoot, host))
		}
	}
	if raw := strings.TrimSpace(getenv("CARLOS_SECRETS_KEY")); raw != "" {
		if priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY"); err == nil {
			cfg.ConfigPriv, cfg.ConfigPub = priv, priv.PublicKey()
			if cfg.SecretsKey == "" {
				cfg.SecretsKey = raw
			}
		}
	}
	if cfg.RenderHeld == nil {
		cfg.RenderHeld = heldRenderer(writes, cfg.ConfigPriv, dataDir, cfg.Box)
	}
	optDir := cfg.OptDir
	cfg.LiveVersion = func(host string) (string, error) { return release.LiveVersion(optDir, host) }
	pub, perr := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	cfg.ChannelVersion = func(ctx context.Context, s release.Store, rt registry.Route) (string, error) {
		if perr != nil {
			return "", perr
		}
		return channelVersion(ctx, s, pub, rt)
	}
	if cfg.Now == nil {
		cfg.Now = time.Now
	}
}

// channelVersion is the build the row's channel points at, as adoption
// resolves it: the account-qualified pointer of the row's account, app and
// channel ("stable" when the row names none), verified under the release
// key.
func channelVersion(ctx context.Context, s release.Store, pub ed25519.PublicKey, rt registry.Route) (string, error) {
	channel := rt.Channel
	if channel == "" {
		channel = "stable"
	}
	ptr, err := release.Channel(ctx, s, pub, release.RouteAccount(rt), rt.App, channel)
	if err != nil {
		return "", err
	}
	return ptr.Version, nil
}

// movesDataDir is the data directory the moves worker is built with,
// resolved once to an absolute, cleaned path (final re-review 1, I2): the
// outboxes under it sync every directory from an entry's own up to it, and
// that walk ends only at an absolute root. A CARLOS_DATA_DIR of "." or any
// relative path names the same directory as before. If the working
// directory cannot be read the path is kept as given and logged; the
// outboxes then refuse every write (outboxCreate), so nothing is
// acknowledged that is not durable.
func movesDataDir(dd string) string {
	abs, err := filepath.Abs(dd)
	if err != nil {
		log.Printf("agent: moves: the data directory %q cannot be made absolute, so the moves outboxes keep nothing: %v", dd, err)
		return dd
	}
	return abs
}

// stagingLeaseKey is the staging lease of a lane in targetRegion, the
// lane's placement code (the record's To, or its From for a return lane):
// moves.StagingLeasePrefix keyed by the lane's engine region, so a lane in
// home is park.HomeRegion and never an empty segment
// (control/staging-leases//<host>; home-region parks, part 1). Named codes
// keep their keys. A code that is no placement refuses.
func stagingLeaseKey(targetRegion, host string) (string, error) {
	r := park.PlacementRegion(targetRegion)
	if r == "" {
		return "", fmt.Errorf("%w: %q is not a placement code", park.ErrUnsafe, targetRegion)
	}
	return moves.StagingLeasePrefix(r, host), nil
}

// errHeldRenderUnpinned is the held render under the claim refusing, before
// anything is selected or written, a merge whose inputs are not the
// landing's pins: the configured step turns it into the landing's
// "configuration changed" refusal (refuseConfigChanged), never the policy's
// refusal of the write (the probe's rule, fix round 1 finding 4).
var errHeldRenderUnpinned = errors.New("moves: the held render's inputs are not the pinned ones")

// heldRenderer is the held row's render (§9.1; Task 7, Q2.2). With the claim
// off (writes.Enabled false) it is appconfig.RenderHeldBefore, unchanged.
// Under the claim it is appconfig.RenderHeldUnderBefore (platform-1e's
// answer to Q2.2): once the merged versions are known, a pin that is not
// them is errHeldRenderUnpinned, and otherwise the write is selected as
// held-config for the host (writes.Select) under the identity of exactly
// the merged configuration (HeldConfigIdentity), which MovesClaimPolicy
// grants at configured only when it is the pinned one (or, for a return,
// the recovery's). before still journals the rendered bytes' sha256 before
// the selection writes them (carry (c)).
func heldRenderer(writes hostwrite.Config, priv *ecdh.PrivateKey, dataDir, box string) func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
	if !writes.Enabled {
		return func(ctx context.Context, s release.Store, rt registry.Route, op string, _ *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
			return appconfig.RenderHeldBefore(ctx, s, priv, dataDir, rt, op, box, before)
		}
	}
	return func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
		return appconfig.RenderHeldUnderBefore(ctx, s, priv, dataDir, rt, op, box, func(pre appconfig.HeldRender) (hostwrite.Work, error) {
			if pin != nil && (!sameBundle(pre.BundleETags, pin.BundleETags) || pre.OverlayETag != pin.OverlayETag) {
				return nil, errHeldRenderUnpinned
			}
			w, err := writes.Select(ctx, rt.Host, movesActionHeldConfig)
			if err != nil {
				return nil, err
			}
			return heldIdentityWork{Work: w, ID: HeldConfigIdentity(pre.BundleETags, pre.OverlayETag)}, nil
		}, before)
	}
}

// armPhase1 wires the Phase 1 hooks (moves design §7.3). Only with the
// switch on: off, none is set and eb.Strict stays nil. writes is the
// agent's host-write configuration (the claim wiring's): the landing's
// held timers import, held render and held release select their Work
// through it.
func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, getenv func(string) string) {
	cfg.Box = metricsBoxLabel(getenv, readMachineID())
	cfg.OptDir = release.OptDir()
	cfg.DeployBucket = strings.TrimSpace(getenv("CARLOS_DEPLOYMENT_BUCKET"))
	if cfg.DeployBucket != "" {
		cfg.DeployBucketRegion = edgesFileRegion(getenv, readPlacementRegion())
		bucket, region := cfg.DeployBucket, cfg.DeployBucketRegion
		cfg.StagingLease = func(ctx context.Context, targetRegion, host string) (moves.SafeUntilGuard, func(context.Context) error, error) {
			key, err := stagingLeaseKey(targetRegion, host)
			if err != nil {
				return nil, nil, err
			}
			h, err := lease.AcquireTTL(ctx, bucket, key, region, moves.StagingLeaseTTL)
			if err != nil {
				return nil, nil, err
			}
			return h, h.Release, nil
		}
	}
	cfg.Staging = func(ctx context.Context, dstRegion string) (lease.StagingWriter, error) {
		return lease.NewStagingMirror(ctx, hib.Region, dstRegion)
	}
	cfg.Versioned = func(ctx context.Context, bucket, region string) (lease.VersionedStore, error) {
		return lease.NewVersionedS3(ctx, map[string]string{bucket: region})
	}
	if raw := strings.TrimSpace(getenv("CARLOS_SECRETS_KEY")); raw != "" {
		if priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY"); err != nil {
			log.Printf("agent: moves: CARLOS_SECRETS_KEY does not parse; Phase 1 rounds refuse: %v", err)
		} else {
			cfg.SecretsKey, cfg.SecretsPriv, cfg.SecretsPub = raw, priv, priv.PublicKey()
		}
	}
	cfg.FilesDir = movesFilesDirFor(dataDir)
	// The cut (movescut.go).
	cfg.FinalCheckpoint = finalCheckpoint(hib.Bucket, hib.Prefix, hib.Region)
	cfg.LocalPosition = localLTXPosition
	cfg.InputVersions = func(ctx context.Context, s release.Store, rec provision.Record) (map[string]string, string, error) {
		_, v, err := appconfig.LoadMergedHostVersions(ctx, s, rec.Account, rec.App, rec.Environment, rec.Host)
		if err != nil {
			return nil, "", err
		}
		return map[string]string{"default": v.Default, "environment": v.Environment}, v.Overlay, nil
	}
	if reg != nil {
		cfg.HoldMoving = holdMoving(reg, holdStageRequester(dataDir))
		cfg.Timers = reg.TimersFor
	}
	// The landing (movesland.go). Everything that exists on main is wired
	// here; ParkLeases, ParkView and ParkAuthorize come from the claim's
	// MoveParks (armMovesParks, Plan 5 Task 9a, in armMovesPass), with
	// park.Engine.RunHeld and the MoveParks local authorizer, and are set
	// only once the wake-source gate opens (movesWakeSourceRegistered, Plan 5
	// Task 9b: the moves wake source registered with both switches on); until
	// then landingArmed is false and the landing pass only logs. The held writes select their Work from writes (ruling
	// G16; Task 7, Q2): the timers through schedule.SelectedTimers as
	// held-timers (heldTimersReplacer), the render as held-config
	// (heldRenderer) and the release as held-release (heldReleaseHooks).
	// With CARLOS_HOST_CLAIM off each is exactly what it was; with it on,
	// the claim's selection and hostbox.Policy.Moves (MovesClaimPolicy)
	// decide. The landing still refuses at checking under the flag until
	// the wake-source gate opens (movesland.go).
	cfg.Guard = hostguard.Guard{DataDir: dataDir}
	cfg.Local = park.Local{DataDir: dataDir, Box: cfg.Box}
	cfg.OwnLease = hostbox.RenewingLeases{Region: hib.Region}.Acquire
	cfg.RenderHeld = heldRenderer(writes, cfg.SecretsPriv, dataDir, cfg.Box)
	cfg.RemoveVerifyEntry = dbverify.RemoveHost
	cfg.WriteGen = hostbox.WriteGenerationMarker
	if reg != nil {
		cfg.ReplaceTimers = heldTimersReplacer(writes, reg)
		cfg.SetReplication = reg.SetReplication
		cfg.SetNextDue = reg.SetNextDue
		// The source's finalization (movesfinal.go): a plain registry call
		// under the host's hold lock, not a claim-selected write (ruling
		// H5(d)). Unarmed with ParkLeases, as the landing is.
		cfg.DeleteHeld = reg.DeleteIfHoldWithTimers
		// The rollback's release hands the row's hold from the move to the
		// rollback, touching the stage-requested marker (ruling J3).
		cfg.HandHold = handHold(reg, holdStageRequester(dataDir))
		// The recovery's open ends the rollback's hold on the source's row,
		// a plain registry call under the host's hold lock (J12).
		cfg.ReleaseHeld = releaseHeld(reg, holdStageRequester(dataDir))
		// The revert's hold goes from moving to restoring under the move
		// id, a conditional status change under the host's hold lock
		// (movesrevert.go, Task 6).
		cfg.RestatusHold = restatusHold(reg, holdStageRequester(dataDir))
	}
	if eb != nil {
		// A box with no delegated cgroup has no manager, and Reconstruct
		// refuses it: such a box cannot produce stop evidence, so it never
		// cuts (Task 3). The lease step reads the activator's own lock, the
		// one its writer holds (ExecBackend.Acquire); the cut has refused
		// any holder whose location is not that prefix before it gets
		// here (ruling T5-lease), and a location that still differs is an
		// error, never evidence.
		cfg.Evidence = func(ctx context.Context, host, location string, dbPaths []string) (moves.StoppedReport, error) {
			writerPrefix := activator.HibernatePrefix(hib.Prefix, host)
			if strings.TrimSuffix(location, "/") != writerPrefix {
				return moves.StoppedReport{}, fmt.Errorf("%w: %s is at %q, not the writer's lease %q", errHolderLocation, host, location, writerPrefix)
			}
			return stopevidence.Reconstruct(ctx, stopevidence.ReconstructInputs{Host: host, Cgroups: eb.Cgroups, DBPaths: dbPaths,
				LeaseFree: func(ctx context.Context) (bool, error) {
					held, _, _, err := lease.LockState(ctx, hib.Bucket, writerPrefix, hib.Region)
					return !held, err
				}})
		}
	}
	cfg.DiskFree = diskBytes
	cfg.ProbeStore = probeStores
	cfg.LaneProbe = laneProbe(cfg.Versioned)
	cfg.MeasureDisk = measureDiskIn(dataDir, time.Now)
	cfg.MeasureS3 = measureS3(func(ctx context.Context) (lease.MirrorStore, error) { return w.mirror(ctx, hib.Region) }, hib.Bucket, time.Now)
	if reg != nil {
		// D3-held: the held path's own gate, for the life of the worker, so
		// its miss cache holds between passes; its bake directory is the
		// healthy history adoption reads.
		//
		// Settings spec rev 11 §4.3a: the held stage installs only a build a
		// promotion record or a pointer verified under the release key, and
		// admitted by the agent's pointer digest ratchet, authorizes. A key
		// that does not parse leaves pub nil, and every held stage is
		// refused: fail closed.
		pub, perr := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
		if perr != nil {
			log.Printf("moves: held stages are refused: %v", perr)
			pub = nil
		}
		//
		// Paul's T7 ruling (2026-10-12): under the claim T7 stages and
		// verifies without publishing, and the configured step publishes
		// under the held-release selection (heldReleaseHooks). With the
		// claim off both are heldReleaseReady, as before.
		//
		// Task 7b: under the claim the agent cannot write /opt/carlos, so a
		// blocked stage and every publication are asked of root over its
		// control socket (heldReleaseHooksVia), which re-reads the claim,
		// the record and the render journal and publishes only inside its
		// own held-release selection.
		via := heldReleaseVia{Region: cfg.Region}
		if writes.Enabled {
			via.Relay = newHeldReleaseRootClient(controlsock.Path)
		}
		cfg.ReleaseReady, cfg.ReleaseLive = heldReleaseHooksVia(via, writes, reg, cfg.OptDir, dataDir, cfg.Box, pub, pointerRatchet(false, cfg.OptDir, dataDir, getenv),
			release.NewRootProofGate(release.ProofPolicyPath, dataDir), stageMarkerRequester(dataDir))
		if svc != nil {
			// S1 asks the instance itself, woken through the activator.
			// An uncommitted import's row (a gate set, or closed by its
			// rollback) is never woken for S1: only its own verify requests
			// reach it (seeded-instances spec §5.4), and it moves only once
			// committed (§5.8). ServeScheduled does not read the gate, so
			// this does, for every round that probes, preparing or ready.
			cfg.ReadyProbe = func(ctx context.Context, host, path string) moves.Check {
				rt, err := reg.Get(host)
				if err != nil || rt == nil || rt.ImportGate != "" {
					return moves.Check{Reason: s1NoAnswer(host)}
				}
				return probeInstance(ctx, svc.ServeScheduled, *rt, path)
			}
		}
	}
	if eb != nil {
		cfg.StrictNext = eb.NextSpawnStrict
		cfg.ComposeHeldEnv = composeHeldEnv(cfg.SecretsPriv, cfg.Region, eb.Cache != nil)
		if reg != nil {
			cfg.LaunchEnv = func(host string) (map[string]string, map[string]bool, error) {
				rt, err := reg.Get(host)
				if err != nil {
					return nil, nil, err
				}
				if rt == nil {
					return nil, nil, fmt.Errorf("%s has no registry row on this box", host)
				}
				env, instance, _, err := eb.LaunchLayers(*rt)
				return env, instance, err
			}
		}
		eb.Strict = w.strictHost
		eb.Phase1 = true
	}
	// Recovery uses the same admitted Group D start path as other claim
	// writers. Preserve the legacy nil hook with the claim switch off.
	if cfg.claimRecords && svc != nil {
		armRecoveryProbe(cfg, w, svc.RecoveryProbe, svc.RecoveryProbeReady)
	}
}

// runMovesPass is the loops' call: inert until armed (no activator = no
// exec instances = nothing this box could ever move).
func runMovesPass(ctx context.Context, s release.Store) {
	w := armedMovesWorker.Load()
	if w == nil || s == nil {
		return
	}
	if err := movesPass(ctx, s, w); err != nil {
		log.Printf("agent: moves pass: %v", err)
	}
}

// moveCfg carries what the worker needs, injectable for tests.
type moveCfg struct {
	registerLanding func(context.Context, hostclaim.Observed) error
	claimRecords    bool   // Captured CARLOS_HOST_CLAIM, independent of the moves switch.
	Region          string // this box's placement code, "" included (the home)
	OwnBucket       string // CARLOS_HIBERNATE_BUCKET
	OwnBucketRegion string // CARLOS_HIBERNATE_REGION
	OwnPrefix       string // CARLOS_HIBERNATE_PREFIX ("instances" in production)
	DataDir         string // /data — where the local DB lives
	Mirror          func(ctx context.Context, dstRegion string) (lease.MirrorStore, error)
	Quiesced        func(host string) bool
	// Fence installs the activator's wake fence for host (G12, the
	// instance-moves design §6.1): status "moving" stops the edge, but the
	// tick runner fires whatever row the registry returns, and a tick that
	// woke the source after the purge would restore it from its own replica
	// and write where the target never looks. A nil error means the fence
	// is installed, whether or not the child has finished stopping. Nil
	// skips (tests without an activator).
	Fence func(ctx context.Context, host string, generation uint64) error
	// LiftFence removes a fence Fence installed, at that generation only.
	LiftFence func(host string, generation uint64) bool
	SetStatus func(host, status string) error
	// OwnsRoute answers whether THIS box's registry serves the host
	// (review round 1): region equality alone would let a standby or
	// second same-region box drive a cut whose child runs on its sibling —
	// each box has its OWN registry, so the sibling's Quiesced is
	// trivially, wrongly true. The box that owns the route owns the move.
	OwnsRoute func(host string) bool
	// Route reads the row itself, so a cutting move can checkpoint the
	// row's OWN DBPath (isolated-exec design §9) instead of guessing the
	// legacy <DataDir>/<host>.db. The guess is wrong for a nested row,
	// whose DB lives under provision.InstanceDir. Nil (a test with no
	// registry in hand) keeps the guess, unchanged.
	Route func(host string) (*registry.Route, error)
	// Checkpoint re-replicates the LOCAL DB to the own bucket and waits
	// (review round 1): the activator's stop logs — and swallows — a failed
	// final checkpoint, because pre-move the local warm cache made that
	// safe. The cut deletes the warm cache, so it must first prove the own
	// replica really carries the tail. Nil skips (tests without S3).
	Checkpoint func(ctx context.Context, dbPath string) error
	Now        func() time.Time
	// ReadyDelta is the readiness bar (#175): a preparing round whose
	// lease.MirrorRound.DeltaBytes (bytes copied plus bytes deferred) is at
	// most this promotes the move to ready. Zero means zero bytes moved: a
	// round that only deleted still qualifies. armMovesPass sets it from
	// moveReadyDeltaFromEnv.
	ReadyDelta int64
	// Enabled is the Phase 1 moves switch (moves.Enabled, set in
	// armMovesPass). Off, the worker runs the legacy protocol only.
	Enabled bool

	// CarryRequest merges this box's consumed request record for host into
	// the build state at bucket/prefix (release.CarryRequest: root's
	// resolution and a failed completion; never a successful completion,
	// never last_executed), under the cut's fence or lease, before a cut's
	// final round copies the state for the seal: the legacy cut, the Phase 1
	// cut's final round and a dirty return's lane (ruling R-SPEC-43A-2; Astra
	// round 1 finding 3). A request this box closed as failed then stays
	// closed on the destination. Nil carries nothing (a box with no
	// build-state store, tests).
	CarryRequest func(ctx context.Context, host, bucket, prefix string) error

	// Phase 1 (moves design §7.3), wired by armMovesPass only with the
	// switch on. A nil hook refuses its step, never falls back.
	//
	// Box is this box's label (metricsBoxLabel), the name the claim's
	// participants and op.intent.target use. OptDir is the version store
	// (the source report's Release is LiveVersion there).
	Box    string
	OptDir string
	// DeployBucket and DeployBucketRegion are where the staging-work lease
	// lives (CARLOS_DEPLOYMENT_BUCKET, CARLOS_AWS_REGION or the IMDS
	// region). StagingLease is nil without a deployment bucket, and then a
	// Phase 1 round refuses (round-5 finding 11).
	DeployBucket, DeployBucketRegion string
	StagingLease                     func(ctx context.Context, targetRegion, host string) (moves.SafeUntilGuard, func(context.Context) error, error)
	// Staging is the staging writer for a target bucket region (Task 6,
	// F17): every staging write goes through it behind the guard.
	Staging func(ctx context.Context, dstRegion string) (lease.StagingWriter, error)
	// Versioned reads a bucket by exact version: the target's T5 in its own
	// bucket and T10's lane in the source's. Never the source against the
	// target's lane.
	Versioned func(ctx context.Context, bucket, region string) (lease.VersionedStore, error)
	// SecretsKey is CARLOS_SECRETS_KEY (the launch report's HMAC key is
	// derived from it); SecretsPub seals the archive and SecretsPriv opens
	// it on the target.
	SecretsKey  string
	SecretsPub  *ecdh.PublicKey
	SecretsPriv *ecdh.PrivateKey
	// LaunchEnv is the source's composed launch environment and the names
	// the instance's own layers supplied (ExecBackend.LaunchLayers over the
	// host's row).
	LaunchEnv func(host string) (env map[string]string, instance map[string]bool, err error)
	// FilesDir is the files directory and <host>.env the archive carries
	// (flat: <DataDir>/<host>/ and <DataDir>/<host>.env).
	FilesDir func(host string) (dir, envPath string)
	// ReadyProbe is S1: GET path from the instance itself, woken through
	// the activator (probeInstance). StrictNext is S2
	// (ExecBackend.NextSpawnStrict).
	ReadyProbe func(ctx context.Context, host, path string) moves.Check
	StrictNext func(host string) moves.Check
	// The target's hooks: T3's composition of the held row, T4's store
	// probe, T6's disk reading, T7's release, T10's lane probe and T11's
	// regional S3 and disk measurements.
	ComposeHeldEnv func(ctx context.Context, s release.Store, rec provision.Record, carriedEnv []byte) (env map[string]string, instance map[string]bool, shadowed []string, in provision.MoveInputs, err error)
	ProbeStore     func(ctx context.Context, env map[string]string, moveID string) (prefix string, err error)
	DiskFree       func(dir string) (total, avail uint64, err error)
	ReleaseReady   func(ctx context.Context, s release.Store, rec provision.Record, version string) moves.Check
	LaneProbe      func(ctx context.Context, bucket, region, moveID string) error
	MeasureS3      func(ctx context.Context, key string) (moves.MeasuredRate, error)
	MeasureDisk    func(ctx context.Context) (moves.MeasuredRate, moves.FilesRate, error)

	// ReleaseLive is the configured step's release (§9.2 as amended
	// 2026-10-12): the release published for the held row and proven. With
	// the claim off it is ReleaseReady itself, which publishes; under the
	// claim T7 only stages, and this publishes under the landing's
	// held-release selection (heldReleaseHooks). Nil is ReleaseReady with
	// the claim off; under the claim a nil ReleaseLive never falls back to
	// T7's hook, and configured waits (moveCfg.releaseLive).
	ReleaseLive func(ctx context.Context, s release.Store, rec provision.Record, version string) moves.Check

	// The cut's hooks (§5.3, §7.4, movescut.go). HoldMoving holds the row
	// "moving" under the move id (holdMoving). TimersDrained waits until
	// no timer delivery or convergence that could have missed the hold is
	// still running (the hold view's drainTimers, Task 6). Evidence is
	// §5.3 steps 1 to 4 (stopevidence.Reconstruct, its lease step read on
	// the activator's own lease, which ruling T5-lease requires to be the
	// claim holder's location). FinalCheckpoint is step 5's checkpoint on
	// that same lease, bound to the claim's lineage, check being the
	// authority read before each replica request (G3). LocalPosition is
	// the stopped database's TXID (the brief's OwnReplicaPosition, G22).
	// Timers reads the host's timer rows. InputVersions reads the current
	// bundle etags (default, environment) and overlay etag, as T3 pinned
	// them.
	HoldMoving      func(host, op string) error
	TimersDrained   func(ctx context.Context, host string) error
	Evidence        func(ctx context.Context, host, location string, dbPaths []string) (moves.StoppedReport, error)
	FinalCheckpoint func(ctx context.Context, host, dbPath string, holder hostclaim.Holder, check func(context.Context) error) error
	LocalPosition   func(dbPath string) (uint64, error)
	Timers          func(host string) ([]registry.Timer, error)
	InputVersions   func(ctx context.Context, s release.Store, rec provision.Record) (bundle map[string]string, overlay string, err error)
	// cutFault is a test seam: called after each side effect of the cut
	// with the point's name, an error being a crash there. Nil in
	// production.
	cutFault func(point string) error

	// The landing's hooks (§7.6, movesland.go). Guard is the host guard
	// over this box's data directory (its Claims are set per pass from the
	// pass's store when nil); Local is park.Local over the same directory
	// (its Guard, Claims and, when nil, Authorize are set per pass).
	// ParkLeases, ParkView and ParkAuthorize are the claim's park engine's
	// (MoveParks, armMovesParks): its leases, view and authorizer over the
	// own bucket. Every arm requires all three (parkHooksArmed) and runs its
	// own check before the authorizer (movesArmAuthorize); nothing stands
	// in. In production they are set only once the wake-source gate opens
	// (movesWakeSourceRegistered, Plan 5 Task 9b), which leaves every arm
	// unarmed there. OwnLease
	// takes the location's lease in the own bucket for the build
	// (hostbox.RenewingLeases over lease.Acquire). RenderHeld is the held
	// row's render (heldRenderer): appconfig.RenderHeldBefore with the claim
	// off, appconfig.RenderHeldUnderBefore under a held-config selection with
	// it on; before journals the render's ownership ahead of its write,
	// carry (c), and pin, when not nil, is the landing's pinned inputs, which
	// a render under the claim must merge before anything is selected or
	// written (errHeldRenderUnpinned); ReplaceTimers is the
	// registry's through schedule.SelectedTimers with a held-timers Work
	// (heldTimersReplacer); SetReplication and SetNextDue are the
	// registry's; RemoveVerifyEntry is
	// dbverify.RemoveHost; WriteGen is hostbox.WriteGenerationMarker, the
	// one <dbPath>.gen writer (ruling G2).
	Guard          hostguard.Guard
	Local          park.Local
	ParkHomeRegion string
	ParkLeases     park.Leases
	ParkView       func(context.Context, hostclaim.Observed) (park.View, error)
	ParkAuthorize  func(context.Context, hostclaim.Observed, hostclaim.Obligation) error
	// ParkRunHeld is park.Engine.RunHeld (movesRunHeld): it runs a chore
	// under a lease the caller holds (M5: the landing's residue park under
	// its own-location lease), borrowed through the claim's leases
	// (AcquireHeld). ParkLocalAuthorize is the MoveParks local authorizer,
	// the owner authorizer for this box's park.Local chores (M9). Nil in
	// production until the wake-source gate opens; armMovesParks sets all
	// five or none.
	ParkRunHeld        func(ctx context.Context, e park.Engine, w park.Work, held park.HeldLease) error
	ParkLocalAuthorize func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error
	OwnLease           func(ctx context.Context, bucket, location string) (park.HeldLease, error)
	RenderHeld         func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(sha256hex string) error) (appconfig.HeldRender, error)
	ReplaceTimers      func(ctx context.Context, host string, rows []registry.Timer) error
	SetReplication     func(host, class string) error
	SetNextDue         func(host string, due int64) error
	RemoveVerifyEntry  func(ctx context.Context, s release.Store, host string) error
	WriteGen           func(path, generation string) error
	// Probe is the landing's probe wake (Plan 3 Task 6): Task 5's
	// activator.Service.ProbeWake. Nil in production, with a standing
	// notice, until claim Plan 11's launcher (R10) and Plan 5 Task 9's
	// arming exist; the landing then waits at restored.
	Probe func(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error)
	// ProbeLive is the activator's LiveProbe: the awake probe entry for a
	// host, its op and admitted session. Nil in production with Probe.
	ProbeLive func(host string) (op string, res activator.ProbeResult, ok bool)
	// MarkProbeOpen is the activator's OpenProbeEntry: the target's open
	// marks the probe's live entry open before it releases the hold, so an
	// edge reload in between keeps it (Plan 3 Task 7). Nil in production
	// with Probe.
	MarkProbeOpen func(host, op string) bool
	// ConfirmProbeOpen is the activator's ConfirmProbeOpened: called only
	// once the open's release of the hold is proved, ending the probe entry's
	// idle protection (final review, ruling 3). Nil in production with Probe.
	ConfirmProbeOpen func(host, op string) bool
	// landFault and parkFault are test seams: the landing's crash points
	// and the park engine's (park.Engine.After). Nil in production.
	landFault func(point string) error
	parkFault func(step string) error

	// The source's finalization (§7.9 step 3, movesfinal.go). DeleteHeld
	// deletes this box's row and its timers rows in one transaction, only
	// while op holds the row (registry.DeleteIfHoldWithTimers, ruling
	// H5(d)). The finalization also uses Guard, Local, ParkLeases,
	// ParkView, ParkAuthorize, Versioned and Route above; ParkLeases nil
	// leaves it unarmed in production (sourceFinalArmed). finalFault is
	// its test seam, nil in production.
	DeleteHeld func(host, op string) (bool, error)
	finalFault func(point string) error

	// The rollback on the target (§7.8, movesrollback.go). HandHold hands
	// the row's hold from the move to its rollback (handHold, ruling J3).
	// LogEvent records a move event (moves.LogEvent, Plan 6 Task 3): in
	// production moveLogWriter, the move log's create-only object and, when
	// this writer created it, the member event through Events (the agent's
	// rollout sink). Nil, the event is a log line. The target's
	// half also uses the cut's stop hooks, the landing's park hooks and
	// DeleteHeld above; ParkLeases nil leaves it unarmed in production
	// (rollbackTargetArmed). rbFault is its test seam, nil in production.
	HandHold func(host, status, toOp, fromOp string, validate func() error) (bool, error)
	LogEvent func(ctx context.Context, s release.Store, rec provision.Record, ev moves.LogEvent) error
	Events   activator.EventSink
	rbFault  func(point string) error

	// Page pages an operator (moves spec §13, K11): PageOperator in
	// production (armMovesPass), nil in tests, where nothing pages. The box
	// arms call it once per rule through the op journal (movespaging.go):
	// where a MoveMarkStuck lands (pageStuck: the rollback's return on the
	// target, and the recovery, its dirty return's install and the revert
	// on the source), where this box closes a rollback or a revert, and
	// where the cut's stop evidence finds a stray process.
	Page func(ctx context.Context, level, subject, body string) error

	// The source's recovery after a rollback (§7.8, movesrecover.go).
	// ReleaseHeld opens the row: registry.ReleaseHold under the op that
	// holds it, touching the stage-requested marker when it landed
	// (releaseHeld); the caller holds provision.WithHostHoldLock (J12). The
	// recovery also uses HandHold, LogEvent, the landing's hooks (the dirty
	// return's install) and Timers above; ParkLeases nil leaves it unarmed
	// in production (recoverySourceArmed). recoveryProbe is the recovery's
	// probe (Task 9, rulings J2 and J18; claim Plan 10 J2): in production
	// armRecoveryProbe binds the activator's Group D entry when host claims
	// are enabled. Missing runtime wiring waits before recording inputs.
	// rcFault is the recovery's test seam, nil in production.
	ReleaseHeld   func(host, op string) (bool, error)
	recoveryProbe func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error)
	rcFault       func(point string) error

	// The revert on the source (§7.4 reverting, movesrevert.go).
	// RestatusHold changes the status of a hold the op already has, moving
	// to restoring under the move id: registry.SetHoldIf expecting the op
	// itself, under the host's hold lock, after the caller's validation,
	// touching the stage-requested marker when it landed (restatusHold). The
	// revert also uses the cut's hooks, ReleaseHeld, Timers, FilesDir,
	// LogEvent and recoveryProbe above; recoveryProbe nil leaves it unarmed in
	// production (revertSourceArmed), and the target region's lane park needs
	// ParkLeases (revertLaneArmed). rvFault is its test seam, nil in
	// production.
	RestatusHold func(host, status, op string, validate func() error) (bool, error)
	rvFault      func(point string) error

	// The abort (§7.11, movesabort.go): the source uses the staging lease,
	// Versioned and its own bucket above; the target region's lane park is
	// laneParkTarget's, armed as the revert's (revertLaneArmed). abFault is
	// its test seam, nil in production.
	abFault func(point string) error

	// The offline operator move (movesoffline.go; docs/superpowers/plans/
	// 2026-10-20-offline-operator-move.md, Tasks 3 to 5). armOffline sets
	// every field below whatever the Phase 1 switch says, and so Box,
	// OptDir, SecretsKey, RenderHeld and Now above; tests set them by hand.
	// The sealer (Task 4) and the target (Task 5) read only these, plus the
	// hibernate coordinates, DataDir, Region, Mirror, Quiesced, Route,
	// Checkpoint and CarryRequest above.
	//
	// OfflineCapable is offlineCapable for this box: false, and every
	// offline step logs one notice per host and does nothing. Offline is the
	// keep view reconcile reads (offlineView.keep); the hold needs none.
	OfflineCapable bool
	Offline        *offlineView
	// FenceAs installs, or meets again, an owned activator fence
	// (activator.Service.FenceAndStopAs under a 2 s deadline,
	// ErrStopUnproven counted as installed: fenceAsWith); LiftFenceAs lifts
	// exactly that fence. Nil (no activator): never fenced.
	FenceAs     func(ctx context.Context, host string, generation uint64, owner string) error
	LiftFenceAs func(host string, generation uint64, owner string) bool
	// ReleaseHold opens a row held by op (registry.ReleaseHold, touching the
	// stage-requested marker when it landed); DeleteIfHold deletes a row
	// held by op, with its timers rows, in one transaction
	// (registry.DeleteIfHoldWithTimers). Both answer false when op does not
	// hold the row. The caller holds provision.WithHostHoldLock.
	ReleaseHold  func(host, op string) (bool, error)
	DeleteIfHold func(host, op string) (bool, error)
	// Registry is this box's registry: TimersFor, ReplaceTimers,
	// SetNextDue, SetReplication and Get.
	Registry *registry.Registry
	// LaunchLayers is the exec backend's composed launch environment for a
	// row (activator.ExecBackend.LaunchLayers).
	LaunchLayers func(rt registry.Route) (env map[string]string, instance map[string]bool, shadowed []string, err error)
	// ConfigPriv and ConfigPub are the deployment's config keypair, parsed
	// from CARLOS_SECRETS_KEY: ConfigPub seals the files archive, ConfigPriv
	// opens it and renders the held config. Nil without the key.
	ConfigPriv *ecdh.PrivateKey
	ConfigPub  *ecdh.PublicKey
	// LiveVersion is the build a row's live link names on this box
	// (release.LiveVersion over OptDir). ChannelVersion is the build the
	// row's channel points at, as adoption resolves it (channelVersion,
	// verified under CARLOS_RELEASE_PUBKEY; an error when that key does
	// not parse).
	LiveVersion    func(host string) (string, error)
	ChannelVersion func(ctx context.Context, s release.Store, rt registry.Route) (string, error)
	// Procs counts the processes left for a host in its exec cgroup and its
	// nested slice (cgroup.HostProcs). Nil, or an error: unknown, which a
	// source report records as -1, not stopped.
	Procs func(host string) (int, error)
}

// defaultMoveReadyDelta is the readiness bar when
// CARLOS_MOVE_READY_DELTA_BYTES is unset or rejected. A busy instance's
// per-round churn of new L0 files sits well under 16 MiB, so it reaches
// ready; a round in which compaction rewrote a large level file (138 MB
// for the 293 MB database in #175) does not, and waits for a quieter one.
// Ready keeps mirroring every pass until the cut, so the bar roughly
// bounds what the cut's final round, run while the instance is down, has
// left to copy.
const defaultMoveReadyDelta int64 = 16 << 20

// moveReadyDeltaEnv overrides defaultMoveReadyDelta with a whole number of
// bytes. Zero is accepted and means a round must move zero bytes.
const moveReadyDeltaEnv = "CARLOS_MOVE_READY_DELTA_BYTES"

// moveReadyDeltaFromEnv reads the readiness bar. An unset, unparsable or
// negative value keeps the default, so an operator's typo never stops the
// agent; ok is false only when a value was SET and rejected, so the caller
// can say so instead of leaving the operator to believe it took.
func moveReadyDeltaFromEnv(getenv func(string) string) (delta int64, ok bool) {
	v := strings.TrimSpace(getenv(moveReadyDeltaEnv))
	if v == "" {
		return defaultMoveReadyDelta, true
	}
	n, err := strconv.ParseInt(v, 10, 64)
	if err != nil || n < 0 {
		return defaultMoveReadyDelta, false
	}
	return n, true
}

// movesPass runs one round over every in-flight move this box owns.
func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
	w.pass.Lock()
	defer w.pass.Unlock()
	// With the change index armed, the record view (refreshed here inside
	// w.pass; lock order w.pass -> RecordView.mu, a leaf) decides only
	// whether this pass has any move work at all. Every move step has side
	// effects in both directions, and a snapshot can be older than a walk a
	// pass made while the view could not vouch (a bump that failed), so a
	// pass with work acts only on its own walk of the bucket, below, as
	// before the index (S3 change-index design §4.3, final review R1
	// finding 1). A pass with none skips the walk.
	idle := w.movesIdle(ctx, s)
	cfg := w.cfg
	if cfg.claimRecords {
		ctx = provision.WithRecordSequence(ctx)
	}
	if cfg.Enabled {
		// The pages the outbox and the index keep undelivered are sent again
		// first, before any record or fleet read (movespaging.go, fix round
		// 3). The move log's kept entries are retried next, whatever their
		// moves now are (moveslogoutbox.go). With the switch off neither
		// outbox is touched: what they keep waits on disk until it is on
		// again (final review I1).
		w.flushPageOutbox(ctx)
		w.flushLogOutbox(ctx, s)
	}
	if idle {
		return nil
	}
	// Strict returns exactly ListAllRecords' records and errors, plus the
	// hosts whose record did not decode, which the offline keep view keeps.
	w.walkedIdle = false
	recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s)
	if err != nil {
		return fmt.Errorf("moves: list records: %w", err)
	}
	var targets map[string]edgesFile // lazy: most passes carry no move
	var fleet *moveFleet             // lazy, Phase 1 only
	loadFleet := func() bool {
		if fleet == nil {
			f, err := loadMoveFleet(ctx, s)
			if err != nil {
				log.Printf("moves: %v. every Phase 1 move waits for the next pass", err)
				return false
			}
			fleet = &f
		}
		return true
	}
	if cfg.Enabled {
		w.lost = map[string]bool{}
		w.setStrict(recs)
		if w.holds != nil {
			w.holds.refresh(ctx, s, recs)
		}
	}
	// The offline keep view, from the list this pass read (movesoffline.go).
	if cfg.Offline != nil {
		cfg.Offline.refresh(recs, unreadable)
	}
	for _, rec := range recs {
		// An offline operator move is routed first, ahead of the Phase 1
		// skip and the lost-box gate below: its id makes it look like a
		// Phase 1 record, which with the switch off is skipped, and it
		// never reads a host claim (movesoffline.go).
		if rec.Move.Offline() {
			if err := w.offlineRound(ctx, s, rec); err != nil {
				w.logTrouble(rec.Host, err)
				continue
			}
			w.trouble.Delete(rec.Host)
			continue
		}
		// A move closed or replaced by another writer is forgotten by the
		// move log's memory (movesevents.go). With the switch off the pass
		// touches no outbox, index or memory of the move log at all (final
		// re-review 1, I1).
		if cfg.Enabled {
			w.retireMoves(rec)
		}
		// A seed moves only once placed, an import only once committed
		// (seeded-instances spec §5.8): a record that slipped past an old
		// console is refused here, before the first copy, in every round
		// (legacy, Phase 1 source, participant). targetReportPass holds
		// the same rule for the target. Only the move's source and target
		// say so: every box lists every record.
		if rec.Move != nil && rec.Move.State == provision.MovePreparing {
			if b := rec.Seed.MoveBlock(); b != "" {
				if rec.Region == cfg.Region || rec.Move.To == cfg.Region {
					w.notice(rec.Host+"#seed", "moves: %s: %s", rec.Host, provision.SeedMoveSentence(b))
				}
				continue
			}
		}
		if rec.Move != nil && !rec.Move.Legacy() {
			// A Phase 1 record. With the switch off it is never touched
			// (none exists unless the switch was on somewhere).
			if !cfg.Enabled {
				if rec.Region == cfg.Region {
					w.notice(rec.Host+"#switch", "moves: %s: move %s waits: %s is off on this box", rec.Host, rec.Move.ID, moves.SwitchEnv)
				}
				continue
			}
			// The lost-box gate (Q9, Plan 4 Task 10): a box the host's claim
			// lists in Lost, or whose claim does not read, runs no move step
			// for the host this pass.
			if w.lostHere(ctx, s, rec) {
				continue
			}
			if rec.Move.State == provision.MoveRollingBack && movesParticipantRecord(cfg, rec) == roleSource {
				// The source's recovery after a rollback (movesrecover.go,
				// Plan 4 Task 5): the record's region is the source's again,
				// and this box is the move's source by the post-flip
				// participant rule (J10), whether or not it still has the row.
				// The transitions another writer made (an operator's retry,
				// a declared loss) are logged first: this branch never
				// reaches phase1MoveRound (final review I3).
				w.observeMove(ctx, s, rec)
				if err := w.recoverSource(ctx, s, rec); err != nil {
					w.logTrouble(rec.Host, err)
					continue
				}
				w.trouble.Delete(rec.Host)
				continue
			}
			if rec.Move.State == provision.MoveReverting && movesParticipantRecord(cfg, rec) == roleSource {
				// The revert on the source (movesrevert.go, Plan 4 Task 6):
				// the record never flipped, and this box is the move's
				// source by the same participant rule (J10). Its entry
				// gives up execution through the cut, which reads the
				// fleet; the rest does not need it. An operator's retry is
				// logged first, as for the recovery above (final review I3).
				w.observeMove(ctx, s, rec)
				var f *moveFleet
				if loadFleet() {
					f = fleet
				}
				if err := w.revertSource(ctx, s, rec, f); err != nil {
					w.logTrouble(rec.Host, err)
					continue
				}
				w.trouble.Delete(rec.Host)
				continue
			}
			if rec.Region != cfg.Region || (cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host)) {
				// After the flip the record's region is the target's, and
				// after the source's finalization deletes its row OwnsRoute
				// no longer holds: a participant that still has the row or
				// the op's journal is visited by the post-flip rule (ruling
				// H5(a), movesfinal.go), never by phase1MoveRound.
				if role := movesParticipantRecord(cfg, rec); role != roleNone {
					if err := w.participantRound(ctx, s, rec, role); err != nil {
						w.logTrouble(rec.Host, err)
						continue
					}
					w.trouble.Delete(rec.Host)
				}
				continue
			}
			if !loadFleet() {
				return nil
			}
			if err := phase1MoveRound(ctx, s, w, rec, *fleet); err != nil {
				w.logTrouble(rec.Host, err)
				continue
			}
			w.trouble.Delete(rec.Host)
			continue
		}
		// A legacy record with the switch on: only its abort completes, by
		// the legacy path (F7(a)); anything else is refused (MV-N7).
		if cfg.Enabled && rec.Move != nil && rec.Move.State != provision.MoveAborting {
			if rec.Region == cfg.Region {
				w.notice(rec.Host+"#legacy", "moves: %s: %s", rec.Host, legacyMoveSentence)
			}
			continue
		}
		if rec.Move == nil || rec.Region != cfg.Region {
			continue
		}
		// The rails the console's moveGate also holds (and this side must
		// re-hold — records have other writers): only an activator-managed
		// exec instance's state is this worker's to ship. A pooled record's
		// data lives on the pool's box, not this one (v1 refuses the whole
		// class), and a non-hibernating instance runs OUTSIDE the activator
		// — Quiesced would answer true over a live, writing process.
		if rec.Pool != "" || !rec.HibernateOrDefault() || (rec.Kind != "" && rec.Kind != "instance") {
			continue
		}
		if cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host) {
			continue // another box's child; its owner drives the move
		}
		if targets == nil {
			if targets, err = moveTargets(ctx, s); err != nil {
				log.Printf("moves: %v — every in-flight move waits for the next pass", err)
				return nil
			}
		}
		if err := legacyMoveRound(ctx, s, w, rec, targets); err != nil {
			// One move's trouble must not starve its neighbours — and a
			// PERSISTENT one (an unapplied staging grant's AccessDenied,
			// review round 1) must not flood the journal at the fast
			// converge loop's cadence: once a minute per host is plenty.
			w.logTrouble(rec.Host, err)
			continue
		}
		w.trouble.Delete(rec.Host)
	}
	switch {
	case !cfg.Enabled:
	case hasPhase1Into(recs, cfg.Region):
		if loadFleet() {
			w.targetReportPass(ctx, s, recs, *fleet)
		}
	default:
		w.evictArchives(nil) // no move into this box: no verdict to keep
		w.retireHeldRefusals(nil)
	}
	if cfg.Enabled {
		// The source's op journals that reached the stamp, whoever's record
		// no longer routes here (movesfinal.go, fix round 1).
		w.sweepFinalJournals(ctx, s, recs)
	}
	if cfg.Enabled {
		// Delivered pages of moves retired on this box are removed, now that
		// every record is listed (movespaging.go). With the switch off the
		// outbox is left as it is (final review I1).
		w.retirePageTombstones(recs)
	}
	w.liftFences(recs)
	w.walkedIdle = len(unreadable) == 0 && !slices.ContainsFunc(recs, func(r provision.Record) bool { return r.Move != nil }) &&
		!w.localMoveState()
	return nil
}

// movesIdle reports whether this pass may skip its walk: the change index is
// armed and the record view can vouch, the last pass walked and found
// nothing (walkedIdle), the view's snapshot has no record with a move and
// none it cannot decode, and this worker holds no move state of its own. Any
// doubt is work: the pass walks the bucket. A move started by a write whose
// bump failed waits for the backstop rebuild (or for any pass that walks).
func (w *movesWorker) movesIdle(ctx context.Context, s release.Store) bool {
	vs, _, ok := viewStore(ctx, s)
	if !ok || !w.walkedIdle {
		return false
	}
	recs, unreadable, err := provision.ListAllRecordsStrict(ctx, vs)
	if err != nil || len(unreadable) > 0 || slices.ContainsFunc(recs, func(r provision.Record) bool { return r.Move != nil }) {
		return false
	}
	return !w.localMoveState()
}

// localMoveState reports whether this worker holds anything a moves pass
// acts on or cleans up from its record walk, whatever the records say: a
// cut's wake fence, a strict spawn set, the hold or offline keep view
// unbuilt or naming a host, T5's archive verdicts, a landing's lease, or,
// on disk, an op journal or a page outbox entry. Under pass.
func (w *movesWorker) localMoveState() bool {
	fenced := false
	w.fenced.Range(func(any, any) bool { fenced = true; return false })
	if fenced {
		return true
	}
	if m := w.strict.Load(); m != nil && len(*m) > 0 {
		return true
	}
	if w.holds != nil {
		if snap := w.holds.snap.Load(); snap == nil || len(snap.claims) > 0 || len(snap.unknown) > 0 {
			return true
		}
	}
	if v := w.cfg.Offline; v != nil {
		if snap := v.snap.Load(); snap == nil || snap.failed || len(snap.moves) > 0 || len(snap.unreadable) > 0 {
			return true
		}
	}
	w.mu.Lock()
	held := len(w.archives) > 0 || len(w.landLeases) > 0
	w.mu.Unlock()
	if held {
		return true
	}
	return movesDirHolds(w.cfg.DataDir)
}

// movesDirHolds reports whether <dataDir>/.moves holds anything below its
// top level: an op journal (.moves/<host>/<op>/, parked journals leave it)
// or a page outbox entry (.moves/.page-outbox/...). An unreadable directory
// holds something, as far as skipping a walk goes.
func movesDirHolds(dataDir string) bool {
	if dataDir == "" {
		return false
	}
	root := filepath.Join(dataDir, ".moves")
	tops, err := os.ReadDir(root)
	if errors.Is(err, os.ErrNotExist) {
		return false
	}
	if err != nil {
		return true
	}
	for _, top := range tops {
		if !top.IsDir() {
			continue
		}
		inner, err := os.ReadDir(filepath.Join(root, top.Name()))
		if err != nil || len(inner) > 0 {
			return true
		}
	}
	return false
}

// legacyMoveSentence is MV-N7, logged (never shown to a user) when the
// switch is on and a record carries a move with no id.
const legacyMoveSentence = "this move was started by an older console and has no id. abort it and request it again"

// hasPhase1Into reports whether any Phase 1 move targets region.
func hasPhase1Into(recs []provision.Record, region string) bool {
	for _, rec := range recs {
		if rec.Move != nil && !rec.Move.Legacy() && !rec.Move.Offline() && rec.Move.To == region {
			return true
		}
	}
	return false
}

// moveFenceGeneration is the cut fence's generation: the move's request
// time, so every pass of one cut installs the same fence (FenceAndStop is
// idempotent at an equal generation) and a later move of the same host
// fences above an earlier one.
func moveFenceGeneration(m *provision.MoveState) uint64 {
	if n := m.RequestedAt.UnixNano(); n > 0 {
		return uint64(n)
	}
	return 1
}

// liftFences lifts each cut fence once nothing on this box needs it: the
// host's route is gone from this box's registry (the flip retired it), or
// the record is at rest here again (a move back landed). Before either,
// the fence holds, including after the flip while the "moving" row waits
// for reconcile to retire it. The fences live in the activator's memory, so
// an agent restart drops them all; ServeScheduled's status gate covers that
// window, and a cut still in flight re-fences on its next pass.
func (w *movesWorker) liftFences(recs []provision.Record) {
	if w.cfg.LiftFence == nil {
		return
	}
	atRest := map[string]bool{}
	for _, rec := range recs {
		if rec.Move == nil && rec.Region == w.cfg.Region {
			atRest[rec.Host] = true
		}
	}
	w.fenced.Range(func(k, v any) bool {
		host, gen := k.(string), v.(uint64)
		held := true
		if w.cfg.Route != nil {
			rt, err := w.cfg.Route(host)
			if err != nil {
				return true // an unreadable registry is not a retired route
			}
			held = rt != nil
		}
		if held && !atRest[host] {
			return true
		}
		w.cfg.LiftFence(host, gen)
		w.fenced.CompareAndDelete(host, gen)
		log.Printf("moves: %s: wake fence lifted", host)
		return true
	})
}

// moveTargets folds the fleet's edges files into placement → coordinates.
// A duplicate placement code keeps the lexically-first box (deterministic
// everywhere, same rule as foldHomes) — and a target with no data bucket is
// simply absent, which legacyMoveRound reports per-move.
func moveTargets(ctx context.Context, s release.Store) (map[string]edgesFile, error) {
	bodies, err := loadEdgesFileBodies(ctx, s)
	if err != nil {
		return nil, err
	}
	out := map[string]edgesFile{}
	for _, body := range bodies {
		var t edgesFile // the writer's own shape — no parallel decl to drift
		if err := json.Unmarshal(body, &t); err != nil || t.V < 1 || t.Placement == "" || t.DataBucket == "" {
			continue
		}
		if prev, dup := out[t.Placement]; dup && prev.Box <= t.Box {
			continue
		}
		out[t.Placement] = t
	}
	return out, nil
}

// legacyMoveRound is the worker's round for a legacy record (no move id),
// main's runMoveRound unchanged. It is the only round with the Phase 1
// switch off; with it on, it completes a legacy aborting record and nothing
// else (F7(a)).
func legacyMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, targets map[string]edgesFile) error {
	cfg := w.cfg
	tgt, ok := targets[rec.Move.To]
	if !ok {
		return fmt.Errorf("target region %s publishes no data bucket yet — waiting", rec.Move.To)
	}
	// The mirror pair is built lazily and memoized per target region: the
	// cut's not-yet-quiesced early return must not pay a credential-chain
	// resolution it never uses.
	mirror := func() (lease.MirrorStore, error) { return w.mirror(ctx, tgt.DataBucketRegion) }
	srcPrefix := activator.HibernatePrefix(cfg.OwnPrefix, rec.Host)
	dstPrefix := activator.MovesPrefix(rec.Host)
	// The copy receipts (#175) live in this box's OWN bucket: the staging
	// grant on the target is push-only, so receipts there could be
	// written but never read back.
	receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: activator.MoveReceiptsKey(rec.Host)}

	switch rec.Move.State {
	case provision.MovePreparing, provision.MoveReady:
		m, err := mirror()
		if err != nil {
			return err
		}
		round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts)
		if err != nil {
			return err
		}
		// Ready means the gap is small, not that it is closed (#175): an
		// awake instance writes new litestream files between any two
		// listings, so a round that copied NOTHING never comes. The cut
		// closes the gap itself (quiesce, fence, final checkpoint, a final
		// round), so this bar only bounds how much that last round, and
		// the downtime around it, has left to copy.
		if rec.Move.State == provision.MovePreparing && round.DeltaBytes() <= cfg.ReadyDelta {
			if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now())); err != nil && !errors.Is(err, provision.ErrMoveConflict) {
				return err
			}
			nudgeSerial(ctx, s)
			log.Printf("moves: %s: staged replica caught up in %s/%s (last round copied %d bytes, deferred %d, in %d changes; bar %d) — ready to cut", rec.Host, tgt.DataBucket, dstPrefix, round.BytesCopied, round.BytesDeferred, round.Changed, cfg.ReadyDelta)
		}
		return nil

	case provision.MoveAborting:
		// Abort's box half: delete the staged copy, THEN clear the move —
		// a lingering staging would win a later landing on the target
		// (staging-wins, activator.promoteStagedReplica).
		m, err := mirror()
		if err != nil {
			return err
		}
		if err := lease.DeletePrefix(ctx, m, tgt.DataBucket, dstPrefix); err != nil {
			return err
		}
		if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) {
			return err
		}
		clearMoveReceipts(ctx, m, receipts, rec.Host)
		nudgeSerial(ctx, s)
		log.Printf("moves: %s: abort complete — staged copy deleted, instance stays in %q", rec.Host, rec.Region)
		return nil

	case provision.MoveCutting:
		// Step 1+2: quiesce. SetStatus is idempotent; a route already
		// "moving" (a resumed cut) just re-writes the word.
		if err := cfg.SetStatus(rec.Host, "moving"); err != nil {
			return fmt.Errorf("set route status: %w", err)
		}
		// Fence before the quiesce check, so the answer Quiesced gives stays
		// true: without it a tick could wake the host between this pass's
		// check and the purge (G12). Every pass re-installs the same fence,
		// which is also how a restarted agent gets it back.
		if cfg.Fence != nil {
			gen := moveFenceGeneration(rec.Move)
			if err := cfg.Fence(ctx, rec.Host, gen); err != nil {
				return fmt.Errorf("fence against wakes: %w", err)
			}
			w.fenced.Store(rec.Host, gen)
		}
		if !cfg.Quiesced(rec.Host) {
			if _, done := w.nudged.LoadOrStore(rec.Host, true); !done {
				nudgeSerial(ctx, s)
			}
			return nil // the retire is in flight; the nudged pass re-enters here
		}
		// A host whose layout migration is in flight (isolated-exec design
		// §8.2) is fenced here exactly as Classify fences it for a wake: its
		// DB may be mid-copy between the flat and nested paths, and neither
		// location is safe to checkpoint or purge until the journal clears.
		// Retried next pass, like every other not-yet-ready state above.
		// JournalPresent Lstats rather than Stats, so a dangling symlink at
		// the journal path still fences: the pass, the activator, SyncExec
		// and litestream all share this one predicate for "held", and a
		// journal this box cannot even read is not proof there isn't one.
		if layout.JournalPresent(cfg.DataDir, rec.Host) {
			log.Printf("moves: %s: layout migration in flight; leaving the move for the next pass", rec.Host)
			return nil
		}
		// Step 3: prove the own replica carries the tail. The activator's
		// stop SWALLOWS a failed final checkpoint (safe pre-move: the warm
		// cache stood behind it; fatal here: step 5 deletes that cache).
		// Re-checkpointing an already-synced DB is cheap and idempotent; a
		// held lease (the child not fully released) or an S3 blip returns
		// an error and the pass retries.
		//
		// dbPath defaults to the legacy flat guess and is overridden by the
		// row's own DBPath when Route is wired (isolated-exec design §9): a
		// nested row's DB never lived at the flat guess at all, so
		// checkpointing and purging (step 5, below) must follow the row.
		dbPath := filepath.Join(cfg.DataDir, rec.Host+".db")
		if cfg.Route != nil {
			if rt, err := cfg.Route(rec.Host); err == nil && rt != nil && rt.DBPath != "" {
				dbPath = rt.DBPath
			}
		}
		if cfg.Checkpoint != nil {
			if _, statErr := os.Stat(dbPath); statErr == nil {
				if err := cfg.Checkpoint(ctx, dbPath); err != nil {
					return fmt.Errorf("final checkpoint: %w", err)
				}
			}
		}
		// Step 3b: this box's request record into the build state, so the
		// final round carries a request it closed as failed (ruling
		// R-SPEC-43A-2). The instance is quiesced and fenced: no wake writes
		// the state now.
		if cfg.CarryRequest != nil {
			if err := cfg.CarryRequest(ctx, rec.Host, cfg.OwnBucket, srcPrefix); err != nil {
				return fmt.Errorf("carry the request record into the build state: %w", err)
			}
		}
		// Step 4: the byte-final mirror round.
		m, err := mirror()
		if err != nil {
			return err
		}
		round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts)
		if err != nil {
			return err
		}
		// A deferred object is one the source changed under this round, so
		// staging is NOT byte-final. Nothing should write a quiesced,
		// fenced, checkpointed source, but the purge below is
		// irreversible, so it waits for a round that proves it.
		if round.Deferred > 0 {
			return fmt.Errorf("final mirror round: %d object(s) changed at the source during the copy; retrying the cut", round.Deferred)
		}
		// Step 5: the warm-cache purge.
		activator.PurgeLocalDBArtifacts(dbPath, true)
		// Step 6: the last act — Region flips, Move clears, one write.
		if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) {
			return err
		}
		clearMoveReceipts(ctx, m, receipts, rec.Host)
		w.nudged.Delete(rec.Host)
		// The flip's nudge is what turns the TARGET's mint from a
		// 5-minute wait into a seconds-cadence converge.
		nudgeSerial(ctx, s)
		log.Printf("moves: %s: cut complete — now homed in %s (route retires on the next reconcile pass)", rec.Host, rec.Move.To)
		return nil
	}
	return nil
}

// clearMoveReceipts deletes a finished or abandoned move's copy receipts.
// Best-effort, after the record write that ended the move: receipts are
// self-checking (a stale one can only fail to match, never vouch for the
// wrong object), and some deployments' store policy denies DeleteObject on
// the data bucket, so a failure costs one small object, never a move.
func clearMoveReceipts(ctx context.Context, m lease.MirrorStore, rc lease.Receipts, host string) {
	if err := m.Delete(ctx, rc.Bucket, rc.Key); err != nil {
		log.Printf("moves: %s: could not delete copy receipts %s/%s: %v (harmless; they no longer match anything)", host, rc.Bucket, rc.Key, err)
	}
}

// nudgeSerial bumps control/serial after a box-side move write, so the
// NEXT actor — this box's own cut continuation, the target's mint, the
// tenant's poll — reacts on the fast-converge path instead of the 5-minute
// floor. The live rehearsal measured that floor three times over in one
// cut (status→quiesce→flip 331s, target mint+wake 700s); with the nudge,
// each hop is a serial-triggered pass. Best-effort like the console's own
// bumpSerial: a failed bump costs cadence, never correctness.
func nudgeSerial(ctx context.Context, s release.Store) {
	if err := release.BumpSerial(ctx, s, time.Now()); err != nil {
		log.Printf("moves: serial nudge: %v (falling back to the timer floor)", err)
	}
}

// movingStatusNudged remembers which hosts this process already nudged for
// the status→quiesce hop, so a child that refuses to die does not turn the
// fast-converge loop into a 2s self-triggering cycle: ONE nudge per host
// per process, then the 5-minute floor stands under a wedged stop.
var movingStatusNudged sync.Map

// logTrouble rate-limits per-host move errors to one line a minute; the
// pass clears a host's entry on its next clean round.
func (w *movesWorker) logTrouble(host string, err error) {
	now := time.Now()
	if last, ok := w.trouble.Load(host); ok && now.Sub(last.(time.Time)) < time.Minute {
		return
	}
	w.trouble.Store(host, now)
	log.Printf("moves: %s: %v", host, err)
}