package hostlife

import (
	"fmt"
	"strings"
)

// worker is one box agent's lifecycle worker: exec-lease renewal, the
// obligations it owns, then its role in the current operation. One call
// performs at most one protocol step.
func (w *world) worker(b *box) { w.workerMode(b, false) }

// obligationRunner is a separate goroutine from the role worker: a box
// runs its chores and its operation role concurrently (CC11 is what keeps
// them from meeting in one namespace).
func (w *world) obligationRunner(b *box) { w.workerMode(b, true) }

func (w *world) workerMode(b *box, obl bool) {
	if b.Old {
		return // a pre-floor executable knows nothing of claims
	}
	c, ok := w.readClaim(b)
	if !ok {
		if obl && w.claim != nil && len(w.claim.Obl) > 0 && w.calm {
			w.logf("box %d: obligation run skipped, the agent cannot read the claim (down=%v paused=%v frozen=%v partitioned=%v destroyed=%v)", b.ID, b.AgentDown, b.AgentPaused, b.Frozen, b.Partitioned, b.Destroyed)
		}
		return
	}
	if obl {
		if c == nil || (contains(c.Lost, b.ID) && on("lost-quarantine")) {
			if c != nil && w.calm {
				w.logf("box %d: obligation run refused, the box is declared lost", b.ID)
			}
			return
		}
		for _, o := range c.Obl {
			if w.ownsObligation(o, b) {
				w.doObligation(b, c, o)
				return
			}
		}
		w.sweep(b, c)
		return
	}
	if b.stale != nil && w.rng.Intn(6) == 0 {
		// a stalled worker of this box resumes with the claim it read long ago
		old := b.stale
		b.stale = nil
		w.resumeStale(b, old)
		return
	} else if w.rng.Intn(10) == 0 {
		b.stale = c.clone()
	}
	changed, pending := w.fenceStep(b, c)
	if changed {
		return // the regional worker's termination fence (region HA §5.9), before its role
	}
	// a relocation participant's fence this worker could not consume yet
	// (a dead target with no spare before its deadline): Worker.Step
	// returns w.replace(...) before the phase switch, so the regional
	// transitions wait; the participant work that does not go through
	// Worker.Step still runs (rev 8, Astra round 5)
	b.fencePending = pending
	w.role(b, c)
	b.fencePending = false
}

// initRecord is claim rev 6.5 §6.4's initialization (R5-4): a box creates
// or refreshes its local stamp record only from a fresh claim read, and
// only when that claim names it as op.exec or as the landing participant;
// it sets active_stamp = holder.stamp. A create's first wake, adoption on
// the serving box, and a lost record on restart all go through here. A
// stamp this box has retired is never made active again.
func (w *world) initRecord(b *box, c *claim) {
	if c == nil || c.Holder == nil || c.Holder.Stamp == 0 || c.Ver != w.claim.Ver {
		return
	}
	inc := c.Holder.Inc
	if !(c.Op.Exec == b.ID || c.Op.Kind == "landing" && c.Op.Tgt == b.ID) {
		return
	}
	if b.stampOf(inc) == c.Holder.Stamp {
		return
	}
	if s := b.Stamps[inc]; s != nil && contains(s.Retired, c.Holder.Stamp) {
		return
	}
	w.activate(b, inc, c.Holder.Stamp, c.Holder.Gen)
	c.readStamp = c.Holder.Stamp
	w.logf("box %d: local record for inc %d initialized from holder.stamp %d", b.ID, inc, c.Holder.Stamp)
}

// activate is activation on box b (claim rev 6.5 §6.4): the old active
// stamp retires, the new one is set, and the artifacts of the generation
// the claim names, installed under the predecessor's stamp, are tagged
// with the new one (R6-4: a dirty rollback's return database is installed
// under S and tagged R when R activates). The oracle refuses a stamp the
// claim issued for another box's installation (INV-6).
func (w *world) activate(b *box, inc, stamp int, gen string) {
	if bx, ok := w.stampBox[stamp]; ok && bx != b.ID {
		w.violate("INV-6: box %d activates stamp %d, issued for box %d's installation (claim rev 6.5: holder.stamp names the installation op.exec should run)", b.ID, stamp, bx)
	}
	b.install(inc, stamp)
	if b.DB != nil && b.DB.Inc == inc && b.DB.Gen == gen {
		b.DB.Stamp = stamp
	}
	if b.Files != nil && b.Files.Inc == inc && b.Files.Gen == gen {
		b.Files.Stamp = stamp
	}
}

// issueStamp is the stamp an installing CAS writes into holder.stamp: the
// gen the CAS produces, recorded as box's (claim rev 6.5 §2).
func (w *world) issueStamp(n *claim, box int) int {
	s := w.nextGen()
	if w.stampBox == nil {
		w.stampBox = map[int]int{}
	}
	w.stampBox[s] = box
	return s
}

func dbStamp(d *localDB) int {
	if d == nil {
		return 0
	}
	return d.Stamp
}

func filesStamp(f *filesDir) int {
	if f == nil {
		return 0
	}
	return f.Stamp
}

// resumeStale is a stalled worker goroutine of b resuming with a claim it
// read long ago (rev 4's actor). With claim C3's revalidation it re-reads
// under its lease, sees the world moved, and stops; the mutation acts on
// the old read. A witness calls this directly (witness_test.go).
func (w *world) resumeStale(b *box, old *claim) {
	c, ok := w.readClaim(b)
	if !ok {
		return
	}
	if on("revalidate-before-effect") && !sameStep(old, c) {
		return
	}
	// the revalidation re-reads the box's own record as well: the stamp the
	// resumed step carries is the one active now (claim rev 6.5 §6.4; seed
	// 67671 / 100,000: a stalled probe step carried the stamp active at its
	// read, before a cold admission's installing CAS, and the box's record
	// refused its own probe)
	old = old.clone()
	old.readStamp = c.readStamp
	w.role(b, old)
}

// role is the worker's exec-lease renewal, quarantine and operation role,
// acting on the claim c it read.
func (w *world) role(b *box, c *claim) {
	w.initRecord(b, c)
	w.finishParticipation(b, c) // a relocation controller whose participation ended (rev 4)
	// exec lease (claim §7.1)
	if w.isParticipant(c, b) && !contains(c.Lost, b.ID) {
		if b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil {
			w.scopedCleanup(b, c) // the relocation controller's lease was lost (1b-next rev 14 finding 1)
		}
		if b.ExecUntil == 0 || 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}
		}
		if c.Op.Kind == "relocating" {
			w.recordAck(b, c) // exec_ack.<box>, durable (rev 4)
			if cur := w.claim; cur != nil && cur.Ver != c.Ver {
				nc := cur.clone()
				nc.readStamp = c.readStamp
				c = nc
			}
		}
	} else if c == nil || !w.isParticipant(c, b) {
		b.ExecUntil = 0
	}
	if !b.Old && b.DB != nil && b.DB.Epoch != b.Epoch && len(b.Procs) > 0 {
		// the agent restarted under a running app whose session's
		// replicator died with the old agent (claim rev 6.10: a new agent
		// process is a new session): the app is stopped, and the next wake
		// is a new admission that declares what it acknowledged meanwhile
		b.Procs = nil
		w.stats["app stopped: its session's agent is gone"]++
		w.logf("box %d: the agent restarted under a running app of a dead session; the app is stopped", b.ID)
	}
	if c == nil {
		return
	}
	if contains(c.Lost, b.ID) && on("lost-quarantine") {
		// a quarantined box retires whatever it kept before anything else
		if b.Row != nil || b.DB != nil || b.Files != nil || len(b.Procs) > 0 {
			if w.evidence(b, "quarantine") {
				inc := 0
				if b.DB != nil {
					inc = b.DB.Inc
				}
				if w.localPark(b, inc) { // the quarantine retires the installation
					b.Row = nil
					b.owedDischarge = "" // a lost box's participation binds nothing (ruling 77)
				}
			}
		}
		return
	}
	if c.Holder == nil {
		return
	}
	if b.open != nil {
		// an admitted session finishing its start (rev 7.11 §7.14); the
		// operation's own step runs once the app answered readiness
		w.resumeStart(b, c)
		if !hasMain(b) && !b.hostFailed {
			return // the operation's own step runs once the app answered readiness, or once the host is marked failed
		}
	}
	switch c.Op.Kind {
	case "moving":
		if b.ID == c.Op.Src {
			w.moveSource(b, c)
		} else if b.ID == c.Op.Tgt {
			w.moveReport(b, c)
		}
	case "reverting":
		if b.ID == c.Op.Src {
			w.revertSource(b, c)
		}
	case "landing":
		if b.ID == c.Op.Tgt {
			w.landTarget(b, c)
		} else if b.ID == c.Op.Src {
			w.holdSourceRow(b, c)
		}
	case "restoring":
		if b.ID == c.Op.Tgt {
			w.releaseTarget(b, c)
		} else if b.ID == c.Op.Src {
			w.recoverSource(b, c)
		}
	case "pushing":
		if b.ID == c.Op.Src {
			w.pushStep(b, c)
		}
	case "relocating":
		w.relocate(b, c)
	case "serving":
		w.completionPass(b, c) // a relocation completed on this box and not yet marked (ruling 80)
	}
}

// convertToRelocating is the termination fence's conversion of a pre-flip
// move (moves rev 7.20 §7.10; claim rev 6.28 §4): one CAS records the
// fence (fence.<box>), sets exec none, keeps holder.stamp and
// holder.generation, starts relocating at pinning in the source's region,
// touches no existing obligation, and queues the staging lane's park for
// the target region when the move had not queued it yet (as an abort
// would). The fenced source is on lost and leaves the participants.
func (w *world) convertToRelocating(c *claim, src *box) {
	op := c.Op
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" || n.Op.Committed {
			return false
		}
		n.Op.Kind, n.Op.Phase, n.Op.Exec, n.Op.FenceBox = "relocating", "pinning", -1, src.ID
		n.Op.RelocReason, n.Op.FenceAt, n.Op.Tgt, n.Op.Src = "failover", w.now, -1, src.ID
		n.Op.Pin, n.Op.Prepared, n.Op.Receipt, n.Op.Release = nil, "", nil, nil
		// the files the relocation installs: region HA §5.6's files
		// publisher kept the source's files directory in S3 as it served
		// (L17); this model has no publisher, so the fence takes the
		// source's directory as published (a stand-in, moves §14.2.6)
		if src.Files != nil {
			n.Op.StopFiles = src.Files.clone()
		}
		var parts []int
		for _, p := range n.Op.Parts {
			if p != src.ID {
				parts = append(parts, p)
			}
		}
		n.Op.Parts = parts
		n.Lost = addUniq(n.Lost, src.ID)
		if c.oblFor(op.ID) == 0 {
			n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc))
		}
		return true
	}) {
		w.stats["relocation: source fenced before the flip"]++
		w.logf("move %s: source box %d terminated before the flip; converted to relocating at pinning in region %d", op.ID, src.ID, op.SrcRegion)
	}
}

