package hostlife

import (
	"fmt"
	"regexp"
	"strings"
)

const host = "h"

// ---------- small helpers ----------

func (w *world) agentUp(b *box) bool {
	return b != nil && !b.Destroyed && !b.Frozen && !b.AgentDown && !b.AgentPaused
}

func (w *world) localNow(b *box) int { return w.now + b.Skew }

// readClaim is a fresh read by a box agent (C4). It fails when the box
// cannot reach the control bucket (C6: a failed read never acts).
func (w *world) readClaim(b *box) (*claim, bool) {
	if !w.agentUp(b) || b.Partitioned {
		return nil, false
	}
	c := w.claim.clone()
	if c != nil && c.Holder != nil {
		c.readStamp = b.stampOf(c.Holder.Inc) // the installation stamp this read carries (claim rev 6.5 §6.4)
	}
	b.snap = c.clone()
	return c, true
}

// admissionClaim is what admission sees: fresh, unless the rule is mutated,
// in which case the box's last cached snapshot.
func (w *world) admissionClaim(b *box) (*claim, bool) {
	if on("fresh-admission") {
		return w.readClaim(b)
	}
	// the mutation: admission decides from an older read (a cache, or a
	// goroutine that read before it paused)
	if b.stale != nil && w.agentUp(b) {
		return b.stale.clone(), true
	}
	if b.snap != nil && w.agentUp(b) {
		return b.snap.clone(), true
	}
	return w.readClaim(b)
}

// cas writes the claim if nobody wrote since `old` was read. On an ETag
// conflict it re-reads once and re-applies, but only if the op id is
// unchanged (C2); the mutation drops that condition.
func (w *world) cas(old *claim, mut func(c *claim) bool) bool {
	cur := w.claim
	curVer := 0
	if cur != nil {
		curVer = cur.Ver
	}
	oldVer := 0
	if old != nil {
		oldVer = old.Ver
	}
	base := old
	if curVer != oldVer {
		if old == nil || cur == nil {
			return false
		}
		if on("claim-cas-opid") && cur.Op.ID != old.Op.ID {
			return false
		}
		base = cur
	}
	n := base.clone()
	if n == nil {
		n = &claim{}
	}
	if !mut(n) {
		return false
	}
	n.Ver = curVer + 1
	w.claim = n
	return true
}

func (w *world) isParticipant(c *claim, b *box) bool {
	return c != nil && c.Op.Kind != "serving" && c.Op.Kind != "" && contains(c.Op.Parts, b.ID)
}

func liveProcs(b *box) []*proc {
	if b.Destroyed {
		return nil
	}
	return b.Procs
}

func hasMain(b *box) bool {
	for _, p := range b.Procs {
		if p.Main {
			return true
		}
	}
	return false
}

func setKey(xs []int) string { return intsKey(xs) }

func (w *world) timerDigest(b *box) string {
	if on("timer-canonical") {
		return fmt.Sprintf("%s/%d", setKey(b.Timers), b.Attempt)
	}
	return fmt.Sprintf("%s/%s/%d", setKey(b.Timers), setKey(b.History), b.Attempt)
}

func (w *world) digestOf(b *box) digest {
	d := digest{Valid: true}
	if b.DB != nil {
		d.Writes = sum(b.DB.State)
	}
	if b.Files != nil {
		d.Files = fmt.Sprintf("%d/%d/%d", b.Files.Inc, b.Files.Key, b.Files.Env)
	}
	d.Timers = w.timerDigest(b)
	return d
}

// ---------- stop evidence (claim §6.1) ----------

// evidence kills the host's process tree on b and reports whether stop
// evidence holds. It reads only current system state unless mutated.
func (w *world) evidence(b *box, opID string) bool {
	if b.Destroyed {
		return false
	}
	var keep []*proc
	killed := len(b.Procs)
	for _, p := range b.Procs {
		if on("evidence-whole-tree") {
			continue // cleanup-only reopen kills the cgroup and every identified stray
		}
		if p.Strict || p.Main {
			continue // cgroup kill, and SIGTERM to the main
		}
		keep = append(keep, p) // an escaped descendant survives both
	}
	b.Procs = keep
	if !on("evidence-from-state") {
		// evidence only from this agent's memory: a stop it watched, or a
		// remembered history. After a restart neither exists, and an empty
		// box answers ErrStopUnproven forever (R2-2).
		if b.stopped == nil {
			b.stopped = map[string]bool{}
		}
		if killed > 0 {
			b.stopped[opID] = true
		}
		if !b.stopped[opID] {
			return false
		}
	}
	ok := !on("evidence-whole-tree") || len(b.Procs) == 0
	if ok {
		if b.StopEvidence == nil {
			b.StopEvidence = map[string]bool{}
		}
		b.StopEvidence[opID] = true
	}
	return ok // the mutation certifies an empty cgroup, blind to escapees
}

// ---------- restore and spawn ----------

// restoreCalc is the restore contract as a pure calculation (moves spec
// §5.1 steps 4, 5 and 8; rev 7.7, R6-1): it reads region r's bucket and
// writes nothing, so every oracle may call it. It returns the database the
// contract yields, with Pos the position it restored to, or nil when the
// contract refuses. With restore-by-generation off it is the mutation: the
// newest object under the location.
func (w *world) restoreCalc(r int, inc int, loc, gen string) *localDB {
	return w.restoreWith(r, inc, loc, gen, "", 0, -1)
}

// restorePinned is the restore of an operation's prepared generation before
// its marker is published (rev 7.10 §7.14): the snapshot identity comes
// from the operation's build journal instead of generation.json; tx 0 is
// an empty generation.
func (w *world) restorePinned(r int, inc int, loc, gen, snapKey string, snapVer, tx int) *localDB {
	return w.restoreWith(r, inc, loc, gen, snapKey, snapVer, tx)
}

func (w *world) restoreWith(r int, inc int, loc, gen, pinKey string, pinVer, pinTX int) *localDB {
	bk := w.buckets[r]
	db := &localDB{Inc: inc, Loc: loc, Gen: gen, State: map[string]int{}}
	if !on("restore-by-generation") {
		var best *version
		for _, k := range bk.keys(loc) {
			if !isDataKey(k) {
				continue
			}
			for _, v := range bk.objs[k] {
				if !v.Marker && (best == nil || v.ID > best.ID) {
					best = v
				}
			}
		}
		if best != nil {
			apply(db.State, best.B)
			db.Pos, db.Key = best.B.Max, best.B.SealKey
			db.Log = txsOf(best.B)
			db.TX = w.lineageTX(best.B)
			if best.B.Inc != 0 {
				db.Inc = best.B.Inc
			}
		}
		return db
	}
	prefix := dataKey(loc, gen, "")
	cur := 0
	var applied []*blob
	pinnedTX := 0
	if pinTX >= 0 {
		pinnedTX = pinTX // the operation's build journal pins the root
	} else if gen != "" {
		m := bk.current(genMarkerKey(loc))
		if m == nil || m.B.Gen != gen {
			return nil // the marker does not name this generation: refuse
		}
		pinnedTX = m.B.TX
	}
	// the root: the newest current snapshot of the generation at or past the
	// pinned one (for the origin generation, which has no pinned snapshot,
	// litestream's own rule: the newest snapshot); the pinned version only
	// when no successor is current
	var root *blob
	if on("root-advance") || gen == "" {
		for _, s := range w.currentLTX(bk, prefix, 9) {
			if s.B.Max >= pinnedTX && (root == nil || s.B.Max > root.Max) {
				root = s.B
			}
		}
	}
	if root == nil && gen != "" && pinnedTX > 0 {
		key, ver := pinKey, pinVer
		if pinTX < 0 {
			m := bk.current(genMarkerKey(loc))
			key, ver = m.B.SnapKey, m.B.Base
		}
		snap := bk.byID(key, ver)
		if snap == nil || snap.Marker || snap.B.Max != pinnedTX {
			return nil // the snapshot version is missing or not at the pinned TXID
		}
		root = snap.B
	}
	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
		}
	}
	files, gap := w.plan(bk, prefix, cur)
	for _, f := range files {
		apply(db.State, f)
		cur = f.Max
		applied = append(applied, f)
		if db.Key == 0 {
			db.Key = f.SealKey
		}
		if f.Inc != 0 {
			db.Inc = f.Inc
		}
	}
	if gap && on("restore-gap-closed") {
		return nil // a gap after cur: refuse (litestream: "non-contiguous ltx files")
	}
	if len(applied) == 0 && gen != "" && pinnedTX > 0 {
		return nil
	}
	// the oracle's bookkeeping: the writes this database holds, which its
	// own later snapshots and anchors carry forward
	db.Log = txsOf(applied...)
	db.TX = w.lineageTX(applied...)
	db.Pos = cur
	return db
}

// restoreInit is the activator's cold restore: the calculation above, then
// the restore-to-replication initialization (§5.1 step 7, R5-1, R6-1).
// Litestream's open takes a fresh box's position from the newest remote
// level-0 file, or 0. When the restored position is above that (a
// snapshot-only install), the activator writes the anchor, every page as
// the level-0 file [cur, cur], and opens replication only once that PUT
// is acknowledged (rule anchor-acked); the position is then read again
// from the remote listing, never assumed. A PUT the client has waited
// uploadTimeout steps for is retried with the same bytes. It returns nil
// while the anchor is outstanding: the wake tries again. The mutation of
// restore-anchor writes no anchor; the mutation of anchor-acked opens at
// once with whatever the listing shows.
func (w *world) restoreInit(b *box, inc int, loc, gen string) *localDB {
	db := w.restoreCalc(b.Region, inc, loc, gen)
	if db == nil {
		w.logf("box %d restore refused under %s", b.ID, dataKey(loc, gen, ""))
		w.stats["restore refused"]++
		return nil
	}
	db.Stamp = b.stampOf(inc)
	bk := w.buckets[b.Region]
	prefix := dataKey(loc, gen, "")
	cur := db.Pos
	remote := w.maxL0(bk, prefix)
	if cur > remote && on("restore-anchor") {
		key := ltxKey(loc, gen, 0, cur, cur)
		acked := w.ackedPut(b, key, anchorOf(db, inc), false)
		w.stats["restore anchors"]++
		if !acked && on("anchor-acked") {
			w.logf("box %d waits for its anchor %s to be acknowledged before opening replication", b.ID, key)
			w.stats["anchor waits"]++
			return nil
		}
		remote = w.maxL0(bk, prefix) // durable evidence: what the listing shows now
	}
	db.FirstFull = true // litestream's first sync after a cold initialization is a full upload (fact R5-1c)
	db.RemotePos = remote
	db.Pos = remote
	w.logf("box %d restored %s: TXID %d at position %d (key %d, stamp %d)", b.ID, prefix, db.TX, db.Pos, db.Key, db.Stamp)
	return db
}

// stampOK is admission's local check (claim rev 6.5 §6.4): the worker's
// read of the claim carries the installation stamp active on this box at
// that read; it is admitted only if that is still the active stamp. A
// landing probe refused here by its own installation is the fault R3-3
// named: the oracle records it rather than let the move roll back.
func (w *world) stampOK(b *box, c *claim, probe string) bool {
	if b.admitsStamp(c.Holder.Inc, c.readStamp) {
		return true
	}
	if probe != "" && c.Op.Kind == "landing" && c.Op.Tgt == b.ID && c.Ver == w.claim.Ver {
		w.violate("INV-3: the landing probe of %s on box %d was refused by the box's own installation record (stamp %d, active %d)", c.Op.ID, b.ID, c.readStamp, b.stampOf(c.Holder.Inc))
	}
	return false
}

// bucketGen reads generation.json at loc in b's own bucket: the generation
// and incarnation it names, or origin when absent.
func (w *world) bucketGen(b *box, loc string) (string, int) {
	v := w.buckets[b.Region].current(genMarkerKey(loc))
	if v == nil {
		return "", 0
	}
	return v.B.Gen, v.B.Inc
}

