package hostlife

import (
	"fmt"
	"sort"
	"strings"
)

// ---------- relocation (region HA spec 1 §5.8; claim rev 6.30 §4, L1; plan 1b-next rev 16) ----------
//
// A same-region executor change. The model's steps are abstract time:
// probeStartWindow (30) is a probe attempt's readiness window, so the
// probe deadline holds three attempts; sigAgeH (45) is the 15-minute
// signature age, so the failover pin delay (30 s) is two steps. Each
// guard names the plan 1b-next rev 16 function it mirrors.

const (
	failoverPinDelay = 2                       // §5.8: pin no earlier than the fence's observed_at plus 30 s (hostclaim.FailoverPinDelay)
	probeDeadline    = 3*probeStartWindow + 10 // §5.8: at most three attempts within the probe deadline (hostrelocate.ProbeDeadline)
	releaseDeadline  = probeDeadline           // 1b-next ruling 14: a release not journaled this long after releasing or probe-failed began is stuck (hostrelocate.ReleaseDeadline, 10 minutes like the probe deadline)
	maxReplacements  = 2                       // §5.8: after two replacements of the target, stuck (hostclaim.MaxTargetReplacements)
	enrollDelay      = 20                      // H16, H17: a launched box is confirmed and enrolled by a worker some time later
	replaceDelay     = 30                      // H5: the group launches a replacement for a terminated box
)

// userRelocate is a scale-in drain (§5.8, reason drain; spec 4 owns the
// decision): the serving host's executor is asked to give the host to
// another eligible box of its region. The start CAS (StartRelocation)
// records the intent, op.exec stays on the source, and both participants
// acknowledge by writing execution leases (role()). The two-box step
// draws it. A host that never admitted a writer is not drained (1b-next
// ruling 11, Worker.Start; rule reloc-drain-needs-writer): its release has
// no writer session to stop, so root refuses its stop evidence and the
// drain could only wait for its release deadline; a failover relocates it.
func (w *world) userRelocate() {
	c := w.claim
	if c == nil || c.Holder == nil || c.Op.Kind != "serving" || w.haOff {
		return
	}
	from := w.execBox(c)
	if from == nil || from.Destroyed || from.Old || !from.Enrolled {
		return
	}
	// Worker.Start's order (rev 7): the source unfenced, a writer to
	// release, an eligible target, the admission registration; the CAS
	// rechecks the admission and applies the start guard
	if on("reloc-drain-needs-writer") && c.Holder.Session == 0 {
		w.stats["drain refused: the host never had a writer"]++
		return
	}
	to := w.chooseBoxExcept(from.Region, c.Holder.App, from.ID)
	if to == nil || to.Old {
		return
	}
	if ad := w.admit[c.Holder.App]; ad != nil && ad.Closed {
		return // a box-started relocation needs the app's admission registration (1b-next ruling 4): a purge closed it
	}
	id := w.newID("rl")
	if w.cas(c, func(n *claim) bool {
		if n.Holder == nil || n.Op.Kind != "serving" || n.Op.Exec != from.ID || contains(n.Lost, to.ID) {
			return false
		}
		if ad := w.admit[n.Holder.App]; ad != nil && ad.Closed {
			return false // Admission.Recheck
		}
		if !w.relocationStartAllowed(n, from, to) {
			return false
		}
		n.Op = opState{Kind: "relocating", ID: id, Phase: "starting", Exec: from.ID, Parts: []int{from.ID, to.ID},
			Src: from.ID, Tgt: to.ID, SrcRegion: from.Region, RelocReason: "drain", ReqAt: w.now, App: n.Holder.App,
			SrcGen: n.Holder.Gen, SrcStamp: n.Holder.Stamp}
		return true
	}) {
		w.stats["relocation started: drain"]++
		w.logf("relocation %s: drain of inc %d from box %d to box %d in region %d", id, c.Holder.Inc, from.ID, to.ID, from.Region)
	}
}

// relocationStartAllowed is the claim's start guard as L11 amends it for
// relocating (RelocationStartGuard): a park-replica or any other
// whole-location chore in the region blocks (CC11), a park-local on the
// target or on a live source blocks, and a park-generation in the region
// does not: relocation writes only its own new prepared generation, never
// one being retired. The mutation of reloc-start-guard starts over
// anything.
func (w *world) relocationStartAllowed(c *claim, from, to *box) bool {
	return w.relocationStartGuard(c, from, to, false)
}

// relocationStartGuard is RelocationStartGuard(c, region, source, target,
// sourceFenced): a failover's fenced source's park-local does not block.
func (w *world) relocationStartGuard(c *claim, from, to *box, sourceFenced bool) bool {
	fenced := -1
	if sourceFenced {
		fenced = from.ID
	}
	if w.completionBarrier(c, fenced) {
		return false // ruling 77, before the chores (RelocationStartGuard)
	}
	if !on("reloc-start-guard") {
		return true
	}
	for _, o := range c.Obl {
		switch {
		case o.Kind == "park-generation":
			continue // L11
		case o.Kind == "park-local":
			if o.Box == to.ID || o.Box == from.ID && !sourceFenced && !contains(c.Lost, from.ID) {
				return false
			}
		case o.Region == from.Region:
			return false
		}
	}
	return true
}

// relocExcept is every box the operation may not name as its target
// (replaceTargetSlot, Worker.next): the source, unless an operator's
// resolve made it eligible again (model ruling 13), every replaced target,
// and the extra boxes given.
func relocExcept(op opState, extra ...int) []int {
	var out []int
	if !op.Resolved {
		out = append(out, op.Src)
	}
	out = append(out, op.Excluded...)
	return append(out, extra...)
}

// relocRelease is the release sequence (§5.8, A3-2; 1b-next Task 7's
// Release) a live box runs to give the host up: the source in releasing,
// a failed target in probe-failed. One step per call; it returns the
// journal entry once every step is done, nil while one is owed.
//
//  1. stop every writer: admission closed (the row held), the process
//     tree stopped and empty; an escaped descendant fails the step (L10);
//  2. flush: the final checkpoint and SyncAndWait, passing only once the
//     listing shows the stopped database's position N (moves §5.1 step
//     1). It needs the session's own replicator and the box's own files
//     directory, or the positions this box recorded for this stop (rule
//     reloc-release-needs-flush, rev 9). When the replicator died with
//     its agent (claim rev 6.10) and nothing was recorded there is no
//     release, and the release deadline makes the operation stuck; the
//     mutation journals what the remote holds, Flushed false;
//  3. publish every pending file and wake-state mutation as one final
//     manifest of the generation (ReleaseFlushFiles, Task 4);
//  4. record the positions, then the producers' drain and root's
//     receipt; recorded positions finish a release only for the
//     controller that recorded them or with that receipt (rule
//     reloc-release-needs-receipt, rev 11).
func (w *world) relocRelease(b *box, c *claim) *releaseEntry {
	h := c.Holder
	// StopWriters (L18 joint round 2 finding 3): the current attempt's and
	// session's launch journal closes first, so the stopped session is never
	// relaunched; root's durable admission fence (relFence) comes only with
	// Collect, after the flush and the recorded positions (ruling 28). In
	// between, no wake admits a writer: AdmitWriter's relocatingAdmission
	// admits nobody in starting or releasing (rule reloc-admits-probe-only).
	// The row hold is reconcile's alone (reloc-row-cases, the catch-up hold)
	w.closeLaunches(b, c.Op.ID, h.Session)
	if !w.evidence(b, c.Op.ID) {
		w.stats["release: a process survives the stop"]++
		return nil
	}
	if r := b.relRecord; r != nil && on("reloc-release-needs-flush") && r.Op == c.Op.ID && r.Phase == c.Op.Phase && r.E.Session == h.Session && r.E.Gen == h.Gen {
		// a flush this box recorded for this stop (ReleasePositions.Recorded):
		// never flushed or guessed again (rev 9). Rev 11 (Astra round 8):
		// the positions are not proof of the stop. The controller that
		// recorded them resumes at the producers' drain, and root persists
		// the receipt (step 4); after a restart only root's durable receipt
		// for the recorded identity finishes the release, and with none
		// RecoveryReleaseDrain refuses: no release, and the deadline
		// makes the operation stuck. The mutation of
		// reloc-release-needs-receipt replays the positions alone, as rev
		// 10 did
		switch {
		case r.Receipt || !on("reloc-release-needs-receipt"):
			if r.Ctl != b.Epoch {
				w.stats["release: replayed from the recorded positions"]++
			}
		case r.Ctl == b.Epoch:
			r.Receipt = true // the producers drain; root persists the receipt
			w.stats["release: receipt persisted"]++
			b.relFence = c.Op.ID // Collect: root's durable admission fence, after the flush and the positions
		default:
			w.stats["release: recorded positions without a receipt, refused"]++
			return nil
		}
		b.relFence = c.Op.ID // root's fence stands with the receipt (Collect)
		e := r.E
		e.Files, e.Timers = r.E.Files.clone(), append([]int(nil), r.E.Timers...)
		return &e
	}
	prefix := dataKey(h.Location, h.Gen, "")
	remote := w.remoteMax(w.buckets[b.Region], prefix)
	n, flushed := remote, false
	if on("reloc-release-needs-flush") && !(b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == h.Gen && b.DB.Epoch == b.Epoch && b.DB.Session == h.Session) {
		// relocationFlushDB needs the session's own replicator, and
		// nothing recorded a flush for this stop: no release (1b-next Task
		// 7; rev 9, Astra round 6). The release deadline makes the
		// operation stuck; the mutation journals the remote position with
		// Flushed false, as rev 8 did
		w.stats["release: no replicator and no recorded flush, refused"]++
		return nil
	}
	if db := b.DB; db != nil && db.Inc == h.Inc && db.Gen == h.Gen {
		if db.Epoch == b.Epoch && db.Session == h.Session {
			if len(db.Pending) > 0 {
				bl := w.deltaOf(db, db.Inc)
				db.FirstFull = false
				db.Queue = append(db.Queue, bl)
				db.Pos, db.Pending = bl.Max, nil
			}
			w.ship(b)
			if len(db.Queue) > 0 || db.InFlight != nil || remote < db.Pos {
				w.stats["release: waits for the flush"]++
				return nil
			}
			n, flushed = db.Pos, true
		} else {
			w.stats["release: the session's replicator is gone, unflushed"]++
		}
	}
	// the files the box holds, or, when it holds none of this incarnation's
	// (a failed target whose unopened install a scoped cleanup retired),
	// the set its own installation applied: a release never invents a
	// files identity
	files := b.Files.clone()
	if files == nil || files.Inc != h.Inc {
		if on("reloc-release-needs-flush") {
			// ReleaseFlushFiles publishes this box's own files directory;
			// there is none to publish (rev 9)
			w.stats["release: no files directory to flush, refused"]++
			return nil
		}
		files = w.relocFiles(c).clone()
	}
	// 3. (Task 4) publish every pending file and wake-state mutation,
	// whatever the app's files_durability, as one final manifest of the
	// generation (ReleaseFlushFiles): create-only, acknowledged before the
	// entry is journaled, its sequence journaled as manifest_seq
	seq := 0
	if w.perRegion > 1 {
		if b.pubGen != h.Gen {
			b.pubGen, b.pubSeq, b.releaseSeq = h.Gen, 0, 0
		}
		if !w.ensureObject(b, h, files) {
			w.stats["release: waits for its files object"]++
			return nil
		}
		if b.releaseSeq == 0 {
			b.pubSeq++
			b.releaseSeq = b.pubSeq
		}
		w.issued++
		bl := &blob{Kind: "files", Inc: h.Inc, Gen: h.Gen, Seq: b.releaseSeq, Files: files.clone(), Timers: append([]int(nil), b.Timers...), Base: w.issued}
		if !w.ackedPut(b, manifestKey(h.Location, h.Gen, b.releaseSeq), bl, true) {
			w.stats["release: waits for its manifest"]++
			return nil
		}
		seq, b.releaseSeq = b.releaseSeq, 0
		b.published = filesState(b.Timers, files)
	}
	e := &releaseEntry{Box: b.ID, Session: h.Session, Gen: h.Gen, FinalTX: n, ManifestSeq: seq, Flushed: flushed,
		Files: files, Timers: append([]int(nil), b.Timers...), At: w.now}
	if on("reloc-release-needs-flush") {
		// ReleasePositions.Record: durable before the producers close
		b.relRecord = &releaseRecord{Op: c.Op.ID, Phase: c.Op.Phase, E: *e, Ctl: b.Epoch}
		b.relRecord.E.Files, b.relRecord.E.Timers = files.clone(), append([]int(nil), b.Timers...)
		if on("reloc-release-needs-receipt") {
			// the producers' drain and root's receipt are the next step:
			// a crash here leaves positions and no receipt (rev 11)
			return nil
		}
	}
	if b.relFence != c.Op.ID {
		b.relFence = c.Op.ID // Collect, on the mutations' paths that skip the record or the receipt
	}
	return e
}

// pinSource is the generation the pin reads (StorePinner's genselect.Select
// over the admitted chain): holder.generation when it holds an object or
// the marker names it; else the baseline the recovery walk selects for an
// unfinished run of cold admissions (claim rev 6.22 §6.2a), as a cold
// admission would restore it. An unexplained empty generation refuses:
// absence is never evidence (§5.9).
func (w *world) pinSource(b *box, c *claim) (string, bool) {
	h := c.Holder
	loc := h.Location
	if w.hasFiles(b, loc, h.Gen) {
		return h.Gen, true
	}
	bg, _ := w.bucketGen(b, loc)
	if bg == h.Gen {
		return h.Gen, true
	}
	run, ok := w.unfinishedRun(c, bg)
	if !ok {
		return "", false
	}
	return w.runBaseline(b, c, run, bg)
}