// relocate is the relocation's worker on a box of the operation's region
// (region HA §5.8's phase table; claim rev 6.30 §4): every live, enrolled
// box of the region that is not on lost may act, and each phase's CAS is
// conditional on the operation, the phase and the ETag, so two workers
// racing make the same CAS and one fails. The phases live in
// regionha_test.go. A stuck operation waits for the operator.
func (w *world) relocate(b *box, c *claim) {
	op := c.Op
	if b.Region != op.SrcRegion || b.Destroyed || contains(c.Lost, b.ID) || !b.Enrolled || op.Stuck != "" || w.haOff {
		return // with region HA off no relocation pass runs, only the completion pass (ruling 82)
	}
	if b.fencePending && on("reloc-fence-pending-blocks") {
		// Worker.Step returned w.replace(...) before its phase switch (rev
		// 8, Astra round 5): no regional transition while a participant's
		// fence is pending. A live executor's own release sequence is the
		// participant's (1b-next Task 7), not Worker.Step's, so it runs.
		// The mutation goes on into the phase switch, as rev 7 did
		w.stats["relocation: phase held, a participant's fence is pending"]++
		if (op.Phase == "releasing" || op.Phase == "probe-failed") && b.ID == op.Exec && (op.Release == nil || op.Release.Box != op.Exec) {
			w.relocJournalRelease(b, c, true)
		}
		return
	}
	switch op.Phase {
	case "starting":
		w.relocStarting(b, c)
	case "releasing":
		w.relocReleasing(b, c)
	case "pinning":
		w.relocPinning(b, c)
	case "installing":
		w.relocInstalling(b, c)
	case "installed":
		w.relocInstalled(b, c)
	case "probing":
		w.relocProbing(b, c)
	case "probe-failed":
		w.relocProbeFailed(b, c)
	case "probed":
		w.relocSettle(b, c)
	}
}

type execLease struct {
	Op      string
	Expires int
}

func (w *world) leasesSeen(c *claim) bool {
	for _, id := range c.Op.Parts {
		l, ok := w.execLease[id]
		if !ok || l.Op != c.Op.ID || l.Expires < w.now {
			return false
		}
	}
	return true
}

// stagingLease: the shared staging-work lease (claim C3), per destination
// region: the target's for a move's staging, the source's for a dirty
// rollback's return lane.
func (w *world) takeStaging(c *claim, b *box) bool {
	return w.takeStagingFor(c.Op.TgtRegion, c.Op.ID, b)
}

func (w *world) takeStagingFor(region int, opID string, b *box) bool {
	l := w.staging[region]
	if l.Op != "" && l.Op != opID && l.Until >= w.now {
		return false
	}
	w.staging[region] = stagingLease{Op: opID, Box: b.ID, Until: w.now + stagingTTL}
	return true
}

type stagingLease struct {
	Op    string
	Box   int
	Until int
}

// snapshotBlob is a staging round: the source's final checkpoint over its
// replica, which is what the mirror copies (§7.3). With no local database
// (a lost disk, or never woken) it is the replica alone.
func (w *world) snapshotBlob(b *box, c *claim) (*blob, *blob) {
	return w.snapshotBlobOf(b, c, c.Holder.Gen)
}

func (w *world) snapshotBlobOf(b *box, c *claim, gen string) (*blob, *blob) {
	// the sealed transfer is a verified final snapshot (rev 7.5 §5.1, R4-1):
	// the local database after its final checkpoint, every page, TXIDs 1
	// to its current one; with no local database (a lost disk, or never
	// woken) the replica restored as it stands
	var db *blob
	if b.DB != nil {
		// the final checkpoint writes the pending writes as the next local
		// level-0 file and uploads what it can (litestream SyncAndWait);
		// the snapshot is then at that position, and carries every page
		// whether or not the upload has landed. While the admitted session's
		// start has not opened replication there is no replicator to sync:
		// the snapshot carries the pages as they stand and nothing is
		// uploaded into the live generation (rev 7.11; seed 17056 / 30,000:
		// a round's forced sync wrote the first upload's key with other
		// bytes while the start's own PUT was still in flight)
		if b.open == nil && (b.Old || b.DB.Epoch == b.Epoch) {
			if len(b.DB.Pending) > 0 {
				bl := w.deltaOf(b.DB, b.DB.Inc)
				b.DB.Queue = append(b.DB.Queue, bl)
				b.DB.Pos, b.DB.Pending = bl.Max, nil
			}
			w.ship(b)
		}
		db = snapshotOf(b.DB, b.DB.Inc)
	} else if sv := w.sealVersion(c.Op.SealDB); sv != nil && c.Op.Kind == "restoring" {
		// a target that never restored a database returns the seal it was
		// given, as staged, pinned by version (rev 7.11 §7.8)
		cp := *sv.B
		db = &cp
	} else {
		// no local database (a lost disk): the remote as the next admission
		// would restore it, a lagging marker read through the journal (rev
		// 7.11; seeds 3987 and 14322 / 30,000 sealed an empty database over
		// a generation whose marker had not landed, and the push lost the
		// legacy host's writes)
		var r *localDB
		if gen == c.Holder.Gen {
			r = w.restoreAsNext(b, c)
		} else {
			r = w.restoreCalc(b.Region, c.Holder.Inc, c.Holder.Location, gen)
		}
		if r != nil {
			db = snapshotOf(r, c.Holder.Inc)
		} else if !w.hasFiles(b, c.Holder.Location, gen) && !w.hasFiles(b, c.Holder.Location, c.Holder.Gen) {
			db = &blob{Kind: "snap", Level: 9, Min: 1, Inc: c.Holder.Inc, Pages: map[string]int{}} // nothing was ever written
		} else {
			w.stats["seal refused: nothing to restore"]++
			return nil, nil // the remote cannot be read: no seal
		}
	}
	files := &blob{Kind: "files", Inc: c.Holder.Inc, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...)}
	if b.Files == nil {
		if fv := w.sealVersion(c.Op.SealFiles); fv != nil && c.Op.Kind == "restoring" {
			cp := *fv.B
			files = &cp
		}
	}
	return db, files
}

// sealVersion is the staged seal object a journal ref names, or nil.
func (w *world) sealVersion(r ref) *version {
	if r.Ver == 0 {
		return nil
	}
	for _, v := range w.buckets[r.Region].objs[r.Key] {
		if v.ID == r.Ver {
			return v
		}
	}
	return nil
}

func (w *world) moveSource(b *box, c *claim) {
	op := c.Op
	switch op.Phase {
	case "leases":
		if w.rec != nil && w.rec.Inc == c.Holder.Inc && w.rec.MoveID == "" {
			// the record half of the request, after the claim's intent (C2):
			// a crash between is finished here or by the recovery pass
			w.rec.MoveID, w.rec.MoveState = op.ID, ""
			return
		}
		if w.leasesSeen(c) {
			w.cas(c, func(n *claim) bool { n.Op.Phase = "preparing"; return sameOp(n.Op.ID, op.ID) })
		}
	case "preparing", "ready":
		if !contains(c.Regions, op.TgtRegion) {
			w.cas(c, func(n *claim) bool { n.Regions = addUniq(n.Regions, op.TgtRegion); return true })
			return
		}
		srcReady := !on("source-readiness") || (b.Config == c.Holder.Inc && b.Live == c.Holder.App && b.LiveInc == c.Holder.Inc && b.HasKey)
		if op.Phase == "ready" && op.ReportOK && w.now-op.ReportAt < 60 && op.Auto && srcReady {
			// auto-cut: admission is atomic with abort eligibility
			if w.rec == nil || w.rec.MoveID != op.ID {
				return
			}
			deadline := w.now + cutBudget
			if !on("cut-deadline") {
				deadline = 1 << 30
			}
			w.cas(c, func(n *claim) bool {
				if !sameOp(n.Op.ID, op.ID) || n.Op.CutAt != 0 || n.Op.Kind != "moving" {
					return false
				}
				n.Op.CutAt = w.now
				n.Op.CutDeadline = deadline
				n.Op.PinVer = w.cfgVer[c.Holder.App] // move.inputs, pinned at admission
				n.Op.Phase = "stopping"
				return true
			})
			return
		}
		if w.now-op.ReqAt > readyWait {
			// readiness never came: abort on its own (no human)
			w.logf("abort %s: phase=%s report=%v@%d srcReady=%v auto=%v", op.ID, op.Phase, op.ReportOK, op.ReportAt, srcReady, op.Auto)
			w.abortMove(c, b)
			return
		}
		if !w.takeStaging(c, b) {
			return
		}
		db, files := w.snapshotBlob(b, c)
		r1 := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), b: db}
		r2 := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), b: files}
		if w.rng.Intn(3) == 0 {
			// this round revalidated under the staging-work lease just now,
			// and its goroutine pauses before signing (J-N2): it may sign
			// after the final round and the seal, but only before its lease's
			// SafeUntil (claim C3)
			until := w.staging[op.TgtRegion].Until
			for _, r := range []*request{r1, r2} {
				w.stalled = append(w.stalled, &stalledSign{box: b.ID, epoch: b.Epoch, req: r, signAt: w.now + w.pauseSign(), until: until})
			}
			return
		}
		v1 := w.issue(b.ID, r1)
		v2 := w.issue(b.ID, r2)
		if v1 > 0 && v2 > 0 && op.Phase == "preparing" {
			w.cas(c, func(n *claim) bool { n.Op.Phase = "ready"; return sameOp(n.Op.ID, op.ID) })
		}
	case "aborting":
		if c.oblFor(op.ID) == 0 {
			w.closeMove(c, op.Src)
		}
	case "stopping":
		if b.Row != nil {
			b.Row.Status, b.Row.HoldOp = "moving", op.ID
		}
		if w.now > op.CutDeadline {
			w.startRevert(c)
			return
		}
		if !w.evidence(b, op.ID) {
			return
		}
		d := w.digestOf(b)
		files := b.Files.clone()
		repl, next := "", 0
		if b.Row != nil {
			repl, next = b.Row.Repl, b.Row.Next
		}
		w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Phase != "stopping" {
				return false
			}
			n.Op.Phase, n.Op.StopAt, n.Op.StopDigest, n.Op.StopFiles = "stopped", w.now, d, files
			n.Op.CutSession = n.Holder.Session // the session src_covered speaks for (rev 7.12)
			n.Op.Repl, n.Op.Next = repl, next
			n.Op.Exec = -1 // K3: the source gives up execution, citing its evidence
			return true
		})
	case "stopped":
		if w.now > op.CutDeadline || b.FailArchive {
			w.startRevert(c)
			return
		}
		if !w.takeStaging(c, b) {
			return
		}
		db, files := w.snapshotBlob(b, c)
		if db == nil {
			w.logf("%s: the source has no database and its remote cannot be restored; reverting", op.ID)
			w.startRevert(c)
			return
		}
		if op.SealDB.Ver == 0 {
			r := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), b: db}
			w.maybeStallDuplicate(b, op, r)
			v := w.issue(b.ID, r)
			if v > 0 {
				w.cas(c, func(n *claim) bool {
					n.Op.SealDB = ref{Region: op.TgtRegion, Key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), Ver: v}
					return sameOp(n.Op.ID, op.ID)
				})
			}
			return
		}
		r := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), b: files}
		w.maybeStallDuplicate(b, op, r)
		v := w.issue(b.ID, r)
		if v > 0 {
			// does the source's own replica cover the stopped database? The
			// final checkpoint's upload may still be in flight or lost; the
			// seal holds every page regardless. A clean rollback restores the
			// source from its replica only when it does (rev 7.11, INV-15;
			// seed 17056 / 30,000)
			covered := b.DB == nil || b.open == nil && len(b.DB.Pending) == 0 && w.remoteMax(w.buckets[b.Region], dataKey(c.Holder.Location, c.Holder.Gen, "")) >= b.DB.Pos
			if !covered {
				w.stats["seal: source replica short of the stop"]++
			}
			w.cas(c, func(n *claim) bool {
				if !sameOp(n.Op.ID, op.ID) {
					return false
				}
				n.Op.SrcCovered = covered
				n.Op.SealFromDB = b.DB != nil
				n.Op.SealFiles = ref{Region: op.TgtRegion, Key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), Ver: v}
				n.Op.SealQuiet = w.now + stagingTTL + horizonH
				if !on("staging-quiet") {
					n.Op.SealQuiet = w.now
				}
				n.Op.Phase = "sealed"
				return true
			})
		}
	case "sealed":
		// the flip, record first (§7.5): one CAS on the record, then the
		// claim follows on a later step. A crash between leaves nobody
		// with execution, which is safe; whoever next sees the record
		// completes the claim (rev 6 §19 K4).
		if w.rec != nil && w.rec.Inc == c.Holder.Inc && w.rec.MoveID == op.ID && w.rec.MoveState == "landing" {
			w.followRecord(c)
			return
		}
		if w.rec != nil && w.rec.Inc == c.Holder.Inc && w.rec.MoveID == op.ID && w.rec.MoveState == "" && op.StopAt != 0 {
			w.rec.Region, w.rec.MoveState = op.TgtRegion, "landing"
			w.rec.Ver = w.newVer() // every record write has a fresh ETag
			w.logf("flip %s: record names region %d", op.ID, op.TgtRegion)
		}
	}
}