// spawn starts the app on b for the holder incarnation. want is the
// generation the claim expects: the holder's current generation for an
// ordinary wake, or the generation an operation is installing for its
// probe. It returns true if the app answered its readiness path.
func (w *world) spawn(b *box, c *claim, want string, probe bool) bool {
	h := c.Holder
	loc := h.Location
	var adm admitInput // the admission this call decided (claim rev 6.9 §6.4a)
	if b.Old {
		// an old executable: no claim, no markers, no locations; it
		// restores whatever is newest at the legacy root
		if b.DB == nil {
			b.DB = &localDB{Inc: h.Inc, Loc: "instances/" + host + "/", State: map[string]int{}, Epoch: b.Epoch}
			var newest *version
			for _, k := range w.buckets[b.Region].keys("instances/" + host + "/ltx/") {
				if v := w.buckets[b.Region].current(k); v != nil && (newest == nil || v.ID > newest.ID) {
					newest = v
				}
			}
			if newest != nil {
				apply(b.DB.State, newest.B)
				b.DB.Pos, b.DB.Key = newest.B.Max, newest.B.SealKey
				b.DB.Log = txsOf(newest.B)
				b.DB.TX = w.lineageTX(newest.B)
				if newest.B.Inc != 0 {
					b.DB.Inc = newest.B.Inc
				}
			}
		}
	} else {
		if b.actFence != "" {
			// the activator's fence on the host (a probe stop's, ruling 75)
			// refuses every start until its owner lifts it
			w.stats["start refused: the activator's fence"]++
			return false
		}
		if b.foreignFence != "" {
			w.violate("INV-32: box %d started the host while %s's activator fence, which its owner never lifted, closed it", b.ID, b.foreignFence)
			return false
		}
		bg, binc := w.bucketGen(b, loc)
		if on("start-check") && bg != "" && binc != h.Inc {
			w.logf("box %d start check refused: generation.json %q inc %d, claim inc %d", b.ID, bg, binc, h.Inc)
			return false
		}
		// every path that opens replication is admitted here (claim rev 6.9
		// §6.4a, R8-2): an ordinary wake, a committed landing's serving, a
		// probe and a restarted probe, a reverted source, an opened rollback
		// recovery, a push's writer. A writer still closing holds the lease:
		// the wake waits for it
		// the retry policy for failed starts (rev 7.11 §7.14, claim rev
		// 6.10): a host marked failed gets no new session and no further
		// attempt until an operator clears it; a failed start within a
		// session waits out its backoff
		if b.hostFailed {
			w.stats["start refused: host failed"]++
			return false
		}
		if w.now < b.nextStart {
			w.stats["start: backoff"]++
			return false
		}
		if o := b.open; o != nil {
			// an admitted session whose start is in progress continues it
			// without another CAS; an admission that moved on (its session
			// or generation changed) abandons it
			if o.Gen != h.Gen || o.Session != h.Session {
				b.open = nil
			} else {
				return w.resumeOpen(b, c, o)
			}
		}
		// the launch environment: config.env rendered and the release live
		// for this incarnation; until then the agent admits nothing (a
		// config pass has not run yet, not an app failure)
		if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) {
			w.stats["start: environment not ready"]++
			return false
		}
		if b.DB != nil && b.DB.Closing {
			w.closeStep(b)
			if b.DB.Closing {
				w.stats["wake waited for a clean close"]++
				w.logf("box %d: start waits for the previous session's clean close", b.ID)
				return false
			}
		}
		var ok bool
		if adm, ok = w.admitWriter(b, c, want, probe); !ok {
			return false
		}
		// the generation to restore from is holder.generation. generation.json
		// may lag it: an admitted writer that never published (it died between
		// its CAS and its marker, or the marker PUT was not acknowledged, rev
		// 7.10 §7.14). A lagging marker for an operation's prepared generation
		// is published again by op.exec before anything else, from the
		// operation's pinned snapshot; a generation with no file at all is an
		// abandoned fresh one, left to the sweep, and the baseline is the
		// generation the marker names, its parent, which holds the data
		source := h.Gen
		pin := -1 // the marker's generation, through generation.json
		prepared := adm.Mode == "cold" && adm.NewGeneration == want && want != h.Gen
		if prepared {
			// a cold admission into the operation's prepared generation: the
			// marker names the predecessor and the admission publishes the
			// new one; the generation clause has nothing to compare
			bg = h.Gen
		}
		if !on("start-check-claim-gen") {
			// rev 5's start check: the box's own marker decides, so the
			// generation it restores is the one its region's generation.json
			// names, whatever the claim says
			source = bg
			bg = h.Gen
		}
		if bg != h.Gen && on("start-check") {
			switch {
			case (h.WriterBox == b.ID || c.Op.Kind == "restoring" && b.ID == c.Op.Src) && w.ancestorGen(loc, h.Gen, bg):
				// the marker lags behind this box's own admission, or behind
				// the return that named the source's generation again:
				// handled below. Another box's lagging marker is the start
				// check's refusal (the generation clause, J2-R2-8): a stale
				// box whose region holds none of holder.generation's files is
				// not the admitted writer that died between its CAS and its
				// marker
			case w.ancestorGen(loc, bg, h.Gen):
				// the marker names a descendant: a clean rollback returned the
				// claim to the source's generation and abandoned the target's;
				// the marker names the source's generation again, as it did
				// before, so the restore contract reads it (rev 7.10 §7.14)
				if !w.republishMarker(b, loc, h.Gen) {
					return false
				}
				bg = h.Gen
			default:
				w.logf("box %d start check refused: generation.json %q, claim %q", b.ID, bg, h.Gen)
				return false
			}
		}
		if bg != h.Gen && on("start-check") {
			if !w.hasFiles(b, loc, h.Gen) {
				// an empty holder.generation: only an unfinished admission's,
				// journalled, whose marker still names its predecessor, or
				// the predecessor of a chain of them (claim rev 6.10); the
				// baseline is the nearest generation of the chain that holds
				// a file, else the marker's, and every generation of the chain
				// is dead once the new marker lands. Any other empty
				// generation is unexplained missing data: refuse and alert
				chain, ok := w.unfinishedChain(c, bg)
				returned := c.Op.Kind == "restoring" && b.ID == c.Op.Src && h.Gen == c.Op.SrcGen
				if !returned && !ok {
					w.alert("box %d: generation %q is empty and is not the latest admission's unfinished one (marker %q); the start is refused", b.ID, h.Gen, bg)
					w.stats["start refused: unexplained empty generation"]++
					return false
				}
				// the recovery walk (claim rev 6.22 §6.2a, R13-1): newest
				// first over the run's links, a foreign link passed over, a
				// covered local one refused, the nearest local one holding an
				// object the baseline, else the marker's
				run, _ := w.unfinishedRun(c, bg)
				src, ok := w.runBaseline(b, c, run, bg)
				if !ok {
					return false
				}
				source = src
				if src != bg && on("restore-selected-baseline") {
					// a generation of the run that holds an object, selected
					// while the marker still names the run's baseline: restored
					// as the oracle restores it, rooted at [1, 1], never through
					// the marker (R14-1: rev 7.14 selected it and then refused it)
					pin = 0
					w.stats["restore: run generation rooted"]++
				}
				_ = chain
				w.stats["abandoned empty generation skipped"]++
			} else if !w.republishMarker(b, loc, h.Gen) {
				// no earlier version names it: the admitted writer's marker
				// never landed; from the operation's build journal for its
				// prepared generation, else a fresh generation rooted at [1,1]
				ver, key, tx := 0, "", 0
				if h.Gen == installGen(&c.Op) && c.Op.BaseKey != "" {
					ver, key, tx = c.Op.BaseVer, c.Op.BaseKey, c.Op.BaseTX
				}
				if w.publishGen(b, loc, h.Gen, h.Inc, ver, key, tx) <= 0 {
					w.logf("box %d: generation.json for %q not acknowledged", b.ID, h.Gen)
					return false
				}
				w.stats["marker republished"]++
				w.logf("box %d: generation.json published again for %q, which the admitted writer never published", b.ID, h.Gen)
				// the marker names holder.generation now: the dead writer's
				// predecessor, and the chain's, are retired as its own marker
				// stage would have (seed 8528 / 30,000: a warm successor
				// continued the generation and nothing ever retired rec19).
				// The walk over holder.admissions finds them (claim rev 6.14)
				if !w.publishGeneration(b, c, c.Op.ID, phaseOf(&c.Op), h.Session, h.Gen, b.Region, w.newID("ob")) {
					return false // the claim moved: next time
				}
			}
		}
		// the admitting CAS first (claim rev 6.10 §6.4a): before the purge,
		// the restore and the app start, so nothing the app acknowledges is
		// ever outside a session (R10-1). Everything after it is the admitted
		// session's start, finished by the agent's own loop (resumeOpen)
		if !on("admit-before-start") && w.startHook != nil {
			// the mutation (rev 7.10's order): the app is already running
			// and may acknowledge work before the admitting CAS; a crash
			// here leaves the acknowledgement in a database the next warm
			// admission purges silently (R10-1)
			if w.startHook(b) {
				return false // the witness's crash
			}
		}
		if !w.admitCommit(b, c, adm) {
			return false
		}
		nh := w.claim.Holder
		o := &openState{Gen: nh.Gen, Want: want, Session: nh.Session, Mode: adm.Mode, Prepared: prepared, Probe: probe, Source: source, SourcePin: pin, RemoteMax: adm.RemoteMaxTXID, stage: "restore"}
		b.open = o
		return w.resumeOpen(b, c, o)
	}
	return w.startProc(b, c, want, probe, "legacy")
}

// appStart is the app's start on its restored database and its readiness
// (§7.7): what the app sees, the launch environment and the readiness
// answer. It starts the processes on success.
func (w *world) appStart(b *box, c *claim, want string, probe bool, mode string) bool {
	h := c.Holder
	w.foreignWrites(b.DB, h.Inc, fmt.Sprintf("box %d start", b.ID))
	if b.DB.Inc != 0 && b.DB.Inc != h.Inc {
		w.violate("box %d started inc %d over a database of inc %d", b.ID, h.Inc, b.DB.Inc)
	}
	if b.Files != nil && b.Files.Inc != h.Inc {
		w.violate("box %d started inc %d with files (instance.key) of inc %d", b.ID, h.Inc, b.Files.Inc)
	}
	if b.Config != 0 && b.Config != h.Inc {
		w.violate("box %d started inc %d with config.env rendered for inc %d", b.ID, h.Inc, b.Config)
	}
	if b.Files == nil {
		w.nextID++
		b.Files = &filesDir{Inc: h.Inc, Key: w.nextID, Gen: b.DB.Gen, Env: w.nextID, Done: true, Stamp: b.stampOf(h.Inc)}
	}
	if b.DB.Key == 0 {
		b.DB.Key = b.Files.Key // unsealed data (never booted) adopts the key
	}
	if b.DB.Key != b.Files.Key {
		w.violate("INV-4: box %d started inc %d with instance.key %d over data sealed with %d", b.ID, h.Inc, b.Files.Key, b.DB.Key)
		return false
	}
	if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) {
		w.stats["start: environment not ready"]++
		return false // not an app failure: the agent waits for its config pass
	}
	// FailApp is the injected "the target cannot run it" fault: landing probes only
	healthy := !(probe && b.FailApp && (c.Op.Kind == "landing" || c.Op.Kind == "relocating" && b.ID != c.Op.Src)) // and relocation probes (region HA, Task 5)
	// §7.7: the probe checks the launch environment at spawn. The forward
	// path requires the inputs pinned at admission. A recovery probe (the
	// source's revert or rollback) runs on the inputs it rendered now and
	// records them (rev 6, J2-R2-5); the mutation keeps rev 5's rule, which
	// pins recoveries too.
	if probe && c.Op.PinVer != 0 && b.CfgVer != c.Op.PinVer {
		switch {
		case c.Op.Kind == "landing":
			healthy = false
		case (c.Op.Kind == "reverting" || c.Op.Kind == "restoring") && !on("recovery-current-inputs"):
			healthy = false
		}
	}
	if !healthy {
		w.logf("box %d unhealthy start inc %d: cfg=%d live=%s/%d fail=%v kind=%s", b.ID, h.Inc, b.Config, b.Live, b.LiveInc, b.FailApp, c.Op.Kind)
		return false
	}
	return w.startProc(b, c, want, probe, mode)
}

// the retry policy for failed app starts within a session (rev 7.11
// §7.14): up to startAttemptsMax consecutive failures, the next attempt
// after a doubling backoff from startBackoff0 steps; then the host is
// marked failed, alerted, and gets no new sessions until an operator
// clears it. The sum of the backoffs is 15 steps; a start that fails
// five times in a row is the release or the configuration, which only
// an operator or a push changes, not the dependency that was slow.
const (
	startAttemptsMax = 5
	startBackoff0    = 1
)