// selectPin is StorePinner.Pin against the observation c: the source
// generation and its listing's newest position N, after a release only
// once N reaches the release's final_txid (rule reloc-pin-reaches-release),
// and an empty listing only with evidence (ruling 15, rule
// reloc-pin-evidenced-empty): a journaled release that flushed at 0, or a
// holder that never admitted a writer; any other empty listing marks the
// operation stuck for an operator. The pin carries the claim version it
// was selected against (ruling 16, RecoveryPinKey). Task 4 adds the files
// pin. It returns nil while the pin must wait.
func (w *world) selectPin(b *box, c *claim) *pinEntry {
	op, h := c.Op, c.Holder
	src, ok := w.pinSource(b, c)
	if !ok {
		w.stats["pin: the source generation cannot be selected yet"]++
		return nil
	}
	n := w.remoteMax(w.buckets[b.Region], dataKey(h.Location, src, ""))
	r := op.Release
	if r != nil && on("reloc-pin-reaches-release") && n < r.FinalTX {
		w.stats["pin: waits for the release's final txid"]++
		return nil
	}
	if n == 0 && on("reloc-pin-evidenced-empty") && !op.Resolved && h.Session != 0 && !(r != nil && r.Flushed && r.FinalTX == 0) {
		w.markStuck(c, fmt.Sprintf("an empty listing of %q is no evidence of an empty database (session %d)", src, h.Session))
		return nil
	}
	pin := &pinEntry{Gen: src, TX: n, ClaimVer: c.Ver, At: w.now, Files: w.relocFiles(c).clone(), Timers: append([]int(nil), w.relocTimers(c)...)}
	// (Task 4) the files pin: the newest landed manifest of the admitted
	// chain, by exact version, independently of the database baseline
	// (B2-4); after a release it must reach the release's manifest_seq
	if mk, mv, ms, mg, _ := w.newestManifest(b.Region, h.Location, h.Gen); mk != "" {
		if r != nil && r.ManifestSeq > 0 && on("reloc-pin-reaches-release") && (mg != r.Gen || ms < r.ManifestSeq) {
			w.stats["pin: waits for the release's manifest"]++
			return nil
		}
		pin.ManKey, pin.ManVer, pin.ManSeq, pin.ManGen = mk, mv, ms, mg
	}
	// 1b-next ruling 25 as rev 18 amended it (finding F2; rule
	// files-pin-needs-key): when a writer was admitted and the pinned
	// database is nonempty, the pinned manifest must carry a live
	// instance.key, or the operation is stuck; no operator replay accepts
	// it, since a database is never installed without its key. The L21
	// window (the app mints the key after it starts, and its first
	// manifest with the key has not landed) ends here. The mutation pins
	// what it found, rev 16's behaviour. Only the two-box shape has a
	// files publisher (ruling 11)
	if w.perRegion > 1 && on("files-pin-needs-key") && h.Session != 0 && n > 0 && !manifestHasKey(w.pinnedManifestOf(b.Region, pin)) {
		w.stats["pin: files unevidenced, no instance.key"]++
		w.markStuck(c, fmt.Sprintf("the pinned files carry no instance.key for a nonempty database (TXID %d)", n))
		return nil
	}
	if m := w.pinnedManifestOf(b.Region, pin); m != nil && m.Files != nil && m.Files.Obj != "" && !w.originAuthorized(b.Region, c, m.Files.Obj) {
		// filesPin's referencedOrigins (1b-next rev 12 finding 1, R11-1):
		// the object the pinned manifest names must be current in this
		// region's bucket, under this holder's location, at an origin the
		// pin may cite; otherwise the pin refuses (ErrOrigin) and the
		// worker retries on a later pass
		w.stats["pin: files object refused"]++
		w.logf("DEBUG pin refused obj %s cur=%v", m.Files.Obj, w.buckets[b.Region].current(m.Files.Obj) != nil)
		return nil
	}
	return pin
}

// ---------- files objects (rev 4, Astra R1-8; spec §5.6 "Objects are immutable") ----------

// objectKey is a files object: <holder.location>/files/<generation>/objects/<content>,
// written create-only by the generation that first held the content.
func objectKey(loc, gen string, f *filesDir) string {
	return dataKey(loc, gen, fmt.Sprintf("files/objects/%d-%d", f.Key, f.Env))
}

// isFilesKey: a manifest or an object of the files tree.
func isFilesKey(k string) bool {
	return strings.Contains(k, "files/m-") || strings.Contains(k, "files/objects/")
}

// objectCurrent: f names an object holding its content, in b's region
// (the region binding; the mutation of files-objects-region-bound accepts
// another region's object, as a moved host's carried reference).
func (w *world) objectCurrent(b *box, f *filesDir) bool {
	if f == nil || f.Obj == "" || !strings.HasSuffix(f.Obj, fmt.Sprintf("/%d-%d", f.Key, f.Env)) {
		return false
	}
	return f.ObjRegion == b.Region || !on("files-objects-region-bound")
}

// ensureObject uploads f's content as an object of the current
// generation when f names none for it in this region (a changed file, or
// a host that came from another region), and reports once one is
// acknowledged; an unchanged file keeps naming the object an older
// generation wrote, which is what keeps that generation's objects live.
func (w *world) ensureObject(b *box, h *holder, f *filesDir) bool {
	if f == nil || w.objectCurrent(b, f) {
		return true
	}
	k := objectKey(h.Location, h.Gen, f)
	if !w.ackedPut(b, k, &blob{Kind: "files-object", Inc: h.Inc, Gen: h.Gen, Files: f.clone()}, true) {
		return false
	}
	f.Obj, f.ObjRegion = k, b.Region
	w.stats["files objects written"]++
	return true
}

// originAuthorized is the pin's check of one referenced object: current
// in region r's bucket under the holder's location, and at an authorized
// origin. With the rule (files-origins-referenced) the origins are the
// generations the pinned manifest references, so any such object is
// authorized; the mutation derives them from holder.generation and the
// retained admissions, as rev 11 did, which the trim to eight shortens.
func (w *world) originAuthorized(r int, c *claim, obj string) bool {
	h := c.Holder
	if !strings.HasPrefix(obj, h.Location) || w.buckets[r].current(obj) == nil {
		return false
	}
	if on("files-origins-referenced") {
		return true
	}
	g := genOf(h.Location, obj)
	if g == h.Gen || c.Op.Prepared != "" && g == c.Op.Prepared {
		return true
	}
	for _, e := range h.Admissions {
		if e.Gen == g || e.Predecessor.Gen == g {
			return true
		}
	}
	return false
}

// pinnedManifestOf is the manifest pin names, read by version.
func (w *world) pinnedManifestOf(r int, pin *pinEntry) *blob {
	if pin.ManKey == "" {
		return nil
	}
	if v := w.buckets[r].byID(pin.ManKey, pin.ManVer); v != nil && !v.Marker {
		return v.B
	}
	return nil
}

// manifestHasKey: the manifest carries a live instance.key (InstanceKeyPath,
// liveEntry): its files directory names one.
func manifestHasKey(m *blob) bool { return m != nil && m.Files != nil && m.Files.Key != 0 }

// restoreUpTo restores generation gen of region r from its current
// listing, capped at TXID upto: the newest current snapshot at or below
// upto as the root, then the planner's rule (ltx.IsContiguous, highest
// max, lowest min, higher level) over the current level-0 and level-1
// files whose range ends at or below upto. It is §5.8's build from the
// pinned recovery set: nothing past the pin is restored, and a listing
// that does not reach upto refuses (nil), as a pin short of its release
// would. It reads the bucket and writes nothing.
func (w *world) restoreUpTo(r, inc int, loc, gen string, upto int) *localDB {
	bk := w.buckets[r]
	prefix := dataKey(loc, gen, "")
	db := &localDB{Inc: inc, Loc: loc, Gen: gen, State: map[string]int{}}
	var root *blob
	for _, s := range w.currentLTX(bk, prefix, 9) {
		if s.B.Max <= upto && (root == nil || s.B.Max > root.Max) {
			root = s.B
		}
	}
	cur := 0
	var applied []*blob
	if root != nil {
		apply(db.State, root)
		cur = root.Max
		applied = append(applied, root)
		db.Key = root.SealKey
		if root.Inc != 0 {
			db.Inc = root.Inc
		}
	}
	var cands []*blob
	for _, level := range []int{0, 1} {
		for _, f := range w.currentLTX(bk, prefix, level) {
			if f.B.Max <= upto {
				cands = append(cands, f.B)
			}
		}
	}
	for cur < upto {
		var best *blob
		for _, f := range cands {
			if f.Min <= cur+1 && f.Max > cur && (best == nil || f.Max > best.Max || f.Max == best.Max && (f.Min < best.Min || f.Min == best.Min && f.Level > best.Level)) {
				best = f
			}
		}
		if best == nil {
			return nil // the listing does not reach upto: a gap, or objects still landing
		}
		apply(db.State, best)
		cur = best.Max
		applied = append(applied, best)
		if db.Key == 0 {
			db.Key = best.SealKey
		}
		if best.Inc != 0 {
			db.Inc = best.Inc
		}
	}
	db.Log = txsOf(applied...)
	db.TX = w.lineageTX(applied...)
	db.Pos = cur
	return db
}

// relocFiles is the files directory the local install applies: the
// pinned set's (the pin carries the files and wake state it was taken
// with), else the release's (a drain, a nested relocation), else the
// converted move's or the failover's stand-in for the files publisher
// (moves §14.2.6). Task 4 puts the pinned manifest first.
func (w *world) relocFiles(c *claim) *filesDir {
	if m := w.pinnedManifest(c); m != nil && m.Files != nil {
		return m.Files
	}
	if p := c.Op.Pin; p != nil && p.Files != nil {
		return p.Files
	}
	if r := c.Op.Release; r != nil && r.Files != nil {
		return r.Files
	}
	return c.Op.StopFiles
}

// relocTimers is the wake state the local install replaces timers with.
func (w *world) relocTimers(c *claim) []int {
	if m := w.pinnedManifest(c); m != nil {
		return m.Timers
	}
	if p := c.Op.Pin; p != nil && p.Files != nil {
		return p.Timers
	}
	if r := c.Op.Release; r != nil {
		return r.Timers
	}
	return nil
}

// nameReplacement names next as the operation's target in n, counting a
// replacement when it replaces another box (replaceTargetSlot): the
// replaced box is excluded for this host, and past maxReplacements the
// operation is marked stuck at the current phase instead (rule
// reloc-replacement-cap). Inside the CAS it refuses a box that owes a
// park-local for the host (rule reloc-replace-not-owing, Task 3). It
// reports whether next was named.
func (w *world) nameReplacement(n *claim, next int, phase string) bool {
	if on("reloc-replace-not-owing") {
		if b := w.boxByID(next); b != nil && w.owesLocal(n, b) {
			w.stats["replacement refused: the box owes a park-local"]++
			return false
		}
	}
	old := n.Op.Tgt
	if old >= 0 && old != next {
		if on("reloc-replacement-cap") && n.Op.Replacements >= maxReplacements {
			n.Op.Stuck = fmt.Sprintf("two replacements of the target used; a third was needed at %s", n.Op.Phase)
			w.stats["relocation stuck: replacement cap"]++
			return false
		}
		n.Op.Replacements++
		n.Op.Excluded = addUniq(n.Op.Excluded, old)
		var parts []int
		for _, p := range n.Op.Parts {
			if p != old {
				parts = append(parts, p)
			}
		}
		n.Op.Parts = parts
	}
	n.Op.Tgt = next
	n.Op.Parts = addUniq(n.Op.Parts, next)
	n.Op.Phase = phase
	return true
}

// markStuck is MarkStuck: a journal entry, never a phase; every relocation
// step refuses afterwards until the operator resolves it. It never stops
// an operation between installed and a probe while P is unopened: that
// live target's safe point is its probe.
func (w *world) markStuck(c *claim, why string) {
	id := c.Op.ID
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, id) || n.Op.Kind != "relocating" || n.Op.Stuck != "" {
			return false
		}
		if n.Op.Phase == "installed" && n.Op.Receipt != nil && !preparedOpened(n) {
			return false
		}
		n.Op.Stuck = why
		return true
	}) {
		w.alerts++
		w.stats["relocation stuck"]++
		w.logf("relocation %s stuck: %s", id, why)
	}
}

// relocationLeftovers queues, inside the serving CAS (CompleteRelocation),
// the park-local chore every abandoned installation owes (claim §4: "every
// abandoned local installation receives an exact park-local"): the
// source, when it is alive and its fence did not discharge it, and every
// other box this operation handed op.exec to.
func (w *world) relocationLeftovers(n *claim, op opState) {
	// every prepared generation the operation abandoned unopened
	// (abandoned_prepared) is a dead generation of holder.location: its
	// remote build is retired by a park-generation queued here, beside the
	// park-local chores (1b-next ruling 27, CompleteRelocation; plan ruling
	// 16). An opened P is part of the writer lineage and never listed
	if h := n.Holder; h != nil {
		for _, g := range op.Abandoned {
			if g == h.Gen || w.ancestorGen(h.Location, h.Gen, g) {
				continue
			}
			n.Obl = append(n.Obl, w.scoped(obligation{ID: w.newID("ob"), Op: op.ID, Kind: "park-generation", Region: op.SrcRegion, Box: -1, Prefix: h.Location, Ret: g, Quiet: w.now + horizonH}, w.now, h.Inc))
			w.stats["relocation: abandoned prepared generation parked"]++
		}
	}
	// the pinned manifest of a superseded generation (rev 5): the files
	// live set kept it while the operation ran, so a park-generation of
	// that generation that ran meanwhile left it; once the operation
	// completes it is a dead generation's key, parked here. This is the
	// model's abstraction, not a CompleteRelocation guarantee (rev 6,
	// Astra round 3 finding 2): in 1b-next rev 19 completion queues only
	// the local chores and the abandoned prepared generations; an obsolete
	// files manifest is the at-rest sweep's (claim §6.2a), and the
	// pinner's retained control object is the storage follow-up's (1b-next
	// ruling 21). The model parks it at once so INV-10 can check the
	// residue at completion
	if h := n.Holder; h != nil && op.Pin != nil && op.Pin.ManKey != "" && op.Pin.ManGen != h.Gen {
		ret := op.Pin.ManGen
		if ret == "" {
			ret = "@root"
		}
		queued := false
		for _, o := range n.Obl {
			if o.Kind == "park-generation" && o.Region == op.SrcRegion && o.Ret == ret && o.Prefix == h.Location {
				queued = true
			}
		}
		if !queued {
			n.Obl = append(n.Obl, w.scoped(obligation{ID: w.newID("ob"), Op: op.ID, Kind: "park-generation", Region: op.SrcRegion, Box: -1, Prefix: h.Location, Ret: ret, Quiet: w.now + horizonH}, w.now, h.Inc))
			w.stats["relocation: the pinned manifest's generation parked"]++
		}
	}
	owed := append([]int{op.Src}, op.InstalledOn...)
	for _, id := range owed {
		if id < 0 || id == op.Exec || contains(op.Discharged, id) {
			continue
		}
		if b := w.boxByID(id); b == nil || w.owesLocal(n, b) {
			continue
		}
		n.Obl = append(n.Obl, obligation{ID: w.newID("ob"), Op: op.ID, Kind: "park-local", Box: id, Region: -1, Ret: op.ID})
	}
}