func (w *world) moveReport(b *box, c *claim) {
	op := c.Op
	if op.Phase != "preparing" && op.Phase != "ready" {
		return
	}
	if w.now-op.ReportAt < 20 {
		return
	}
	ok := b.HasKey || !on("readiness-key")
	if on("readiness-quiet") && w.now < c.Quiet[op.TgtRegion] {
		ok = false // T12: the landing would wait out quiet_after inside the cut window
	}
	w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) {
			return false
		}
		n.Op.ReportAt, n.Op.ReportOK = w.now, ok
		return true
	})
}

func (w *world) abortMove(c *claim, by *box) {
	op := c.Op
	w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" || n.Op.CutAt != 0 {
			return false
		}
		n.Op.Phase = "aborting"
		if on("abort-target-owner") {
			n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc))
		}
		return true
	})
	if !on("abort-target-owner") && by != nil {
		// the source tries to retire remote staging itself
		w.issue(by.ID, &request{kind: rDelVer, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), ver: 1})
	}
}

func (w *world) startRevert(c *claim) {
	op := c.Op
	w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" {
			return false
		}
		n.Op.Kind, n.Op.Exec = "reverting", op.Src
		if op.StopAt == 0 {
			n.Op.RevPhase = "evidence"
		} else {
			n.Op.RevPhase = "verify"
		}
		n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc))
		return true
	})
}

func (w *world) holdRow(b *box, status, holdOp string, c *claim) {
	if b.Row == nil {
		b.Row = &row{Inc: c.Holder.Inc, App: c.Holder.App}
	}
	b.Row.Status, b.Row.HoldOp = status, holdOp
	if w.rng.Intn(25) == 0 {
		// a paused goroutine's duplicate of this write, executed later
		b.deferred = append(b.deferred, &localWrite{kind: "hold", status: status, holdOp: holdOp, epoch: b.Epoch, at: w.now + 1 + w.pauseLen()})
	}
}

// openRow is the conditional open (moves §5.2).
func (w *world) openRow(b *box, holdOp string) bool {
	if w.rng.Intn(6) == 0 {
		d := &localWrite{kind: "open", holdOp: holdOp, epoch: b.Epoch, at: w.now + 1 + w.pauseLen()}
		if w.rng.Intn(2) == 0 {
			// the adversary keeps this one paused until its worst moment,
			// or a long stall's end
			d.worst, d.at = true, w.now+4000
		}
		b.deferred = append(b.deferred, d)
	}
	return w.openNow(b, holdOp)
}

func (w *world) openNow(b *box, holdOp string) bool {
	if b.Row == nil {
		return false
	}
	if on("hold-op-open") && b.Row.HoldOp != holdOp {
		return false
	}
	b.Row.Status, b.Row.HoldOp = "ready", ""
	return true
}

// localWrite is a host-specific registry write whose goroutine passed its
// revalidation and then paused: it lands later, and only the conditions
// on the write itself (hold_op) stand between it and a newer operation.
type localWrite struct {
	kind   string // "open" or "hold"
	status string
	holdOp string
	epoch  int
	at     int  // the latest it lands
	worst  bool // the scheduler wakes it at the worst moment before at
}

// pauseLen is how long a paused goroutine sleeps before its registry
// write lands: mostly short, sometimes as long as a whole operation. A
// crash or reboot (b.Epoch) kills it first.
func (w *world) pauseLen() int {
	if w.rng.Intn(4) == 0 {
		return w.rng.Intn(2000)
	}
	return w.rng.Intn(150)
}

// pauseSign is how long a staging goroutine pauses between revalidating
// under the staging-work lease and signing: it may outlast the lease, and
// then SafeUntil (claim C3) is all that stops the signature.
func (w *world) pauseSign() int { return w.rng.Intn(2*stagingTTL + horizonH + 1) }

// maybeStallDuplicate models a lost-response retry in the cut's own
// staging round: the first attempt's goroutine paused before signing and
// the agent retried. The paused attempt may sign after the seal.
func (w *world) maybeStallDuplicate(b *box, op opState, r *request) {
	w.maybeStallDuplicateFor(b, op.TgtRegion, r)
}

func (w *world) maybeStallDuplicateFor(b *box, region int, r *request) {
	if w.rng.Intn(3) != 0 {
		return
	}
	dup := *r
	w.stalled = append(w.stalled, &stalledSign{box: b.ID, epoch: b.Epoch, req: &dup, signAt: w.now + w.pauseSign(), until: w.staging[region].Until})
}

type stalledSign struct {
	box       int
	epoch     int
	req       *request
	signAt    int
	until     int
	checked   bool // passed its SafeUntil check and signed, not yet sent
	checkedAt int
	retry     bool // a retry of a send that failed transiently
}

// runStalled lets paused goroutines resume: deferred registry writes and
// staging rounds that pause between revalidation and signing.
func (w *world) runStalled() {
	var keep []*stalledSign
	for _, s := range w.stalled {
		b := w.boxByID(s.box)
		if b == nil || b.Destroyed || b.Epoch != s.epoch {
			continue // the goroutine died with its agent
		}
		if s.signAt > w.now || b.Frozen || b.AgentPaused {
			keep = append(keep, s)
			continue
		}
		if s.checked {
			// the request is sent now, after a pause that a whole-VM
			// freeze can stretch without bound. Signed at the check, its
			// timestamp is the check's, and S3 refuses it once older than
			// the signature-age limit (model 18); signed at send time, it
			// is accepted however late
			resigned := s.retry && !on("sdk-no-resign") // the SDK's retry refreshed the signature
			if on("sign-at-check") && !resigned && w.now > s.checkedAt+sigAge {
				w.stats["stale signature refused"]++
				continue
			}
			if !s.retry && w.rng.Intn(3) == 0 {
				// a transient error: the SDK retries the same request later,
				// after a pause a frozen box can stretch (CC19)
				s.retry, s.signAt = true, w.now+1+w.pauseLen()
				keep = append(keep, s)
				continue
			}
			w.issue(b.ID, s.req)
			continue
		}
		margin := maxSkew + maxPause
		if on("sign-at-check") {
			margin = maxSkew + sigAge // SafeUntil leaves room for the signature's whole valid life
		}
		if on("sign-before-safeuntil") && w.localNow(b) > s.until-margin {
			continue // checked SafeUntil immediately before signing: too late
		}
		// the check passed and the request is signed; the goroutine may
		// still pause before sending it
		s.checked, s.checkedAt, s.signAt = true, w.now, w.now+w.rng.Intn(maxPause+1)
		keep = append(keep, s)
	}
	w.stalled = keep
	for _, b := range w.boxes {
		if b.Destroyed || b.Frozen || b.AgentPaused || b.AgentDown {
			continue
		}
		var rest []*localWrite
		for _, d := range b.deferred {
			if d.epoch != b.Epoch {
				continue
			}
			// an adversarial wake-up: a stale open resumes while a newer
			// operation holds the row, at its worst moment: after that
			// operation committed and stopped re-asserting the hold
			worst := d.worst && d.kind == "open" && b.Row != nil && b.Row.HoldOp != "" && b.Row.HoldOp != d.holdOp &&
				w.claim != nil && w.claim.Op.Committed && w.rng.Intn(3) == 0
			if d.at > w.now && !worst {
				rest = append(rest, d)
				continue
			}
			switch d.kind {
			case "open":
				w.openNow(b, d.holdOp)
			case "hold":
				if b.Row != nil && (!on("hold-op-open") || b.Row.HoldOp == "" || b.Row.HoldOp == d.holdOp) {
					b.Row.Status, b.Row.HoldOp = d.status, d.holdOp
				}
			}
		}
		b.deferred = rest
	}
}