// resumeOpen advances an admitted session's start (rev 7.11 §7.14, claim
// rev 6.10): purge and restore, app start and readiness, then the PUTs
// before replication opens (a fresh generation's full first upload, the
// marker, a warm restore's anchor), each acknowledged before the next,
// retried with the same bytes after the client's timeout, and every one
// of them in the session's ledger (R10-7). It returns true once replication
// is open. The retire chore for a fresh generation's predecessor is queued
// only once the marker names the new generation (R10-2).
func (w *world) resumeOpen(b *box, c *claim, o *openState) bool {
	cur := w.claim
	if cur == nil || cur.Holder == nil || cur.Holder.Gen != o.Gen || cur.Holder.Session != o.Session || on("exec-writer") && cur.Op.Exec != b.ID {
		// the session moved on, or op.exec was handed to another box by a
		// CAS citing this box's stop evidence (a rollback after the retry
		// policy marked the host failed, seed 25581 / 30,000): the start is
		// abandoned, never resumed
		b.open = nil
		w.stats["open: abandoned"]++
		return false
	}
	h := cur.Holder
	loc := h.Location
	bk := w.buckets[b.Region]
	for {
		switch o.stage {
		case "restore":
			// a box never reuses its local database (claim rev 6.9, R9-1):
			// the purge, error-returning; then the restore from the remote.
			// The mutation of remote-baseline keeps a local database of the
			// generation it opens, as rev 7.9 and today's activator do
			if b.DB != nil && (on("remote-baseline") || b.DB.Inc != h.Inc || b.DB.Gen != o.Gen) {
				if b.PurgeFails {
					b.PurgeFails = false // a transient error: the next attempt's purge succeeds
					w.stats["purge failed: writer not started"]++
					w.logf("box %d: the purge of its local database failed; the start waits", b.ID)
					return false
				}
				b.DB = nil
				b.Parked++
				w.stats["local database purged"]++
			}
			db := b.DB
			if db == nil {
				switch {
				case o.Prepared:
					db = w.restorePinned(b.Region, h.Inc, loc, o.Gen, cur.Op.BaseKey, cur.Op.BaseVer, cur.Op.BaseTX)
				default:
					// the same selection semantics as the oracle's restoreAsNext
					// (R14-1): the marker's generation through generation.json, or
					// a run generation the walk selected, rooted at [1, 1]
					db = w.restoreWith(b.Region, h.Inc, loc, o.Source, "", 0, o.SourcePin)
					if db != nil && o.Mode == "warm" && db.Pos < o.RemoteMax {
						w.alert("box %d: the restore of %q reached position %d, the certificate says %d; the remote is incomplete", b.ID, o.Source, db.Pos, o.RemoteMax)
						db = nil
					}
				}
				if db == nil {
					w.stats["restore refused"]++
					w.logf("box %d: nothing to restore for a %s admission from %q", b.ID, o.Mode, o.Source)
					if o.startedAt == 0 {
						o.startedAt = w.now
					}
					if o.Probe && on("probe-window") && w.now-o.startedAt >= probeStartWindow {
						// a probe attempt whose restore cannot proceed within its
						// window fails like one whose app will not start (§7.7)
						b.open = nil
						w.stats["probe attempt failed: restore"]++
						w.probeAttemptFailed(b, cur)
					}
					return false
				}
				w.logf("box %d restored %s: TXID %d at position %d, %d writes (%s admission into %q)", b.ID, dataKey(loc, o.Source, ""), db.TX, db.Pos, len(db.Log), o.Mode, o.Gen)
			}
			db.Gen, db.Session, db.Stamp, db.Epoch = o.Gen, o.Session, h.Stamp, b.Epoch
			db.Pending, db.Queue, db.InFlight, db.Issued, db.Uncertain, db.Closing, db.CloseReq = nil, nil, nil, nil, false, false, nil
			db.RemotePos, db.FirstFull = -1, true // litestream's first sync after a cold initialization is a full upload (fact R5-1c)
			if b.Files != nil {
				b.Files.Gen, b.Files.Stamp = o.Gen, h.Stamp // the carried files, retagged (§6.4)
			}
			switch {
			case o.Mode == "cold" && o.Prepared:
				db.Pos = w.maxL0(bk, dataKey(loc, o.Gen, "")) // the prepared generation's anchor
			case o.Mode == "cold":
				db.Pos = 0 // a fresh generation: empty until the first upload
			}
			b.DB = db
			o.DB = db
			o.stage = "app"
		case "app":
			// the app starts on the restored database and answers its
			// readiness path; replication is not open yet, so nothing it
			// acknowledges is outside the session (R10-1). A failed start is
			// retried by the same agent under the same session, with backoff,
			// up to the limit
			if b.DB == nil {
				o.stage = "restore" // the database went with a lost disk: restore again
				continue
			}
			if !hasMain(b) {
				if o.startedAt == 0 {
					o.startedAt = w.now
					if o.Probe {
						b.lastProbe = probeTiming{startedAt: w.now} // INV-19 reads it at the probe's acceptance
					}
				}
				late := o.Probe && on("probe-window") && w.now-o.startedAt >= probeStartWindow
				if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) || w.now < o.startedAt+b.envUnreadyFor {
					if late {
						// a probe attempt's readiness window has passed with the
						// environment never ready: the attempt fails (§7.7)
						b.open = nil
						b.startFails, b.nextStart = 0, 0
						w.stats["probe attempt failed: environment"]++
						w.probeAttemptFailed(b, cur)
					}
					return false // the environment: wait, no failure counted
				}
				if late {
					// the deadline is checked before the start (rev 7.13 §7.7,
					// R12-2): an attempt whose window passed while the agent was
					// paused, or between a failed start and its retry, does not
					// start the app at all; a start that succeeded here would
					// have made an expired attempt the admitted writer
					b.open = nil
					b.startFails, b.nextStart = 0, 0
					w.stats["probe attempt failed: deadline before the start"]++
					w.probeAttemptFailed(b, cur)
					return false
				}
				if !w.appStart(b, cur, o.Want, o.Probe, o.Mode) {
					b.startFails++
					b.nextStart = w.now + startBackoff0<<(b.startFails-1)
					w.stats["start failed"]++
					if o.Probe && (b.startFails >= startAttemptsMax || on("probe-window") && w.now-o.startedAt >= probeStartWindow) {
						// a probe attempt's readiness window is the budget's 30 s
						// (§7.3, §7.7): the retry policy's restarts share it, and
						// when the fifth fails or the window passes the attempt
						// fails; the operation counts one of its budgeted tries,
						// and the host is not marked failed by a probe
						b.open = nil
						b.startFails, b.nextStart = 0, 0
						w.stats["probe attempt failed: start"]++
						w.probeAttemptFailed(b, cur)
						return false
					}
					if b.startFails >= startAttemptsMax {
						b.hostFailed, b.failedAt = true, w.now
						b.open = nil // no session continues on a failed host; the operator's clearing leads to a fresh admission
						w.stats["host marked failed"]++
						w.alert("box %d: the host failed to start %d times; marked failed, no new sessions until an operator clears it", b.ID, b.startFails)
					}
					return false
				}
				b.startFails = 0
				if o.Probe {
					// and before readiness is accepted (R12-2): the app answered
					// after the deadline; it is stopped, the attempt fails, and
					// nothing it did is a session's (replication is not open)
					b.lastProbe.readyAt = w.now
					if on("probe-window") && w.now-o.startedAt >= probeStartWindow {
						var keep []*proc
						for _, p := range b.Procs {
							if !p.Main && !p.Strict {
								keep = append(keep, p)
							}
						}
						b.Procs = keep
						b.open = nil
						b.startFails, b.nextStart = 0, 0
						w.stats["probe attempt failed: readiness after the deadline"]++
						w.probeAttemptFailed(b, cur)
						return false
					}
				}
				if on("admit-before-start") && w.startHook != nil {
					if w.startHook(b) { // a witness's background work right after the start, inside the session
						return false // the witness's crash
					}
				}
			}
			db := o.DB
			switch {
			case o.Mode == "cold" && !o.Prepared && (len(db.Log) > 0 || len(db.State) > 0 || len(db.Pending) > 0):
				o.stage = "first"
			case o.Mode == "cold":
				o.stage = "marker"
			case db.Pos > w.maxL0(bk, dataKey(loc, o.Gen, "")) && on("restore-anchor"):
				o.stage = "anchor" // the restored position is above the level-0 maximum: the anchor [N, N] (§5.1 step 7)
			default:
				o.stage = "open"
			}
		case "first", "anchor":
			db := o.DB
			var key string
			if o.blob == nil {
				// the bytes are captured once; a retry carries the same bytes
				// (R10-7). Rev 7.11's first build rebuilt the file at the retry
				// and so carried a write the original did not; when the
				// original landed last it replaced the retry's version and the
				// write, durable in the retry's, was gone from the restore
				// (seed 7574 / 30,000)
				if o.stage == "first" {
					o.blob = w.deltaOf(db, h.Inc)
				} else {
					o.blob = anchorOf(db, h.Inc)
				}
				o.pendAt = len(db.Pending)
			}
			if o.stage == "first" {
				key = ltxKey(loc, o.Gen, 0, 1, 1)
			} else {
				key = ltxKey(loc, o.Gen, 0, db.Pos, db.Pos)
			}
			mk := func() *request { return &request{kind: rPut, region: b.Region, key: key, b: o.blob} }
			clientOK := o.req != nil && o.req.done && o.req.doneAt-o.req.at < uploadTimeout
			if o.req == nil || (!clientOK && w.now-o.req.at >= uploadTimeout) {
				if o.req != nil {
					// the client gave up (a timeout, or a completion it never
					// saw: a lost response, seed 571 / 30,000) and retries the
					// same bytes: the session is uncertain for good (R10-7, R9-5)
					db.Uncertain = true
					w.stats["open: PUT retried"]++
				}
				r := mk()
				if o.stage == "anchor" {
					w.stats["restore anchors"]++
				} else {
					w.stats["first uploads"]++
				}
				v := w.issue(b.ID, r)
				db.track(r, v)
				o.req = r
				if v > 0 {
					o.req = nil
				}
			}
			if o.req != nil && !(o.req.done && o.req.doneAt-o.req.at < uploadTimeout) {
				// the client has not seen success: a server completion after
				// the wait is a lost response, not an acknowledgement
				if o.stage == "anchor" && !on("anchor-acked") {
					// the mutation opens before the anchor is acknowledged
				} else {
					w.stats["open: waits for its "+o.stage]++
					return false
				}
			}
			if cv := bk.current(key); cv != nil {
				w.markDurable(key, cv.B)
			}
			// the writes acknowledged since the capture are still pending
			if o.pendAt <= len(db.Pending) {
				db.Pending = append([]int(nil), db.Pending[o.pendAt:]...)
			}
			if o.stage == "first" {
				db.Pos, db.FirstFull = 1, false
				o.stage = "marker"
			} else {
				o.stage = "open"
			}
			o.req, o.blob, o.pendAt = nil, nil, 0
		case "marker":
			db := o.DB
			bg, _ := w.bucketGen(b, loc)
			if bg != o.Gen {
				ver, k, tx := 0, "", 0
				if o.Prepared {
					ver, k, tx = cur.Op.BaseVer, cur.Op.BaseKey, cur.Op.BaseTX
				}
				if o.req == nil || o.req.done || w.now-o.req.at >= uploadTimeout {
					if o.req != nil && !o.req.done {
						w.stats["open: marker PUT retried"]++
					}
					v := w.publishGen(b, loc, o.Gen, h.Inc, ver, k, tx)
					o.req = w.lastIssued
					if v > 0 {
						o.req = nil
					}
					w.logf("box %d: marker PUT for %q issued (acknowledged %v)", b.ID, o.Gen, v > 0)
				}
				if bg, _ = w.bucketGen(b, loc); bg != o.Gen {
					w.stats["open: waits for the marker"]++
					w.logf("box %d: admitted into %q, waiting for its marker", b.ID, db.Gen)
					return false
				}
			}
			o.req = nil
			// the marker names the new generation: every cold admission,
			// a prepared generation's included, publishes it (claim rev
			// 6.14 §6.4a item 5; rev 7.12 left prepared generations to the
			// operation, R12-3). The walk over holder.admissions covers what
			// this region may retire: the predecessor, the empty
			// generations of a chain of unfinished admissions, an earlier
			// dead writer's predecessor (seed 899 / 30,000); a predecessor
			// whose objects are another region's stays uncovered for that
			// region's whole-location park
			if o.Mode == "cold" {
				if !w.publishGeneration(b, cur, cur.Op.ID, phaseOf(&cur.Op), o.Session, o.Gen, b.Region, w.newID("ob")) {
					return false // the claim moved: next time
				}
			}
			o.stage = "open"
			if w.openYield != nil && w.openYield(b) {
				return false // the marker is published; replication is not open yet
			}
		case "open":
			if !hasMain(b) && b.DB != nil {
				// the app was stopped idle while the start was still
				// finishing its PUTs (seed 2663 / 30,000): the agent's
				// replicator ships what the app wrote, then closes cleanly
				b.open = nil
				w.flush(b)
				b.DB.Closing = true
				w.closeStep(b)
				w.stats["replication opened"]++
				return false
			}
			b.open = nil
			w.stats["replication opened"]++
			return true
		}
	}
}

// queueGenerationRetire is claim rev 6.10's QueueGenerationRetire
// (observedSession, publishedGeneration, region, retireGeneration, newID):
// once generation.json names holder.generation, the admitted writer queues
// the park-generation chore for the generation it replaced, owned by the
// region, under the session's op id "session:<holder.session>". The
// mutation of baseline-live queues it at the admitting CAS instead, as
// rev 7.10 did, before the new generation has a root.
// publishGeneration is claim rev 6.14's PublishGeneration(observedOpID,
// observedPhase, observedSession, publishedGeneration, region, newID)
// (§6.4a item 5): the one CAS that retires generations, made after
// generation.json names the published one, bound to the observed
// operation and phase like the admission, legal while serving and inside
// any operation a writer runs in (R11-5). It walks every uncovered cold
// record of holder.admissions: a predecessor that is the published
// generation is covered with no chore (it is live again: a clean return);
// one with no objects anywhere (predecessor_region "") is covered with no
// chore; one whose objects are this region's gets a deduplicated
// park-generation chore, op = op.id inside an operation and "session:<n>"
// while serving, and is covered; one whose objects are another region's
// stays uncovered, with no chore, for that region's whole-location park
// (CoverRegionPredecessors) or its own publication: a chore queued into
// the source's region from the target's publication would park the
// generation the rollback restores from (seeds 33646, 40148, 1296 / rev
// 7.12; R12-3). The mutation of foreign-region-uncovered queues it there.
// Only predecessors recorded in holder.admissions are ever retired, and
// "" and "@root" are one origin (R11-2). Retention is conservative (claim
// rev 6.14 item 8): covered records are trimmed from the front only, to
// eight, and an uncovered record is never dropped, nor anything behind it.
func (w *world) publishGeneration(b *box, c *claim, observedOpID, observedPhase string, session int, published string, region int, newID string) bool {
	return w.cas(c, func(n *claim) bool {
		if n.Holder == nil || n.Holder.Session != session || n.Holder.Gen != published {
			return false
		}
		named := false
		for _, e := range n.Holder.Admissions {
			if e.Gen == published {
				named = true
			}
		}
		if !named {
			return false // an entry must name the published generation (claim rev 6.22 item 5; any entry, as E reads it)
		}
		if n.Op.ID != observedOpID || phaseOf(&n.Op) != observedPhase {
			return false // bound to the operation and phase it observed
		}
		id := n.Op.ID
		if n.Op.Kind == "serving" || id == "" {
			id = fmt.Sprintf("session:%d", session)
		}
		// the walk is over the uncovered entries up to and including the
		// published one, warm records included (claim rev 6.21 item 5, as
		// E processes them; rev 7.13 skipped warm records): the publishing
		// session is the latest admitted, so no entry lies beyond it
		for i := range n.Holder.Admissions {
			e := &n.Holder.Admissions[i]
			if e.Session > session {
				break // beyond the published entry: the bound, never reached under the CAS's session check
			}
			if e.Covered {
				continue
			}
			switch {
			case e.Predecessor.Gen == published:
				// the predecessor is the live generation again (a clean
				// return's source generation, a republished lagging marker):
				// left uncovered, no chore. Covered here, as claim rev 6.14
				// item 5's first bullet reads, nothing would retire it once a
				// later admission supersedes it, since that admission's
				// predecessor is the last record, not the generation it
				// restored from (seed 41914 / 100,000: the source's
				// republished marker covered the target's record while the
				// source's generation was live again, and the generation was
				// residue at the end; a reading for the claim)
				w.stats["publish: predecessor live, left uncovered"]++
				continue
			case e.PredRegion < 0:
				e.Covered = true // it holds nothing anywhere and no writer can add to it
				w.stats["publish: predecessor empty everywhere"]++
				continue
			case e.PredRegion != region && on("foreign-region-uncovered"):
				w.stats["publish: predecessor left to its region"]++
				continue // uncovered: that region's whole-location park covers it
			}
			chore := region
			if e.PredRegion != region {
				chore = e.PredRegion // the mutation: queued into the other region
				w.stats["publish: chore queued into another region (mutation)"]++
			}
			// an empty dead generation still gets its chore: a dead writer's
			// late upload or certificate may land in it within the chore's
			// horizon (seeds 15273 and 20810 / 30,000)
			ret := e.Predecessor.Gen
			if ret == "" {
				ret = "@root" // the sentinel for the origin generation (claim rev 6.10)
			}
			e.Covered = true
			queued := false
			for _, o := range n.Obl {
				if o.Kind == "park-generation" && o.Region == chore && o.Ret == ret && o.Prefix == n.Holder.Location {
					queued = true // already queued
				}
			}
			if queued {
				continue
			}
			oid := newID
			if hasObl(n.Obl, oid) {
				oid = fmt.Sprintf("%s-%s", newID, ret) // one caller id, one chore per dead generation
			}
			// the chore's scope (claim rev 6.15 §2, R14-5): the publication is
			// the evidence that the predecessor is dead, taken now
			n.Obl = append(n.Obl, w.scoped(obligation{ID: oid, Op: id, Kind: "park-generation", Region: chore, Box: -1, Prefix: n.Holder.Location, Ret: ret, Quiet: w.now + horizonH}, w.now, n.Holder.Inc))
		}
		// retention is E's (claim rev 6.14 item 8, R12-6): covered entries
		// are dropped from the front only while more than eight entries
		// remain; an uncovered head stops the trimming
		for len(n.Holder.Admissions) > 8 && n.Holder.Admissions[0].Covered {
			n.Holder.Admissions = n.Holder.Admissions[1:]
		}
		return true
	})
}