// relocStarting: the barrier (claim D6, requireAcks in ToReleasing). A
// drain leaves starting for releasing once both participants' execution
// leases are current (role() writes them on every read of the claim), and
// the CAS records releasing_since for the release deadline; a failover
// never enters starting in this model (it starts at pinning, Task 3, and
// PinFailover's barrier is checked there before the pin is selected:
// rule reloc-failover-ack-before-pin, rev 5 and 6). The mutation of
// reloc-exec-ack-barrier moves on with no acknowledgement.
func (w *world) relocStarting(b *box, c *claim) {
	op := c.Op
	if on("reloc-exec-ack-barrier") && !acked(c) {
		return
	}
	w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "relocating" || n.Op.Phase != "starting" {
			return false
		}
		n.Op.Phase, n.Op.ReleasingSince = "releasing", w.now
		return true
	})
}

// relocReleasing: the source runs the release sequence and journals it
// (JournalRelease); then a worker's CAS (PinAfterRelease) cites the entry
// and leaves for pinning with op.exec none and the stamp kept (claim
// R6-4), once every participant has acknowledged (rule
// reloc-ack-before-pinning: a target replaced in releasing acknowledges
// before the operation leaves the phase; 1b-next rev 4 N6). A source that
// has not journaled its release releaseDeadline after releasing began is
// stuck (rule reloc-release-deadline, Worker.releaseOverdue): nothing
// moves away from a live box without its stop evidence. The mutation of
// reloc-release-first pins without the entry: the source is never
// stopped.
func (w *world) relocReleasing(b *box, c *claim) {
	op := c.Op
	if op.Release == nil {
		if on("reloc-release-deadline") && w.now > op.ReleasingSince+releaseDeadline {
			w.markStuck(c, fmt.Sprintf("box %d has not released the host %d steps after releasing", op.Exec, releaseDeadline))
			return
		}
		if b.ID != op.Exec {
			if !on("reloc-release-first") {
				w.cas(c, func(n *claim) bool {
					ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "releasing"
					if ok {
						n.Op.Phase, n.Op.Exec = "pinning", -1
					}
					return ok
				})
			}
			return
		}
		w.relocJournalRelease(b, c, false)
		return
	}
	if on("reloc-ack-before-pinning") && !acked(c) {
		w.stats["pin after release: waits for the acknowledgements"]++
		return
	}
	w.cas(c, func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "releasing" && n.Op.Release != nil && n.Op.Release.Box == n.Op.Exec
		if ok {
			n.Op.Phase, n.Op.Exec = "pinning", -1
		}
		return ok
	})
}

// relocPinning: a worker selects the pin against its observation and
// journals it (JournalPin) in a CAS bound to that observation's claim
// version (rule reloc-pin-claim-gen, ruling 16: a pin selected against an
// older claim is abandoned, and the next pass selects afresh); a failover
// pins no earlier than the fence's observed_at plus failoverPinDelay (rule
// reloc-failover-pin-delay, a loss bound). Then it names a target when the
// operation has none (NameTarget: a converted move, a resolved stuck
// operation, a target lost with no spare); then, with every participant's
// lease current (rule reloc-exec-ack-barrier), hands op.exec to the
// target for a new installation (ToInstalling): a new stamp, journaled as
// install_stamp, and a fresh prepared generation id (claim R6-4). The
// target must owe no park-local for the host.
//
// An initial failover (op.PinFailover: 1b-next's starting after
// StartRelocation with a fence) first passes PinFailover's barrier (rev
// 5, Astra round 2 finding 1; narrowed in rev 6, round 3 finding 1): the
// target is named, every participant has acknowledged, and the PinFailover
// CAS clears the flag, all before any pin is selected; so a manifest that
// lands while the target's controller is unavailable is in the pin (rule
// reloc-failover-ack-before-pin). The flag is the model's starting
// (ruling 4). A converted move, a nested failover after a replacement and
// an operator's replay enter pinning directly, as in 1b-next, where
// Worker.pinning selects and journals the pin first and checks the
// acknowledgements only for ToInstalling. The mutation selects the pin
// before the barrier, as rev 4 did.
func (w *world) relocPinning(b *box, c *claim) {
	op := c.Op
	if op.PinFailover && on("reloc-failover-ack-before-pin") {
		if op.Tgt < 0 {
			w.relocNameTarget(c)
			return
		}
		if !acked(c) {
			w.stats["failover pin: waits for the acknowledgements"]++
			return
		}
		if w.cas(c, func(n *claim) bool {
			ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "pinning" && n.Op.PinFailover && n.Op.Tgt == op.Tgt && n.Op.Stuck == ""
			if ok {
				n.Op.PinFailover = false
			}
			return ok
		}) {
			w.logf("relocation %s: PinFailover, box %d acknowledged", op.ID, op.Tgt)
		}
		return
	}
	if op.Tgt < 0 && on("reloc-name-before-pin") {
		// Worker.pinning names a missing target before it selects any pin
		// (1b-next rev 19; rev 7, Astra round 4 finding 1), on every path
		// that enters pinning without one: a converted move, an operator's
		// replay. With no eligible box it waits, and a manifest that lands
		// meanwhile is in the later pin. The mutation pins first, as rev 6
		// did for these paths
		w.relocNameTarget(c)
		return
	}
	if op.Pin == nil {
		if on("reloc-failover-pin-delay") && (op.Release == nil || on("reloc-pin-delay-every-fence")) && w.now < latestObservation(op)+failoverPinDelay {
			return
		}
		pin := w.selectPin(b, c)
		if pin == nil {
			return
		}
		if w.cas(c, func(n *claim) bool {
			ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "pinning" && n.Op.Pin == nil && n.Op.Stuck == ""
			if ok && on("reloc-pin-claim-gen") && n.Ver != pin.ClaimVer {
				w.stats["pin: selected against an older claim, abandoned"]++
				ok = false
			}
			if ok {
				n.Op.Pin = pin
			}
			return ok
		}) {
			w.logf("relocation %s: pinned %q at TXID %d (selected against claim version %d)", op.ID, pin.Gen, pin.TX, pin.ClaimVer)
		}
		return
	}
	if op.Tgt < 0 {
		w.relocNameTarget(c)
		return
	}
	if on("reloc-exec-ack-barrier") && !acked(c) {
		return
	}
	to := op.Tgt
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "relocating" || n.Op.Phase != "pinning" || n.Op.Tgt != to || n.Op.Pin == nil || n.Op.Stuck != "" || contains(n.Lost, to) {
			return false
		}
		if t := w.boxByID(to); t == nil || w.owesLocal(n, t) {
			return false
		}
		w.handExec(n, to, "installing", w.newID("P"))
		n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = "", 0, 0
		return true
	}) {
		w.logf("relocation %s: box %d installs with stamp %d", op.ID, to, w.claim.Holder.Stamp)
	}
}

// relocNameTarget is NameTarget at pinning: choose() in the source region
// over the eligible boxes, the excluded ones left out. A converted move's
// own target (1b-next's move_target_box, in the region the move was
// going to) leaves the participants in the same CAS (rule
// reloc-name-drops-move-target, 1b-next 151261fe): readiness binds to the
// box named here (moves §7.10) and a failover's participant is its
// target alone (§5.8); the move's target never held op.exec on the
// converted operation. A relocation's participants are otherwise all in
// its region, so the model drops every participant outside op.SrcRegion,
// which only a converted move leaves there. The mutation keeps it, and
// the barrier waits for a box that may never acknowledge: a stopped move
// target holds the relocation at pinning with no deadline and no stuck
// (a liveness guard: TestConvertedMoveDropsTheMoveTargetFromTheBarrier).
func (w *world) relocNameTarget(c *claim) {
	op := c.Op
	to := w.chooseBoxExcept(op.SrcRegion, c.Holder.App, relocExcept(op)...)
	if to == nil {
		return // no eligible box: the recovery pass alerts once (moves §7.10)
	}
	dropped := false
	if w.cas(c, func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "pinning" && n.Op.Tgt < 0
		if ok && on("reloc-name-drops-move-target") {
			var parts []int
			for _, p := range n.Op.Parts {
				if x := w.boxByID(p); x == nil || x.Region == n.Op.SrcRegion {
					parts = append(parts, p)
				}
			}
			dropped = len(parts) < len(n.Op.Parts)
			n.Op.Parts = parts
		}
		if ok {
			ok = w.nameReplacement(n, to.ID, "pinning")
		}
		return ok
	}) && dropped {
		w.stats["name target: the move's target left the participants"]++
	}
}

// abandonPrepared is abandonPrepared(c) in hostclaim (1b-next ruling
// 27, rev 18 finding 3): before an unopened P is replaced or dropped, it
// is journaled in abandoned_prepared, so the serving CAS parks it. Every
// path that replaces or drops P calls it: install, the replacement
// table's nested branch, the nested relocation after probe-failed and the
// operator's replay. The mutation of reloc-abandoned-prepared records
// nothing, which reopens finding F3.
func (w *world) abandonPrepared(n *claim) {
	p := n.Op.Prepared
	if p == "" || preparedOpened(n) || hasStr(n.Op.Abandoned, p) || !on("reloc-abandoned-prepared") {
		return
	}
	n.Op.Abandoned = append(n.Op.Abandoned, p)
}

// handExec is install() in hostclaim: op.exec to box at phase with
// prepared generation P, a new stamp (this CAS's), journaled as
// install_stamp, no receipt, the probe attempts reset for the new
// executor (BeginProbeAttempt counts per box), and the box noted as owing
// a park-local should it be abandoned.
func (w *world) handExec(n *claim, box int, phase, prepared string) {
	if prepared != n.Op.Prepared {
		w.abandonPrepared(n) // install replaces an unopened P (1b-next ruling 27)
	}
	n.Op.Exec, n.Op.Phase, n.Op.Prepared = box, phase, prepared
	n.Holder.Stamp = w.issueStamp(n, box)
	n.Op.InstallStamp = n.Holder.Stamp
	if on("reloc-install-receipt") {
		n.Op.Receipt = nil
	}
	n.Op.Attempts = 0
	n.Op.ProbeStop, n.Op.ProbePassedAt = nil, 0
	n.Op.InstalledOn = addUniq(n.Op.InstalledOn, box)
	n.Boxes = addUniq(n.Boxes, box)
}

// relocInstalling: the target builds P, the remote prepared generation,
// exactly as a landing builds its generation (moves §7.6, L16): the
// snapshot [1, N] of the pinned generation, create-only and acknowledged,
// then the level-0 anchor [N, N], acknowledged, then the installed CAS
// (ToInstalled, R7-4). A pin at N = 0 builds nothing: the probe is an
// empty bootstrap (claim P6-D3). Only op.exec installs (rule
// reloc-exec-only-installs, HA2: being in the region is never enough).
func (w *world) relocInstalling(b *box, c *claim) {
	op, h := c.Op, c.Holder
	if on("reloc-exec-only-installs") && op.Exec != b.ID {
		return
	}
	loc, g := h.Location, op.Prepared
	bk := w.buckets[b.Region]
	installed := func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "installing" && n.Op.Exec == op.Exec && n.Op.Prepared == g
		if ok && on("reloc-exec-ack-barrier") && !acked(n) {
			ok = false // ToInstalled's requireAcks (rev 4, Astra R1-4)
		}
		if ok {
			n.Op.Phase = "installed"
		}
		return ok
	}
	if !w.buildPreparedFiles(b, c) {
		return // Task 4: P's prepared files manifest first
	}
	if op.Pin.TX == 0 {
		w.cas(c, installed)
		return
	}
	if op.BaseKey == "" {
		db := w.restoreUpTo(b.Region, h.Inc, loc, op.Pin.Gen, op.Pin.TX)
		if db == nil || db.Pos != op.Pin.TX {
			w.stats["build: the pinned set cannot be restored yet"]++
			return
		}
		key := snapKey(loc, g, op.Pin.TX)
		if !w.ackedPut(b, key, snapshotOf(db, h.Inc), true) {
			return
		}
		v := bk.current(key)
		if v == nil {
			return
		}
		w.cas(c, func(n *claim) bool {
			ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "installing" && n.Op.Exec == op.Exec && n.Op.Prepared == g && n.Op.BaseKey == ""
			if ok {
				n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = key, v.ID, op.Pin.TX
			}
			return ok
		})
		return
	}
	if !w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) {
		return
	}
	if w.cas(c, installed) {
		w.logf("relocation %s: P %q built on box %d at TXID %d", op.ID, g, b.ID, op.BaseTX)
	}
}

// buildPreparedFiles writes P's prepared files manifest before anything
// of the database (1b-next Task 8b; CopyPinnedManifest; claim rev 6.30
// R18-1, moves rev 7.22 §7.6): create-only at sequence 0, bound to
// {location, P, op id, install_stamp}, with the pinned manifest's files
// and wake state; the installed CAS waits for its acknowledgement
// (ToInstalled requires files_seq 0 and the pin's source sequence as
// provenance). The first admission consumes it as the session
// publisher's baseline, and the publisher starts at sequence 1 (rule
// files-prepared-seq-zero); the mutation keeps the source manifest's
// sequence, as rev 14 did, so the session's first manifests sort below
// it. In the one-box shape there is no publisher (ruling 11) and nothing
// to write.
func (w *world) buildPreparedFiles(b *box, c *claim) bool {
	if w.perRegion <= 1 {
		return true
	}
	op, h := c.Op, c.Holder
	seq := 0
	if !on("files-prepared-seq-zero") && op.Pin != nil {
		seq = op.Pin.ManSeq
	}
	bl := &blob{Kind: "files", Inc: h.Inc, Gen: op.Prepared, Seq: seq, Files: w.relocFiles(c).clone(), Timers: append([]int(nil), w.relocTimers(c)...), Text: fmt.Sprintf("%s/%d", op.ID, op.InstallStamp), Base: w.issued}
	return w.ackedPut(b, manifestKey(h.Location, op.Prepared, seq), bl, true)
}