func (w *world) revertSource(b *box, c *claim) {
	op := c.Op
	if !on("revert-journal") && !b.mem[op.ID+"rev"] && (op.RevPhase == "probing" || op.RevPhase == "probed") {
		// the journal is not trusted after a restart: verify again
		w.cas(c, func(n *claim) bool { n.Op.RevPhase = "verify"; return sameOp(n.Op.ID, op.ID) })
		return
	}
	if op.RevPhase != "probed" && op.RevPhase != "opened" {
		w.holdRow(b, "restoring", op.ID, c)
	}
	switch op.RevPhase {
	case "evidence":
		if w.evidence(b, op.ID) {
			w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probing"; return sameOp(n.Op.ID, op.ID) })
		}
	case "verify":
		if !w.evidence(b, op.ID) {
			return
		}
		ph := "probing"
		if b.DB == nil {
			// the source lost its disk since it stopped (R8-2): nothing to
			// verify; its writer is admitted like any lost replication
			// state, a warm continuation of the intact generation or a cold
			// admission (claim rev 6.8 §6.4a)
			w.logf("revert: box %d's database is gone; the probe restores it", b.ID)
			w.stats["revert: source database lost"]++
		} else if w.digestOf(b) != op.StopDigest {
			ph = "stuck"
			w.logf("revert: digest mismatch %+v vs %+v", w.digestOf(b), op.StopDigest)
		}
		w.cas(c, func(n *claim) bool { n.Op.RevPhase = ph; return sameOp(n.Op.ID, op.ID) })
	case "probing":
		b.mem[op.ID+"rev"] = true
		if !w.admits(b, c, op.ID) {
			return
		}
		if b.HasKey { // RenderHeld + MaterializeHeld before every probe
			b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc
		}
		if hasMain(b) || w.spawn(b, c, c.Holder.Gen, true) {
			w.probeAccepted(b, c)
			w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probed"; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if b.hostFailed && !b.failCounted {
			b.failCounted = true // a host the retry policy marked failed: one failed try for the operation
		} else if b.open != nil || b.hostFailed {
			return // admitted, its start in progress, or a failed host awaiting the operator: not a failed try
		}
		w.cas(c, func(n *claim) bool {
			n.Op.ProbeTries++
			if n.Op.ProbeTries >= probeTries*3 {
				n.Op.RevPhase = "stuck"
			}
			return sameOp(n.Op.ID, op.ID)
		})
	case "probed":
		if b.open != nil {
			return // the source's session has not opened replication yet (rev 7.11 §7.14)
		}
		if !hasMain(b) {
			w.stats["probe restarted"]++ // the probe died: restarted, admitted like any other (rev 7.11)
			w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probing"; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if w.openRow(b, op.ID) {
			w.cas(c, func(n *claim) bool { n.Op.RevPhase = "opened"; return sameOp(n.Op.ID, op.ID) })
		}
	case "opened":
		if c.oblFor(op.ID) == 0 {
			w.closeMove(c, op.Src)
		}
	}
}

func (w *world) holdSourceRow(b *box, c *claim) {
	if b.Row != nil && !c.Op.Committed {
		b.Row.Status, b.Row.HoldOp = "moving", c.Op.ID
	}
}

// rollback is record first (§7.8): one CAS on the record, landing →
// rolling_back, guarded by move id. The commit is a CAS on the same record
// field, so exactly one of them wins. The claim follows on a later step.
func (w *world) rollback(c *claim, why string) {
	op := c.Op
	r := w.rec
	if r == nil || r.Inc != c.Holder.Inc || r.MoveID != op.ID || r.MoveState != "landing" {
		return
	}
	r.MoveState, r.Region, r.RollbackID, r.Reason = "rolling_back", op.SrcRegion, w.newID("rb"), why
	r.Ver = w.newVer()
	w.logf("rollback %s: %s", op.ID, why)
	w.stats["why: "+why+" @"+op.LandPhase]++
}

// followRecord completes the claim half of a record transition the claim
// has not caught up with: the flip, the commit, a rollback. Any actor that
// reads both may do it (rev 6 §19 K4): the CAS is fully determined by the
// record and is conditional on the op id and the claim's prior state.
func (w *world) followRecord(c *claim) bool {
	r := w.rec
	if c == nil || c.Holder == nil || r == nil || r.Inc != c.Holder.Inc || r.MoveID == "" || r.MoveID != c.Op.ID {
		return false
	}
	op := c.Op
	switch {
	case r.MoveState == "landing" && op.Kind == "moving" && op.Phase == "sealed":
		return w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" || n.Op.Phase != "sealed" {
				return false
			}
			n.Op.Kind, n.Op.Exec, n.Op.LandPhase, n.Op.FlipAt = "landing", op.Tgt, "checking", w.now
			n.Op.SrcStamp = n.Holder.Stamp           // the source installation's stamp, kept in the journal for the return (claim rev 6.5)
			n.Op.SrcGen = n.Holder.Gen               // the source's generation at the cut, which a clean rollback's return names again (rev 7.10)
			n.Holder.Stamp = w.issueStamp(n, op.Tgt) // holder.stamp: the gen this CAS produces, the landing's installation stamp (claim rev 6.5 §2)
			n.Op.FlipGen = n.Holder.Stamp
			return true
		})
	case r.MoveState == "committed" && op.Kind == "landing" && !op.Committed:
		return w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "landing" || n.Op.Committed {
				return false
			}
			n.Op.Committed = true // holder.generation moved at the probe's cold admission (claim rev 6.9)
			q := op.StopAt + horizonH
			if !on("quiet-before-ack") {
				q = 0
			}
			n.Obl = append(n.Obl,
				w.scoped(w.newObl(n, "park-superseded", op.SrcRegion, -1, c.Holder.Location, "", q), op.StopAt, c.Holder.Inc),
				w.newObl(n, "park-local", -1, op.Src, "", "", 0),
				w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", op.SealQuiet), stagingBound(op.SealQuiet), c.Holder.Inc))
			if q > n.Quiet[op.SrcRegion] {
				n.Quiet[op.SrcRegion] = q
			}
			// the source region's whole location is parked: its uncovered
			// predecessors are covered here, in the same CAS, with no chore
			// (claim rev 6.14 §6.4a item 6, CoverRegionPredecessors)
			w.coverRegionPredecessors(n, op.SrcRegion, "commit")
			return true
		})
	case r.MoveState == "rolling_back" && op.Kind == "landing":
		return w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "landing" || n.Op.Committed {
				return false
			}
			n.Op.Kind, n.Op.RollbackID, n.Op.SrcPhase, n.Op.Phase, n.Op.Reason = "restoring", r.RollbackID, "waiting", "releasing", r.Reason
			return true
		})
	}
	return false
}

func (w *world) landTarget(b *box, c *claim) {
	op := c.Op
	if c.Op.Exec != b.ID {
		return
	}
	loc := c.Holder.Location
	g := installGen(&op) // the generation this landing prepares; the probe's cold admission makes it holder.generation (claim rev 6.9)
	if !op.Committed {
		if w.rec != nil && w.rec.MoveID == op.ID && w.rec.MoveState != "landing" {
			w.followRecord(c) // a commit or rollback the claim has not caught up with
			return
		}
		w.holdRow(b, "landing", op.ID, c)
		if w.now > op.CutDeadline {
			w.rollback(c, "cut deadline") // N15: every phase before the commit
			return
		}
	}
	switch op.LandPhase {
	case "checking":
		// the new installation sets this box's active stamp to its own
		// captured stamp, the claim gen of the CAS that made the landing
		// current, under the guard, on a fresh read showing its operation
		// current and this box the landing participant (claim rev 6.5
		// §6.4, R3-3, R4-2): a move back to a region the instance left is
		// admitted at its own stamp while the retired installation's
		// writers stay refused
		if fc, ok := w.readClaim(b); ok && fc.Op.ID == op.ID && fc.Op.Kind == "landing" && fc.Op.Tgt == b.ID {
			retiredHere := b.Stamps[c.Holder.Inc] != nil && len(b.Stamps[c.Holder.Inc].Retired) > 0
			if on("fence-by-installation") || !retiredHere {
				b.install(c.Holder.Inc, fc.Holder.Stamp)
				w.logf("box %d: installation stamp %d for inc %d (%s), from holder.stamp", b.ID, fc.Holder.Stamp, c.Holder.Inc, op.ID)
			}
		}
		w.cas(c, func(n *claim) bool {
			n.Boxes = addUniq(n.Boxes, b.ID)
			n.Op.LandPhase = "install-park"
			return sameOp(n.Op.ID, op.ID)
		})
	case "install-park":
		if !w.evidence(b, op.ID) {
			return
		}
		if w.now < c.Quiet[b.Region] && !b.mem[op.ID+"quietwait"] {
			// INV-9 (T12): an admitted cut never waits out quiet_after; the
			// readiness report said the region was quiet
			b.mem[op.ID+"quietwait"] = true
			w.violate("INV-9: the cut of %s waits on quiet_after[%d] until %d", op.ID, b.Region, c.Quiet[b.Region])
		}
		// an earlier stay of this incarnation, in its location here
		if w.parkStep(b, loc, fmt.Sprintf("superseded/%s/%s/install/", op.ID, host), c.Quiet[b.Region], w.opScope(c, loc), w.keepFilesLive(b.Region, confine(loc))) {
			w.cas(c, func(n *claim) bool { n.Op.LandPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) })
		}
	case "install-copy":
		src := op.SealDB
		if cur := w.buckets[b.Region].current(src.Key); cur != nil && cur.ID != src.Ver {
			w.stats["diag: staging newer than seal at install"]++
			if sv := w.buckets[b.Region].byID(src.Key, src.Ver); sv != nil && cur.B.Max < sv.B.Max {
				w.stats["diag: ...and older content"]++
			}
		}
		if !on("seal-by-version") {
			cur := w.buckets[b.Region].current(src.Key)
			if cur == nil {
				w.rollback(c, "staging missing")
				return
			}
			src.Ver = cur.ID
		}
		// into this move's own generation, as its snapshot at the sealed
		// TXID: a copy that completes late can only ever add an object to
		// this generation
		sv := w.buckets[b.Region].byID(src.Key, src.Ver)
		if sv == nil {
			w.rollback(c, "staging missing")
			return
		}
		if sv.B.Max == 0 && on("valid-ltx") {
			// an empty source (R8-3): no LTX file can carry TXID 0, so the
			// generation is installed empty, its marker naming no snapshot
			w.cas(c, func(n *claim) bool {
				n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", 0, "", 0
				return sameOp(n.Op.ID, op.ID)
			})
			return
		}
		key := snapKey(loc, g, sv.B.Max)
		v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: key, srcRegion: b.Region, srcKey: src.Key, srcVer: src.Ver})
		if v > 0 || !on("build-completion") {
			// the build journal records the snapshot's key, version and TXID
			// (R3-1, R4-1); the anchor follows, then completion (R7-4). The
			// mutation publishes from the intent, before the artifacts landed
			next := "install-anchor"
			if !on("build-completion") {
				next = "published"
			}
			w.cas(c, func(n *claim) bool {
				n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, sv.B.Max
				return sameOp(n.Op.ID, op.ID)
			})
		}
	case "install-anchor":
		if w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) {
			w.cas(c, func(n *claim) bool { n.Op.LandPhase = "published"; return sameOp(n.Op.ID, op.ID) })
		}
	case "published":
		// the build is complete; generation.json is published by the probe's
		// cold admission, after readiness and before its first upload (claim
		// rev 6.9 §6.4a)
		w.cas(c, func(n *claim) bool { n.Op.LandPhase = "local"; return sameOp(n.Op.ID, op.ID) })
	case "local":
		w.installLocal(b, c, g, op.SealFiles, op.LandPhase, func(n *claim) { n.Op.LandPhase = "config" })
	case "config":
		if !b.HasKey && on("config-before-probe") {
			w.rollback(c, "configuration did not render")
			return
		}
		if b.HasKey {
			b.Config, b.CfgVer = c.Holder.Inc, w.cfgVer[c.Holder.App]
			if b.CfgVer != op.PinVer {
				// `configured`: the bundle changed after admission (§7.6)
				w.rollback(c, "configuration changed during the move")
				return
			}
		}
		b.Live, b.LiveInc = c.Holder.App, c.Holder.Inc
		w.cas(c, func(n *claim) bool { n.Op.LandPhase = "restore"; return sameOp(n.Op.ID, op.ID) })
	case "restore":
		// the installed digest, for the settle check's dirty verdict, from a
		// restore of the prepared generation computed and not kept: the
		// probe's admission restores the database it starts on (claim rev
		// 6.9: a box never reuses a local database)
		tmp := *b
		tmp.DB = w.restorePinned(b.Region, c.Holder.Inc, loc, g, op.BaseKey, op.BaseVer, op.BaseTX)
		if tmp.DB == nil {
			w.rollback(c, "restore failed")
			return
		}
		d := w.digestOf(&tmp)
		w.cas(c, func(n *claim) bool { n.Op.Installed, n.Op.LandPhase = d, "probing"; return sameOp(n.Op.ID, op.ID) })
	case "probing":
		if w.now > op.CutDeadline {
			w.rollback(c, "cut deadline")
			return
		}
		if !w.admits(b, c, op.ID) {
			return
		}
		if hasMain(b) || w.spawn(b, c, g, true) {
			w.probeAccepted(b, c)
			if !on("commit-before-open") {
				w.openRow(b, op.ID)
			}
			w.cas(c, func(n *claim) bool { n.Op.LandPhase, n.Op.ProbedAt = "probed", w.now; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if b.hostFailed {
			w.rollback(c, "target would not start") // the retry policy marked the host failed (rev 7.11 §7.14)
			return
		}
		if b.open != nil {
			return // admitted, its start in progress or in backoff: not a failed try
		}
		tries := op.ProbeTries + 1
		w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) })
		if tries >= probeTries {
			w.rollback(c, "target would not start")
		}
	case "probed":
		if !op.Committed {
			if w.now > op.CutDeadline {
				w.rollback(c, "cut deadline")
				return
			}
			if !hasMain(b) {
				// the probe died in its settle window (a crash, an agent
				// restart): the landing restarts only the probe (§7.14), a
				// writer admitted like any other (claim rev 6.9 §6.4a, R8-2):
				// cold, into a fresh generation restored from the dead probe's
				if w.spawn(b, c, c.Holder.Gen, true) {
					w.stats["probe restarted"]++
					w.cas(c, func(n *claim) bool { n.Op.ProbedAt = w.now; return sameOp(n.Op.ID, op.ID) })
					return
				}
				if b.hostFailed {
					w.rollback(c, "probe died")
					return
				}
				if b.open != nil {
					return // admitted, its start in progress: not a failed try
				}
				tries := op.ProbeTries + 1
				w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) })
				if tries >= probeTries {
					w.rollback(c, "probe died")
				}
				return
			}
			if b.open != nil {
				return // the probe's session has not opened replication yet (rev 7.11 §7.14): the settle window waits
			}
			if w.now < op.ProbedAt+15 {
				if w.fail["lostreturn"] && w.rng.Intn(2) == 0 {
					w.traffic(b) // the biased schedule: the app does real work in the settle window
				}
				return // the settle window between the two readiness checks
			}
			// the settle window (§7.7): an app that answered once, did real
			// work, and then fails its second readiness check. The biased
			// schedule (model 27) makes every target such an app, so the
			// rollback is dirty and a return lane follows
			if (b.FailSettle || w.fail["lostreturn"]) && hasMain(b) && w.digestOf(b) != op.Installed && w.rng.Intn(2) == 0 {
				w.logf("box %d failed the settle check for %s", b.ID, op.ID)
				w.rollback(c, "settle check failed")
				return
			}
			// the commit, record first (§7.9): a CAS on the same record
			// field the rollback uses, conditional on the deadline
			r := w.rec
			if r != nil && r.MoveID == op.ID && r.MoveState == "landing" && w.now <= op.CutDeadline {
				r.MoveState = "committed"
				r.Ver = w.newVer()
				w.logf("commit %s: record committed", op.ID)
				if w.now > op.CutAt+cutBudget {
					w.violate("INV-9: %s committed at %d, cut at %d (budget %d)", op.ID, w.now, op.CutAt, cutBudget)
				}
			}
			return
		}
		if !op.Opened {
			if w.openRow(b, op.ID) || b.Row != nil && b.Row.Status == "ready" {
				w.cas(c, func(n *claim) bool { n.Op.Opened = true; return sameOp(n.Op.ID, op.ID) })
			}
			return
		}
		if c.oblFor(op.ID) == 0 {
			w.closeMove(c, op.Tgt)
		}
	}
}