// coverRegionPredecessors is claim rev 6.14's CoverRegionPredecessors
// (§6.4a item 6), composed into the CAS that queues a region's
// whole-location park (the commit's park of the source region, a
// rollback's park of the target region): every uncovered cold record
// whose predecessor's objects are that region's is covered, with no
// chore, since the park retires the whole location there (R12-3, note 5).
// The mutation of cover-at-park leaves them uncovered, and a later
// publication in the region (or none) is their only cover.
func (w *world) coverRegionPredecessors(n *claim, region int, why string) {
	if n.Holder == nil || !on("cover-at-park") {
		return
	}
	for i := range n.Holder.Admissions {
		e := &n.Holder.Admissions[i]
		if !e.Covered && e.PredRegion == region && e.Predecessor.Gen != n.Holder.Gen {
			e.Covered = true
			w.stats["admissions covered at the "+why+" park"]++
		}
	}
}

// sweptPredecessors is the generation set the at-rest sweep parks in a
// region, read from the uncovered records whose predecessors are there:
// at the queuing CAS, the predecessors with a sweepable object; at the
// acknowledgement CAS (retired true), those whose objects it retired on
// the way (had objects, has none). An empty predecessor, with nothing to
// sweep, is in neither set and waits for its publication's chore.
func (w *world) sweptPredecessors(n *claim, region int, retired bool) []string {
	if n.Holder == nil {
		return nil
	}
	bk := w.buckets[region]
	loc := n.Holder.Location
	var gens []string
	for _, e := range n.Holder.Admissions {
		if e.Covered || e.PredRegion != region || e.Predecessor.Gen == n.Holder.Gen || hasStr(gens, e.Predecessor.Gen) {
			continue
		}
		if retired {
			if w.hadFilesIn(region, loc, e.Predecessor.Gen) && !w.hasFilesIn(region, loc, e.Predecessor.Gen) {
				gens = append(gens, e.Predecessor.Gen)
			}
			continue
		}
		for _, k := range bk.keys(dataKey(loc, e.Predecessor.Gen, "")) {
			if genOf(loc, k) == e.Predecessor.Gen && w.sweepable(n, region, k) {
				gens = append(gens, e.Predecessor.Gen)
				break
			}
		}
	}
	return gens
}

// coverParkedPredecessors is claim rev 6.21's CoverParkedPredecessors(op,
// phase, region, generations) (§6.4a item 6): the at-rest sweep covers
// what it parks, selectively, in the CAS that queues its park-sweep and in
// the one that acknowledges it; only the records whose predecessors are
// among the generations it parks in the region are covered, never every
// predecessor in the region.
func (w *world) coverParkedPredecessors(n *claim, region int, gens []string) {
	if n.Holder == nil || !on("cover-at-park") {
		return
	}
	for i := range n.Holder.Admissions {
		e := &n.Holder.Admissions[i]
		if !e.Covered && e.PredRegion == region && e.Predecessor.Gen != n.Holder.Gen && hasStr(gens, e.Predecessor.Gen) {
			e.Covered = true
			w.stats["admissions covered at the sweep park"]++
		}
	}
}

// hadFilesIn: the generation held an object in region r at some time: a
// key under its prefix, whatever versions remain (a parked key keeps its
// entry with none). With hasFilesIn it tells a parked generation from one
// that never had anything (INV-17).
func (w *world) hadFilesIn(r int, loc, gen string) bool {
	for _, k := range w.buckets[r].keys(dataKey(loc, gen, "")) {
		if genOf(loc, k) == gen && !isFilesKey(k) { // a files manifest is no database object (region HA §5.6)
			return true
		}
	}
	return false
}

// hasFilesIn: the generation holds a current object (an LTX file or a
// certificate, not a delete marker) in region r.
func (w *world) hasFilesIn(r int, loc, gen string) bool {
	bk := w.buckets[r]
	for _, k := range bk.keys(dataKey(loc, gen, "")) {
		if genOf(loc, k) == gen && bk.current(k) != nil && !isFilesKey(k) { // a files manifest is no database object (region HA §5.6)
			return true
		}
	}
	return false
}

func hasObl(obl []obligation, id string) bool {
	for _, o := range obl {
		if o.ID == id {
			return true
		}
	}
	return false
}

// hasFiles: the generation holds at least one object (an LTX file or a
// certificate) in b's region; for the origin generation "", the objects
// directly under the location, not those of its g-<id>/ generations.
func (w *world) hasFiles(b *box, loc, gen string) bool {
	return w.hasFilesIn(b.Region, loc, gen)
}

// runBaseline walks an unfinished run in b's region (claim rev 6.22 §6.2a,
// R13-1): each record's predecessor is a link below it, newest first. A
// link whose generation keeps its objects in another region is passed
// over, covered or not (a clean return's fresh source admission after the
// target's park was queued walks R, then the target's T in the other
// region, then the source's S); a link in this region must be uncovered,
// since a covered local generation is retired or being retired here, so
// reaching one as a needed baseline refuses, with an alert. The baseline
// is the newest link in this region that holds an object, else the
// marker's generation (bg, the last link). The mutation of
// covered-baseline-refused ignores the covered flag, as rev 7.13 did, and
// selects by objects alone.
func (w *world) runBaseline(b *box, c *claim, run []admissionRecord, bg string) (string, bool) {
	loc := c.Holder.Location
	for _, r := range run {
		link := r.Predecessor.Gen
		if r.PredRegion != b.Region {
			continue // another region's, or holding nothing anywhere: passed over
		}
		if r.Covered && on("covered-baseline-refused") {
			w.alert("box %d: generation %q, a baseline the restore of %q needs, is covered in this region (retired or being retired); the start is refused", b.ID, link, c.Holder.Gen)
			w.stats["start refused: covered local baseline"]++
			return "", false
		}
		if w.hasFiles(b, loc, link) {
			return link, true
		}
	}
	return bg, true
}

func hasStr(xs []string, s string) bool {
	for _, x := range xs {
		if x == s {
			return true
		}
	}
	return false
}

// restoreAsNext is what the next admission on e would restore (the oracle's
// baseline, rev 7.11): holder.generation when the marker names it, or when
// it holds objects (the admitted writer republishes its marker first, from
// the operation's build journal for a prepared generation); behind an
// explained lagging marker, the nearest generation of the chain that holds
// objects, else the marker's.
func (w *world) restoreAsNext(e *box, c *claim) *localDB {
	h := c.Holder
	loc := h.Location
	bg, _ := w.bucketGen(e, loc)
	if bg == h.Gen {
		return w.restoreCalc(e.Region, h.Inc, loc, h.Gen)
	}
	if w.hasFiles(e, loc, h.Gen) {
		ver, key, tx := 0, "", 0
		if h.Gen == installGen(&c.Op) && c.Op.BaseKey != "" {
			ver, key, tx = c.Op.BaseVer, c.Op.BaseKey, c.Op.BaseTX
		}
		return w.restoreWith(e.Region, h.Inc, loc, h.Gen, key, ver, tx)
	}
	chain, ok := w.unfinishedChain(c, bg)
	if !ok {
		return w.restoreCalc(e.Region, h.Inc, loc, h.Gen) // unexplained: the restore refuses, as the start does
	}
	_ = chain
	run, _ := w.unfinishedRun(c, bg)
	src, fine := w.runBaseline(e, c, run, bg)
	if !fine {
		return nil // the start refuses: a covered local baseline (R13-1)
	}
	if src != bg {
		return w.restoreWith(e.Region, h.Inc, loc, src, "", 0, 0) // a fresh generation rooted at [1, 1]
	}
	return w.restoreCalc(e.Region, h.Inc, loc, bg)
}

// unfinishedRun is unfinishedChain's records: the cold admissions of the
// run, newest first, each one's predecessor the link below it; the last
// record's predecessor is the marker's generation.
func (w *world) unfinishedRun(c *claim, bg string) ([]admissionRecord, bool) {
	h := c.Holder
	if bg == h.Gen {
		return nil, true
	}
	gen := h.Gen
	var run []admissionRecord
	for i := len(c.Holder.Admissions) - 1; i >= 0; i-- {
		e := c.Holder.Admissions[i]
		if e.Gen != gen {
			return nil, false
		}
		if e.Mode != "cold" {
			continue
		}
		run = append(run, e)
		if e.Predecessor.Gen == bg {
			return run, true
		}
		gen = e.Predecessor.Gen
	}
	return nil, false
}

// unfinishedChain explains a marker that lags holder.generation (claim rev
// 6.10, R10-2): the writers journal must show holder.generation to be the
// latest admission's, and every generation between it and the one the
// marker names (bg) to be an earlier cold admission's whose predecessor is
// the next; each died before its marker (seed 28216 / 30,000: two in a
// row, the second's purge failed after its CAS). The chain is returned
// newest first, holder.generation included, bg excluded; ok is false when
// the disagreement is unexplained.
func (w *world) unfinishedChain(c *claim, bg string) ([]string, bool) {
	h := c.Holder
	if bg == h.Gen {
		return nil, true
	}
	gen := h.Gen
	var chain []string
	for i := len(c.Holder.Admissions) - 1; i >= 0; i-- {
		e := &c.Holder.Admissions[i]
		if e.Gen != gen {
			return nil, false // the journal does not explain holder.generation
		}
		if e.Mode != "cold" {
			continue // a warm session of the same generation
		}
		chain = append(chain, gen)
		if e.Predecessor.Gen == bg {
			return chain, true
		}
		gen = e.Predecessor.Gen
	}
	return nil, false
}

// probeAttemptFailed counts a failed probe attempt against the operation's
// budget (§7.3, §7.7): a landing rolls back at probeTries (three), a
// revert or a rollback's source is stuck at three times that (nine, rev
// 7.13 R12-5); the next attempt, if any, is a new cold admission into a
// fresh generation. A probe never marks the host failed (claim rev 6.14):
// the retry limit of §7.14 belongs to a serving wake.
func (w *world) probeAttemptFailed(b *box, c *claim) {
	id := c.Op.ID
	w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, id) {
			return false
		}
		n.Op.ProbeTries++
		switch n.Op.Kind {
		case "reverting":
			if n.Op.ProbeTries >= probeTries*3 {
				n.Op.RevPhase = "stuck"
			}
		case "restoring":
			if n.Op.ProbeTries >= probeTries*3 {
				n.Op.SrcPhase = "stuck"
			}
		}
		return true
	})
	if cur := w.claim; cur != nil && sameOp(cur.Op.ID, id) && cur.Op.Kind == "landing" && cur.Op.ProbeTries >= probeTries {
		w.rollback(cur, "target would not start")
	}
	w.logf("box %d: probe attempt %d failed within its readiness window (%s/%s)", b.ID, w.claim.Op.ProbeTries, c.Op.Kind, id)
}

// resumeStart is the agent's own loop finishing an admitted session's
// start (rev 7.11 §7.14): the activator process holds the admission and
// advances the stages without a wake, within the retry policy's backoff.
// An admission the claim moved past is abandoned.
func (w *world) resumeStart(b *box, c *claim) {
	o := b.open
	if o == nil || c == nil || c.Holder == nil {
		return
	}
	if o.Gen != c.Holder.Gen || o.Session != c.Holder.Session {
		b.open = nil
		w.stats["open: abandoned"]++
		return
	}
	if w.now < b.nextStart || b.hostFailed {
		return // the retry policy's backoff, or a failed host
	}
	w.resumeOpen(b, c, o)
}