// localInstall advances the executor's local install by one step (§5.8,
// B1-2; 1b-next Task 8c's Install) and reports whether it is whole: park
// what an earlier stay left as retire-artifacts of its stamp (claim §6.4),
// extract the files, replace the timers with the pinned wake state (the
// files Done marker last), render and materialize with current inputs,
// and write the box's install journal (its receipt) under install_stamp.
// EnsureLocal reuses it to restore a retired install under the same
// stamp.
func (w *world) localInstall(b *box, c *claim) bool {
	op, h := c.Op, c.Holder
	g := op.Prepared
	if b.Files == nil || b.Files.Gen != g {
		for _, st := range []int{dbStamp(b.DB), filesStamp(b.Files)} {
			if st != 0 && st != b.stampOf(h.Inc) {
				w.retireArtifacts(b, h.Inc, st)
			}
		}
		if b.DB != nil {
			b.DB = nil // never reused (claim R6); the probe's admission restores
			b.Parked++
		}
		src := w.relocFiles(c)
		if src == nil {
			w.nextID++
			src = &filesDir{Inc: h.Inc, Key: w.nextID, Env: w.nextID}
		}
		f := src.clone()
		f.Inc, f.Gen, f.Done, f.Stamp = h.Inc, g, false, b.stampOf(h.Inc)
		if !on("markers-last") {
			f.Done = true
		}
		b.Files = f
		b.pendingTimers = append([]int(nil), w.relocTimers(c)...)
		return false
	}
	if !b.Files.Done {
		b.Timers = append([]int(nil), b.pendingTimers...) // ReplaceTimers: the pinned wake state
		b.Attempt, b.History = 0, nil
		b.Files.Done = true
		b.installEpoch = b.Epoch
		return false
	}
	if b.HasKey {
		b.Config, b.CfgVer = h.Inc, w.cfgVer[h.App]
	}
	b.Live, b.LiveInc = h.App, h.Inc
	b.receipt = &receiptEntry{Op: op.ID, Box: b.ID, Stamp: op.InstallStamp}
	return true
}

// relocInstalled: the executor's local install, then its receipt by CAS
// (RecordLocalInstall), bound to this box and install_stamp; probing
// needs that exact receipt (ToProbing; rule reloc-install-receipt), and
// the install is verified, and restored if a lost lease's scoped cleanup
// retired it, before the probing CAS (EnsureLocal; rule
// reloc-ensure-local). The mutation of reloc-install-receipt lets a
// replacement inherit its predecessor's receipt and skip its install.
func (w *world) relocInstalled(b *box, c *claim) {
	op := c.Op
	if on("reloc-exec-only-installs") && op.Exec != b.ID {
		return
	}
	probing := func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "installed" && n.Op.Exec == op.Exec && n.Op.Receipt != nil
		if ok && on("reloc-exec-ack-barrier") && !acked(n) {
			ok = false // ToProbing's requireAcks (rev 4, Astra R1-4)
		}
		if ok {
			n.Op.Phase, n.Op.ProbingSince = "probing", w.now
		}
		return ok
	}
	if r := op.Receipt; r != nil {
		mine := r.Box == b.ID && r.Stamp == op.InstallStamp
		if !mine && !on("reloc-install-receipt") {
			w.cas(c, probing) // the mutation: any receipt will do, the install is skipped
			return
		}
		if mine {
			if w.ensureLocal(b, c) {
				w.cas(c, probing)
			}
			return
		}
	}
	if op.Exec != b.ID || !w.localInstall(b, c) {
		return
	}
	w.cas(c, func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "installed" && n.Op.Exec == b.ID && n.Op.InstallStamp == op.InstallStamp && !preparedOpened(n)
		if ok {
			n.Op.Receipt = &receiptEntry{Op: op.ID, Box: b.ID, Stamp: op.InstallStamp}
		}
		return ok
	})
}

// ensureLocal is EnsureLocal (1b-next rev 15 finding 2): before P's first
// admission the executor's local install must be whole, so it is
// verified before the probing CAS and before every attempt, and a
// retired one (the scoped cleanup of a lost controller lease retires an
// unopened P's local install, role()) is restored under the same
// install_stamp. It never touches an opened P. It reports whether the
// install is whole now. The mutation goes on without verifying, so P's
// first admission can meet a missing install.
func (w *world) ensureLocal(b *box, c *claim) bool {
	op := c.Op
	if preparedOpened(c) || !on("reloc-ensure-local") {
		return true
	}
	if b.Files != nil && b.Files.Gen == op.Prepared && b.Files.Done && b.receipt != nil && b.receipt.Op == op.ID && b.receipt.Stamp == op.InstallStamp {
		return true
	}
	if w.localInstall(b, c) {
		w.stats["local install restored before the first admission"]++
		w.logf("relocation %s: box %d restored its retired local install under stamp %d", op.ID, b.ID, op.InstallStamp)
		return true
	}
	return false
}

// verifyReceipt is VerifyReceipt (1b-next rev 16 finding 1): once P is
// opened the local install is never remade, and before every attempt the
// box's durable receipt (its install journal) must still be the one the
// claim journaled; a missing or different one marks the operation stuck
// and nothing is admitted. Files, env and timers the admitted writers
// changed are never compared. The mutation goes on to the next attempt.
func (w *world) verifyReceipt(b *box, c *claim) bool {
	op := c.Op
	if r := b.receipt; r != nil && r.Op == op.ID && r.Box == b.ID && r.Stamp == op.InstallStamp {
		return true
	}
	if !on("reloc-verify-receipt") {
		w.stats["receipt missing after P opened, attempt made anyway (mutation)"]++
		return true
	}
	w.markStuck(c, fmt.Sprintf("box %d's local install receipt is missing after P was admitted", b.ID))
	return false
}

// preparedOpened is PreparedOpened: an admission has named P (B2-5). The
// admission that opens P journals prepared_opened in the same CAS
// (admitCommit), so the fact outlives holder.admissions' trim to eight
// (1b-next rev 12 S1; rule reloc-prepared-opened-journal); the mutation
// reads the retained admissions alone, as rev 11 did.
func preparedOpened(c *claim) bool {
	if c.Op.Prepared == "" || c.Holder == nil {
		return false
	}
	if on("reloc-prepared-opened-journal") && c.Op.PreparedOpened == c.Op.Prepared {
		return true
	}
	for _, e := range c.Holder.Admissions {
		if e.Gen == c.Op.Prepared {
			return true
		}
	}
	return false
}

// relocProbing: each attempt is counted in the claim before it is made
// (BeginProbeAttempt, ruling 19; rule reloc-probe-attempts: a crash, a
// lost response or a restart never buys a fourth), then one cold
// AdmitWriter by op.exec into P (the first) or a fresh generation
// restored from the previous attempt's (claim §6.4a; spawn and admitWriter
// do both), then the probe. The fourth is refused and the target enters
// probe-failed, as it does past the probe deadline. Before P's first
// admission the local install is verified and restored (ensureLocal);
// after it, the receipt alone is verified (verifyReceipt) and the install
// is never remade (rule reloc-no-reinstall; its mutation reapplies the
// pinned wake state on every agent restart, as rev 4.2's install marker
// did, over the timers the admitted session registered).
func (w *world) relocProbing(b *box, c *claim) {
	op, h := c.Op, c.Holder
	failed := func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == "probing" && n.Op.Exec == op.Exec
		if ok {
			n.Op.Phase, n.Op.FailedAt, n.Op.Release = "probe-failed", w.now, nil
		}
		return ok
	}
	if op.Exec != b.ID {
		if w.now > op.ProbingSince+probeDeadline && op.Exec >= 0 {
			w.stats["probe-failed: deadline"]++
			w.cas(c, failed) // the worker's: ToProbeFailed past the probe deadline
		}
		return
	}
	if b.Row == nil || b.Row.Status == "ready" {
		w.holdRow(b, "relocating", op.ID, c)
	}
	opened := preparedOpened(c)
	if !opened {
		if !w.ensureLocal(b, c) {
			return
		}
	} else if !w.verifyReceipt(b, c) {
		return
	}
	if !on("reloc-no-reinstall") && opened && b.installEpoch != b.Epoch {
		b.Timers = append([]int(nil), w.relocTimers(c)...) // the mutation: the install marker re-applied after a restart
		b.installEpoch = b.Epoch
		w.stats["local install remade after admission (mutation)"]++
	}
	if !w.admits(b, c, op.ID) {
		return
	}
	if b.HasKey {
		b.Config, b.CfgVer, b.Live, b.LiveInc = h.Inc, w.cfgVer[h.App], h.App, h.Inc
	}
	probed := func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Phase == "probing" && n.Op.Exec == b.ID
		if ok {
			n.Op.Phase, n.Op.ProbePassedAt = "probed", w.now // RecordProbe's At (ruling 67's bound)
		}
		return ok
	}
	if hasMain(b) {
		w.probeAccepted(b, c)
		w.cas(c, probed)
		return
	}
	if w.now > op.ProbingSince+probeDeadline {
		w.stats["probe-failed: deadline"]++
		w.cas(c, failed)
		return
	}
	tag := fmt.Sprintf("%s/%d", op.ID, op.Attempts)
	if b.open == nil {
		// a new attempt: counted before it is made. attemptTag is the
		// attempt this agent counted (tag) or made (tag+"+"); one made and
		// ended, or none counted, is counted now
		if on("reloc-probe-attempts") && b.attemptTag != tag {
			if preparedOpened(c) && w.c11Fallback() {
				// Preflight's ErrLaterAttemptUnlaunchable (ruling 58, until
				// claim ask C11): root pins every relocating launch to
				// install_stamp, so a cold attempt after P's opening is not
				// made; probe-failed now, before any attempt is counted
				w.stats["probe-failed: a later attempt root cannot launch (C11 open)"]++
				w.cas(c, failed)
				return
			}
			if op.Attempts >= probeTries {
				w.stats["probe-failed: attempts"]++
				w.cas(c, failed)
				return
			}
			if w.cas(c, func(n *claim) bool {
				ok := sameOp(n.Op.ID, op.ID) && n.Op.Phase == "probing" && n.Op.Exec == b.ID && n.Op.Attempts == op.Attempts
				if ok {
					n.Op.Attempts++
				}
				return ok
			}) {
				b.attemptTag = fmt.Sprintf("%s/%d", op.ID, op.Attempts+1)
			}
			return
		}
		if !on("reloc-probe-attempts") && op.ProbeTries >= probeTries {
			w.stats["probe-failed: attempts"]++
			w.cas(c, failed) // the mutation: failures counted after the fact, as rev 1 did
			return
		}
	}
	if b.open == nil {
		b.attemptTag = tag + "+" // made now
	}
	if w.spawn(b, c, op.Prepared, true) {
		w.probeAccepted(b, c)
		w.cas(c, probed)
	}
}

// relocProbeFailed: the live target runs its own release sequence (§5.8
// probe-failed; JournalRelease), then a worker starts the nested
// relocation from pinning (PinAfterRelease with a next target), pinned to
// at least the target's final TXID (the release entry, which
// relocPinning reads), the failed target excluded and counted as a
// replacement. Exhaustion is checked before a replacement is looked for
// (1b-next R1-16): with the replacements used the operation is stuck. A
// target that cannot finish its release within releaseDeadline goes to
// stuck (ruling 14): nothing moves away from a live box without its stop
// evidence. With no eligible box for releaseDeadline the operation is
// stuck too (model ruling 14: 1b-next waits for a spare with no bound,
// and in a two-box region the source is never a spare).
func (w *world) relocProbeFailed(b *box, c *claim) {
	op, h := c.Op, c.Holder
	if op.Release == nil || op.Release.Box != op.Exec {
		if on("reloc-release-deadline") && w.now > op.FailedAt+releaseDeadline {
			w.markStuck(c, fmt.Sprintf("box %d cannot finish its release sequence", op.Exec))
			return
		}
		if b.ID != op.Exec {
			return
		}
		w.relocJournalRelease(b, c, false)
		return
	}
	if on("reloc-replacement-cap") && op.Replacements >= maxReplacements {
		w.markStuck(c, fmt.Sprintf("two replacements of the target used; a third was needed at %s", op.Phase))
		return
	}
	next := w.chooseBoxExcept(op.SrcRegion, h.App, relocExcept(op, op.Exec)...)
	if next == nil {
		// 1b-next ruling 26 (rev 17, finding F1; rule reloc-no-spare-bound):
		// with no spare ReleaseDeadline after probe_failed_since, stuck;
		// never the source, which the operator's resolve alone may name.
		// The mutation waits for a spare for ever, rev 16's behaviour
		if on("reloc-no-spare-bound") && w.now > op.FailedAt+releaseDeadline {
			w.markStuck(c, "no eligible box for the nested relocation")
		}
		return
	}
	w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "relocating" || n.Op.Phase != "probe-failed" || n.Op.Release == nil || n.Op.Release.Box != n.Op.Exec {
			return false
		}
		if !w.nameReplacement(n, next.ID, "pinning") {
			return n.Op.Stuck != "" // stuck: the CAS records it
		}
		w.abandonPrepared(n)
		n.Op.Exec = -1
		n.Op.Pin, n.Op.Prepared, n.Op.Receipt = nil, "", nil
		n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = "", 0, 0
		n.Op.ProbeTries, n.Op.Attempts = 0, 0
		w.stats["nested relocation after probe-failed"]++
		return true
	})
}

// relocSettle is the end of a relocation as 1b-next builds it (L18 joint
// round finding 2; regionha_completion_test.go): at probed (1b-next's
// probing with the probe journaled) op.exec stops the probe's session with
// root's receipt (relocStopProbe, ruling 49), bounded by ruling 67 (no
// stop ReleaseDeadline after the probe passed is stuck for an operator);
// then any worker of the region makes the serving CAS (relocComplete,
// CompleteRelocation), which records the completion. The row stays held:
// the completion pass releases it after the discharge, and the first
// serving wake is a separate admission (N6). Rev 7.20's settle window and
// open are gone: a probe that dies after it passed is stopped all the same.
func (w *world) relocSettle(b *box, c *claim) {
	op := c.Op
	if op.Phase != "probed" {
		return
	}
	if op.ProbeStop == nil && on("reloc-probe-stop") {
		if w.now > op.ProbePassedAt+releaseDeadline {
			w.markStuck(c, fmt.Sprintf("box %d passed its probe but has no root stop receipt for session %d", op.Exec, c.Holder.Session))
			return
		}
		if b.ID == op.Exec {
			w.relocStopProbe(b, c)
		}
		return
	}
	w.relocComplete(c)
}