// installLocal is the local install, journaled so the files marker is
// written last (R2-6). seal names the files archive version.
func (w *world) installLocal(b *box, c *claim, gen string, seal ref, _ string, next func(n *claim)) {
	if b.Files != nil && b.Files.Gen == gen && b.Files.Done && b.Files.Inc == c.Holder.Inc {
		if w.perRegion > 1 && !w.preparedManifestHere(b, c, gen) {
			return // moves §7.6 (L17, R18-1) in the two-box shape: the generation's prepared manifest first
		}
		// the marker vouches for everything before it
		w.cas(c, func(n *claim) bool { next(n); return true })
		return
	}
	if b.Files == nil || b.Files.Gen != gen {
		// fresh extraction: park whatever is there first, as
		// retire-artifacts of the predecessor's stamp (claim rev 6.5
		// §6.4): the new installation's active stamp is untouched. A
		// landing parks an earlier stay, never its own installation; a
		// dirty rollback's install on the source parks the pre-move
		// database while its stamp is still the active one (R6-4)
		for _, st := range []int{dbStamp(b.DB), filesStamp(b.Files)} {
			if st == 0 {
				continue
			}
			if c.Op.Kind == "restoring" && st != b.stampOf(c.Holder.Inc) && !b.stampRetired(c.Holder.Inc, st) {
				// a retired stamp's leftover is fair game: a cold admission that
				// issued a stamp and then failed its purge leaves the old
				// installation's database under a newer active stamp (rev
				// 7.11, the CAS before the purge; seed 65090 / 100,000)
				w.violate("INV-6: the dirty install on box %d parks an artifact tagged %d while the active stamp is %d (claim rev 6.5 §6.4)", b.ID, st, b.stampOf(c.Holder.Inc))
			}
			if c.Op.Kind == "landing" && st == b.stampOf(c.Holder.Inc) {
				w.violate("INV-6: landing on box %d would park an artifact tagged with its own active stamp %d", b.ID, st)
			}
			w.retireArtifacts(b, c.Holder.Inc, st)
		}
		b.DB = nil
		b.Files = nil
		src := w.buckets[b.Region].byID(seal.Key, seal.Ver)
		if !on("seal-by-version") {
			src = w.buckets[b.Region].current(seal.Key)
		}
		if src == nil || src.B.Files == nil {
			return
		}
		if !on("files-carried") {
			// the kpmg outage: nothing carries the files directory, and the
			// landing carries on regardless
			b.pendingTimers = append([]int(nil), src.B.Timers...)
			b.Timers = append([]int(nil), b.pendingTimers...)
			w.cas(c, func(n *claim) bool { next(n); return true })
			return
		}
		f := src.B.Files.clone()
		f.Gen, f.Done, f.Stamp = gen, false, b.stampOf(c.Holder.Inc)
		if !on("markers-last") {
			f.Done = true
		}
		b.Files = f
		b.pendingTimers = append([]int(nil), src.B.Timers...)
		return
	}
	// files extracted, marker not yet written: env, timers, columns, marker
	b.Timers = append([]int(nil), b.pendingTimers...)
	if src := w.buckets[b.Region].byID(seal.Key, seal.Ver); src != nil {
		b.Timers = append([]int(nil), src.B.Timers...)
	}
	b.Attempt = 0
	b.History = nil
	if b.Row != nil {
		b.Row.Repl, b.Row.Next = c.Op.Repl, c.Op.Next
	}
	b.Files.Done = true
	if w.perRegion > 1 && !w.preparedManifestHere(b, c, gen) {
		return // the first branch finishes once the prepared manifest is acknowledged
	}
	w.cas(c, func(n *claim) bool { next(n); return true })
}

// moveClose reports whether op is a move's close: a committed landing, an
// opened rollback or revert, or an abort. Pushes and relocations release on
// their own terms.
func moveClose(op opState) bool {
	switch op.Kind {
	case "landing", "restoring", "reverting":
		return true
	case "moving":
		return op.Phase == "aborting"
	}
	return false
}

// closeExec is the box a move's close releases to: the target after a
// landing, the source otherwise (§19 K8).
func closeExec(op opState) int {
	if op.Kind == "landing" {
		return op.Tgt
	}
	return op.Src
}

// closeMove is a move's close by the box that runs it, record first (moves
// rev 7.23 §7.5 and §19 K8, Paul's ruling M4). With nothing owed under the
// op, the record's CAS clears move into last_move and names the executor's
// region; only then does the claim release. The box may stop between the
// halves (every third step here, a deterministic schedule that draws
// nothing from the seed's stream). It never comes back to a record without a
// move, so a closed record whose claim still holds the op is the recovery
// pass's to release, from last_move (fixRecord).
func (w *world) closeMove(c *claim, execBox int) {
	op := c.Op
	if c.oblFor(op.ID) != 0 {
		return
	}
	if !on("record-first-close") {
		// rev 7.22: the claim first; the recovery pass wrote the record
		// half after the release
		w.release(c, execBox)
		return
	}
	if !w.closeRecord(c, execBox) {
		return // closed already (last_move): not the box's any more
	}
	if w.now%3 == 0 {
		w.stats["close: stopped between the record and the release"]++
		return
	}
	w.release(c, execBox)
}

// closeRecord is the close's record half: the CAS that clears move into
// last_move and names the executor's region, conditional on the record still
// carrying this move for the claim's incarnation.
func (w *world) closeRecord(c *claim, execBox int) bool {
	r, op := w.rec, c.Op
	if r == nil || c.Holder == nil || r.Inc != c.Holder.Inc || r.MoveID != op.ID {
		return false
	}
	if b := w.boxByID(execBox); b != nil {
		r.Region = b.Region
	}
	r.MoveID, r.MoveState, r.RollbackID, r.LastMove = "", "", "", op.ID
	r.Ver = w.newVer()
	w.logf("close %s: record closed (last_move)", op.ID)
	return true
}