// startProc starts the host's processes on b for the admitted writer.
func (w *world) startProc(b *box, c *claim, want string, probe bool, mode string) bool {
	h := c.Holder
	strict := !b.Old && (on("strict-spawn") || w.rng.Intn(2) == 0)
	gen := want
	if b.DB != nil {
		gen = b.DB.Gen
	}
	b.Procs = append(b.Procs, &proc{Main: true, Strict: strict, Inc: h.Inc, Gen: gen, Session: h.Session, Box: b.ID})
	if w.rng.Intn(2) == 0 {
		b.Procs = append(b.Procs, &proc{Main: false, Strict: strict, Inc: h.Inc, Gen: gen, Session: h.Session, Box: b.ID})
	}
	w.logf("box %d spawned inc %d gen %q (%s, probe=%v strict=%v) claim=%s/%s exec=%d", b.ID, h.Inc, gen, mode, probe, strict, c.Op.Kind, c.Op.ID, c.Op.Exec)
	return true
}

// ---------- admission (C4, §19.1 of the moves spec) ----------

func (w *world) admits(b *box, c *claim, probe string) bool {
	if c == nil || c.Holder == nil || b.Row == nil || b.Row.Inc != c.Holder.Inc {
		return false
	}
	if on("lost-quarantine") && contains(c.Lost, b.ID) {
		return false
	}
	if on("exec-admission") && c.Op.Exec != b.ID {
		return false
	}
	if !w.stampOK(b, c, probe) {
		return false // a retired installation's worker (claim rev 6.5 §6.4)
	}
	if w.isParticipant(c, b) && b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil {
		return false // self-fenced until the lease is renewed
	}
	if !w.isParticipant(c, b) {
		b.ExecUntil = 0 // a fresh read with no operation ends the fence (§7.1)
	}
	op := c.Op
	// the admission table's role column: in this model each phase's role
	// is a box id, and it is op.exec in every phase, so exec-admission
	// governs both
	role := func(id int) bool { return !on("exec-admission") || b.ID == id }
	switch op.Kind {
	case "serving":
		return probe == "" && b.Row.Status == "ready"
	case "moving":
		return probe == "" && role(op.Src) && op.CutAt == 0 && b.Row.Status == "ready"
	case "reverting":
		if probe != "" {
			return probe == op.ID && role(op.Src) && op.RevPhase == "probing"
		}
		return role(op.Src) && op.RevPhase == "opened" && b.Row.Status == "ready"
	case "landing":
		if probe != "" {
			return probe == op.ID && role(op.Tgt) && op.LandPhase == "probing"
		}
		return role(op.Tgt) && op.Committed && b.Row.Status == "ready"
	case "restoring":
		if probe != "" {
			return probe == op.RollbackID && role(op.Src) && op.SrcPhase == "probing"
		}
		return role(op.Src) && op.SrcPhase == "opened" && b.Row.Status == "ready"
	case "pushing":
		return probe == op.ID && op.Phase == "probing"
	case "relocating":
		if op.Stuck != "" && on("reloc-stuck-refuses-admission") {
			return false // relocationAdmission: a stuck operation admits nobody, the source's wake included (ruling 9, R1-16; rev 4, Astra R1-2)
		}
		if probe != "" {
			return probe == op.ID && role(op.Tgt) && op.Phase == "probing"
		}
		if role(op.Src) && (op.Phase == "starting" || op.Phase == "releasing" && op.Release == nil) {
			return b.Row.Status == "ready" && b.relFence != op.ID // the source serves until its admission fence closes (§5.3, §5.8)
		}
		return role(op.Tgt) && op.Phase == "opened" && b.Row.Status == "ready"
	}
	return false
}

// certificate is the clean-close certificate of the observed session
// (claim rev 6.9 §6.4a): closed/<holder.session>.json exists for
// holder.generation and its max_txid is the generation's remote maximum
// TXID over every level. Any op.exec box may continue on it (claim rev
// 6.8.2: the certificate proves every upload of the session completed,
// whichever box made them). The mutation of writer-session takes the
// newest certificate of any session (rev 7.8's single marker); the
// mutation of closed-marker ignores max_txid.
func (w *world) certificate(b *box, c *claim) *version {
	h := c.Holder
	bk := w.buckets[b.Region]
	prefix := dataKey(h.Location, h.Gen, "")
	var m *version
	if on("writer-session") {
		m = bk.current(closedKey(h.Location, h.Gen, h.Session))
	} else {
		for _, k := range bk.keys(prefix) {
			if isClosedKey(k) {
				if v := bk.current(k); v != nil && (m == nil || v.ID > m.ID) {
					m = v
				}
			}
		}
	}
	if m == nil {
		w.logf("box %d: no certificate for session %d of %s", b.ID, h.Session, prefix)
		return nil
	}
	if m.B.Seq != b.ID {
		w.stats["certificate of another box"]++ // continued all the same (claim rev 6.8.2)
	}
	if on("closed-marker") && m.B.TX != w.remoteMax(bk, prefix) {
		w.logf("box %d: certificate %d for %s is stale (max_txid %d, remote %d)", b.ID, m.B.Session, prefix, m.B.TX, w.remoteMax(bk, prefix))
		w.stats["stale certificate"]++
		return nil
	}
	return m
}

// installGen is the generation an operation has prepared for its writer: a
// landing's or a push's g-<op id>/, a dirty rollback's return generation.
// Its probe's cold admission makes it holder.generation (claim rev 6.9:
// holder.generation is the only authoritative generation, in every kind
// and phase).
func installGen(op *opState) string {
	switch op.Kind {
	case "restoring":
		return op.RollbackID
	case "relocating":
		return op.Prepared // P, fresh per installation (region HA §5.8)
	}
	return op.ID
}

// republishMarker writes generation.json for gen again, with the content
// its last marker had: a clean rollback's return to the source generation.
func (w *world) republishMarker(b *box, loc, gen string) bool {
	bk := w.buckets[b.Region]
	var last *version
	for _, v := range bk.objs[genMarkerKey(loc)] {
		if !v.Marker && v.B != nil && v.B.Gen == gen && (last == nil || v.ID > last.ID) {
			last = v
		}
	}
	if last == nil && gen != "" {
		w.logf("box %d: no earlier marker names %q", b.ID, gen)
		return false
	}
	inc, base, key, tx := w.claim.Holder.Inc, 0, "", 0
	if last != nil {
		inc, base, key, tx = last.B.Inc, last.B.Base, last.B.SnapKey, last.B.TX
	}
	if w.publishGen(b, loc, gen, inc, base, key, tx) <= 0 {
		w.logf("box %d: generation.json for %q not acknowledged", b.ID, gen)
		return false
	}
	w.stats["marker republished after a clean rollback"]++
	w.logf("box %d: generation.json names %q again after the rollback", b.ID, gen)
	return true
}

// parentGen is the generation gen was admitted from within loc, as newGen
// recorded it; ok is false at the origin.
func (w *world) parentGen(loc, gen string) (string, bool) {
	p, ok := w.genParent[dataKey(loc, gen, "")]
	if !ok {
		return "", false
	}
	if p == loc {
		return "", true
	}
	return strings.TrimSuffix(strings.TrimPrefix(p, loc+"g-"), "/"), true
}

// ancestorGen: anc is gen or one of its ancestors within loc.
func (w *world) ancestorGen(loc, gen, anc string) bool {
	for {
		if gen == anc {
			return true
		}
		p, ok := w.parentGen(loc, gen)
		if !ok {
			return false
		}
		gen = p
	}
}

// phaseOf is the operation phase an admission observes and binds to (R9-4).
func phaseOf(op *opState) string {
	return op.Phase + "/" + op.LandPhase + "/" + op.SrcPhase + "/" + op.RevPhase
}

// admitInput is claim rev 6.9's AdmitWriter input (Admission in the claim library): what the box observed,
// the mode it decided and the generation it opens.
type admitInput struct {
	ObservedOpID         string
	ObservedPhase        string
	ObservedSession      int
	ObservedGeneration   string
	Box                  int
	Mode                 string // warm or cold; "unguarded" is the rev 7.7 mutation
	NewGeneration        string // cold only; "" for warm (claim rev 6.10)
	PredecessorCertified bool
	RemoteMaxTXID        int
	// KeepStamp (claim rev 6.9.1): a cold admission keeps holder.stamp only
	// when no writer was admitted since it was issued (stamp > session)
	KeepStamp bool
	// Coverage (claim rev 6.12) replaces rev 6.9.1's BaselineCoversPredecessor
	// of the stopped predecessor, so nothing is lost and no writes_lost is
	// recorded; the journal entry says baseline: sealed
	// "" (writes_lost unless certified), "sealed" (a from_db seal of the
	// stopped predecessor), "clean-return" (a clean rollback with
	// src_covered: the unchanged target and the source's complete replica
	// establish preservation, R11-4)
	Coverage string
	// Region is where the new writer keeps its generation's objects (its
	// box's region); PredecessorRegion where the predecessor's are, -1 for
	// a predecessor holding nothing anywhere (claim rev 6.14 §6.4a item 1)
	Region            int
	PredecessorRegion int
}

// admitWriter decides the admission for a start on b under the claim it
// read (claim rev 6.9 §6.4a, R8-2, R9-3): it makes no CAS and issues
// nothing. It requires op.exec to be this box (rule exec-writer) and
// refuses every kind in which no writer may open (deleting, parking,
// purging, creating; R9-4). The mode is warm when the certificate of the
// observed session is intact and the generation to open is
// holder.generation; else cold, into the operation's prepared generation
// (want, when it is not holder.generation) or a fresh one.
func (w *world) admitWriter(b *box, c *claim, want string, probe bool) (admitInput, bool) {
	h := c.Holder
	a := admitInput{ObservedOpID: c.Op.ID, ObservedPhase: phaseOf(&c.Op), ObservedSession: h.Session, ObservedGeneration: h.Gen, Box: b.ID, Region: b.Region, PredecessorRegion: w.predecessorRegion(h)}
	if on("exec-writer") && c.Op.Exec != b.ID {
		w.stats["admission refused: not op.exec"]++
		w.logf("box %d: admission refused, op.exec is %d (%s/%s)", b.ID, c.Op.Exec, c.Op.Kind, c.Op.ID)
		return a, false
	}
	switch c.Op.Kind {
	case "creating", "deleting", "parking", "purging", "placing":
		w.stats["admission refused: op kind"]++
		return a, false
	}
	if c.Op.Kind == "relocating" && c.Op.Stuck != "" && on("reloc-stuck-refuses-admission") {
		w.stats["admission refused: relocation stuck"]++
		return a, false
	}
	if c.Op.Kind == "relocating" && !(probe && c.Op.Phase == "probing") && on("reloc-admits-probe-only") {
		// relocatingAdmission (1b-next generation.go): a relocating claim
		// admits only op.exec's cold probe in probing; the source in
		// starting or releasing keeps the session it has and admits no new
		// one (L18 joint round 2 finding 3)
		w.stats["admission refused: relocating admits only the probe"]++
		return a, false
	}
	a.KeepStamp = h.Stamp > h.Session
	if !on("admit-everywhere") && c.Op.Kind != "serving" {
		a.Mode, a.NewGeneration = "unguarded", want // the mutation: writer sessions for ordinary wakes only (rev 7.8)
		if want == "" {
			a.NewGeneration = h.Gen
		}
		return a, true
	}
	bk := w.buckets[b.Region]
	a.RemoteMaxTXID = w.remoteMax(bk, dataKey(h.Location, h.Gen, ""))
	a.PredecessorCertified = w.certificate(b, c) != nil
	if w.completedPrior(b, c) {
		// the first serving wake after a completion (N6) is claimServingOwners'
		// Select: always cold into a fresh generation, PredecessorCertified
		// refused, the predecessor's writes_lost recorded; the probe's stop
		// receipt serves the handover alone (L18 joint round 2 finding 1)
		a.PredecessorCertified = false
	}

	if want == "" || want == h.Gen || w.ancestorGen(h.Location, h.Gen, want) {
		// holder.generation, or an operation's prepared generation that a
		// writer already opened (it is holder.generation or an ancestor of
		// it): never re-entered, since a dead writer's uploads may still
		// land in it (seeds 43 and 87 / 300)
		if a.PredecessorCertified || !on("recovery-generation") {
			a.Mode, a.NewGeneration = "warm", "" // warm: NewGeneration empty (claim rev 6.10)
			return a, true
		}
		a.Mode, a.NewGeneration = "cold", w.newID("rec")
		if l := lastAdmission(h); c.Op.Kind == "restoring" && c.Op.DirtyKnown && !c.Op.Dirty && c.Op.LostAt == 0 && c.Op.SrcCovered && b.ID == c.Op.Src && h.Gen == c.Op.SrcGen &&
			l != nil && (l.Box == c.Op.Tgt || l.Session == c.Op.CutSession) && on("clean-return-identity") {
			// (a return made clean by a declared-lost target is not covered:
			// the target may have written, and its session is declared; and
			// only the target's probe session or the source's own cut
			// session is covered by the seal's src_covered: a later session
			// of the recovered source that dies uncertified is declared as
			// any other, seed 95491 / 100,000)
			// a clean rollback's return: the target wrote nothing and the
			// source's replica covered the stop, so the predecessor's writes
			// are preserved and nothing is declared (claim rev 6.12, R11-4)
			a.Coverage = "clean-return"
		}
		return a, true
	}
	// the operation's prepared generation, opened for the first time; built
	// from a seal of the stopped predecessor (a landing from the source's
	// seal, a dirty return from the target's, a push from its own), so the
	// baseline covers it (claim rev 6.9.1) when the seal was built from the
	// stopped database itself; a seal built from the remote because the
	// disk was gone stops at the remote's position, and the predecessor's
	// acknowledged writes beyond it are declared (rev 7.11; seed 34260 /
	// 100,000)
	covers := c.Op.SealFromDB
	if c.Op.Kind == "restoring" {
		covers = c.Op.ReturnFromDB
	}
	a.Mode, a.NewGeneration = "cold", want
	if c.Op.Kind == "relocating" {
		// released (claim §6.4a item 2, L9): only on the admission that
		// opens P, and only when the predecessor is exactly the box and
		// session whose flushed release this operation journaled and whose
		// final_txid the pin reached; a failover's first admission and every
		// later attempt declare (coverage ""). The mutation of
		// reloc-released-coverage grants it to every first opening of P
		r := c.Op.Release
		predBox, predSession, predGen := h.WriterBox, h.Session, h.Gen // the predecessor's identity as admitCommit journals it
		if l := lastAdmission(h); l != nil {
			predBox, predSession, predGen = l.Box, l.Session, l.Gen
		}
		switch {
		case !on("reloc-released-coverage"):
			a.Coverage = "released"
		case r != nil && r.Flushed && c.Op.Pin != nil && c.Op.Pin.TX >= r.FinalTX && predBox == r.Box && predSession == r.Session && predGen == r.Gen:
			a.Coverage = "released"
		}
		return a, true
	}
	if covers {
		a.Coverage = "sealed"
	}
	return a, true
}