// scopedCleanup is the relocation controller's scoped cleanup when its
// lease was lost (1b-next rev 14 finding 1, Task 7's live): an unopened
// P's local install is retired and its receipt (the install journal)
// kept, so EnsureLocal restores it before the first admission. role()
// calls it when the executor's execution lease expired before it renewed
// it (a freeze, a pause, a partition longer than the TTL).
func (w *world) scopedCleanup(b *box, c *claim) {
	if c == nil || c.Holder == nil || c.Op.Kind != "relocating" || c.Op.Exec != b.ID || preparedOpened(c) {
		return
	}
	if f := b.Files; f != nil && f.Gen == c.Op.Prepared && f.Gen != "" {
		w.retireArtifacts(b, c.Holder.Inc, f.Stamp)
		w.stats["relocation controller lost: unopened local install retired"]++
		w.logf("box %d: relocation %s's controller lease was lost; the scoped cleanup retired its unopened local install", b.ID, c.Op.ID)
	}
}

// ---------- termination, enrollment and the fence (region HA §5.9; H1, H10, H11, H16, H17, H19) ----------

// terminate is the cloud terminating a box (H1): nothing on it runs
// again, it never returns (a stopped box may; heal skips a terminated
// one), and the requests it issued still complete at their own time,
// which is the late-PUT-after-termination case. The two-box step draws
// it; it never leaves the world without an enrolled live box, since H17's
// cross-region confirmer must exist for a replacement to enroll.
func (w *world) terminate(b *box) {
	if b.Destroyed {
		return
	}
	enrolled := 0
	for _, x := range w.liveBoxes() {
		if x.Enrolled && x != b {
			enrolled++
		}
	}
	if enrolled == 0 {
		return
	}
	b.Destroyed, b.Terminated, b.TerminatedAt = true, true, w.now
	b.Procs, b.plans = nil, nil
	w.stats["terminated"]++
	w.logf("box %d terminated by the cloud (region %d)", b.ID, b.Region)
}

// launchReplacements: the group keeps the region at its minimum (H5): a
// replacement launches replaceDelay steps after the termination that left
// the region short, unenrolled (H16), with its volume flag drawn (two-box
// sweeps only: a witness world's quiet source draws nothing).
func (w *world) launchReplacements() {
	for r := 0; r < nRegions; r++ {
		if len(w.regionBoxes(r)) >= w.regionMin(r) {
			continue
		}
		last := 0
		for _, b := range w.boxes {
			if b.Region == r && b.Terminated && b.TerminatedAt > last {
				last = b.TerminatedAt
			}
		}
		if last == 0 || w.now < last+replaceDelay {
			continue
		}
		w.nextID++
		nb := &box{ID: 100 + w.nextID, Region: r, HasKey: true, EnrollAt: w.now, DeleteOnTermination: w.rng.Intn(8) != 0}
		nb.forget()
		w.boxes = append(w.boxes, nb)
		w.stats["replacements launched"]++
		w.logf("box %d launched in region %d, unenrolled", nb.ID, r)
	}
}

// enrollStep is automatic enrollment (H16, L19): a launched box's boxid
// evidence verifies and an enrolled box confirms its group membership
// enrollDelay steps after launch. Any enrolled box of any region may
// confirm (H17: the first box of an emptied region is confirmed from
// another region); with none, nothing enrolls.
func (w *world) enrollStep() {
	confirmer := false
	for _, b := range w.liveBoxes() {
		if b.Enrolled {
			confirmer = true
		}
	}
	if !confirmer {
		return
	}
	for _, b := range w.liveBoxes() {
		if !b.Enrolled && w.now >= b.EnrollAt+enrollDelay {
			b.Enrolled = true
			w.stats["boxes enrolled automatically"]++
			w.logf("box %d enrolled (region %d)", b.ID, b.Region)
		}
	}
}

// twoBoxStep is the two-box shape's extra step (ruling 1): replacements
// launch and enroll; during the chaos a box is terminated now and then,
// and a drain is asked for now and then.
func (w *world) twoBoxStep() {
	w.launchReplacements()
	w.enrollStep()
	for _, b := range w.liveBoxes() {
		w.renewController(b)
		w.publishFiles(b)
	}
	if !w.calm && w.fail["terminate"] && w.rng.Intn(200) == 0 {
		bs := w.liveBoxes()
		if b := bs[w.rng.Intn(len(bs))]; w.regionHAOwnsLoss(b) {
			w.terminate(b)
		}
	}
	if !w.calm && w.rng.Intn(60) == 0 {
		w.userRelocate()
	}
	if !w.calm && w.rng.Intn(400) == 0 {
		bs := w.liveBoxes()
		w.updateExecutable(bs[w.rng.Intn(len(bs))]) // claim §5.10: refused while a controller record exists (rev 4)
	}
}

// regionHAOwnsLoss: the loss of b is region HA's to recover (model ruling
// 15): the claim holds the host and names b, if at all, only as a serving
// executor, a relocation's participant or a chore's owner; a box holding
// a legacy host's state before its adoption is never terminated. A participant of a move,
// a landing, a rollback, a revert, a push, a create or a delete is
// recovered from a lost box by moves' and the claim's declare-lost paths
// (L14), which the model walks for a destroyed box that is stopped and
// may return; the sweep never terminates one.
func (w *world) regionHAOwnsLoss(b *box) bool {
	c := w.claim
	if c == nil || c.Holder == nil {
		return b.Row == nil && b.DB == nil && b.Files == nil // a legacy host not yet adopted is not region HA's
	}
	switch c.Op.Kind {
	case "serving", "relocating":
		if c.Op.Exec == b.ID && w.perRegion > 1 {
			// the L21 window (plan ruling 17): an executor whose newest
			// landed manifest carries no instance.key (none landed yet, or
			// only the pre-start copy) is unrecoverable by any relocation:
			// its failover is stuck by design (1b-next ruling 25) and never
			// converges, so the sweep leaves it alone; the witness for
			// files-pin-needs-key and TestFailoverBeforeTheFirstManifest
			// show it fail safe
			_, _, _, _, m := w.newestManifest(b.Region, c.Holder.Location, c.Holder.Gen)
			return manifestHasKey(m)
		}
		return true
	}
	return !(c.Op.Exec == b.ID || contains(c.Op.Parts, b.ID) || c.Op.Src == b.ID || c.Op.Tgt == b.ID || c.Op.NewBox == b.ID)
}

// fenceStep is the regional worker's termination fence (§5.9; claim L2
// to L5; Worker.Step's "evidence first"): for a terminated, enrolled box
// of its own region that the claim still names, it writes the fence in
// one CAS (fence). A stopped box is never fenced (rule
// fence-terminated-only: "stopped and stopping do not count, because a
// stopped box can start again with its disk"); the mutation fences any
// destroyed box. An unenrolled box has no roster entry to bind the fence
// to (HA7) and is never fenced. The source's fence is consumed before the
// target's, as Worker.Step orders them.
func (w *world) fenceStep(b *box, c *claim) (changed, pending bool) {
	if c == nil || !b.Enrolled || contains(c.Lost, b.ID) || w.haOff {
		return false, false
	}
	var cands []*box
	for _, x := range w.boxes {
		if x.Region != b.Region || x == b || !x.Destroyed || !x.Enrolled || contains(c.Lost, x.ID) {
			continue
		}
		if !x.Terminated && on("fence-terminated-only") {
			continue
		}
		if !w.claimNames(c, x) {
			continue
		}
		cands = append(cands, x)
	}
	if len(cands) == 0 {
		return false, false
	}
	// Worker.Step's order (rev 8): the source's fence, then the target's,
	// then any other box the claim names (a chore's owner)
	rank := func(x *box) int {
		switch {
		case c.Op.Kind == "relocating" && x.ID == c.Op.Src:
			return 0
		case c.Op.Kind == "relocating" && x.ID == c.Op.Tgt:
			return 1
		}
		return 2
	}
	sort.SliceStable(cands, func(i, j int) bool { return rank(cands[i]) < rank(cands[j]) })
	if x := cands[0]; x.FenceObservedAt == 0 {
		// hostfence.TerminationWorker: the first worker to observe the
		// terminated state writes the box-fences record with that
		// observed_at; every deadline and delay runs from it, never from
		// the termination itself, which no worker saw (rev 4, Astra R1-5)
		x.FenceObservedAt = w.now
		w.logf("box %d observed box %d terminated: fence record, observed_at %d", b.ID, x.ID, w.now)
	}
	before := w.claim.Ver
	w.fence(b, c, cands[0])
	if w.claim.Ver != before {
		return true, false
	}
	// nothing written: for a relocation's source or target the fence is
	// pending (rule reloc-fence-pending-blocks)
	return false, c.Op.Kind == "relocating" && rank(cands[0]) < 2
}

// claimNames: the claim gives x a role, so a fence changes it.
func (w *world) claimNames(c *claim, x *box) bool {
	if c.Op.Exec == x.ID || contains(c.Op.Parts, x.ID) || c.Op.Kind == "creating" && c.Op.NewBox == x.ID || c.Op.Kind == "moving" && c.Op.Src == x.ID {
		return true
	}
	if c.Op.Kind == "relocating" && (c.Op.Src == x.ID || c.Op.Tgt == x.ID) {
		return true
	}
	for _, o := range c.Obl {
		if o.Owner == boxOwner(x.ID) {
			return true
		}
	}
	return w.owesLocal(c, x)
}

// boxOwner is an obligation's owner box:<b> (claim §2): the legacy form a
// regional chore may carry; "" is region:<r>, any eligible worker of the
// region (P6-D5).
func boxOwner(id int) string { return fmt.Sprintf("box:%d", id) }

// fence is the fence CAS for x, observed terminated by worker b. Moves'
// cases keep their owners: a pre-flip move's source converts (moves
// §7.10, the existing CAS); a move's target before the flip, a landing's
// or a rollback's target are moves' declare-lost path, which this model
// gives the operator (L14); a create's box is cancelled by the recovery
// pass (Task 1). Everything else is one CAS here: the box goes on lost
// with the fence reference and observed_at, its chores are discharged if
// its volume evidence qualifies, its regional work passes to its region,
// and the operation's rule applies (StartRelocation's failover,
// FenceSource, ReplaceTarget, FenceExhaustedTarget). A target with no
// spare to name is fenced on a later pass (Worker.next: "no eligible box
// yet").
func (w *world) fence(b *box, c *claim, x *box) {
	op := c.Op
	switch {
	case op.Kind == "moving" && op.Src == x.ID:
		if on("fence-converts-moving") && !op.Committed && !(w.rec != nil && w.rec.MoveID == op.ID && w.rec.MoveState == "landing") {
			w.convertToRelocating(c, x)
		}
		return
	case op.Kind == "landing" && op.Committed && op.Tgt == x.ID && on("fence-converts-committed-landing"):
		// finding F5 (rev 4): the record committed the landing, so no
		// rollback is possible, and its dead target can never finish it.
		// The fence makes it a failover in the target's region, as a
		// serving executor's fence does (plan ruling 20); the mutation
		// leaves it to the operator's path, which has nothing to do. The
		// failover keeps the landing's op id, as hostclaim
		// ConvertCommittedLanding does (and as convertToRelocating does
		// before the flip): the move's admission registration is the move
		// id, and judgeRegistration drains it 90 seconds in unless the
		// claim's operation carries that id (hostops RegisterMove)
		id := op.ID
		if w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, id) || n.Op.Kind != "landing" || !n.Op.Committed || contains(n.Lost, x.ID) {
				return false
			}
			w.startFailover(n, x, nil, id) // the model's conversion (plan ruling 20): its target is named at pinning
			w.fenceBox(n, x)
			return true
		}) {
			w.stats["fences written"]++
			w.stats["relocation started: failover of a committed landing"]++
			w.logf("box %d fenced box %d, the committed landing %s's dead target: failover in region %d", b.ID, x.ID, id, x.Region)
			w.declareTimers(x, w.claim.Holder)
		}
		return
	case (op.Kind == "landing" || op.Kind == "restoring") && op.Tgt == x.ID:
		// claim D2's automatic termination for a move's target (rev 4,
		// claim §11 "automatic target loss racing the record commit CAS"):
		// the same record-first declared loss the operator makes
		w.declareTargetLost(c)
		return
	case op.Kind == "moving" && op.Tgt == x.ID, op.Kind == "creating":
		return
	case op.Kind == "relocating" && op.Stuck != "":
		return // Worker.Step consumes no fence of a stuck operation: the operator's resolve does
	}
	id := op.ID
	var to *box
	if c.Holder != nil && op.Kind == "serving" && op.Exec == x.ID {
		var ok bool
		if to, ok = w.failoverStart(c, x); !ok {
			return // no start yet: the claim, and the fence, wait for a later pass
		}
	}
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, id) || n.Op.Kind != op.Kind || contains(n.Lost, x.ID) {
			return false
		}
		switch {
		case n.Holder != nil && n.Op.Kind == "serving" && n.Op.Exec == x.ID:
			if !w.failoverStartAllowed(n, x, to) {
				w.stats["failover start: refused by the start CAS"]++
				return false
			}
			w.startFailover(n, x, to, w.newID("rl"))
		case n.Op.Kind == "relocating" && n.Op.Stuck != "":
			return false
		case n.Op.Kind == "relocating" && n.Op.Src == x.ID:
			w.fenceSource(n, x)
			if n.Op.Tgt == x.ID && !w.replaceTarget(n, x) {
				// an operator's replay made the source eligible and named it
				// the target (model ruling 13): its fence is a target loss
				// too, or pinning waits on a lost target for ever
				return false // no spare yet
			}
		case n.Op.Kind == "relocating" && n.Op.Tgt == x.ID:
			if !w.replaceTarget(n, x) {
				return false // no spare yet
			}
		}
		w.fenceBox(n, x)
		return true
	}) {
		w.stats["fences written"]++
		w.logf("box %d fenced box %d (terminated at %d) under %s/%s", b.ID, x.ID, x.TerminatedAt, w.claim.Op.Kind, w.claim.Op.ID)
		w.declareTimers(x, w.claim.Holder)
	}
}