func (w *world) release(c *claim, execBox int) {
	op := c.Op
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) {
			return false
		}
		if op.Kind == "relocating" && (n.Op.Kind != "relocating" || n.Op.Phase != op.Phase || n.Op.Exec != execBox) {
			return false // CompleteRelocation's requireRelocation (L18 finding 2)
		}
		// holder.admissions survives the release by living in the holder
		// (claim rev 6.12): an admission whose start is unfinished at the
		// release stays explained (seed 1052 / 30,000)
		if op.Kind == "relocating" {
			w.relocationLeftovers(n, op) // the abandoned installations' park-local chores (claim §4)
		}
		n.Op = opState{Kind: "serving", Exec: execBox, Parts: []int{execBox}}
		if op.Kind == "relocating" {
			// completed_relocation (ruling 56): the box and the probe's
			// session, unmarked until the completion pass discharges it
			n.Op.Completed = &completedEntry{Op: op.ID, Box: execBox, Session: n.Holder.Session}
		}
		return true
	}) {
		w.logf("released %s to serving on box %d", op.ID, execBox)
		if b := w.boxByID(execBox); b != nil && op.Kind == "relocating" {
			b.owedDischarge = op.ID // ground truth for INV-31
		}
		if moveClose(op) && (w.rec == nil || w.rec.LastMove != op.ID || w.rec.MoveID == op.ID) {
			// K8 as rev 7.23 amends it: never a serving claim over a live
			// move. Judged only for a release CAS that succeeded: one the
			// op-id guard refused released nothing (fix round 2)
			w.violate("K8: %s released the claim before its record closed (record first, rev 7.23)", op.ID)
		}
		w.opsDone++
		// a move's record closed before this (closeMove, rev 7.23); a
		// push's or a relocation's record half follows on a later step
		w.checkDurable(w.claim, "at the release of "+op.ID)
		if r := w.residueAtEnd(); len(r) > 0 {
			w.violate("INV-10: %s ended with residue %v", op.ID, r)
		}
		switch {
		case op.Kind == "landing":
			w.stats["move-done"]++
		case op.Kind == "restoring" && op.LostBox != 0:
			w.stats["declared-lost"]++
		case op.Kind == "restoring" && op.Dirty:
			w.stats["rollback-dirty"]++
		case op.Kind == "restoring":
			w.stats["rollback-clean"]++
		case op.Kind == "reverting":
			w.stats["revert"]++
		case op.Kind == "moving":
			w.stats["abort"]++
		case op.Kind == "pushing":
			w.stats["push"]++
		case op.Kind == "relocating":
			w.stats["relocation"]++
		}
	}
}

func (w *world) releaseTarget(b *box, c *claim) {
	op := c.Op
	if op.Exec == b.ID {
		w.holdRow(b, "releasing", op.RollbackID, c)
		if !w.evidence(b, op.RollbackID) {
			return
		}
		dirty := false
		if (op.LandPhase == "probing" || op.LandPhase == "probed") && b.DB != nil {
			// a target without a database never opened: nothing to return.
			// Since rev 7.10 the probe's admission restores the database, so
			// a target whose probe never started has none (seed 62 / 300:
			// a spurious dirty return sealed an empty database and the
			// source restored it)
			dirty = w.digestOf(b) != op.Installed
		}
		if !on("dirty-return") {
			dirty = false
		}
		if !op.SrcCovered && on("seal-return") {
			// the source's replica stops short of its stopped database: only
			// the seal holds every acknowledged write, so the return carries
			// it whether or not the target wrote (rev 7.11 §7.8)
			dirty = true
			w.stats["rollback: return carries the seal"]++
		}
		w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Exec != b.ID {
				return false
			}
			n.Op.Exec, n.Op.DirtyKnown, n.Op.Dirty, n.Op.Phase, n.Op.TgtStopAt = -1, true, dirty, "released", w.now
			q := w.now + horizonH
			if on("target-park") {
				n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-superseded", op.TgtRegion, -1, c.Holder.Location, "", q), w.now, c.Holder.Inc))
				w.coverRegionPredecessors(n, op.TgtRegion, "rollback") // the target region's whole location is parked (claim rev 6.14 item 6)
			}
			n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", op.SealQuiet), stagingBound(op.SealQuiet), c.Holder.Inc))
			if !dirty {
				n.Obl = append(n.Obl, w.newObl(n, "park-local", -1, b.ID, "", "", 0))
			}
			if q > n.Quiet[op.TgtRegion] {
				n.Quiet[op.TgtRegion] = q
			}
			return true
		})
		return
	}
	if op.Dirty && op.Phase == "released" && op.Exec == -1 {
		// return the target's writes over the approved lane (M17), under
		// the staging-work lease of the source's region, with the same
		// paused signers and SafeUntil rule as the forward path
		if !w.takeStagingFor(op.SrcRegion, op.RollbackID, b) {
			return
		}
		db, files := w.snapshotBlobOf(b, c, op.ID)
		if db == nil {
			return // nothing to return yet: the remote could not be read
		}
		if op.ReturnDB.Ver == 0 {
			r := &request{kind: rPut, region: op.SrcRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.RollbackID), b: db}
			w.maybeStallDuplicateFor(b, op.SrcRegion, r)
			v := w.issue(b.ID, r)
			if v > 0 {
				fromDB := b.DB != nil
				w.cas(c, func(n *claim) bool {
					n.Op.ReturnDB = ref{Region: op.SrcRegion, Key: fmt.Sprintf("moves/%s/%s/db", host, op.RollbackID), Ver: v}
					n.Op.ReturnFromDB = fromDB
					return sameOp(n.Op.ID, op.ID)
				})
			}
			w.loseReturning(b) // the first PUT of the lane is out, accepted or in flight
			return
		}
		r := &request{kind: rPut, region: op.SrcRegion, key: fmt.Sprintf("moves/%s/%s/files", host, op.RollbackID), b: files}
		w.maybeStallDuplicateFor(b, op.SrcRegion, r)
		v := w.issue(b.ID, r)
		if v > 0 {
			w.cas(c, func(n *claim) bool {
				if !sameOp(n.Op.ID, op.ID) {
					return false
				}
				n.Op.ReturnFiles = ref{Region: op.SrcRegion, Key: fmt.Sprintf("moves/%s/%s/files", host, op.RollbackID), Ver: v}
				n.Op.ReturnQuiet = w.now + stagingTTL + horizonH // the forward path's staging horizon (§7.3)
				if !on("staging-quiet") {
					n.Op.ReturnQuiet = w.now
				}
				n.Op.Phase = "returned"
				n.Obl = append(n.Obl, w.newObl(n, "park-local", -1, b.ID, "", "", 0))
				return true
			})
			return
		}
		w.loseReturning(b) // the second PUT is in flight and the return is not yet journaled
	}
}

// loseReturning is the biased schedule for model 27 (rev 7.2). In half the
// seeds the cloud stops the target the moment it has begun its dirty
// return and before the return is complete: a return PUT is accepted, or
// still in flight, in the source's region, and the box that would finish
// the lane is gone. Declare-lost then meets a partial return lane, which
// the uniform schedule reached once in 100,000 seeds on rev 7 and never on
// rev 7.1. A stopped box is what the operator's compute fencing verifies
// (§7.10), so it is declared lost on the operator's next call.
func (w *world) loseReturning(b *box) {
	if !w.fail["lostreturn"] || w.rng.Intn(2) != 0 {
		return
	}
	b.Destroyed, b.Procs = true, nil
	w.stats["biased: target stopped mid-return"]++
	w.logf("box %d stopped mid-return (biased schedule)", b.ID)
}

func (w *world) recoverSource(b *box, c *claim) {
	op := c.Op
	if !on("rollback-journal") && !b.mem[op.RollbackID] && op.SrcPhase != "waiting" && op.SrcPhase != "opened" && op.SrcPhase != "stuck" {
		// without a durable journal a restarted source starts over
		w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "claimed"; return sameOp(n.Op.ID, op.ID) })
		b.mem[op.RollbackID] = true
		return
	}
	b.mem[op.RollbackID] = true
	switch op.SrcPhase {
	case "waiting":
		if b.Row != nil && !op.Committed {
			b.Row.Status, b.Row.HoldOp = "restoring", op.RollbackID
		}
		if op.Exec == -1 && op.DirtyKnown && (!op.Dirty || op.Phase == "returned") {
			w.cas(c, func(n *claim) bool {
				if !sameOp(n.Op.ID, op.ID) || n.Op.Exec != -1 {
					return false
				}
				n.Op.Exec, n.Op.SrcPhase = op.Src, "claimed"
				// return to source (claim rev 6.5): the CAS that makes the
				// source's existing installation current again writes its
				// stamp back, from the journal; never the target's F
				if on("return-stamp") {
					n.Holder.Stamp = op.SrcStamp
				}
				// and, for a clean rollback, the source's generation (rev
				// 7.10 §7.14): the target's probe moved holder.generation to
				// its own; the target wrote nothing, and the source's probe
				// restores from the predecessor generation's remote in its
				// own region
				if !op.Dirty && n.Holder.Gen != op.SrcGen {
					n.Holder.Gen = op.SrcGen // "" is the origin generation
				}
				return true
			})
		}
	case "claimed":
		w.holdRow(b, "restoring", op.RollbackID, c)
		next := "verify"
		if op.Dirty {
			next = "install-park"
		}
		w.cas(c, func(n *claim) bool { n.Op.SrcPhase = next; return sameOp(n.Op.ID, op.ID) })
	case "verify":
		if !w.evidence(b, op.RollbackID) {
			return
		}
		ph := "config"
		if b.DB == nil {
			// the source lost its disk since it stopped (R8-2): see the
			// revert's verify above
			w.logf("recover: box %d's database is gone; the probe restores it", b.ID)
			w.stats["rollback: source database lost"]++
		} else if w.digestOf(b) != op.StopDigest {
			ph = "stuck"
			w.logf("recover: digest mismatch %+v vs %+v", w.digestOf(b), op.StopDigest)
		}
		w.cas(c, func(n *claim) bool { n.Op.SrcPhase = ph; return sameOp(n.Op.ID, op.ID) })
	case "install-park":
		if !w.evidence(b, op.RollbackID) {
			return
		}
		if on("dirty-return-needs-admission") && !targetAdmitted(c) {
			// movesinstall.go's parking (fix rounds 3 and 4): the target never
			// admitted this move, so holder.generation is still the pre-move
			// one, the source's live state; there is no residue, and nothing
			// is parked. The mutation parks as if the target had admitted
			w.stats["rollback: dirty return parks nothing, the target never admitted"]++
			w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if w.parkStep(b, c.Holder.Location, fmt.Sprintf("superseded/%s/%s/return/", op.ID, host), op.StopAt+horizonH, w.opScope(c, c.Holder.Location), w.keepFilesLive(b.Region, confine(c.Holder.Location))) {
			w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) })
		}
	case "install-copy":
		sv := w.buckets[b.Region].byID(op.ReturnDB.Key, op.ReturnDB.Ver)
		if sv == nil {
			return // the lane is gone: wait (stuck, in time)
		}
		g := installGen(&op)
		if sv.B.Max == 0 && on("valid-ltx") {
			w.cas(c, func(n *claim) bool {
				n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", 0, "", 0
				return sameOp(n.Op.ID, op.ID)
			})
			return
		}
		key := snapKey(c.Holder.Location, g, sv.B.Max)
		v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: key, srcRegion: b.Region, srcKey: op.ReturnDB.Key, srcVer: op.ReturnDB.Ver})
		if v > 0 || !on("build-completion") {
			next := "install-anchor"
			if !on("build-completion") {
				next = "published"
			}
			w.cas(c, func(n *claim) bool {
				n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, sv.B.Max
				return sameOp(n.Op.ID, op.ID)
			})
		}
	case "install-anchor":
		if w.anchorAcked(b, c.Holder.Location, installGen(&op), op.BaseKey, op.BaseVer, op.BaseTX) {
			w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "published"; return sameOp(n.Op.ID, op.ID) })
		}
	case "published":
		w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "local"; return sameOp(n.Op.ID, op.ID) }) // the marker is the probe's admission's (claim rev 6.9)
	case "local":
		w.installLocal(b, c, installGen(&op), op.ReturnFiles, "", func(n *claim) { n.Op.SrcPhase = "config" })
	case "config":
		b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc
		w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probing"; return sameOp(n.Op.ID, op.ID) })
	case "probing":
		if !w.admits(b, c, op.RollbackID) {
			return
		}
		want := c.Holder.Gen
		if op.Dirty {
			want = installGen(&op) // the return generation prepared from the target's sealed copy
		}
		if hasMain(b) || w.spawn(b, c, want, true) {
			w.probeAccepted(b, c)
			w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probed"; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if b.hostFailed && !b.failCounted {
			b.failCounted = true // a host the retry policy marked failed: one failed try for the operation
		} else if b.open != nil || b.hostFailed {
			return // admitted, its start in progress, or a failed host awaiting the operator: not a failed try
		}
		w.logf("box %d: the rollback's probe did not start (try %d, generation %q, db=%v, host failed %v)", b.ID, op.ProbeTries+1, want, b.DB != nil, b.hostFailed)
		w.cas(c, func(n *claim) bool {
			n.Op.ProbeTries++
			if n.Op.ProbeTries >= probeTries*3 {
				n.Op.SrcPhase = "stuck"
			}
			return sameOp(n.Op.ID, op.ID)
		})
	case "probed":
		if b.open != nil {
			return // the source's session has not opened replication yet (rev 7.11 §7.14)
		}
		if !hasMain(b) {
			w.stats["probe restarted"]++ // the probe died: restarted, admitted like any other (rev 7.11)
			w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probing"; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if op.Dirty && !op.ReturnParked {
			// holder.generation and holder.stamp moved at the probe's cold
			// admission (claim rev 6.9); the return lane's park is owed
			w.cas(c, func(n *claim) bool {
				n.Op.ReturnParked = true
				n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.SrcRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.RollbackID), "", op.ReturnQuiet), stagingBound(op.ReturnQuiet), c.Holder.Inc))
				return sameOp(n.Op.ID, op.ID)
			})
			return
		}
		if w.openRow(b, op.RollbackID) {
			w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "opened"; return sameOp(n.Op.ID, op.ID) })
		}
	case "opened":
		if c.oblFor(op.ID) == 0 {
			w.closeMove(c, op.Src)
		}
	}
}