// predecessorRegion is the next admission's predecessor_region (claim rev
// 6.14 item 1): the region of the last admitted writer's box, where its
// generation's objects are; before any record, the adopted or created
// host's writer box, else wherever holder.generation holds an object. It
// is -1, nothing to retire, when it holds none anywhere. A generation
// whose objects were already parked, or that never held one, keeps its
// region: a dead writer's late upload may still land in it, and the chore
// a later publication queues there is what parks it (seeds 15273 and
// 20810 / 30,000).
func (w *world) predecessorRegion(h *holder) int {
	box, gen := h.WriterBox, h.Gen
	if l := lastAdmission(h); l != nil {
		box, gen = l.Box, l.Gen
	}
	for _, b := range w.boxes {
		if b.ID == box && box >= 0 {
			return b.Region
		}
	}
	for x := range w.buckets {
		if w.hasFilesIn(x, h.Location, gen) {
			return x
		}
	}
	return -1
}

// lastAdmission is the newest entry of holder.admissions, or nil: the next
// admission's predecessor (claim rev 6.12).
func lastAdmission(h *holder) *admissionRecord {
	if h == nil || len(h.Admissions) == 0 {
		return nil
	}
	return &h.Admissions[len(h.Admissions)-1]
}

// durableInLineage: W is durable in gen or one of its ancestors within loc.
func (w *world) durableInLineage(loc, gen string, x int) bool {
	seen := map[string]bool{}
	for p := dataKey(loc, gen, ""); p != "" && !seen[p]; p = w.genParent[p] {
		seen[p] = true
		if w.durable[p][x] {
			return true
		}
	}
	return false
}

// alert raises an operator alert (counted; the oracle checks every
// writes_lost record raised one, R10-6).
func (w *world) alert(format string, args ...any) {
	w.alerts++
	w.logf("alert: "+format, args...)
}

// admitCommit is the AdmitWriter CAS (claim rev 6.10 §6.4a), made before
// the purge, the restore and the app start. It is bound to the observed op.id and
// phase (R9-4; rule admit-bind-op), the observed session and generation,
// and op.exec. It writes holder.session, holder.writer_box and the journal
// entry; in cold mode also holder.generation, holder.stamp (the installing
// CAS), a writes_lost record when the observed session had no certificate
// (R9-2), and a park-generation chore for a fresh generation's predecessor.
func (w *world) admitCommit(b *box, c *claim, a admitInput) bool {
	if a.Mode == "unguarded" {
		return true
	}
	h := c.Holder
	var lost *writesLost
	var lostTgt []*writesLost
	var last *admissionRecord // the last admitted writer before this CAS, the predecessor's identity (claim rev 6.12, R11-4)
	var entry admissionRecord
	ok := w.cas(c, func(n *claim) bool {
		if n.Holder == nil || n.Holder.Session != a.ObservedSession || n.Holder.Gen != a.ObservedGeneration {
			return false
		}
		if on("admit-bind-op") {
			// bound to the observed op.id and phase (serving's "" compared
			// literally) and to the kinds in which a writer may open: a
			// closure computed under serving fails once any operation
			// started, a delete included (R9-4)
			if n.Op.ID != a.ObservedOpID || phaseOf(&n.Op) != a.ObservedPhase {
				return false
			}
			switch n.Op.Kind {
			case "creating", "deleting", "parking", "purging", "placing":
				return false
			}
		}
		if on("exec-writer") && n.Op.Exec != b.ID {
			return false
		}
		// the predecessor is the last writer in holder.admissions, with its
		// own session, box and generation; never the generation an
		// operation chose for the restore (claim rev 6.12, R11-4: a clean
		// return resets holder.generation to the source's, but the
		// predecessor is still the target's writer). The mutation takes
		// rev 7.11's holder fields and journals an identity that never
		// admitted a writer
		last = nil
		if l := lastAdmission(n.Holder); l != nil {
			cp := *l
			last = &cp
		}
		p := pred{Session: a.ObservedSession, Box: h.WriterBox, Gen: h.Gen, Stamp: h.Stamp}
		if last != nil && on("clean-return-identity") {
			p = pred{Session: last.Session, Box: last.Box, Gen: last.Gen, Stamp: last.Stamp}
		}
		n.Holder.Session = w.nextGen()
		n.Holder.WriterBox = b.ID
		gen := a.NewGeneration
		if a.Mode != "cold" {
			gen = n.Holder.Gen
		}
		e := admissionRecord{Session: n.Holder.Session, Box: b.ID, Gen: gen, Stamp: n.Holder.Stamp, Mode: a.Mode, Coverage: a.Coverage, Predecessor: p, Region: a.Region, PredRegion: a.PredecessorRegion}
		if a.Mode != "cold" {
			// born covered (claim rev 6.22 item 5): a warm entry's predecessor
			// is its own generation, which nothing retires; the next cold
			// admission's record names it as predecessor and queues its chore
			e.Covered = true
			w.stats["admissions born covered: warm"]++
		}
		if on("cover-at-park") {
			// born covered (claim rev 6.22 item 5, rev 7.13's reading): the
			// claim already holds a whole-location park for the
			// predecessor's region, so the predecessor is being retired there
			// and this record needs no chore. A rollback's return is admitted
			// after the release queued the target region's park, and
			// CoverRegionPredecessors in that CAS could not see this record
			for _, o := range n.Obl {
				// narrowed (claim rev 6.24 item 5): a park scoped exactly to
				// holder.location, or an unscoped park-replica; a scoped park of
				// an old incarnation or of a staging prefix does not count
				exact := o.Kind == "park-superseded" && o.Prefix == n.Holder.Location && (o.Scope == nil || o.Scope.Location == scopeLocation(n.Holder.Location))
				unscopedReplica := o.Kind == "park-replica" && o.Scope == nil
				if (exact || unscopedReplica) && o.Region == e.PredRegion {
					e.Covered = true
					w.stats["admissions born covered: whole-location park pending"]++
					break
				}
			}
		}
		if a.Mode == "cold" {
			w.newGen(n.Holder.Location, h.Gen, a.NewGeneration)
			n.Holder.Gen = a.NewGeneration
			if !a.KeepStamp {
				// a new installation: a new stamp (the installing CAS, claim
				// rev 6.9 §6.4a, §6.4). KeepStamp (claim rev 6.9.1): the stamp
				// stays only when no writer was admitted since it was issued
				// (holder.stamp > holder.session): a landing's first probe
				// after the flip, whose record the target activated before the
				// probe; a restarted probe, a push and a dirty return take a
				// new one
				n.Holder.Stamp = w.issueStamp(n, b.ID)
			}
			e.Stamp = n.Holder.Stamp
			if a.Coverage == "" && !a.PredecessorCertified && p.Box >= 0 && on("writes-lost") {
				// only an uncovered, uncertified predecessor is declared
				// (claim rev 6.12): a sealed or clean-return coverage holds
				// every write it acknowledged. A created incarnation's first
				// admission has no predecessor writer (holder.writer_box is
				// empty before the first writer, claim rev 6.24): nothing is
				// declared, since a declaration always names a box (INV-24)
				lost = &writesLost{Box: p.Box, Gen: p.Gen, Session: p.Session, RemoteMaxTXID: a.RemoteMaxTXID}
				e.Lost = lost
			} else if a.Coverage == "" && !a.PredecessorCertified && p.Box < 0 {
				w.stats["admission: no predecessor writer, nothing declared"]++
			}
			if n.Op.Kind == "restoring" && n.Op.LostAt != 0 && b.ID == n.Op.Src && on("writes-lost") {
				// additional_writes_lost (claim rev 6.22 item 2, R13-3, R14-2):
				// beside the ordinary writes_lost for the predecessor, every
				// session the lost target was admitted into since the cut (a
				// probe session that certified and a warm continuation that
				// did not are two; no return lane carries either's writes,
				// seed 33373 / 100,000), and the source's cut session when
				// src_covered is false (the seal in the lost region was the
				// one holder of its writes beyond the replica, §7.10; seeds
				// 16395 and 27191 / 30,000). The list and the ordinary record
				// are unique by (box, session): a target destroyed before its
				// first admission leaves the source's cut session as the
				// predecessor itself, declared once (R14-2; rev 7.14 declared
				// it twice through a separate field). The mutation of
				// unique-declarations is rev 7.14's separate path
				lostTgt = nil
				add := func(x *writesLost) {
					if on("unique-declarations") {
						if lost != nil && lost.Box == x.Box && lost.Session == x.Session {
							return
						}
						for _, y := range lostTgt {
							if y.Box == x.Box && y.Session == x.Session {
								return
							}
						}
					}
					lostTgt = append(lostTgt, x)
				}
				for _, q := range n.Holder.Admissions {
					if q.Box == n.Op.Tgt && q.Session > n.Op.CutSession {
						add(&writesLost{Box: q.Box, Gen: q.Gen, Session: q.Session, RemoteMaxTXID: a.RemoteMaxTXID})
					}
				}
				if !n.Op.SrcCovered && !n.Op.Dirty && h.Gen == n.Op.SrcGen {
					// the persisted cut-session identity, never a search by generation
					add(&writesLost{Box: n.Op.Src, Gen: n.Op.SrcGen, Session: n.Op.CutSession, RemoteMaxTXID: a.RemoteMaxTXID})
					w.stats["writes_lost: source cut session behind a lost target"]++
				}
				e.AdditionalWritesLost = lostTgt
			}
			// no retire chore here: the predecessor is the baseline of an
			// unfinished admission until generation.json names the new
			// generation; the chore is queued then (claim rev 6.10, R10-2).
			// The mutation of baseline-live queues it now, as rev 7.10 did
			if !on("baseline-live") && a.Coverage != "sealed" && a.Coverage != "released" {
				ret := h.Gen
				if ret == "" {
					ret = "@root"
				}
				n.Obl = append(n.Obl, w.scoped(obligation{ID: w.newID("ob"), Op: fmt.Sprintf("session:%d", n.Holder.Session), Kind: "park-generation", Region: b.Region, Box: -1, Prefix: n.Holder.Location, Ret: ret, Quiet: w.now + horizonH}, w.now, n.Holder.Inc))
			}
		}
		if n.Op.Kind == "relocating" && n.Op.Prepared != "" && gen == n.Op.Prepared {
			n.Op.PreparedOpened = gen // prepared_opened, journaled with the admission that opens P (1b-next rev 12 S1)
		}
		n.Holder.Admissions = append(n.Holder.Admissions, e)
		entry = e
		return true
	})
	if !ok {
		w.logf("box %d: %s admission CAS refused (session %d, op %s/%s exec %d)", b.ID, a.Mode, h.Session, c.Op.Kind, c.Op.ID, c.Op.Exec)
		w.stats["admission CAS refused"]++
		return false
	}
	w.stats["admissions: "+a.Mode]++
	if last != nil && (entry.Predecessor.Session != last.Session || entry.Predecessor.Box != last.Box || entry.Predecessor.Gen != last.Gen) {
		// INV-16 (rev 7.12, R11-4): the journalled predecessor is the last
		// admitted writer; a tuple that never admitted a writer (the
		// target's session and box with the source's generation, after a
		// clean return) is a false record
		w.violate("INV-16: box %d journalled predecessor {session %d, box %d, generation %q}, but the last admitted writer was {session %d, box %d, generation %q}", b.ID, entry.Predecessor.Session, entry.Predecessor.Box, entry.Predecessor.Gen, last.Session, last.Box, last.Gen)
	}
	if a.Coverage != "" {
		w.stats["admissions covered: "+a.Coverage]++
	}
	switch w.claim.Op.Kind {
	case "creating", "deleting", "parking", "purging", "placing":
		// INV-13 (R9-4): no writer is admitted while the host is being
		// created, deleted, parked, purged or placed
		w.violate("INV-13: box %d was admitted as a writer while the host is %s (%s)", b.ID, w.claim.Op.Kind, w.claim.Op.ID)
	}
	if a.Mode == "cold" {
		w.logf("box %d: cold admission %s from generation %q (session %d %scertified, op %s/%s)", b.ID, a.NewGeneration, h.Gen, h.Session, map[bool]string{true: "", false: "un"}[a.PredecessorCertified], c.Op.Kind, c.Op.ID)
	}
	if lost != nil {
		// the writes the predecessor acknowledged and never uploaded are
		// lost, declared and alerted (claim rev 6.9), never silently
		n := w.exemptDeclared(b, lost, h, a.NewGeneration, true)
		w.stats["writes_lost declared"]++
		w.stats["writes declared lost"] += n
		if d := c.Op.Completed; n > 0 && c.Op.Kind == "serving" && d != nil && d.Box == b.ID && lost.Session == d.Session && lost.Box == d.Box {
			// INV-34 (ruling 87): the probe stop flushed every write its
			// session acknowledged, so the first serving wake's writes_lost,
			// bounded by the remote's last shipped transaction, names none
			w.violate("INV-34: box %d's first serving wake after relocation %s declared %d writes the stopped probe session %d acknowledged lost (ruling 87: the probe stop flushes first)", b.ID, d.Op, n, d.Session)
		}
		w.lostAlert++
		w.alert("box %d: writes_lost {box %d, generation %q, session %d, remote_max_txid %d}: %d acknowledged writes", b.ID, lost.Box, lost.Gen, lost.Session, lost.RemoteMaxTXID, n)
	}
	for _, x := range lostTgt {
		// one alert per declaration (claim rev 6.22 item 2)
		n := w.exemptDeclared(b, x, h, a.NewGeneration, true)
		w.stats["writes_lost declared"]++
		w.stats["writes declared lost"] += n
		if x.Box == w.claim.Op.Tgt {
			w.stats["writes_lost: another session of a lost target"]++
		}
		w.lostAlert++
		w.alert("box %d: writes_lost {box %d, generation %q, session %d, remote_max_txid %d}: %d acknowledged writes (additional_writes_lost behind the lost target)", b.ID, x.Box, x.Gen, x.Session, x.RemoteMaxTXID, n)
	}
	if lost != nil || len(lostTgt) > 0 {
		// INV-21 (rev 7.15, R14-2): the declarations of one admission are
		// unique by (box, session), the ordinary record and the list together.
		// INV-24 (claim rev 6.24, R16-3): every declaration names a box; the
		// legacy identity, session 0, comes only with generation "" (@root),
		// the adopted host's pre-claim writer, and is valid in both fields
		seen := map[[2]int]bool{}
		all := append([]*writesLost{}, lostTgt...)
		if lost != nil {
			all = append(all, lost)
		}
		for _, x := range all {
			k := [2]int{x.Box, x.Session}
			if seen[k] {
				w.violate("INV-21: box %d declared {box %d, session %d} twice in one admission (writes_lost and additional_writes_lost)", b.ID, x.Box, x.Session)
			}
			seen[k] = true
			if x.Box < 0 || x.Session == 0 && x.Gen != "" {
				w.violate("INV-24: box %d declared {box %d, generation %q, session %d}: a declaration names a box, and session 0 is the legacy identity in the origin generation only", b.ID, x.Box, x.Gen, x.Session)
			}
			if x.Session == 0 {
				w.stats["writes_lost: the legacy identity (session 0)"]++
			}
		}
	}
	return true
}