// fenceBox is plan 1b-next's fenceBox: lost with the fence reference
// (FenceBox, FenceAt); the box's regional work (a park-generation or
// park-replica owned by box:<x>) passed to region:<r> keeping its
// identity (rule fence-reassigns-regional, 1b-next rev 6 finding 2, as
// DeclareLost does for a declared loss); and the box's park-local chores
// discharged only when its volume evidence qualifies (L3, H19: every
// attached volume DeleteOnTermination while it ran); otherwise they stay
// open for the operator path, and nothing about recovery waits on them.
// The mutation of fence-discharge-qualified discharges regardless (INV-26
// is the oracle); the mutation of fence-reassigns-regional leaves the
// regional work with the dead box, where nothing runs it.
func (w *world) fenceBox(n *claim, x *box) {
	n.Lost = addUniq(n.Lost, x.ID)
	at := x.FenceObservedAt
	if at == 0 {
		at = w.now
	}
	n.Op.FenceBox, n.Op.FenceAt = x.ID, at
	n.Op.Observed = append(n.Op.Observed, at) // fence.<box>.observed_at, retained for every later pin (JournalPin)
	var parts []int
	for _, p := range n.Op.Parts {
		if p != x.ID {
			parts = append(parts, p)
		}
	}
	n.Op.Parts = parts
	if on("fence-reassigns-regional") {
		for i := range n.Obl {
			if o := &n.Obl[i]; o.Kind != "park-local" && o.Owner == boxOwner(x.ID) {
				o.Owner = "" // region:<r>, the chore's id, operation, retirement and scope kept
				w.stats["fence: regional work passed to the region"]++
			}
		}
	}
	// a fence exists only for a terminated box (TerminationWorker verifies
	// terminated), and only a terminated box's volume evidence says
	// anything: a stopped box keeps its disk, so its park-local chores
	// stay owed whatever DeleteOnTermination says (claim §7.3 D5, H19;
	// rev 10, Astra round 7). The operator's declared loss of a stopped
	// box reaches this function too, and discharges nothing
	qualified := x.Terminated && x.DeleteOnTermination
	if !on("fence-discharge-qualified") {
		qualified = true
	}
	if !qualified {
		w.stats["fence: park-local chores left open (volume evidence)"]++
		return
	}
	var keep []obligation
	for _, o := range n.Obl {
		if o.Kind == "park-local" && o.Box == x.ID {
			if !x.Terminated || !x.DeleteOnTermination {
				w.violate("INV-26: box %d's park-local chore %s discharged without qualifying volume evidence (§5.9, H19; terminated %v)", x.ID, o.ID, x.Terminated)
			}
			w.stats["fence: park-local chores discharged"]++
			continue
		}
		keep = append(keep, o)
	}
	n.Obl = keep
	if !x.Terminated || !x.DeleteOnTermination {
		// INV-26 (rev 10): a discharge recorded for the operation's
		// leftovers is a discharge too, though no chore exists yet
		w.violate("INV-26: box %d's local retirement discharged without qualifying volume evidence (§5.9, H19, claim §7.3 D5; terminated %v)", x.ID, x.Terminated)
	}
	n.Op.Discharged = addUniq(n.Op.Discharged, x.ID)
}

// declareTimers: the timers acknowledged on a fenced box and not yet
// delivered are declared lost with it and alerted (moves §7.10: no return
// lane carries them; §5.9: the loss record's timer knowledge is partial).
// Task 4 narrows it to the timers its last published manifest does not
// carry. The mutation of fence-declares-timers drops them silently.
func (w *world) declareTimers(x *box, h *holder) {
	if h == nil || !on("fence-declares-timers") {
		return
	}
	n := 0
	for i := range w.acks {
		a := &w.acks[i]
		if a.Box == x.ID && a.Inc == h.Inc && a.Kind == "timer" && !a.Exempt && !w.delivered[a.W] && !w.timerCarried(x, h, a.W) {
			a.Exempt = true
			n++
		}
	}
	if n > 0 {
		w.stats["timers declared lost"] += n
		w.alert("box %d fenced: %d acknowledged timers registered on it are lost with it", x.ID, n)
	}
}

// timerCarried: the newest landed manifest of the admitted chain carries
// the timer (Task 4), so the install that follows restores it and the
// fence does not declare it. INV-28 is the oracle beside it: a timer the
// manifest published last of the generation carries is never declared,
// whatever the sequences say.
func (w *world) timerCarried(x *box, h *holder, id int) bool {
	_, _, _, _, m := w.newestManifest(x.Region, h.Location, h.Gen)
	if m != nil && contains(m.Timers, id) {
		return true
	}
	if last := w.lastLandedManifest(x.Region, h.Location, h.Gen); last != nil && contains(last.Timers, id) {
		w.violate("INV-28: box %d's timer %d is declared lost, but the manifest published last in generation %q carries it (seq %d)", x.ID, id, last.Gen, last.Seq)
	}
	return false
}

// deadBoxFiles is what a fenced box's files directory contributes to the
// recovery: in the one-box shape, which has no files publisher (ruling
// 11), its local directory stands in for what it published (L17), as
// convertToRelocating's does; in the two-box shape nothing, because a
// dead box's disk is never read: the files come from the pinned manifest
// alone.
func (w *world) deadBoxFiles(x *box) *filesDir {
	if w.perRegion > 1 {
		return nil
	}
	return x.Files.clone()
}

// startFailover is the fence's start of a relocation for a serving
// executor (§5.8 reason failover, StartRelocation with a fence; spec 3's
// placer supplies the decision): op.exec none, the stamp kept, at pinning
// with op.PinFailover standing for 1b-next's starting (ruling 4). The
// start names the target to and makes it the only participant (rev 7,
// Astra round 4 finding 3); only the mutation of
// reloc-failover-start-guard starts with none, and pinning names one.
// The target acknowledges before the pin is selected (PinFailover's
// barrier, rule reloc-failover-ack-before-pin, rev 5 and 6), and the pin
// waits out failoverPinDelay from the fence. id is the operation's id: a
// fresh one for a serving executor's failover, the landing's own for a
// committed landing's conversion (ConvertCommittedLanding keeps it).
func (w *world) startFailover(n *claim, x *box, to *box, id string) {
	n.Op = opState{Kind: "relocating", ID: id, Phase: "pinning", Exec: -1, Src: x.ID, Tgt: -1, SrcRegion: x.Region,
		RelocReason: "failover", ReqAt: w.now, App: n.Holder.App, SrcGen: n.Holder.Gen, SrcStamp: n.Holder.Stamp, StopFiles: w.deadBoxFiles(x), PinFailover: true}
	if to != nil {
		// StartRelocation names the target and makes it the only
		// participant (rev 7, Astra round 4 finding 3)
		n.Op.Tgt, n.Op.Parts = to.ID, []int{to.ID}
	}
	w.stats["relocation started: failover"]++
}

// failoverStart is Worker.Start for a failover (1b-next rev 19; rev 7,
// Astra round 4 finding 3), the checks before the start CAS in its order:
// the claim serves on the fenced source, an eligible target exists
// (Eligible), and the app admits the registration (Admission.Register).
// It returns the target, or nil when the start must wait for a later
// pass; the CAS rechecks the admission and applies RelocationStartGuard
// with the source fenced (failoverStartAllowed). The mutation of
// reloc-failover-start-guard starts at once with no target, as rev 6 did.
func (w *world) failoverStart(c *claim, x *box) (*box, bool) {
	if !on("reloc-failover-start-guard") {
		return nil, true
	}
	to := w.chooseBoxExcept(x.Region, c.Holder.App, x.ID)
	if to == nil {
		w.stats["failover start: no eligible target yet"]++
		return nil, false
	}
	if ad := w.admit[c.Holder.App]; ad != nil && ad.Closed {
		w.stats["failover start: app admission closed"]++
		return nil, false
	}
	return to, true
}

// failoverStartAllowed is the start CAS's own checks for a failover:
// the admission recheck (Admission.Recheck), the target not on lost, and
// RelocationStartGuard with the source fenced, so the source's own
// park-local does not block (L11) and a park-replica in the region does.
func (w *world) failoverStartAllowed(n *claim, x, to *box) bool {
	if !on("reloc-failover-start-guard") {
		return true
	}
	if ad := w.admit[n.Holder.App]; ad != nil && ad.Closed {
		return false
	}
	return to != nil && !contains(n.Lost, to.ID) && w.relocationStartGuard(n, x, to, true)
}

// fenceSource is §5.8's source loss (B1-6, FenceSource): a drain whose
// source is terminated before its release is journaled becomes a
// failover in the same CAS, op.exec none, the stamp kept, the source off
// the barrier, its files directory the publisher's stand-in; after its
// release only its liveness changes. (1b-next keeps a source lost in
// starting at starting until the remaining participant acknowledges,
// then PinFailover; the model goes to pinning at once with
// op.PinFailover set for a source lost in starting, so the target
// acknowledges before the pin is selected, rule
// reloc-failover-ack-before-pin, and the pin waits out the delay.) The
// mutation of reloc-source-loss takes a source lost in releasing as
// released at what the remote holds, so its unuploaded writes go
// undeclared.
func (w *world) fenceSource(n *claim, x *box) {
	switch {
	case n.Op.Phase == "starting", n.Op.Phase == "releasing" && n.Op.Release == nil:
		if !on("reloc-source-loss") && n.Op.Phase == "releasing" {
			h := n.Holder
			n.Op.Release = &releaseEntry{Box: x.ID, Session: h.Session, Gen: h.Gen, FinalTX: w.remoteMax(w.buckets[x.Region], dataKey(h.Location, h.Gen, "")), Flushed: true, Files: w.deadBoxFiles(x), Timers: append([]int(nil), x.Timers...), At: w.now}
			n.Op.Exec, n.Op.Phase = -1, "pinning"
			return
		}
		n.Op.PinFailover = n.Op.Phase == "starting" // 1b-next keeps it in starting until PinFailover (rev 6)
		n.Op.RelocReason, n.Op.Exec, n.Op.Phase = "failover", -1, "pinning"
		n.Op.Release, n.Op.StopFiles = nil, w.deadBoxFiles(x)
		w.stats["relocation: source lost before its release, converted to failover"]++
	default:
		w.stats["relocation: source lost after its release"]++
	}
}

// replaceTarget is §5.8's replacement table (ReplaceTarget) for a target
// terminated by its fence, within the fence's CAS:
//
//	starting, releasing, pinning      -> the same phase, to' named
//	installing                        -> installing: to' is op.exec, a new stamp, a fresh P'
//	installed, or probing before P    -> installed: to' is op.exec, a new stamp, the same P, no receipt
//	after P's first admission, or     -> pinning: a nested failover from holder.generation as observed
//	probe-failed
//
// Exhaustion is checked first (Worker.replace, 1b-next R1-16): with the
// replacements used the fence goes to fenceExhaustedTarget and no spare
// is looked for. With no eligible spare it reports false and the fence
// waits for a later pass (Worker.next). The mutation of
// reloc-replacement-table resumes every late loss at installed with the
// same P.
func (w *world) replaceTarget(n *claim, x *box) bool {
	phase := n.Op.Phase
	if on("reloc-replacement-cap") && n.Op.Replacements >= maxReplacements {
		w.fenceExhaustedTarget(n, x)
		return true
	}
	next := w.chooseBoxExcept(n.Op.SrcRegion, n.Holder.App, relocExcept(n.Op, x.ID)...)
	if next == nil {
		if on("reloc-dead-target-bound") && x.FenceObservedAt > 0 && w.now > observedAt(x)+releaseDeadline {
			// FenceTargetWithoutSpare (1b-next rev 18 finding 2): the fence
			// is consumed ReleaseDeadline after its observed_at and the
			// operation is stuck; the mutation waits for a spare for ever
			w.stats["relocation: dead target with no spare fenced"]++
			w.fenceTargetStuck(n, x, fmt.Sprintf("box %d terminated and no spare in %d steps", x.ID, releaseDeadline))
			return true
		}
		w.stats["relocation: target lost, no spare yet"]++
		return false
	}
	if !w.nameReplacement(n, next.ID, phase) {
		return n.Op.Stuck != ""
	}
	w.stats["relocation: target replaced at "+phase]++
	late := preparedOpened(n) || phase == "probe-failed" || phase == "probed" || phase == "opened"
	switch {
	case phase == "starting" || phase == "releasing" || phase == "pinning":
	case phase == "installing":
		n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = "", 0, 0
		w.handExec(n, next.ID, "installing", w.newID("P"))
	case !late || !on("reloc-replacement-table"):
		w.handExec(n, next.ID, "installed", n.Op.Prepared) // P was never opened: to' makes its own local install and opens it
	default:
		w.abandonPrepared(n)
		n.Op.Exec, n.Op.Phase = -1, "pinning"
		n.Op.Pin, n.Op.Prepared, n.Op.Receipt, n.Op.Release = nil, "", nil, nil
		n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = "", 0, 0
		n.Op.ProbeTries, n.Op.Attempts = 0, 0
		w.stats["nested failover after P was admitted"]++
	}
	return true
}

// fenceExhaustedTarget is FenceExhaustedTarget (1b-next ruling 23 as rev
// 9 amended it; rule reloc-exhausted-fence): a target terminated after
// the operation used its replacements is fenced (the caller's fenceBox)
// and the operation is stuck, with no third target chosen. While another
// box is op.exec (the live source in starting or releasing) only stuck is
// new: the target's fence is no evidence against the source, whose
// execution identity, phase and release stay, so its own stop evidence
// still decides. Otherwise (op.exec the dead target, or none) the dead
// installation's authority is revoked: op.exec none at pinning, its
// receipt and release dropped, and P's pin and remote build kept when P
// was never opened, dropped for a nested pinning when it was. The
// mutation is rev 8's: always op.exec none at pinning, which revokes a
// live source without its stop evidence.
func (w *world) fenceExhaustedTarget(n *claim, x *box) {
	w.stats["relocation: exhausted target fenced"]++
	w.fenceTargetStuck(n, x, fmt.Sprintf("box %d terminated after %d target replacements", x.ID, maxReplacements))
}