func (w *world) pushStep(b *box, c *claim) {
	op := c.Op
	loc := c.Holder.Location
	g := installGen(&op) // the generation this push prepares; its probe's cold admission makes it holder.generation (claim rev 6.9)
	switch op.Phase {
	case "stopping":
		w.holdRow(b, "moving", op.ID, c)
		if !w.evidence(b, op.ID) {
			return
		}
		// the push's snapshot: the replica under the final checkpoint, plus
		// the push's own mark, one more acknowledged transaction
		snap, _ := w.snapshotBlobOf(b, c, c.Holder.Gen)
		if snap == nil {
			// nothing to seal: the push is abandoned before it changed
			// anything (rev 7.11)
			w.stats["push abandoned: nothing to seal"]++
			w.logf("push %s abandoned: the box has no database and its remote cannot be restored", op.ID)
			w.release(c, b.ID)
			return
		}
		snap.Gen = g
		w.nextW++
		mark := w.nextW
		w.acks = append(w.acks, ack{W: mark, Inc: c.Holder.Inc, Box: b.ID, Session: c.Holder.Session, Gen: c.Holder.Gen, At: w.now, Kind: "write"})
		w.writeInc[mark] = c.Holder.Inc
		snap.Max++
		snap.Pages[fmt.Sprintf("k%d", mark)] = mark
		snap.TXs = append(snap.TXs, mark)
		w.txs[mark] = tx{W: mark, TX: snap.Max, Key: fmt.Sprintf("k%d", mark), Val: mark}
		w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Phase != "stopping" {
				return false
			}
			n.Op.Phase, n.Op.StopAt, n.Op.PushSnap, n.Op.SealFromDB = "park", w.now, snap, b.DB != nil
			return true
		})
	case "park":
		if w.parkStep(b, loc, fmt.Sprintf("superseded/%s/%s/push/", op.ID, host), op.StopAt+horizonH, w.opScope(c, loc), w.keepFilesLive(b.Region, confine(loc))) {
			w.cas(c, func(n *claim) bool { n.Op.Phase = "install"; return sameOp(n.Op.ID, op.ID) })
		}
	case "install":
		if on("quiet-before-ack") && w.now < c.Quiet[b.Region] {
			return // installs wait out quiet_after (claim §6.2a)
		}
		snap := op.PushSnap
		if b.Files != nil {
			snap.SealKey = b.Files.Key
		}
		key := snapKey(loc, g, snap.Max)
		v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: key, b: snap})
		if v > 0 || !on("build-completion") {
			next := "install-anchor"
			if !on("build-completion") {
				next = "published"
			}
			w.cas(c, func(n *claim) bool {
				n.Op.Phase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, snap.Max
				return sameOp(n.Op.ID, op.ID)
			})
		}
	case "install-anchor":
		if w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) {
			w.cas(c, func(n *claim) bool { n.Op.Phase = "published"; return sameOp(n.Op.ID, op.ID) })
		}
	case "published":
		w.cas(c, func(n *claim) bool { n.Op.Phase = "restore"; return sameOp(n.Op.ID, op.ID) }) // the marker is the probe's admission's (claim rev 6.9)
	case "restore":
		// the probe's admission restores the database (claim rev 6.9); the
		// files are unchanged and carried (F7)
		// every probe renders config and materializes the release for the
		// held row first (the passes skip held rows): RenderHeld, MaterializeHeld
		if !b.HasKey {
			return
		}
		b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc
		w.cas(c, func(n *claim) bool { n.Op.Phase = "probing"; return sameOp(n.Op.ID, op.ID) })
	case "probing":
		if !w.admits(b, c, op.ID) {
			return
		}
		if hasMain(b) || w.spawn(b, c, g, true) {
			w.probeAccepted(b, c)
			w.cas(c, func(n *claim) bool {
				n.Op.Phase = "probed" // holder.generation and holder.stamp moved at the probe's cold admission (claim rev 6.9)
				q := op.StopAt + horizonH
				if q > n.Quiet[b.Region] {
					n.Quiet[b.Region] = q
				}
				return sameOp(n.Op.ID, op.ID)
			})
		}
	case "probed":
		if b.open != nil {
			return // the admitted session's start is not complete: replication is not open yet (rev 7.11 §7.14)
		}
		if !hasMain(b) {
			// the probe died before its session opened replication or in
			// its settle (seed 487 / 30,000: a push released over a dead
			// probe whose marker never landed): the probe restarts, a cold
			// admission like any other
			w.stats["probe restarted"]++
			w.cas(c, func(n *claim) bool { n.Op.Phase = "probing"; return sameOp(n.Op.ID, op.ID) })
			return
		}
		if w.openRow(b, op.ID) || b.Row != nil && b.Row.Status == "ready" {
			w.release(c, b.ID)
		}
	}
}

// sameStep: the operation and every phase the worker branches on are what
// the stale read said (claim C3's revalidation under the lease).
func sameStep(a, b *claim) bool {
	if a == nil || b == nil {
		return a == b
	}
	x, y := a.Op, b.Op
	if !(x.ID == y.ID && x.Kind == y.Kind && x.Phase == y.Phase && x.LandPhase == y.LandPhase &&
		x.SrcPhase == y.SrcPhase && x.RevPhase == y.RevPhase && x.Exec == y.Exec && x.Committed == y.Committed &&
		x.Opened == y.Opened && len(a.Obl) == len(b.Obl) && (a.Holder == nil) == (b.Holder == nil)) {
		return false
	}
	// the writer state is part of the step (claim rev 6.8 §6.4a, R8-2): a
	// stalled worker that read a cold admission in progress must not replay
	// its build once the admission completed and the writer runs, since the
	// build takes stop evidence and replaces the database (seed 10382 /
	// 30,000)
	if a.Holder != nil && (a.Holder.Session != b.Holder.Session || a.Holder.Gen != b.Holder.Gen || a.Holder.Stamp != b.Holder.Stamp) {
		return false
	}
	if a.Holder != nil && a.Holder.WriterBox != b.Holder.WriterBox {
		return false
	}
	return a.Holder == nil || len(a.Holder.Admissions) == len(b.Holder.Admissions)
}

// sweep is claim rev 5's at-rest sweep (§6.2a, CC15): for a host with no
// operation running, each region's box lists the locations of the host's
// retired incarnations and every generation within holder.location other
// than holder.generation, and parks whatever it finds as a further
// retirement event, under a park-sweep obligation it owns. The claim's
// wording leaves out `moves/<host>/`; `sweep-staging` (CC20) adds it.
func (w *world) sweep(b *box, c *claim) {
	holderless := c != nil && c.Holder == nil && c.Op.Kind == "parking" && c.Op.ParkSweep != ""
	if !on("straggler-sweep") || c == nil || !(c.Op.Kind == "" || c.Op.Kind == "serving" || holderless) {
		return // a running holderless sweep admits another region's chores (R14-4); any other operation suppresses sweeps
	}
	for _, o := range c.Obl {
		if o.Kind == "park-sweep" && o.Region == b.Region {
			return // its own sweep is still running
		}
	}
	bk := w.buckets[b.Region]
	// one chore per scope location holding residue (claim rev 6.24 §2,
	// R16-2): the sweepable keys are grouped by the location they belong
	// to, an incarnation's own, the legacy instances/<host> (a level
	// directory or generation directly under the host prefix, never a
	// child incarnation) or moves/<host>; each chore's scope names that
	// location and its archive incarnation. The mutation of
	// scope-membership queues rev 7.16's two chores, instances/<host> and
	// moves/<host>, over the whole prefixes
	var prefixes []string
	for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} {
		for _, k := range bk.keys(p) {
			if !hasNonMarker(bk.objs[k]) || !w.sweepable(c, b.Region, k) {
				continue
			}
			loc := p
			if on("scope-membership") {
				loc = scopeOf(k) + "/"
			}
			if !hasStr(prefixes, loc) {
				prefixes = append(prefixes, loc)
			}
		}
	}
	if len(prefixes) == 0 {
		return
	}
	w.cas(c, func(n *claim) bool {
		joining := n.Holder == nil && n.Op.Kind == "parking" && n.Op.ParkSweep != ""
		if !(n.Op.Kind == "" || n.Op.Kind == "serving" || joining) {
			return false // any other operation running suppresses new sweeps (claim rev 6.15 §6.2a)
		}
		// the chore's owner: on a serving host the session, as for
		// PublishGeneration; on a free host a holderless parking op, {kind:
		// parking, intent: {park_sweep: "v1"}}, no exec and no holder, the
		// only holderless claim that may carry obligations, so it blocks
		// free's successor create until its last acknowledgement frees the
		// claim again (FreeAfterPark; claim rev 6.15 §6.2a, R13-4). The
		// mutation of holderless-park queues rev 7.13's bare "sweep" chores
		// on the free claim. Another region's sweep while the op runs joins
		// it with its own chores (a reading, §19.4)
		op := "sweep"
		switch {
		case n.Holder != nil:
			op = fmt.Sprintf("session:%d", n.Holder.Session)
		case joining:
			op = n.Op.ID
			w.stats["sweep joined a running holderless op"]++
		case on("holderless-park"):
			n.Op = opState{Kind: "parking", ID: w.newID("park"), ParkSweep: "v1", Exec: -1, ReqAt: w.now}
			op = n.Op.ID
			w.stats["holderless sweeps"]++
		}
		for _, p := range prefixes {
			// the scope is persisted at the queuing CAS (claim rev 6.23 §2):
			// the sweeping box's complete listing is the evidence, taken now;
			// the location and its archive incarnation (the location's own,
			// the holder's for the legacy location or the staging sweep, the
			// newest retired on a free host; claim rev 6.24 §2)
			n.Obl = append(n.Obl, w.scoped(obligation{ID: w.newID("ob"), Op: op, Kind: "park-sweep", Region: b.Region, Box: -1, Prefix: p, Ret: "sweep"}, w.now, incOf(n, scopeLocation(p))))
		}
		// CoverParkedPredecessors at the queuing CAS (claim rev 6.21 item 6):
		// the predecessors with a sweepable object, no others
		w.coverParkedPredecessors(n, b.Region, w.sweptPredecessors(n, b.Region, false))
		return true
	})
}