// exemptDeclared applies a writes_lost record to the acknowledgements: the
// writes the declared session acknowledged and never uploaded into the
// new lineage are exempt from the loss oracles, since the record and its
// alert declared them (claim rev 6.9, R10-6). It returns how many. With
// inv15 it checks INV-15 for the predecessor's record.
func (w *world) exemptDeclared(b *box, l *writesLost, h *holder, newGen string, inv15 bool) int {
	n := 0
	for i := range w.acks {
		x := &w.acks[i]
		if !w.declaredBy(x, l, h.Inc) || w.durableInLineage(h.Location, newGen, x.W) {
			continue
		}
		if inv15 {
			// a rollback whose target was declared lost has no return lane:
			// what the seal holds in the target's region is the operator's
			// (§7.10), and the source declares what its own replica lacks
			// (seed 8035 / 30,000)
			declaredLost := contains(w.claim.Lost, l.Box) || w.claim.Op.LostAt != 0
			if w.isDurable(x.W) && !declaredLost {
				// INV-15 (rev 7.11): a write the host's remote holds is never
				// declared lost while its writer's box is still with us; the
				// lineage must adopt the generation that holds it (a dirty
				// rollback's return lane), not declare it. A box declared
				// lost may leave such writes behind: the record's
				// remote_max_txid names where an operator finds them
				w.violate("INV-15: box %d declared write %d lost, which is durable in the remote outside the new lineage (box %d's session %d)", b.ID, x.W, l.Box, l.Session)
			}
		}
		x.Exempt = true
		n++
	}
	return n
}

// declaredBy: acknowledgement x is one the writes_lost record covers: a
// write of the host's incarnation on the record's box, acknowledged by
// the writer session the record names, into the generation it names (rev
// 7.13 §14.2.3, R12-1). A declaration for one session exempts nothing
// another session acknowledged: those are undeclared losses still. The
// mutation of session-scoped-loss is rev 7.12's box-wide match.
func (w *world) declaredBy(x *ack, l *writesLost, inc int) bool {
	if x.Box != l.Box || x.Inc != inc || x.Kind != "write" || x.Exempt {
		return false
	}
	if !on("session-scoped-loss") {
		return true
	}
	return x.Session == l.Session && x.Gen == l.Gen
}

// probeAccepted is INV-19 (rev 7.13 §7.7, R12-2), checked where a probe
// is accepted: the attempt whose app this is answered readiness within
// its window of its first start; an attempt that answered at or after
// the deadline is never the probed writer.
func (w *world) probeAccepted(b *box, c *claim) {
	p := b.lastProbe
	if p.startedAt > 0 && p.readyAt >= p.startedAt && p.readyAt-p.startedAt >= probeStartWindow {
		w.violate("INV-19: box %d's probe was accepted %d steps after its first start (%s/%s), beyond the %d-step window", b.ID, p.readyAt-p.startedAt, c.Op.Kind, c.Op.ID, probeStartWindow)
	}
}

// ---------- parking (claim §6.2) ----------

// parkStep advances one park of `prefix` in b's own bucket into dest. It
// returns true when the location is verified empty after the quiet time.
// keep, when set, names keys the park must leave alone.
func (w *world) parkStep(b *box, prefix, dest string, quiet int, scope *parkScope, keep func(string) bool) bool {
	return w.parkDepth(b, prefix, dest, quiet, keep, scope, 0)
}

// member is INV-23's second half (claim rev 6.24 §2; R16-2, R17-1): every
// reference a scoped park archives or deletes, in a manifest it builds
// and in a persisted manifest it replays, must be within the scope. It
// is checked for every reference, whatever rules are on: the listing's
// confinement (rule scope-membership) is the implementation, this is the
// oracle. An out-of-scope reference is a violation and is not acted on.
func (w *world) member(scope *parkScope, id, what string, r ref) bool {
	if scope == nil || inScope(r.Key, scope.Location) {
		return true
	}
	w.violate("INV-23: chore %s (scope %s, incarnation %d) would %s %s@%d, outside its scope", id, scope.Location, scope.Inc, what, r.Key, r.Ver)
	return false
}

func (w *world) parkDepth(b *box, prefix, dest string, quiet int, keep func(string) bool, scope *parkScope, depth int) bool {
	if depth > 8 {
		return false // a park that cannot finish: reported as non-convergence
	}
	bk := w.buckets[b.Region]
	id := dest
	skip := func(k string) bool {
		return (strings.HasSuffix(k, "lock.json") && on("park-skip-lock")) || (keep != nil && keep(k))
	}
	// the parker takes the location's replica lease first (claim §6.2 step 1):
	// its own lock.json, renewed on every visit
	w.issue(b.ID, &request{kind: rPut, region: b.Region, key: prefix + "lock.json", b: &blob{Kind: "lock", Text: fmt.Sprintf("box%d@%d", b.ID, w.now)}})
	manKey := dest + "manifest"
	if !on("retirement-ids") {
		// one destination per operation, shared by every retirement event
		if i := strings.Index(dest, "/"+host+"/"); i >= 0 {
			manKey = dest[:i+len(host)+2] + "manifest"
			dest = dest[:i+len(host)+2]
		}
	}
	man := bk.current(manKey)
	if man == nil {
		// the snapshot (claim rev 4 §6.2 step 2): the archive set is every
		// non-marker version, current and noncurrent; the deletion set is
		// every version, delete markers included. The mutation archives
		// and deletes only current versions, so older ones resurface.
		var archive, dels []ref
		for _, k := range bk.keys(prefix) {
			if skip(k) {
				continue
			}
			vs := bk.objs[k]
			for i, v := range vs {
				current := i == len(vs)-1
				if !on("park-all-versions") && !current {
					continue
				}
				r := ref{Region: b.Region, Key: k, Ver: v.ID}
				if !w.member(scope, id, "delete", r) {
					continue // out of scope: never deleted, never archived
				}
				dels = append(dels, r)
				if !v.Marker {
					archive = append(archive, r)
				}
			}
		}
		// copy each archive-set version by id; a destination keyed by the
		// source version is idempotent, so progress accumulates
		all := true
		for _, r := range archive {
			dk := dest + strings.TrimPrefix(r.Key, "instances/"+host+"/") + fmt.Sprintf("@%d", r.Ver)
			if bk.current(dk) != nil {
				continue
			}
			all = false
			w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: dk, srcRegion: b.Region, srcKey: r.Key, srcVer: r.Ver})
		}
		if !all {
			return false
		}
		// the manifest last, once every copy exists, with both sets
		w.issue(b.ID, &request{kind: rPut, region: b.Region, key: manKey, b: &blob{Kind: "manifest", Refs: archive, Dels: dels}})
		return false
	}
	// replay the manifest's deletion set, by version id, in one batch per
	// visit; a persisted manifest's references are checked against the
	// scope as the fresh set was (R17-1: a manifest naming another
	// incarnation's keys under a valid scope is refused, not replayed)
	issued := false
	for _, r := range man.B.Refs {
		w.member(scope, id, "archive", r)
	}
	for _, r := range man.B.Dels {
		if !w.member(scope, id, "delete", r) || bk.byID(r.Key, r.Ver) == nil {
			continue
		}
		if on("park-by-version") {
			w.issue(b.ID, &request{kind: rDelVer, region: b.Region, key: r.Key, ver: r.Ver})
		} else {
			w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: r.Key})
		}
		issued = true
	}
	if issued {
		return false
	}
	if on("quiet-before-ack") {
		if w.now < quiet {
			return false
		}
		// a straggler written after the snapshot: park it as a new event
		for _, k := range bk.keys(prefix) {
			if skip(k) {
				continue
			}
			for _, v := range bk.objs[k] {
				if !v.Marker {
					return w.parkDepth(b, prefix, dest+"s/", quiet, keep, scope, depth+1)
				}
			}
		}
	}
	return true
}

// localPark retires b's local state for incarnation inc under the host
// guard: a retired marker first, then each artifact renamed aside, one per
// call, so a crash can fall between any two (claim §6.3). It reports
// whether nothing is left.
// localPark is claim rev 6.5's retire-installation (§6.3): the active
// stamp moves into retired_stamps and becomes none, then every artifact is
// parked. It serves a delete, a move away and the quarantine; a landing
// parks its predecessor's leftovers with retire-artifacts instead.
func (w *world) localPark(b *box, inc int) bool {
	if on("hostguard") && b.stampOf(inc) != 0 {
		b.retireStamp(inc) // retire-installation: active moves into retired_stamps (claim rev 6.5 §6.4)
		return false
	}
	switch {
	case b.DB != nil:
		b.DB = nil
	case b.Files != nil:
		b.Files = nil
	case b.Config != 0:
		b.Config = 0
	case b.Live != "":
		b.Live, b.LiveInc = "", 0
	case len(b.Timers) > 0 || len(b.History) > 0 || b.Attempt != 0:
		b.Timers, b.History, b.Attempt = nil, nil, 0
	default:
		b.receipt = nil // a relocation's install journal goes with the installation
		return true
	}
	b.Parked++
	return false
}

// ---------- obligations ----------

func (w *world) ownerOf(o obligation) *box {
	if o.Kind == "park-local" {
		b := w.boxByID(o.Box)
		if b == nil || b.Destroyed {
			return nil
		}
		return b
	}
	return w.regionBox(o.Region)
}