// fenceTargetStuck is 1b-next's fenceTargetStuck, the body
// FenceExhaustedTarget and FenceTargetWithoutSpare share (rev 18): phase
// sensitive as above; an unopened P is kept for the operator.
func (w *world) fenceTargetStuck(n *claim, x *box, why string) {
	if on("reloc-exhausted-fence") && n.Op.Exec >= 0 && n.Op.Exec != x.ID {
		n.Op.Stuck = why
		return
	}
	opened := preparedOpened(n)
	n.Op.Exec, n.Op.Phase = -1, "pinning"
	n.Op.Receipt, n.Op.Release = nil, nil
	n.Op.ProbeTries, n.Op.Attempts = 0, 0
	if opened {
		n.Op.Pin, n.Op.Prepared = nil, ""
		n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = "", 0, 0
	}
	n.Op.Stuck = why
}

// ---------- the files publisher and the files pin (region HA §5.6, L13, L17) ----------

// manifestKey is a files manifest of generation gen within loc, by
// sequence; isManifestKey recognizes one. genOf reads files/ under the
// origin generation as "".
func manifestKey(loc, gen string, seq int) string {
	return dataKey(loc, gen, fmt.Sprintf("files/m-%06d", seq))
}

func isManifestKey(k string) bool { return strings.Contains(k, "files/m-") }

// filesState is what a manifest carries, as the publisher compares it.
func filesState(timers []int, f *filesDir) string {
	if f == nil {
		return fmt.Sprint(timers)
	}
	return fmt.Sprint(timers, f.Key, f.Env)
}

// publishFiles is the files publisher (§5.6, H6: platform-managed files
// and wake state sync continuously on every plan): the executor
// publishes a manifest, files and timers as they stand, whenever that
// state changed since its last one, at the generation's next sequence
// (the publisher of a generation starts at 1; a relocation's P has its
// prepared manifest at 0). The PUT goes through issue, so it may complete
// late, after the box is terminated: the late-file-upload case. A
// relocation's target publishes its probe session's state under P
// (B2-4). Only the two-box shape runs it (ruling 11).
func (w *world) publishFiles(b *box) {
	c := w.claim
	if c == nil || c.Holder == nil || !w.agentUp(b) || b.Row == nil || b.Files == nil || b.Files.Inc != c.Holder.Inc {
		return
	}
	if c.Op.Exec != b.ID || !(c.Op.Kind == "serving" || c.Op.Kind == "relocating" && (b.ID == c.Op.Src || preparedOpened(c))) {
		return // a relocation's target publishes only once P is opened, into its own generations
	}
	h := c.Holder
	if b.pubGen != h.Gen {
		b.pubGen, b.pubSeq, b.releaseSeq = h.Gen, 0, 0
	}
	state := filesState(b.Timers, b.Files)
	if state == b.published && w.objectCurrent(b, b.Files) {
		return
	}
	if !w.ensureObject(b, h, b.Files) {
		return // the object first: a manifest names only durable objects (§5.6)
	}
	b.pubSeq++
	w.issued++
	bl := &blob{Kind: "files", Inc: h.Inc, Gen: h.Gen, Seq: b.pubSeq, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...), Base: w.issued}
	w.issue(b.ID, &request{kind: rPut, region: b.Region, key: manifestKey(h.Location, h.Gen, b.pubSeq), b: bl})
	b.published = state
	w.stats["files manifests published"]++
}

// newestManifest is the files pin's choice (§5.8, B2-4; StorePinner): the
// newest landed manifest in the admitted chain, holder.generation first,
// then each generation it superseded (genParent: the files live set's
// chain, which no trim of holder.admissions shortens; 1b-next rev 12
// finding 1), by sequence; not the database baseline's generation.
// Nothing landed: "" and nil.
func (w *world) newestManifest(r int, loc, gen string) (key string, ver, seq int, mgen string, m *blob) {
	bk := w.buckets[r]
	seen := map[string]bool{}
	for p := dataKey(loc, gen, ""); !seen[p]; {
		seen[p] = true
		var best *blob
		for _, k := range bk.keys(p + "files/") {
			v := bk.current(k)
			if v == nil || v.B == nil || v.B.Kind != "files" {
				continue
			}
			if best == nil || v.B.Seq > seq {
				key, ver, seq, mgen, best = k, v.ID, v.B.Seq, v.B.Gen, v.B
			}
		}
		if best != nil {
			return key, ver, seq, mgen, best
		}
		next, ok := w.genParent[p]
		if !ok {
			break
		}
		p = next
	}
	return "", 0, 0, "", nil
}

// lastLandedManifest is INV-28's ground truth: of the first generation in
// the chain that holds a landed manifest, the one published last (the
// issue order the model records in the blob), whatever its sequence.
func (w *world) lastLandedManifest(r int, loc, gen string) *blob {
	_, _, _, mg, m := w.newestManifest(r, loc, gen)
	if m == nil {
		return nil
	}
	bk := w.buckets[r]
	last := m
	for _, k := range bk.keys(dataKey(loc, mg, "files/")) {
		if v := bk.current(k); v != nil && v.B != nil && v.B.Kind == "files" && v.B.Gen == mg && v.B.Base > last.Base {
			last = v.B
		}
	}
	return last
}

// filesLive is the files live set (L13; claim §6.2a "Files in the live
// set"): a manifest of holder.generation, of a relocation's prepared
// generation, the one a running relocation's pin names, and the newest
// landed manifest of the admitted chain (the live manifest, which an
// unchanged file keeps referencing generations after it was published)
// is never archived by a park-generation or read as residue. It is
// region-qualified (rule files-live-region; joint round 1 finding 1): k
// is the key in region r's bucket. The mutation of files-live-set
// protects nothing, so a park-generation of an older generation archives
// the live or the pinned manifest.
func (w *world) filesLive(c *claim, r int, k string) bool {
	return on("files-live-set") && w.filesLiveIn(c, r, k, on("files-live-region") && on("live-set-region"))
}

// filesLiveSet is the files live set itself, region-qualified, which the
// residue oracle reads whatever is mutated.
func (w *world) filesLiveSet(c *claim, r int, k string) bool {
	return w.filesLiveIn(c, r, k, true)
}

// filesHome: does region r hold the files live set? Only the
// authoritative region does (claim rev 6.5 §6.2a, R4-3): the region of
// the holder's current installation, as the moves spec's §5.1 table
// ("The authoritative region", the claim state by claim state table)
// names it, united with op.exec's region, whose box runs moves' own parks
// (F4's keepFilesLive):
//
//   - serving, pushing, reverting: op.exec's region;
//   - relocating: its source region, for the whole operation (a
//     relocation never leaves it; while op.exec is none, at a failover's
//     pinning or a replay, the chain's newest manifest there is the pin's
//     candidate: two-box seed 1215, stuck for ever under a narrower
//     reading, INV-3);
//   - moving (before the flip, every phase, op.exec none from stopped on):
//     the source's region;
//   - landing: the target's region, op.exec's, throughout; and the
//     source's region until the claim follows the move's commit
//     (op.Committed, followRecord's "committed" case), not only until the
//     target's probe is admitted as §5.1's row for holder.generation
//     reads: the probe's admission commits nothing, and until the commit
//     a clean rollback names op.SrcGen again in the source region
//     (moves_test.go's return CAS, Holder.Gen = op.SrcGen), whose next
//     failover pins the source's chain there (Astra joint round 2). The
//     record's commit (landTarget's MoveState = "committed", the CAS on
//     the field rollback also takes) rules the rollback out; the claim
//     follows it in the CAS that queues the source region's
//     whole-location park-superseded, so the source stays home until
//     that park owns it;
//   - restoring: the source's region while the target releases
//     (releasing, released, the dirty lane's returned: SrcPhase waiting,
//     op.exec the target or none), until the source's return CAS (moves
//     a5e8c13f, L18 finding 1). After a clean return the source is op.exec
//     and names op.SrcGen again. After a dirty return the newest admission
//     (the target's) decides until the source's probe is admitted into the
//     return generation: the source region is not the files home then,
//     even as op.exec's, so the return's install park (moves' D6,
//     install-park) retires the pre-move generations with their files
//     (dirtyReturned). The target region is retired by the release's
//     whole-location park, which never consults the files live set;
//   - creating, deleting, parking, purging, free: none (op.exec is none
//     in each, and no case below names them).
//
// The mutation of files-home-move-phases keeps op.exec's region and a
// relocation's source only, so from a move's stopped phase to the
// commit, and in a rollback's released phase, a park-generation queued
// during the move archives the live manifest in the source region.
func (w *world) filesHome(c *claim, r int) bool {
	if dirtyReturned(c, on("dirty-return-needs-admission")) && r == c.Op.SrcRegion && on("files-home-move-phases") {
		return false // D6: the dirty return's install parks the pre-move generations
	}
	if e := w.execBox(c); e != nil && e.Region == r {
		return true
	}
	if c.Op.Kind == "relocating" && r == c.Op.SrcRegion {
		return true
	}
	if !on("files-home-move-phases") || r != c.Op.SrcRegion {
		return false
	}
	switch c.Op.Kind {
	case "moving", "restoring":
		return true
	case "landing":
		return !c.Op.Committed
	}
	return false
}

// pinWatch is what INV-30 remembers of one region: the manifest version
// its next install would pin, and the object that manifest names when
// the object was current in the region's bucket.
type pinWatch struct {
	inc       int
	loc       string
	key       string
	ver, base int
	obj       string
}

// servableRegions are the regions where the host could still be served
// next (INV-30's own reading of claim §6.2a and moves §5.1, written
// apart from filesHome so that it can catch filesHome): op.exec's
// region; a relocation's source region (it never leaves it); a move's
// source region before the flip; a landing's source region until the
// record commits (landTarget's MoveState = "committed", the CAS on the
// field a rollback also takes: until then a clean rollback names
// op.SrcGen again there); a rollback's source region. Creating,
// deleting, parking and purging have no executor and name none.
func (w *world) servableRegions(c *claim) []int {
	if c == nil || c.Holder == nil || dirtyReturned(c, true) {
		// after a dirty return of a move its target admitted, the source's
		// next install comes from the return lane, never from a pinned
		// manifest of the pre-move generations, which D6 retires by design
		// (moves a5e8c13f). The oracle reads the claim's admissions itself,
		// whatever the rules say: a dirty return the target never admitted
		// keeps the pre-move generation as the source's live state
		return nil
	}
	var out []int
	if e := w.execBox(c); e != nil {
		out = append(out, e.Region)
	}
	src := false
	switch c.Op.Kind {
	case "relocating", "moving", "restoring":
		src = true
	case "landing":
		src = !c.Op.Committed && !(w.rec != nil && w.rec.MoveID == c.Op.ID && w.rec.MoveState == "committed")
	}
	if src && (len(out) == 0 || out[0] != c.Op.SrcRegion) {
		out = append(out, c.Op.SrcRegion)
	}
	return out
}

// checkFilesPreserved is INV-30 (joint round 2; claim §6.2a "Files in the
// live set", moves §5.1): in every region where the host could still be
// served next, the manifest that region's next install would pin (the
// newest landed manifest of the admitted chain there, newestManifest,
// which the failover's pin takes) and the object it names survive: once
// seen, the version is never archived or deleted while the region can
// still serve, unless a newer manifest has landed in its place. It reads
// the bucket each step and remembers the last step's choice, so an
// archive is seen as it happens, before any later read of the bucket
// takes the shortened chain for the truth (the fence's timerCarried and
// INV-28 read the bucket after the archive; a failover's stuck pin is
// INV-3 only much later, and only if a failover happens).
func (w *world) checkFilesPreserved() {
	c := w.claim
	next := map[int]pinWatch{}
	for _, r := range w.servableRegions(c) {
		loc := c.Holder.Location
		bk := w.buckets[r]
		mk, mv, _, _, m := w.newestManifest(r, loc, c.Holder.Gen)
		if p, ok := w.pinWatch[r]; ok && p.inc == c.Holder.Inc && p.loc == loc {
			gone := bk.current(p.key) == nil || bk.byID(p.key, p.ver) == nil
			superseded := m != nil && m.Base > p.base
			if gone && !superseded {
				w.violate("INV-30: region %d could still serve inc %d (%s %s), and the manifest its next install would pin, %s@%d, is gone from its bucket (now %q)", r, c.Holder.Inc, c.Op.Kind, c.Op.ID, p.key, p.ver, mk)
				return
			}
			if !gone && mk == p.key && p.obj != "" && bk.current(p.obj) == nil {
				w.violate("INV-30: region %d could still serve inc %d (%s %s), and the object %s its pinned manifest %s names is gone from its bucket", r, c.Holder.Inc, c.Op.Kind, c.Op.ID, p.obj, p.key)
				return
			}
		}
		if m == nil {
			continue
		}
		pw := pinWatch{inc: c.Holder.Inc, loc: loc, key: mk, ver: mv, base: m.Base}
		if m.Files != nil && m.Files.Obj != "" && bk.current(m.Files.Obj) != nil {
			pw.obj = m.Files.Obj
		}
		next[r] = pw
	}
	w.pinWatch = next
}

// filesLiveIn is the files live set of region r. regional false is the
// region-blind set the rule files-live-region replaced (and every region
// authoritative, as live-set-region's mutation reads the whole live set):
// the newest manifest of the chain in any region is live in every region,
// so a manifest PUT that lands in a former region after a move's cleanup
// protects itself there for ever.
func (w *world) filesLiveIn(c *claim, r int, k string, regional bool) bool {
	if c == nil || c.Holder == nil || !isFilesKey(k) {
		return false
	}
	if !isManifestKey(k) {
		return w.objectReferenced(c, r, k, regional)
	}
	loc := c.Holder.Location
	if !strings.HasPrefix(k, loc) {
		return false
	}
	if regional && !w.filesHome(c, r) {
		return false
	}
	g := genOf(loc, k)
	if g == c.Holder.Gen {
		return true
	}
	if c.Op.Kind == "relocating" {
		if c.Op.Pin != nil && k == c.Op.Pin.ManKey {
			return true
		}
		if c.Op.Prepared != "" && g == c.Op.Prepared {
			return true
		}
	}
	for _, x := range w.filesRegions(r, regional) {
		if mk, _, _, _, _ := w.newestManifest(x, loc, c.Holder.Gen); mk == k {
			return true
		}
	}
	return false
}