// sweepable: is k, in region r, one of the things the at-rest sweep parks?
func (w *world) sweepable(c *claim, r int, k string) bool {
	if strings.HasSuffix(k, "lock.json") {
		return false
	}
	if strings.HasPrefix(k, "moves/") {
		return on("sweep-staging") // CC20
	}
	h := c.Holder
	if !on("sweep-namespace") {
		// claim rev 5's discovery: retired_incarnations (capped) and the
		// legacy flag. A location older than the cap is never listed
		if h != nil && strings.HasPrefix(k, h.Location) {
			return w.deadGeneration(c, r, k)
		}
		for _, inc := range c.Retired {
			if strings.HasPrefix(k, locationFor(inc, false)) {
				return true
			}
		}
		return w.legacyRetired(c)
	}
	// claim rev 6 §6.2a (R3-4): locations come from the namespace itself.
	// Each child prefix of instances/<host>/ is an incarnation location, or
	// the legacy location (the host prefix, litestream's level layout):
	// residue unless it is holder.location, and within holder.location
	// residue unless it is holder.generation
	if h != nil && locationOf(k) == h.Location {
		return w.deadGeneration(c, r, k)
	}
	return true
}

// locationOf is the location a key under instances/<host>/ belongs to:
// instances/<host>/<incarnation>/ for a child prefix, else the legacy
// location instances/<host>/ (a level directory directly under the host).
func locationOf(k string) string {
	p := "instances/" + host + "/"
	rest := strings.TrimPrefix(k, p)
	if i := strings.Index(rest, "/"); i > 0 {
		child := rest[:i]
		// an incarnation id as a child prefix (the claim's 64 hex
		// characters; the model's decimal ids) is an incarnation location.
		// A level directory (0000 to 0009, litestream's layout) or a
		// generation prefix (g-<op>/) directly under the host belongs to
		// the legacy location, the host prefix itself (claim rev 6.5 §6.2a)
		digits := len(child) > 0 && !isLevelDir(child)
		for _, r := range child {
			if r < '0' || r > '9' {
				digits = false
			}
		}
		if digits {
			return p + child + "/"
		}
	}
	return p
}

// deadGeneration: within holder.location, everything but the live set.
// The live set is {authoritative region, holder.location,
// holder.generation} (claim rev 6.5 §6.2a, R4-3): the generation's
// objects are live only in the authoritative region's bucket, the region
// of the holder's current installation; a matching key in any other
// region is residue. The mutation of live-set-region exempts matching keys
// in every region.
func (w *world) deadGeneration(c *claim, r int, k string) bool {
	h := c.Holder
	if w.filesLive(c, r, k) {
		return false
	}
	e := w.execBox(c)
	authoritative := e != nil && e.Region == r
	if !on("live-set-region") {
		authoritative = true
	}
	// an absent marker is the origin generation "" (R11-1): an adopted
	// host's replica has no generation.json until its first cold admission
	// publishes one, and that admission's unfinished chain protects the
	// origin exactly as the residue oracle reads it
	mg := ""
	v := w.buckets[r].current(genMarkerKey(h.Location))
	if v != nil {
		mg = v.B.Gen
	}
	if v == nil && !on("absent-marker-origin") {
		// the mutation (rev 7.11's sweep): an absent marker is no
		// disagreement, so nothing protects the unfinished chain's origin
		mg = h.Gen
	}
	if authoritative && mg != h.Gen {
		// the marker disagrees with holder.generation. Normal while the
		// latest admission is unfinished: the marker names its predecessor,
		// and both generations are live (claim rev 6.10, R10-2). Any other
		// disagreement: touch nothing under the location, and alert (§6.4)
		chain, ok := w.unfinishedChain(c, mg)
		if ok && on("baseline-live") {
			for _, g := range append(chain, mg) {
				if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, g, "") {
					return false
				}
				if isClosedKey(k) && strings.HasPrefix(k, dataKey(h.Location, g, "closed/")) {
					return false
				}
			}
			if k == genMarkerKey(h.Location) {
				return false
			}
		} else if !w.markerAlerted[h.Location] {
			w.markerAlerted[h.Location] = true
			w.alert("box in region %d: generation.json names %q, the claim %q, and no unfinished admission explains it; the location is left untouched", r, mg, h.Gen)
			return false
		} else {
			return false
		}
	}
	if k == genMarkerKey(h.Location) {
		v := w.buckets[r].current(k)
		return v != nil && (v.B.Gen != h.Gen || !authoritative)
	}
	if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, h.Gen, "") {
		return !authoritative
	}
	if isClosedKey(k) {
		if strings.HasPrefix(k, dataKey(h.Location, h.Gen, "closed/")) {
			return !authoritative // the live generation's certificates, every session's
		}
		return true // a dead generation's certificates
	}
	if isFilesKey(k) {
		// a files manifest or object outside the files live set (region HA §5.6, L13):
		// its generation's, as the data keys are
		if genOf(h.Location, k) == h.Gen {
			return !authoritative
		}
		return true
	}
	return isDataKey(k)
}

// genOf is the generation within loc that key k belongs to: "" for the
// origin layout (level directories and closed.json directly under loc),
// the op id for g-<op>/..., and "-" for a key of no generation (markers,
// the lock).
func genOf(loc, k string) string {
	rest := strings.TrimPrefix(k, loc)
	if strings.HasPrefix(rest, "files/") {
		return "" // a files manifest of the origin generation (region HA §5.6)
	}
	if strings.HasPrefix(rest, "g-") {
		if i := strings.Index(rest, "/"); i > 0 {
			return rest[2:i]
		}
		return "-"
	}
	if isDataKey(k) && genPrefixOf(k) == loc || strings.HasPrefix(rest, "closed/") {
		return ""
	}
	return "-"
}

// legacyRetired: the host once held a legacy incarnation, now retired, so
// the bare namespace outside holder.location is a retired location.
func (w *world) legacyRetired(c *claim) bool {
	for _, inc := range c.Retired {
		if w.legacyIncs[inc] {
			return true
		}
	}
	return false
}

// publishGen writes generation.json, conditional on the marker it
// supersedes: If-Match its version, or If-None-Match when there is none
// (moves rev 7.1 §5.1, model 29). generation.json is a reused key, so an
// earlier publication that completes late would otherwise put an older
// generation back, and every wake would refuse.
func (w *world) publishGen(b *box, loc, gen string, inc int, base int, key string, tx int) int {
	// the marker names the generation's snapshot by key, version id and TXID (R3-1, R4-1)
	// INV-11 (R7-4): generation.json is published only from a completed
	// build: the snapshot version it names exists and the anchor at its
	// TXID is current. A marker that names artifacts still in flight would
	// make the completion predicate true before the artifacts exist
	bk := w.buckets[b.Region]
	if tx > 0 {
		if sv := bk.byID(key, base); sv == nil || sv.Marker {
			w.violate("INV-11: generation.json for %s published before its snapshot %s@%d landed (R7-4)", gen, key, base)
		} else if on("restore-anchor") && bk.current(ltxKey(loc, gen, 0, tx, tx)) == nil {
			w.violate("INV-11: generation.json for %s published before its anchor [%d,%d] landed (R7-4)", gen, tx, tx)
		}
	}
	r := &request{kind: rPut, region: b.Region, key: genMarkerKey(loc), b: &blob{Kind: "gen", Gen: gen, Inc: inc, Base: base, SnapKey: key, TX: tx}}
	if w.fail["markerhold"] && w.rng.Intn(2) == 0 {
		// the biased schedule (rev 7.5): this publication is accepted and
		// completes at its worst moment, once a newer marker is current
		r.hold = true
	}
	if on("generation-conditional") {
		r.ifSet = true
		if cur := w.buckets[b.Region].current(r.key); cur != nil {
			r.ifVer = cur.ID
		}
	}
	return w.issue(b.ID, r)
}

// anchorAcked is §5.1 step 7 inside a build (R6-1, R7-4): the anchor
// [tx, tx], every page of the snapshot the build installed, is PUT into the
// generation and the build goes on only once the PUT is acknowledged. The
// mutation of restore-anchor writes none; the mutation of anchor-acked
// does not wait.
func (w *world) anchorAcked(b *box, loc, gen, snapKey string, snapVer, tx int) bool {
	if !on("restore-anchor") || tx == 0 && on("valid-ltx") {
		return true // no transactions yet: there is no position to anchor (the mutation of valid-ltx writes [0,0], which the encoder refuses)
	}
	sv := w.buckets[b.Region].byID(snapKey, snapVer)
	if sv == nil {
		return false
	}
	if w.buckets[b.Region].current(ltxKey(loc, gen, 0, tx, tx)) != nil {
		return true // a replay finds its anchor current and reuses it
	}
	db := &localDB{Inc: sv.B.Inc, Loc: loc, Gen: gen, Key: sv.B.SealKey, Pos: tx, State: cloneState(sv.B.Pages), Log: append([]int(nil), sv.B.TXs...)}
	acked := w.ackedPut(b, ltxKey(loc, gen, 0, tx, tx), anchorOf(db, sv.B.Inc), false)
	w.stats["restore anchors"]++
	return acked || !on("anchor-acked")
}