// doObligation runs one step of an obligation owned by b.
func (w *world) doObligation(b *box, c *claim, o obligation) {
	switch o.Kind {
	case "park-local":
		if !w.evidence(b, o.Op) {
			w.logf("box %d: park-local %s waits for stop evidence", b.ID, o.ID)
			return
		}
		inc := 0
		if c.Holder != nil {
			inc = c.Holder.Inc
		}
		if b.DB != nil && b.DB.Inc != 0 {
			inc = b.DB.Inc
		}
		if !w.localPark(b, inc) {
			if w.calm {
				w.logf("box %d: park-local %s parked an artifact of inc %d (stamp %d, db=%v files=%v cfg=%d live=%q)", b.ID, o.ID, inc, b.stampOf(inc), b.DB != nil, b.Files != nil, b.Config, b.Live)
			}
			return // the next artifact on a later step
		}
		w.logf("box %d: park-local %s done", b.ID, o.ID)
		if b.Row != nil && b.Row.HoldOp == o.Op || b.Row != nil && (c.Holder == nil || b.ID != c.Op.Exec) {
			b.Row = nil
		}
	default:
		quiet := o.Quiet
		if o.Scope != nil && !scopeLocationRe.MatchString(o.Scope.Location) {
			// INV-23 (claim rev 6.23 §2, R15-1): a scope's location is exactly
			// one of the grammar's four forms
			w.violate("INV-23: chore %s (%s) carries a scope location %q outside the claim's grammar", o.ID, o.Kind, o.Scope.Location)
		}
		// membership (claim rev 6.24 §2, R16-1, R16-2): a park executes
		// strictly within its scope's namespace. Every key the chore would
		// archive must be in scope (INV-23's second half, checked on the
		// manifest's sources), and under the rule the chore's listing is
		// confined to the scope: the legacy instances/<host> scope excludes
		// child incarnations, an instances scope never touches moves/, a
		// staging lane's scope only its own lane. The mutation of
		// scope-membership is rev 7.16's listing of the whole park prefix
		var member func(string) bool
		if o.Scope != nil && o.Kind != "park-local" {
			loc := o.Scope.Location
			member = func(k string) bool { return inScope(k, loc) }
		}
		if o.Kind == "park-replica" || o.Kind == "park-superseded" || o.Kind == "park-staging" {
			// the owner first proves the region's box runs nothing for the
			// host, when it is not the current executor (a whole-location
			// park; a staging lane's bound is its seal's, persisted)
			if (o.Kind == "park-replica" || o.Kind == "park-superseded") && (c.Op.Exec != b.ID || c.Holder == nil) {
				if !w.evidence(b, o.Op) {
					return
				}
				if w.perRegion > 1 && on("park-evidence-every-box") {
					// every box of the region publishes its own stop evidence
					// (hostops collectEvidence over the inventory); the owner's
					// alone proves nothing about its sibling. The mutation
					// parks on the owner's evidence alone, and a sibling
					// writer's replicator ships into the parked region
					// (seed 819; wParkEvidenceEveryBox)
					for _, x := range w.regionBoxes(b.Region) {
						if x != b && !contains(c.Lost, x.ID) && !x.StopEvidence[o.Op] {
							return
						}
					}
				}
			}
			// a replicator's PUT signed before that evidence can land up to
			// the horizon later: verify only after it (this model's finding).
			// The evidence time is the chore's persisted park scope (claim
			// rev 6.15 §2, R13-4), bound in a CAS before the first step runs,
			// so a retry after an agent restart keeps the original time; the
			// mutation keeps rev 7.13's box-local map, which the restart
			// clears, and the horizon starts again from the restart's clock
			if o.Scope == nil && on("persisted-park-scope") {
				inc := 0
				if c.Holder != nil {
					inc = c.Holder.Inc
				}
				// the chore was queued unscoped (a delete's or adoption's
				// park-replica): the evidence is the stop evidence first in
				// hand, now; a staging lane never comes here under the rule
				scope := &parkScope{Location: scopeLocation(o.Prefix), Inc: inc, EvidenceAt: w.now, RequestLimit: requestLimitH}
				w.cas(c, func(n *claim) bool {
					for i := range n.Obl {
						if n.Obl[i].ID == o.ID && n.Obl[i].Scope == nil {
							n.Obl[i].Scope = scope
							return true
						}
					}
					return false
				})
				w.stats["park scope bound"]++
				return // the next step runs under the persisted scope
			}
			if b.evidAt == nil {
				b.evidAt = map[string]int{}
			}
			if _, ok := b.evidAt[o.ID]; !ok {
				b.evidAt[o.ID] = w.now
			}
			// the horizon after the evidence is H, the signature-age
			// allowance plus the request bound (R14-3): the scope persists the
			// request bound, never the whole horizon. One deadline for every
			// scope, a staging lane's included (claim rev 6.23 §2, R15-1): a
			// listing later than evidence_at + H, the evidence being the
			// persisted original bound, never the cleanup's start; rev 7.15
			// bound a staging park at cleanup and then ignored the bound for
			// the seal's quiet time
			deadline := b.evidAt[o.ID] + horizonH
			if o.Scope != nil {
				deadline = o.Scope.deadline()
			}
			if (on("quiet-before-ack") || o.Kind == "park-staging") && deadline > quiet {
				quiet = deadline
			}
		}
		if o.Kind == "park-generation" && o.Scope != nil && on("quiet-before-ack") {
			// a dead generation's chore under its scope (R14-5): the
			// publication's evidence time plus H, never less than the
			// queued quiet time
			if q := o.Scope.deadline(); q > quiet {
				quiet = q
			}
		}
		if o.Kind != "park-local" && o.Scope == nil && on("persisted-park-scope") {
			// INV-22 (rev 7.15, R14-5): every remote chore runs its steps under
			// a persisted scope, bound at its queuing or before its first step
			w.violate("INV-22: remote chore %s (%s %s) is about to run with no persisted park scope", o.ID, o.Kind, o.Prefix)
		}
		var keep func(string) bool
		if member != nil && on("scope-membership") {
			keep = func(k string) bool { return !member(k) } // confined to the scope
		}
		if o.Kind == "park-sweep" {
			inner := keep
			keep = func(k string) bool { return !w.sweepable(w.claim, b.Region, k) || inner != nil && inner(k) }
			quiet = 0
			if o.Scope != nil && on("quiet-before-ack") {
				// the sweep's scope (claim rev 6.23 §6.2a): a park-sweep is a
				// retirement event, parked at once and acknowledged only after a
				// verification listing later than evidence_at + H, H the
				// signature-age allowance plus the request bound; the straggler
				// pass is that listing; every sweep acknowledgement waits it
				quiet = o.Scope.deadline()
			}
		}
		if o.Kind == "park-generation" {
			// a dead generation within the live location (a cold recovery's
			// old generation, claim rev 6.6 §6.4a): only its own keys
			retire := o.Ret
			if retire == "@root" {
				retire = "" // the origin generation's sentinel (claim rev 6.10)
			}
			inner := keep
			keep = func(k string) bool {
				if genOf(o.Prefix, k) != retire || inner != nil && inner(k) {
					return true
				}
				if w.filesLive(w.claim, b.Region, k) {
					w.markSweepOwed(b.Region, k)
					return true
				}
				return false
			}
		}
		if !w.parkStep(b, o.Prefix, o.Dest(), quiet, o.Scope, keep) {
			return
		}
	}
	deadline := horizonH
	if b.evidAt != nil {
		deadline = b.evidAt[o.ID] + horizonH
	}
	if o.Scope != nil {
		deadline = o.Scope.deadline()
	}
	w.cas(c, func(n *claim) bool {
		for i, x := range n.Obl {
			if x.ID == o.ID {
				if (o.Kind == "park-replica" || o.Kind == "park-superseded") && on("quiet-before-ack") && o.Region >= 0 && deadline > n.Quiet[o.Region] {
					n.Quiet[o.Region] = deadline // quiet_after survives free
				}
				if o.Kind == "park-superseded" && n.Holder != nil && o.Prefix == n.Holder.Location {
					// INV-17 (rev 7.13, R12-3): the whole location in this region
					// is retired now; every record whose predecessor kept its
					// objects here was covered by the CAS that queued the park
					// (item 6) or born covered under it. One still uncovered
					// would wait for a publication in this region that may never
					// come, with a chore for a prefix already parked
					for _, e := range n.Holder.Admissions {
						if !e.Covered && e.PredRegion == o.Region && e.Predecessor.Gen != n.Holder.Gen {
							w.violate("INV-17: the whole-location park %s of region %d retired admission %d's predecessor %q, and the record is uncovered", o.ID, o.Region, e.Session, e.Predecessor.Gen)
							break
						}
					}
				}
				if o.Kind == "park-sweep" && n.Holder != nil {
					// CoverParkedPredecessors at the acknowledgement (claim rev 6.21
					// item 6): what the sweep retired on its way
					w.coverParkedPredecessors(n, o.Region, w.sweptPredecessors(n, o.Region, true))
				}
				n.Obl = append(n.Obl[:i:i], n.Obl[i+1:]...)
				if o.Ret == "legacy" && n.oblFor("adopt") == 0 {
					n.Complete = true // every region has acknowledged
				}
				if n.Holder == nil && n.Op.Kind == "parking" && n.Op.ParkSweep != "" && n.oblFor(n.Op.ID) == 0 {
					// FreeAfterPark (claim rev 6.15 §6.2a): the holderless
					// sweep's last acknowledgement frees the claim again in the
					// same CAS, keeping quiet_after
					n.Op = opState{}
					w.stats["free after park"]++
				}
				return true
			}
		}
		return false
	})
	w.logf("box %d finished obligation %s %s %s", b.ID, o.ID, o.Kind, o.Prefix)
}

func (o obligation) Dest() string {
	if o.Kind == "park-sweep" {
		if o.Scope != nil {
			// the archive names the scope's incarnation (claim rev 6.24 §2;
			// R16-2: rev 7.16's destination ignored it)
			return fmt.Sprintf("parked/sweep/%s/%d/%s/", host, o.Scope.Inc, o.ID)
		}
		return fmt.Sprintf("parked/sweep/%s/%s/", host, o.ID)
	}
	if o.Kind == "park-replica" {
		return fmt.Sprintf("parked/%s/%s/%s/", o.Ret, host, o.ID)
	}
	return fmt.Sprintf("superseded/%s/%s/%s/", o.Op, host, o.ID)
}

func (w *world) newObl(c *claim, kind string, region, boxID int, prefix, ret string, quiet int) obligation {
	return obligation{ID: w.newID("ob"), Op: c.Op.ID, Kind: kind, Region: region, Box: boxID, Prefix: prefix, Ret: ret, Quiet: quiet}
}

// scoped attaches a chore's persisted park scope at its queuing CAS (claim
// rev 6.23 §2): the location is the park prefix without its trailing "/",
// evidenceAt the original evidence or fencing bound the caller cites, inc
// the archive incarnation. Under the mutation of persisted-park-scope the
// chore is queued unscoped and the runner binds the cleanup's own clock,
// as rev 7.13 did.
func (w *world) scoped(o obligation, evidenceAt, inc int) obligation {
	if !on("persisted-park-scope") {
		return o
	}
	o.Scope = &parkScope{Location: scopeLocation(o.Prefix), Inc: inc, EvidenceAt: evidenceAt, RequestLimit: requestLimitH}
	return o
}

// scopeLocation is the scope grammar's form of a park prefix.
func scopeLocation(prefix string) string { return strings.TrimSuffix(prefix, "/") }

// opScope is the scope an operation's own park runs under (a landing's
// install park of the target region's earlier stay, a dirty return's and
// a push's park of what their install supersedes): holder.location, the
// holder's incarnation, the operation's own evidence time (R16-1, R17-1).
func (w *world) opScope(c *claim, prefix string) *parkScope {
	inc := 0
	if c.Holder != nil {
		inc = c.Holder.Inc
	}
	return &parkScope{Location: scopeLocation(prefix), Inc: inc, EvidenceAt: w.now, RequestLimit: requestLimitH}
}

// confine is the keep function that confines a park's listing to its
// scope: a legacy instances/<host> scope excludes child incarnations.
func confine(prefix string) func(string) bool {
	loc := scopeLocation(prefix)
	return func(k string) bool { return !inScope(k, loc) }
}

// scopeOf is the scope location a key belongs to (claim rev 6.24 §2): a
// staging key's is moves/<host> (or its own lane, moves/<host>/<op id>,
// which inScope tests by prefix); an instance key's is its incarnation
// location, or the legacy instances/<host> for a level directory or
// generation directly under the host prefix. A child incarnation never
// belongs to the legacy scope, and nothing under moves/ belongs to an
// instances scope.
func scopeOf(k string) string {
	if strings.HasPrefix(k, "moves/"+host+"/") {
		return "moves/" + host
	}
	return scopeLocation(locationOf(k))
}

// inScope: key k is within the scope location loc.
func inScope(k, loc string) bool {
	if strings.HasPrefix(loc, "moves/"+host+"/") {
		return strings.HasPrefix(k, loc+"/") // one staging lane
	}
	return scopeOf(k) == loc
}

// incOf is the archive incarnation of a scope location (claim rev 6.24
// §2): an incarnation location's own; for the legacy location or the
// staging sweep, the holder's incarnation, else the newest retired.
func incOf(c *claim, loc string) int {
	if strings.HasPrefix(loc, "instances/"+host+"/") {
		n := 0
		fmt.Sscanf(strings.TrimPrefix(loc, "instances/"+host+"/"), "%d", &n)
		if n > 0 {
			return n
		}
	}
	if c.Holder != nil {
		return c.Holder.Inc
	}
	if len(c.Retired) > 0 {
		return c.Retired[len(c.Retired)-1]
	}
	return 0
}

// scopeLocationRe is claim rev 6.23's grammar for a scope's location
// (the model's incarnation ids are decimal, the claim's 64 hex): exactly
// instances/<host>, instances/<host>/<incarnation>, moves/<host>/<op id>
// or moves/<host>, nothing else, no trailing "/".
var scopeLocationRe = regexp.MustCompile(`^(instances/` + host + `(/[0-9]+)?|moves/` + host + `(/[A-Za-z0-9-]+)?)$`)

// stagingBound is a staging lane's evidence_at: the seal bound, the time
// after which no staging writer, its lease holder included, can still
// write; the lane's quiet time is that bound plus H, so the bound is the
// quiet time less H (the staging-quiet mutation's quiet time of "now"
// gives a bound already past).
func stagingBound(quiet int) int { return quiet - horizonH }

func (c *claim) oblFor(op string) int {
	n := 0
	for _, o := range c.Obl {
		if o.Op == op {
			n++
		}
	}
	return n
}

// sameOp is the op.id condition every transition carries (claim C2). The
// mutation drops it everywhere at once, CAS retry and closures alike.
func sameOp(a, b string) bool { return !on("claim-cas-opid") || a == b }