// filesRegions is where the live set's newest manifest is read for a key
// of region r: region r alone; the region-blind set reads every region.
func (w *world) filesRegions(r int, regional bool) []int {
	if regional {
		return []int{r}
	}
	all := make([]int, nRegions)
	for i := range all {
		all[i] = i
	}
	return all
}

// pinnedManifest is the manifest the local install applies: the exact
// version the pin names (rule files-pin-exact; the model's versioned
// bucket reads it by version id, where the real store, which has none,
// reads the pinner's retained copy, 1b-next ruling 21); the mutation
// reads the newest manifest at install time, which a late PUT from the
// dead box may have changed since the pin.
func (w *world) pinnedManifest(c *claim) *blob {
	p := c.Op.Pin
	if p == nil || p.ManKey == "" || c.Holder == nil {
		return nil
	}
	if !on("files-pin-exact") {
		_, _, _, _, m := w.newestManifest(c.Op.SrcRegion, c.Holder.Location, c.Holder.Gen)
		return m
	}
	if v := w.buckets[c.Op.SrcRegion].byID(p.ManKey, p.ManVer); v != nil && !v.Marker {
		return v.B
	}
	return nil
}

// regionMin is the region's minimum box count (H5): the shape's, or one
// for a witness's one-box region.
func (w *world) regionMin(r int) int {
	if w.oneBoxRegion[r] {
		return 1
	}
	return w.perRegion
}

// ---------- acknowledgements (rev 4, Astra R1-4) ----------

// acked is requireAcks: every current participant has a durable
// exec_ack.<box> entry in the journal. It reads only the claim; whether a
// participant's controller is ready now is the participant's own check
// before its effects (recordAck, and role()'s lease), never the barrier.
func acked(c *claim) bool {
	for _, p := range c.Op.Parts {
		if !contains(c.Op.Acks, p) {
			return false
		}
	}
	return true
}

// recordAck is AcknowledgeExecution: a participant whose controller is
// ready (its execution lease current, role() renews it first) journals
// exec_ack.<box> once; the entry outlives a lost lease or an agent
// restart (1b-next rev 2 finding 10: the claim keeps the first
// acknowledgement). A box replaced out of the participants keeps its
// entry, which the barrier no longer reads.
func (w *world) recordAck(b *box, c *claim) {
	if c == nil || c.Op.Kind != "relocating" || !contains(c.Op.Parts, b.ID) || contains(c.Op.Acks, b.ID) || contains(c.Lost, b.ID) {
		return
	}
	if b.ExecUntil == 0 || w.localNow(b) > b.ExecUntil {
		return // no ready controller: nothing to acknowledge with
	}
	id := c.Op.ID
	// the controller's durable record exists before the acknowledgement
	// (1b-next Task 7: journaled before the open), and stays until a
	// successful scoped cleanup clears it (finishParticipation)
	b.execRecord, b.cleanupPending = id, true
	if w.cas(c, func(n *claim) bool {
		ok := sameOp(n.Op.ID, id) && n.Op.Kind == "relocating" && contains(n.Op.Parts, b.ID) && !contains(n.Op.Acks, b.ID)
		if ok {
			n.Op.Acks = append(n.Op.Acks, b.ID)
		}
		return ok
	}) {
		w.logf("relocation %s: box %d acknowledged execution", id, b.ID)
	}
}

// finishParticipation is the participant's Sweep and Finish (1b-next
// Task 6 and 7, rulings 20): once the claim no longer runs the operation
// with this box as a live participant, the scoped cleanup runs (it
// retires an unopened P's local install of that operation, and is a
// no-op on a box that admitted a writer or serves the host); only when it
// succeeds is the durable record cleared (rule reloc-clear-after-cleanup;
// the mutation clears it after a failed cleanup, as rev 4 of 1b-next
// did). A box whose cleanup fails keeps the record, and root refuses an
// executable update while it exists (claim §5.10).
func (w *world) finishParticipation(b *box, c *claim) {
	if b.execRecord == "" || c == nil {
		return
	}
	if d := c.Op.Completed; c.Op.Kind == "serving" && d != nil && !d.Discharged && d.Box == b.ID && b.execRecord == d.Op {
		return // the completion box's participation is C10's to discharge (completionPass), first
	}
	if c.Op.Kind == "relocating" && c.Op.ID == b.execRecord && contains(c.Op.Parts, b.ID) && !contains(c.Lost, b.ID) {
		return // still a participant
	}
	if b.cleanupFails {
		w.stats["relocation controller: scoped cleanup failed"]++
		if on("reloc-clear-after-cleanup") {
			return // the record stays, and so does the refusal of updates
		}
		b.execRecord = "" // the mutation: cleared anyway
		return
	}
	if f := b.Files; f != nil && f.Gen != "" && c.Holder != nil && f.Inc == c.Holder.Inc && strings.HasPrefix(f.Gen, "P") && !(c.Op.Kind == "serving" && c.Op.Exec == b.ID) && f.Gen != c.Holder.Gen && !w.ancestorGen(c.Holder.Location, c.Holder.Gen, f.Gen) {
		w.retireArtifacts(b, f.Inc, f.Stamp) // an unopened P's local install of the finished operation
	}
	b.execRecord, b.cleanupPending = "", false
	w.stats["relocation controller: record cleared"]++
}

// updateExecutable is an executable update on b (claim §5.10, region HA
// §5.10): root refuses it while a relocation controller's record exists.
// INV-29 is the oracle: an update never runs on a box whose relocation
// participation's scoped cleanup has not succeeded.
func (w *world) updateExecutable(b *box) bool {
	if b.Destroyed {
		return false
	}
	if b.execRecord != "" {
		w.stats["executable update refused: controller record"]++
		return false
	}
	if b.cleanupPending {
		w.violate("INV-29: box %d updated its executable while a relocation's scoped cleanup had not succeeded (claim §5.10)", b.ID)
	}
	w.stats["executable updated"]++
	return true
}

// latestObservation is the newest observed_at the operation's journal
// retains (JournalPin checks every fence.<box>.observed_at, a release or
// not: rev 4, Astra R1-5); FenceAt covers a converted move's fence.
func latestObservation(op opState) int {
	at := op.FenceAt
	for _, o := range op.Observed {
		if o > at {
			at = o
		}
	}
	return at
}

// observedAt is when the fence record says the box was observed
// terminated (FenceTargetWithoutSpare's deadline basis); the mutation of
// reloc-deadline-from-observed-at uses the termination itself, which no
// worker may have seen yet.
func observedAt(x *box) int {
	if !on("reloc-deadline-from-observed-at") {
		return x.TerminatedAt
	}
	return x.FenceObservedAt
}

// objectReferenced: a live manifest (the chain's newest, the pinned one,
// holder.generation's or P's) names the object, or a live box's files
// directory does (its next manifest will): the files live set is every
// object the live manifests reference (claim §6.2a). Region-qualified as
// the manifests are: k is the key in region r's bucket, region r must be
// the files home (filesHome), only region r's manifests are read, and a
// box's directory names k only when it names region r's object. The
// region-blind set (regional false) reads every region's manifests and
// every box's directory, for a key of any region.
func (w *world) objectReferenced(c *claim, r int, k string, regional bool) bool {
	loc := c.Holder.Location
	names := func(m *blob) bool { return m != nil && m.Files != nil && m.Files.Obj == k }
	if regional && !w.filesHome(c, r) {
		return false
	}
	for _, x := range w.filesRegions(r, regional) {
		if _, _, _, _, m := w.newestManifest(x, loc, c.Holder.Gen); names(m) {
			return true
		}
		bk := w.buckets[x]
		for _, g := range []string{c.Holder.Gen, c.Op.Prepared} {
			if g == "" && g != c.Holder.Gen {
				continue
			}
			for _, mk := range bk.keys(dataKey(loc, g, "files/m-")) {
				if v := bk.current(mk); v != nil && names(v.B) {
					return true
				}
			}
		}
	}
	if c.Op.Kind == "relocating" && c.Op.Pin != nil && c.Op.Pin.ManKey != "" && (!regional || r == c.Op.SrcRegion) {
		if names(w.pinnedManifestOf(c.Op.SrcRegion, c.Op.Pin)) {
			return true
		}
	}
	for _, b := range w.liveBoxes() {
		if b.Files != nil && b.Files.Obj == k && (!regional || b.Files.ObjRegion == r) {
			return true
		}
	}
	return false
}

// keepFilesLive wraps a moves park's keep (a push's whole-location park,
// a landing's install park, a rollback's return park) so it never
// archives the files live set (finding F4, rev 4: claim §6.2a's live set
// binds every park; moves' parks predate region HA's files tree). Only
// the two-box shape has files keys, so the one-box shape is unchanged.
// The mutation of files-live-set-moves-parks archives them with the rest.
func (w *world) keepFilesLive(r int, keep func(string) bool) func(string) bool {
	return func(k string) bool {
		if keep(k) {
			return true
		}
		if on("files-live-set-moves-parks") && w.filesLive(w.claim, r, k) {
			w.markSweepOwed(r, k)
			return true
		}
		return false
	}
}

// markSweepOwed: a park kept k only because the files live set named it;
// once obsolete it is the at-rest sweep's (claim §6.2a, 1b-next ruling 21),
// so INV-10 at an operation's end exempts it as it exempts a straggler.
func (w *world) markSweepOwed(r int, k string) {
	for i := range w.buckets {
		if r >= 0 && i != r {
			continue
		}
		for _, v := range w.buckets[i].objs[k] {
			v.SweepOwed = true
		}
	}
}

// renewController is a relocation controller's own renewal loop (hostexec
// renews its lease on its own clock, not when the regional worker happens
// to run; rev 4): an agent that is up and holds a current lease for a
// relocation it participates in keeps it current. A lease is still lost
// when the agent is down, paused, frozen or partitioned past the TTL,
// which is the scoped cleanup's case. Two-box shape only.
func (w *world) renewController(b *box) {
	c := w.claim
	if c == nil || c.Op.Kind != "relocating" || !contains(c.Op.Parts, b.ID) || !w.agentUp(b) || b.Partitioned {
		return
	}
	if b.ExecUntil > 0 && w.localNow(b) <= b.ExecUntil && b.ExecUntil-w.localNow(b) < execRenew {
		b.ExecUntil = w.localNow(b) + execTTL
		w.execLease[b.ID] = execLease{Op: c.Op.ID, Expires: w.now + execTTL}
	}
}

// preparedManifestHere is moves rev 7.22 §7.6's one install routine (L16,
// L17; claim R18-1) for a landing or a return in the two-box shape (rev
// 4, claim §11 "move landing, source cleanup, then target failure with an
// unchanged file"): the generation's prepared files manifest, create-only
// at sequence 0 in this region, naming objects in this region's bucket
// (an unchanged file the host carried from another region is uploaded
// here first), acknowledged before the install's CAS. Without it a
// target lost right after its commit would leave no files in its region.
func (w *world) preparedManifestHere(b *box, c *claim, gen string) bool {
	h := c.Holder
	if !w.ensureObject(b, h, b.Files) {
		return false
	}
	bl := &blob{Kind: "files", Inc: h.Inc, Gen: gen, Seq: 0, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...), Text: c.Op.ID, Base: w.issued}
	return w.ackedPut(b, manifestKey(h.Location, gen, 0), bl, true)
}

// relocJournalRelease is the participant's release (1b-next Task 7): the
// live executor runs its release sequence and journals it
// (JournalRelease), the source in releasing, a failed target in
// probe-failed. relocReleasing and relocProbeFailed call it for their
// release branch, and relocate calls it by itself while a fence is
// pending (rev 8; pending marks the trace line). The CAS is the union of
// the two phases' conditions: in releasing no release is journaled yet;
// in probe-failed the journaled release, if any, is another box's.
func (w *world) relocJournalRelease(b *box, c *claim, pending bool) {
	op := c.Op
	phase := op.Phase
	if phase == "probe-failed" && w.c1Fallback() {
		// QueueRelocationRelease needs the source in releasing: with claim
		// ask C1 open there is no failed-target obligation, so nothing is
		// queued, stopped or journaled, and the release deadline makes the
		// operation stuck (finding 4)
		w.stats["release refused: no failed-target obligation (C1 open)"]++
		return
	}
	r := w.relocRelease(b, c)
	if r == nil {
		return
	}
	if w.cas(c, func(n *claim) bool {
		ok := sameOp(n.Op.ID, op.ID) && n.Op.Kind == "relocating" && n.Op.Phase == phase && n.Op.Exec == b.ID && (n.Op.Release == nil || phase == "probe-failed" && n.Op.Release.Box != b.ID)
		if ok {
			n.Op.Release = r
		}
		return ok
	}) {
		if phase == "releasing" {
			w.stats["release journaled: source"]++
		} else {
			w.stats["release journaled: failed target"]++
		}
		note := ""
		if pending {
			note = ", a fence pending"
		}
		w.logf("relocation %s: box %d released the host at TXID %d (flushed=%v)%s", op.ID, b.ID, r.FinalTX, r.Flushed, note)
	}
}

// dirtyReturned: a rollback whose source has taken the host back dirty
// (its return CAS written: SrcPhase past waiting) and whose probe has not
// yet been admitted into the return generation. From that CAS the newest
// admission decides (moves' movesAuthoritative, a5e8c13f): the target's,
// so the source region is not authoritative, and the return's install
// park retires the pre-move generations (D6).
func dirtyReturned(c *claim, needAdmission bool) bool {
	if c == nil || c.Holder == nil || c.Op.Kind != "restoring" || !c.Op.Dirty || c.Op.SrcPhase == "waiting" || c.Holder.Gen == c.Op.RollbackID {
		return false
	}
	return !needAdmission || targetAdmitted(c)
}

// targetAdmitted is movesinstall.go's targetAdmitted (Astra round 4, item
// 2): the claim's recorded admissions hold one of this move's generation
// (the landing's, installGen: the move's op id). A dirty return can come
// without one: src_covered false makes the return dirty even when the
// target never admitted (releaseTarget's seal-return), and then
// holder.generation is still the pre-move one, the source's live state,
// which the return's install keeps and parks nothing of. Decided from the
// claim's admissions only, never from the landing phase.
func targetAdmitted(c *claim) bool {
	if c == nil || c.Holder == nil {
		return false
	}
	for _, a := range c.Holder.Admissions {
		if a.Gen == c.Op.ID {
			return true
		}
	}
	return false
}