package hostlife

import (
	"fmt"
	"sort"
)

// consolePlan is a multi-step console operation. A console crash drops
// every plan; the claim-recovery pass must then finish or undo the work
// from the claim alone (claim C2).
type consolePlan struct {
	Kind   string // create, delete, purge
	Step   int
	App    string
	OpID   string
	Host   string // "h" or "h2" for a create
	Inc    int
	Region int
	Read   bool // a create's registration re-read has passed
	Stall  int  // the console is paused on this plan until then
	// an orphan cleanup's observations (claim rev 6 §5.1 step 5): the
	// record's ETag and incarnation, the claim read, and the verdict
	RecVer int
	RecInc int
	Snap   *claim
	Orphan bool
	// the claim's ETag as the create read it when it registered: its
	// claim CAS is conditional on it (CC22)
	SeenVer    int
	SeenAbsent bool
}

// doorPlan is one pull-door listing (claim rev 4 §6.5): O1 is the claim
// read it authorized against; it selects versions under O1's location after
// a pause, reads the claim again (O2), and only then signs handles, which a
// file request opens after another pause.
type doorPlan struct {
	Region int
	Loc    string
	Gen    string
	Inc    int    // O1's holder incarnation
	OpID   string // O1's op
	Key    string
	Ver    int
	Signed bool
	Adv    bool // the biased schedule stalls this one until its worst moment
	Until  int  // a stalled step may resume until then (one in four stall long)
}

func (w *world) execBox(c *claim) *box {
	if c == nil {
		return nil
	}
	return w.boxByID(c.Op.Exec)
}

// fixRecord is the recovery owner for the record halves of a move (claim
// C2): the request's `move` field after the claim's intent; the claim half
// of a record-first transition (followRecord); the claim's release after a
// move's record-first close, from last_move (rev 7.23 §7.5, ruling M4: the
// closing box may stop between the halves and never comes back); and, after
// any other release (a push, a relocation, or rev 7.22's claim-first close
// under its mutation), the CAS that clears `move` and names the executor's
// region.
func (w *world) fixRecord() {
	c := w.claim
	if c == nil || c.Holder == nil || w.rec == nil || w.rec.Inc != c.Holder.Inc {
		return
	}
	if moveClose(c.Op) && w.rec.LastMove == c.Op.ID && w.rec.MoveID == "" {
		if c.oblFor(c.Op.ID) == 0 {
			w.stats["recovery: release from last_move"]++
			w.release(c, closeExec(c.Op))
		}
		return
	}
	switch c.Op.Kind {
	case "serving":
		if b := w.execBox(c); b != nil && (w.rec.MoveID != "" || w.rec.Region != b.Region) {
			w.rec.Region, w.rec.MoveID, w.rec.MoveState, w.rec.RollbackID = b.Region, "", "", ""
			w.rec.Ver = w.newVer()
		}
	case "moving":
		if w.rec.MoveID == "" {
			w.rec.MoveID, w.rec.MoveState = c.Op.ID, ""
			w.rec.Ver = w.newVer()
		}
	}
	w.followRecord(c)
}

func (w *world) liveBoxes() []*box {
	var out []*box
	for _, b := range w.boxes {
		if !b.Destroyed {
			out = append(out, b)
		}
	}
	return out
}

// ---------- user operations ----------

// userCreate starts a create as a console plan: registration, the claim
// CAS, the record and serving are separate steps on separate objects, and
// the console may pause or crash between any two (J2-R2-2, J2-R2-7).
func (w *world) userCreate() {
	app := app0
	if w.rng.Intn(2) == 0 {
		app = app1
	}
	if w.rng.Intn(3) == 0 {
		if w.h2.Kind != "" {
			return
		}
		w.nextInc++
		w.console = append(w.console, &consolePlan{Kind: "create", Host: "h2", App: app, OpID: w.newID("cr"), Inc: w.nextInc})
		return
	}
	c := w.claim
	if c != nil && (c.Holder != nil || (c.Op.Kind != "" && c.Op.Kind != "serving")) {
		return
	}
	if w.rec != nil || w.recPend != nil {
		return
	}
	r := w.rng.Intn(nRegions)
	var inc int
	if on("fresh-incarnation") || c == nil || len(c.Retired) == 0 {
		w.nextInc++
		inc = w.nextInc
	} else {
		inc = c.Retired[len(c.Retired)-1] // a caller-supplied CreatedAt repeats an identity
	}
	w.console = append(w.console, &consolePlan{Kind: "create", Host: "h", App: app, OpID: w.newID("cr"), Inc: inc, Region: r})
}

func (w *world) createSteps(p *consolePlan) bool {
	ad := w.admit[p.App]
	abandon := func() bool {
		delete(ad.Reg, p.OpID)
		return true
	}
	// 412 on the claim CAS: something wrote the claim since the read the
	// registration took, a purge fence among the possibilities. The create
	// starts again from step 0, with a new registration and a new read; it
	// never re-reads and retries the CAS alone (claim rev 5.2 §5.1, CC22)
	restart := func() bool {
		delete(ad.Reg, p.OpID)
		w.logf("create %s: claim CAS 412, restarting from step 0", p.OpID)
		w.stats["create restarted on 412"]++
		w.nextInc++
		p.OpID, p.Inc = w.newID("cr"), w.nextInc
		p.Step, p.Read, p.Stall, p.SeenVer, p.SeenAbsent = 0, false, 0, 0, false
		return false
	}
	if e, ok := ad.Reg[p.OpID]; ok && p.Step > 0 {
		// the creating console renews its registration lease (CC16)
		e.Until = w.now + 60
		ad.Reg[p.OpID] = e
	}
	switch p.Step {
	case 0:
		// app admission (claim rev 4 §5.1 step 0): register while open.
		// The mutation is rev 3's check of the purge reservation, a read of
		// another object than the claim the create then writes.
		if on("admission-registration") {
			if ad.Closed {
				return true
			}
			ad.Reg[p.OpID] = regEntry{Host: p.Host, Until: w.now + 60}
			w.logf("create %s registered for app %s (%s)", p.OpID, p.App, p.Host)
		} else if w.purgeRes[p.App] {
			return true
		}
		w.stores[p.App] = true // the app exists, or exists again after a purge
		// the host claim's ETag, read with the registration, or its absence
		// (claim rev 5.2 §5.1 step 0): the claim CAS is conditional on it
		// (CC22), so any purge fence placed after this, even one since
		// released, refuses the create
		if p.Host == "h2" {
			p.SeenVer = w.h2Epoch
		} else if w.claim == nil {
			p.SeenAbsent = true
		} else {
			p.SeenVer = w.claim.Ver
		}
		p.Step = 1
		return false
	case 1:
		if !p.Read {
			// the re-read, the last step before the claim CAS (claim rev
			// 5.1 §5.3): stop if the registration is gone or admission is
			// closed. Cheap, and not sufficient: the console can pause
			// between this read and the CAS
			if _, ok := ad.Reg[p.OpID]; on("admission-registration") && on("registration-check") && (!ok || ad.Closed) {
				return abandon()
			}
			p.Read = true
			if w.fail["pausecreate"] && w.rng.Intn(2) == 0 {
				// the biased schedule: the console pauses here, past the
				// registration's expiry, while a purge may run
				p.Stall = w.now + 100 + w.rng.Intn(200)
				w.logf("create %s: console paused before its claim CAS", p.OpID)
			}
			return false
		}
		if p.Host == "h2" {
			if w.h2.Kind != "" {
				return abandon()
			}
			if on("fence-etag") && w.h2Epoch != p.SeenVer {
				return restart() // If-Match the epoch read at registration (If-None-Match: * when it was 0)
			}
			w.h2 = liteHost{Inc: p.Inc, App: p.App, Kind: "creating", OpID: p.OpID, At: w.now, Box: w.h2Box(p.App)}
			w.h2Epoch++ // every write changes the claim's ETag
			w.logf("create %s: second host claimed for app %s", p.OpID, p.App)
			p.Step = 2
			return false
		}
		c := w.claim
		if c != nil && c.Holder == nil && c.Op.Kind == "parking" && c.Op.ParkSweep != "" {
			return false // the holderless sweep blocks free's successor create until FreeAfterPark (claim rev 6.15 §6.2a)
		}
		if c != nil && (c.Holder != nil || (c.Op.Kind != "" && c.Op.Kind != "serving")) || w.recPend != nil {
			return abandon()
		}
		if on("fence-etag") && (p.SeenAbsent != (c == nil) || c != nil && c.Ver != p.SeenVer) {
			// If-Match the ETag read at registration (If-None-Match: * if
			// the claim was absent then): a fence placed and released
			// meanwhile changed it, and the 412 restarts the create
			return restart()
		}
		if c != nil && len(c.Obl) > 0 && on("holderless-park") {
			return false // a pending chore finishes first (CC11); since rev 7.14 a free host's sweep is the parking op above
		}
		if w.rec != nil {
			// the record key is occupied: an orphan is removed by the
			// cleanup of claim rev 6 §5.1 step 5, run by this console as
			// its own plan, and the create retries (CC17); anything else
			// refuses the create
			if c != nil && w.orphanOf(c, w.rec.Inc) && on("cancelled-creates") {
				if !w.hasPlan("orphan", "create") {
					w.logf("create %s: cleaning up an orphan record inc %d", p.OpID, w.rec.Inc)
					w.console = append(w.console, &consolePlan{Kind: "orphan", Host: "create"})
				}
				return false
			}
			return abandon()
		}
		// the executor, chosen once and recorded in the intent (region HA
		// §5.2, H12, L8; claim P5-D6 selected_executor)
		b := w.chooseBox(p.Region, p.App)
		if b == nil || (c != nil && contains(c.Lost, b.ID)) {
			return abandon()
		}
		if on("fresh-incarnation") && c != nil && contains(c.Retired, p.Inc) {
			return abandon()
		}
		w.serial++
		if !w.cas(c, func(n *claim) bool {
			if len(n.Obl) > 0 {
				// INV-20 (rev 7.14, R13-4): a create never claims a host while a
				// park of its namespace is outstanding; the holderless sweep's
				// op blocks it until FreeAfterPark
				w.violate("INV-20: create %s claimed the host while %d obligation(s) were outstanding (%s %s)", p.OpID, len(n.Obl), n.Obl[0].Kind, n.Obl[0].Prefix)
			}
			n.Holder = &holder{Inc: p.Inc, App: p.App, Serial: w.serial, Location: locationFor(p.Inc, false), WriterBox: -1} // no writer yet
			n.Op = opState{Kind: "creating", ID: p.OpID, Exec: -1, NewInc: p.Inc, NewApp: p.App, NewRegion: p.Region, NewBox: b.ID, ReqAt: w.now}
			n.Regions, n.Boxes, n.Complete, n.Obl = []int{p.Region}, []int{b.ID}, true, nil
			return true
		}) {
			return abandon()
		}
		w.logf("create %s inc %d app %s in region %d", p.OpID, p.Inc, p.App, p.Region)
		p.Step = 2
		return false
	case 2:
		if p.Host == "h2" {
			if w.h2.OpID != p.OpID || w.h2.Kind != "creating" {
				return abandon()
			}
			w.h2.Rec = true
			w.h2Epoch++ // every write changes the claim's ETag
			p.Step = 3
			return false
		}
		c := w.claim
		if c == nil || c.Op.ID != p.OpID || c.Op.Kind != "creating" {
			return abandon()
		}
		if w.rec == nil && w.recPend == nil {
			rec := &record{Ver: w.newVer(), Inc: c.Holder.Inc, App: c.Holder.App, Region: c.Op.NewRegion}
			switch n := w.rng.Intn(10); {
			case n < 7:
				w.rec = rec
			case n < 9:
				// the response is lost and the PUT lands later, rarely after
				// cancelled_creates' window
				at := w.now + 1 + w.rng.Intn(2*horizonH)
				if w.rng.Intn(beyondH) == 0 {
					at = w.now + 4*horizonH + w.rng.Intn(10*horizonH)
				}
				w.recPend = &pendingRec{rec: rec, at: at}
			}
			// otherwise it failed outright; the next step retries
		}
		p.Step = 3
		return false
	default:
		if p.Host == "h2" {
			if w.h2.OpID != p.OpID || w.h2.Kind != "creating" {
				return abandon()
			}
			w.h2.Kind = "serving" // a CAS from creating, like the main host's
			w.h2Epoch++           // every write changes the claim's ETag
			w.logf("create %s: second host serving", p.OpID)
			delete(ad.Reg, p.OpID)
			return true
		}
		c := w.claim
		if c == nil || c.Op.ID != p.OpID || c.Op.Kind != "creating" {
			return abandon()
		}
		if w.rec == nil || w.rec.Inc != c.Holder.Inc {
			if w.rec == nil && w.recPend == nil {
				p.Step = 2 // publish again
			}
			return false
		}
		b := w.createExecutor(c)
		if b == nil {
			return false
		}
		if w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, p.OpID) {
				return false
			}
			n.Op = opState{Kind: "serving", Exec: b.ID, Parts: []int{b.ID}}
			n.Holder.Stamp = w.issueStamp(n, n.Op.Exec) // create step 4: holder.stamp is this CAS's gen (claim rev 6.5 §2)
			return true
		}) {
			delete(ad.Reg, p.OpID)
			return true
		}
		return false
	}
}

func (w *world) userDelete() {
	c := w.claim
	if c == nil || c.Holder == nil {
		return
	}
	if on("claim-op-exclusive") && c.Op.Kind != "serving" {
		return
	}
	id := w.newID("del")
	if e := w.execBox(c); e != nil {
		w.lastRegion = e.Region
	}
	if w.fail["recreate"] && c.Op.Kind == "serving" {
		w.doorAuthorize() // a door listing that will race the delete and a recreate
		if e := w.execBox(c); e != nil && e.DB != nil && hasMain(e) && w.rng.Intn(2) == 0 {
			// the replicator's last PUT before the delete's stop evidence:
			// accepted, then completing after the host is recreated
			bl := snapshotOf(e.DB, e.DB.Inc)
			bl.Kind, bl.Level = "ltx", 0 // a level-0 file covering everything so far
			w.pending = append(w.pending, &request{kind: rPut, region: e.Region, key: ltxKey(e.DB.Loc, e.DB.Gen, 0, 1, bl.Max), b: bl,
				doneAt: w.now + 40*horizonH, beyond: true, by: fmt.Sprintf("box:%d", e.ID)})
		}
	}
	recVer := 0
	if w.rec != nil {
		recVer = w.rec.Ver
	}
	if w.cas(c, func(n *claim) bool {
		if w.completionBarrier(n, w.verifiedCompletionFence(n)) {
			return false // StartGuard's completion barrier (ruling 77), a verified fence exempt (ruling 86)
		}
		n.Op = opState{Kind: "deleting", ID: id, Exec: -1, ReqAt: w.now, RecVer: recVer} // intent: the record's ETag (claim §5.2 step 1)
		return true
	}) {
		w.console = append(w.console, &consolePlan{Kind: "delete", OpID: id})
		w.logf("delete %s inc %d", id, c.Holder.Inc)
	}
}

func (w *world) userMove() {
	c := w.claim
	if c == nil || c.Holder == nil || c.Op.Kind != "serving" {
		return
	}
	if src := w.execBox(c); src == nil || src.Destroyed {
		return
	}
	w.moveTo(w.rng.Intn(nRegions)) // drawn only when a move is possible, as before rev 7.3
}

// h2Box is the executor the second host's create names (region HA §5.2):
// choose() in region 0, where the model keeps the control-plane-only
// host. With one box per region it is box 0, as the model always implied.
func (w *world) h2Box(app string) int {
	if b := w.chooseBoxFor("h2", 0, app); b != nil {
		return b.ID
	}
	return 0
}

// createExecutor is the box a create's step 4 hands op.exec to: the one
// its intent named, never another (region HA §5.2, claim P5-D2: recovery
// never substitutes a newly available box); nil while that box is gone.
// The mutation of create-choice-once chooses again, as rev 7.22's
// region-wide pick did.
func (w *world) createExecutor(c *claim) *box {
	if w.perRegion <= 1 {
		return w.regionBox(c.Op.NewRegion) // the one-box shape: today's region-wide pick, unchanged (spec §4; rev 4, Astra R1-1)
	}
	if !on("create-choice-once") {
		return w.chooseBox(c.Op.NewRegion, c.Holder.App)
	}
	b := w.boxByID(c.Op.NewBox)
	if b == nil || b.Destroyed {
		return nil
	}
	return b
}

// moveTo requests a move of the serving host to region r. A witness calls
// it directly (witness_test.go).
func (w *world) moveTo(r int) {
	c := w.claim
	if c == nil || c.Holder == nil || c.Op.Kind != "serving" {
		return
	}
	src := w.execBox(c)
	if src == nil || src.Destroyed {
		return
	}
	if r == src.Region {
		return
	}
	tgt := w.chooseBox(r, c.Holder.App) // the target box, chosen in the start CAS (region HA L15)
	if tgt == nil || contains(c.Lost, tgt.ID) || tgt.Old {
		return
	}
	if on("obligations-block-ops") {
		for _, o := range c.Obl {
			if o.Region == r || o.Box == tgt.ID || o.Box == src.ID || o.Region == src.Region {
				return // an operation never starts over a pending chore in its regions
			}
		}
	}
	id := w.newID("mv")
	if w.cas(c, func(n *claim) bool {
		if n.Op.Kind != "serving" || w.completionBarrier(n, -1) {
			return false
		}
		n.Op = opState{Kind: "moving", ID: id, Phase: "leases", Exec: src.ID, Parts: []int{src.ID, tgt.ID},
			Src: src.ID, Tgt: tgt.ID, SrcRegion: src.Region, TgtRegion: r, ReqAt: w.now, Auto: true, App: c.Holder.App,
			SrcGen: c.Holder.Gen} // the source's generation, which a clean rollback's return names again (rev 7.10)
		return true
	}) {
		w.logf("move %s inc %d from box %d to box %d", id, c.Holder.Inc, src.ID, tgt.ID)
	}
}

func (w *world) userPush() {
	c := w.claim
	if c == nil || c.Holder == nil || c.Op.Kind != "serving" {
		return
	}
	b := w.execBox(c)
	if b == nil || b.FailApp || b.Terminated { // a terminated executor is fenced and failed over, never pushed (region HA §5.9)
		return
	}
	if on("obligations-block-ops") {
		for _, o := range c.Obl {
			if o.Region == b.Region || o.Box == b.ID {
				return
			}
		}
	}
	id := w.newID("push")
	if w.cas(c, func(n *claim) bool {
		if n.Op.Kind != "serving" || w.completionBarrier(n, -1) {
			return false
		}
		n.Op = opState{Kind: "pushing", ID: id, Phase: "stopping", Exec: b.ID, Parts: []int{b.ID}, Src: b.ID, Tgt: b.ID, SrcRegion: b.Region, TgtRegion: b.Region, ReqAt: w.now}
		return true
	}) {
		w.logf("push %s on box %d", id, b.ID)
	}
}

func (w *world) userPurge() {
	c := w.claim
	app := app0
	if c != nil && c.Holder != nil {
		app = c.Holder.App
	}
	for _, p := range w.console {
		if p.Kind == "purge" {
			return
		}
	}
	w.console = append(w.console, &consolePlan{Kind: "purge", App: app})
	w.logf("purge app %s", app)
}

func (w *world) parkingObligations(n *claim, inc int) {
	regions := append([]int(nil), n.Regions...)
	var boxes []int
	boxes = append(boxes, n.Boxes...)
	if !n.Complete && on("touched-incomplete") {
		regions = nil
		for r := 0; r < nRegions; r++ {
			regions = append(regions, r)
		}
		for _, b := range w.liveBoxes() {
			boxes = addUniq(boxes, b.ID)
		}
	}
	// the deleted incarnation's own location (the legacy instances/<host>
	// for a legacy host, which excludes child incarnations; claim rev
	// 6.24 §2) and the staging namespace, each its own chore; an older
	// incarnation's residue is the at-rest sweep's (R3-4). The mutation of
	// scope-membership parks the whole host prefix, as rev 7.16 did
	loc := "instances/" + host + "/"
	if n.Holder != nil && on("scope-membership") {
		loc = n.Holder.Location
	}
	for _, r := range regions {
		for _, p := range []string{loc, "moves/" + host + "/"} {
			n.Obl = append(n.Obl, w.newObl(n, "park-replica", r, -1, p, fmt.Sprint(inc), 0))
		}
	}
	for _, id := range boxes {
		if b := w.boxByID(id); b != nil && !b.Destroyed {
			n.Obl = append(n.Obl, w.newObl(n, "park-local", -1, id, "", fmt.Sprint(inc), 0))
		}
	}
	n.Retired = retire(n.Retired, inc)
}

// consoleStep runs the next step of one console plan, chosen at random:
// the console's requests run concurrently.
func (w *world) consoleStep() {
	if len(w.console) == 0 {
		return
	}
	i := w.rng.Intn(len(w.console))
	if w.console[i].Stall > w.now {
		return // this request's console goroutine is paused
	}
	if w.runPlan(w.console[i]) {
		w.console = append(w.console[:i:i], w.console[i+1:]...)
	}
}

func (w *world) runPlan(p *consolePlan) bool {
	c := w.claim
	switch p.Kind {
	case "create":
		return w.createSteps(p)
	case "delete":
		if c == nil || c.Op.ID != p.OpID {
			return true
		}
		return w.deleteSteps(c)
	case "purge":
		return w.purgeSteps(p)
	case "orphan":
		return w.orphanSteps(p)
	}
	return true
}

// hasPlan: a console plan of this kind, by this owner, is in flight.
func (w *world) hasPlan(kind, owner string) bool {
	for _, p := range w.console {
		if p.Kind == kind && p.Host == owner {
			return true
		}
	}
	return false
}

// orphanSteps is the orphan cleanup of claim rev 6 §5.1 step 5 (R3-2) as
// a console plan, so the console can pause between its observations and
// its delete: read the record (ETag E, incarnation I); read the claim
// fresh, afterwards; it is an orphan only if that claim names neither I
// as holder nor a creating operation for I; DeleteIfMatch(E). The
// mutation reads the claim first and the record second, and decides on
// the stale claim: a successor that claimed and published in between is
// then deleted.
func (w *world) orphanSteps(p *consolePlan) bool {
	switch p.Step {
	case 0:
		if on("orphan-order") {
			if w.rec == nil {
				return true
			}
			p.RecVer, p.RecInc = w.rec.Ver, w.rec.Inc
		} else {
			p.Snap = w.claim.clone()
		}
		p.Step = 1
		return false
	case 1:
		if on("orphan-order") {
			p.Orphan = w.orphanOf(w.claim, p.RecInc) // fresh, after the record
		} else {
			if w.rec == nil {
				return true
			}
			p.RecVer, p.RecInc = w.rec.Ver, w.rec.Inc
			p.Orphan = w.orphanOf(p.Snap, p.RecInc)
		}
		p.Step = 2
		return false
	default:
		if !p.Orphan || w.rec == nil || w.rec.Ver != p.RecVer {
			return true // 412, or nothing to do
		}
		if c := w.claim; c != nil && (c.Holder != nil && c.Holder.Inc == w.rec.Inc || c.Op.Kind == "creating" && c.Op.NewInc == w.rec.Inc) {
			w.violate("claim/record: orphan cleanup deleted the record of incarnation %d, which the claim names", w.rec.Inc)
		}
		w.logf("orphan cleanup (%s): removed record inc %d", p.Host, w.rec.Inc)
		w.rec = nil
		return true
	}
}

// deleteSteps: record DeleteIfMatch on the version the intent names, then
// parking with obligations, then free once every obligation is
// acknowledged (claim §5.2). Under a purge (claim rev 6.5 §5.3 step 4,
// R4-4) the same steps run under the purge's own claim: the record version
// is recorded in op.intent first, the delete is conditional on it, and
// purging → parking is legal only with the record authoritatively gone
// and the purge journal at torn-down or later. A purged host never returns
// to serving.
func (w *world) deleteSteps(c *claim) bool {
	switch c.Op.Kind {
	case "deleting", "purging":
		if c.Op.Kind == "purging" && on("purge-journal") && w.purgeStep[c.Holder.App] < 4 {
			return false // the journal is not at torn-down yet
		}
		if w.rec != nil && w.rec.Inc == c.Holder.Inc {
			if c.Op.RecVer == 0 {
				// the intent: the exact record version to delete
				ver := w.rec.Ver
				w.cas(c, func(n *claim) bool {
					if !sameOp(n.Op.ID, c.Op.ID) || n.Op.Kind != c.Op.Kind {
						return false
					}
					n.Op.RecVer = ver
					return true
				})
				return false
			}
			if w.rec.Ver == c.Op.RecVer {
				w.rec = nil // DeleteIfMatch
			}
			return false
		}
		if w.rec != nil {
			return false // C6: a mismatched record means wait
		}
		inc := c.Holder.Inc
		w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, c.Op.ID) || n.Op.Kind != c.Op.Kind {
				return false
			}
			n.Op.Kind = "parking"
			w.parkingObligations(n, inc)
			return true
		})
		return false
	case "parking":
		if len(c.Obl) > 0 {
			return false
		}
		w.free(c)
		return true
	}
	return true
}

func (w *world) free(c *claim) {
	app := ""
	if c.Holder != nil {
		app = c.Holder.App
	}
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, c.Op.ID) || len(n.Obl) > 0 {
			return false
		}
		n.Holder, n.Op, n.Regions, n.Boxes, n.Complete = nil, opState{}, nil, nil, true
		return true
	}) {
		w.logf("host freed")
		w.opsDone++
		if r := w.residueAtEnd(); len(r) > 0 {
			w.violate("INV-10: the host was freed with residue %v", r)
		}
		w.stats["freed"]++
		if w.fail["recreate"] && w.rng.Intn(2) == 0 && !w.calm {
			// the biased schedule: recreate the host at once, where it last
			// served, while the old incarnation's requests are in flight
			w.nextInc++
			w.console = append(w.console, &consolePlan{Kind: "create", Host: "h", App: app, OpID: w.newID("cr"), Inc: w.nextInc, Region: w.lastRegion})
		}
		if w.purgeRes[app] && !on("purge-journal") {
			w.purgeRes[app] = false // freeing the host drops the reservation
		}
	}
}

func (w *world) purgeSteps(p *consolePlan) bool {
	c := w.claim
	reserve := func() {
		w.purgeRes[p.App] = true
	}
	claimIt := func() bool {
		if c == nil && w.rec != nil && w.rec.App == p.App && on("purge-adopts") {
			// an unclaimed legacy record of this app: adopt it first (§9),
			// so the purge decides over claims, not over their absence
			w.adopt()
			return false
		}
		if c == nil || c.Holder == nil || c.Holder.App != p.App {
			return true // nothing of this app holds the host
		}
		if c.Op.Kind == "purging" || c.Op.Kind == "parking" || c.Op.Kind == "deleting" {
			return true
		}
		if c.Op.Kind != "serving" {
			return false // retry later; the reservation stays
		}
		id := w.newID("pg")
		return w.cas(c, func(n *claim) bool {
			if n.Op.Kind != "serving" || w.completionBarrier(n, w.verifiedCompletionFence(n)) {
				return false // ruling 86: a completion box whose termination fence verifies is exempt
			}
			n.Op = opState{Kind: "purging", ID: id, Exec: -1, ReqAt: w.now}
			return true
		})
	}
	if on("purge-journal") {
		if st, ok := w.purgeStep[p.App]; ok && st > p.Step {
			p.Step = st // resume from the reservation's journal
		}
		defer func() {
			if w.purgeRes[p.App] {
				w.purgeStep[p.App] = p.Step
			}
		}()
	}
	ad := w.admit[p.App]
	if p.OpID == "" {
		// the purge id lives in the reservation, so a resumed purge
		// releases its own fences
		if id, ok := w.purgeOp[p.App]; ok {
			p.OpID = id
		} else {
			p.OpID = w.newID("pg")
			w.purgeOp[p.App] = p.OpID
		}
	}
	switch p.Step {
	case 0:
		if on("purge-reserve-first") {
			reserve()
		} else {
			w.stores[p.App] = false // tear down first
		}
		p.Step = 1
		return false
	case 1:
		// close admission (claim rev 4 §5.3 step 1b), so no new create can
		// register
		if on("admission-registration") {
			ad.Closed = true
			w.logf("purge %s: admission closed", p.App)
		}
		p.Step = 2
		return false
	case 2:
		// drain: every registered create reaches serving or is cancelled
		// before the host set is fixed; an expired registration is dead
		if on("admission-registration") {
			// in create-id order: Go's map order would make the drain, and
			// with it every seed after a purge, differ from run to run
			// (found in rev 7.3; the sole source of nondeterminism)
			ids := make([]string, 0, len(ad.Reg))
			for id := range ad.Reg {
				ids = append(ids, id)
			}
			sort.Strings(ids)
			for _, id := range ids {
				e := ad.Reg[id]
				// an expired registration whose host's claim does not name
				// its create is dead: the create checks the expiry right
				// before its claim CAS, so it can no longer claim (CC16)
				claimed := e.Host == "h2" && w.h2.OpID == id && w.h2.Kind == "creating" ||
					e.Host == "h" && w.claim != nil && w.claim.Op.ID == id && w.claim.Op.Kind == "creating"
				if on("registration-recovery") && w.now > e.Until && !claimed {
					// fence the host's claim first, so a paused creator's
					// CAS from free or absent fails (CC21); then remove
					if on("purge-claim-fence") && !w.fenceHost(e.Host, p) {
						return false
					}
					delete(ad.Reg, id)
					continue
				}
				return false
			}
		}
		p.Step = 3
		return false
	case 3:
		// the host set: records, plus registered in-flight creates (drained
		// above). A second host the purge cannot see is not in it.
		if !claimIt() {
			return false
		}
		if w.h2.App == p.App && w.h2.Rec {
			switch w.h2.Kind {
			case "serving":
				w.h2.Kind = "purging"
				w.h2Epoch++ // every write changes the claim's ETag
			case "purging":
			default:
				return false
			}
		}
		p.Step = 4
		return false
	case 4:
		if on("purge-reserve-first") {
			w.logf("purge %s: store torn down", p.App)
			w.stores[p.App] = false
		} else {
			reserve()
		}
		p.Step = 5
		return false
	default:
		if w.h2.App == p.App && w.h2.Kind == "purging" {
			w.h2 = liteHost{} // record deleted, claim freed
			w.h2Epoch++       // every write changes the claim's ETag
		}
		if c == nil || c.Holder == nil || c.Holder.App != p.App {
			// done: release this purge's fences by writing a free claim,
			// never by deleting the object (CC21; claim rev 5.2 §5.3, C5a).
			// The mutation deletes a fence that found no claim, so the
			// claim is absent again and a paused creator's If-None-Match: *
			// passes
			if c != nil && c.Holder == nil && c.Op.Kind == "purging" && c.Op.ID == p.OpID {
				if c.Op.FencedAbsent && !on("claim-never-deleted") {
					w.claim = nil
					w.logf("purge %s: fence on host %s deleted (the claim is absent again)", p.App, host)
				} else {
					w.cas(c, func(n *claim) bool {
						if n.Holder != nil || n.Op.ID != p.OpID {
							return false
						}
						n.Op = opState{}
						return true
					})
				}
			}
			if w.h2.Kind == "fenced" && w.h2.OpID == p.OpID {
				if w.h2.FencedAbsent && !on("claim-never-deleted") {
					w.h2, w.h2Epoch = liteHost{}, 0 // deleted: absent again
				} else {
					w.h2 = liteHost{}
					w.h2Epoch++ // every write changes the claim's ETag
				}
			}
			delete(w.purgeOp, p.App)
			w.purgeRes[p.App] = false
			delete(w.purgeStep, p.App)
			ad.Closed = false // the purge's completion reopens admission
			return true
		}
		w.deleteSteps(c)
		return false
	}
}

// fenceHost CASes a host's claim from free or absent to a purge fence with
// no holder (claim rev 5.1 §5.3, CC21). A claim held by anyone already
// refuses a creator's CAS from free, so it needs no fence. A fence that
// found no claim is the one create-only write of a host's claim (claim
// rev 5.2 C5a); the fence remembers it, because its release must write
// free rather than restore the absence.
func (w *world) fenceHost(h string, p *consolePlan) bool {
	if h == "h2" {
		switch {
		case w.h2.Kind == "":
			w.h2 = liteHost{Kind: "fenced", OpID: p.OpID, App: p.App, FencedAbsent: w.h2Epoch == 0}
			w.h2Epoch++ // every write changes the claim's ETag
			w.logf("purge %s: fenced the second host", p.App)
		case w.h2.Kind == "fenced" && w.h2.OpID != p.OpID:
			return false // another purge's fence: wait for it
		}
		return true
	}
	c := w.claim
	if c != nil && (c.Holder != nil || c.Op.Kind != "") {
		return !(c.Holder == nil && c.Op.Kind == "purging" && c.Op.ID != p.OpID)
	}
	ok := w.cas(c, func(n *claim) bool {
		if n.Holder != nil || n.Op.Kind != "" {
			return false
		}
		n.Op = opState{Kind: "purging", ID: p.OpID, App: p.App, Exec: -1, FencedAbsent: c == nil}
		return true
	})
	if ok {
		w.logf("purge %s: fenced host %s", p.App, host)
	}
	return ok
}

// resumePurges: the recovery pass restarts a purge whose plan was lost,
// from its reservation's journal.
func (w *world) resumePurges() {
	if !on("purge-journal") {
		return
	}
	for _, app := range []string{app0, app1} {
		if !w.purgeRes[app] {
			continue
		}
		running := false
		for _, p := range w.console {
			if p.Kind == "purge" && p.App == app {
				running = true
			}
		}
		if !running {
			w.console = append(w.console, &consolePlan{Kind: "purge", App: app, Step: w.purgeStep[app]})
		}
	}
}

// ---------- the door (claim §6.5) ----------

func (w *world) doorAuthorize() {
	c := w.claim // O1
	if c == nil || c.Holder == nil || c.Op.Kind != "serving" {
		return
	}
	b := w.execBox(c)
	if b == nil {
		return
	}
	// the listing will select the generation's newest LTX object
	bk := w.buckets[b.Region]
	key := ""
	for _, k := range bk.keys(dataKey(c.Holder.Location, c.Holder.Gen, "")) {
		if !isDataKey(k) || genPrefixOf(k) != dataKey(c.Holder.Location, c.Holder.Gen, "") {
			continue
		}
		if v := bk.current(k); v != nil && (key == "" || v.ID > bk.current(key).ID) {
			key = k
		}
	}
	if key == "" {
		return
	}
	d := &doorPlan{Region: b.Region, Loc: c.Holder.Location, Gen: c.Holder.Gen, Inc: c.Holder.Inc, OpID: c.Op.ID,
		Key: key, Until: w.now + 1 + w.pauseLen()}
	if w.fail["recreate"] && w.rng.Intn(2) == 0 {
		d.Adv = true
		if w.rng.Intn(2) == 0 {
			d.Until = w.now + 40*horizonH // stalls before selecting
		}
	}
	w.doorReads = append(w.doorReads, d)
}

// doorOpen advances one stalled listing or file request. A stalled step
// resumes at its worst moment if one comes before its stall ends: when the
// claim's holder has changed since O1. Otherwise it waits out its stall.
func (w *world) doorOpen() {
	if len(w.doorReads) == 0 {
		return
	}
	i := w.rng.Intn(len(w.doorReads))
	worst := -1
	for j, d := range w.doorReads {
		// its worst moment: the object its key names now belongs to
		// another incarnation
		if cur := w.buckets[d.Region].current(d.Key); cur != nil && !cur.Marker && cur.B.Inc != 0 && cur.B.Inc != d.Inc {
			worst = j
			break
		}
	}
	switch {
	case worst >= 0:
		i = worst
	case w.doorReads[i].Until > w.now:
		return
	}
	d := w.doorReads[i]
	bk := w.buckets[d.Region]
	if !d.Signed {
		// select versions only under O1's location (step 3)
		if on("door-by-version") {
			v := bk.current(d.Key)
			if v == nil {
				w.doorReads = append(w.doorReads[:i:i], w.doorReads[i+1:]...)
				return
			}
			d.Ver = v.ID
		}
		// O2: refuse if the holder or the op changed since O1 (step 5)
		if c := w.claim; on("door-reread") && (c == nil || c.Holder == nil || c.Holder.Inc != d.Inc || c.Op.ID != d.OpID) {
			w.doorReads = append(w.doorReads[:i:i], w.doorReads[i+1:]...)
			return
		}
		d.Signed, d.Until = true, w.now+1+w.pauseLen()
		if d.Adv {
			d.Until = w.now + 40*horizonH // stalls between signing and the open
		}
		return
	}
	w.doorReads = append(w.doorReads[:i:i], w.doorReads[i+1:]...)
	var v *version
	if d.Ver != 0 {
		v = bk.byID(d.Key, d.Ver)
	} else {
		v = bk.current(d.Key)
	}
	if v != nil && v.B.Inc != 0 && v.B.Inc != d.Inc {
		w.violate("door: a read authorized for inc %d returned data of inc %d", d.Inc, v.B.Inc)
	}
}

// ---------- the console's claim-recovery pass (claim §8) ----------

func (w *world) recoveryPass() {
	if !on("recovery-pass") {
		return
	}
	w.resumePurges()
	// the second host's create (control plane only)
	if h := w.h2; h.Kind == "creating" && w.now-h.At >= 15 {
		if h.Rec {
			w.h2.Kind = "serving"
			w.h2Epoch++ // every write changes the claim's ETag
		} else {
			w.h2 = liteHost{}
			w.h2Epoch++ // every write changes the claim's ETag
		}
		w.logf("recovery: second host's create %s -> %q", h.OpID, w.h2.Kind)
		delete(w.admit[h.App].Reg, h.OpID)
	}
	c := w.claim
	if c == nil {
		return
	}
	// a terminated source of a pre-flip move (moves rev 7.20 §7.10, claim
	// rev 6.28 §4): the termination fence converts the operation to
	// relocating at pinning. The flip is the record's decision, so a record
	// at landing refuses the conversion and §7.10's own recovery applies.
	if on("fence-converts-moving") && c.Holder != nil && c.Op.Kind == "moving" && !c.Op.Committed &&
		!(w.rec != nil && w.rec.MoveID == c.Op.ID && w.rec.MoveState == "landing") {
		if src := w.boxByID(c.Op.Src); src != nil && src.Destroyed {
			w.convertToRelocating(c, src)
			return
		}
	}
	// a relocation with no box in its region (a one-box region whose only
	// box was fenced): safe, nothing runs, the data is pinned; not
	// recovered until region HA enrolls a replacement. The operator is
	// told once (moves rev 7.20 §7.10, §13)
	if c.Op.Kind == "relocating" && c.Op.Phase == "pinning" && c.Op.Exec == -1 && !c.Op.Alerted && w.regionBox(c.Op.SrcRegion) == nil {
		if w.cas(c, func(n *claim) bool { n.Op.Alerted = true; return sameOp(n.Op.ID, c.Op.ID) && n.Op.Kind == "relocating" }) {
			w.alerts++
			w.stats["relocation: waiting for a box in the region"]++
			w.logf("relocation %s waits at pinning: no box in region %d", c.Op.ID, c.Op.SrcRegion)
		}
		return
	}
	// a late record PUT from a cancelled create: an orphan, never admitted
	// (C1); removed by the ordered cleanup of claim rev 6 §5.1 step 5, as
	// a plan of this console (J2-R2-7, R3-2)
	if w.rec != nil && on("cancelled-creates") && w.orphanOf(c, w.rec.Inc) {
		if !w.hasPlan("orphan", "recovery") {
			w.logf("recovery: cleaning up an orphan record inc %d", w.rec.Inc)
			w.console = append(w.console, &consolePlan{Kind: "orphan", Host: "recovery"})
		}
		return
	}
	switch c.Op.Kind {
	case "creating":
		if w.now-c.Op.ReqAt < 15 {
			return
		}
		if w.rec != nil && w.rec.Inc == c.Holder.Inc {
			b := w.createExecutor(c)
			if b == nil {
				if nb := w.boxByID(c.Op.NewBox); w.perRegion > 1 && on("create-choice-once") && (nb == nil || nb.Destroyed) {
					// the chosen box cannot complete the create (the two-box
					// shape only; the one-box shape waits for its region's
					// box, as the baseline does): recovery cancels it (claim
					// §5.1 step 5, P5-D6), retiring the
					// incarnation; the record is an orphan the cleanup removes
					inc, app, id := c.Holder.Inc, c.Holder.App, c.Op.ID
					if w.cas(c, func(n *claim) bool {
						if !sameOp(n.Op.ID, id) {
							return false
						}
						n.Holder, n.Op = nil, opState{}
						n.Regions, n.Boxes = nil, nil
						n.Retired = retire(n.Retired, inc)
						n.Cancelled = append(n.Cancelled, cancelledCreate{Inc: inc, Until: w.now + 4*horizonH})
						return true
					}) {
						delete(w.admit[app].Reg, id)
						w.stats["create cancelled: its box is gone"]++
						w.logf("recovery: cancelled create %s inc %d, its chosen box %d is gone", id, inc, c.Op.NewBox)
					}
				}
				return
			}
			if w.cas(c, func(n *claim) bool {
				if !sameOp(n.Op.ID, c.Op.ID) {
					return false
				}
				n.Op = opState{Kind: "serving", Exec: b.ID, Parts: []int{b.ID}}
				n.Holder.Stamp = w.issueStamp(n, n.Op.Exec) // the recovery pass finishing create step 4 writes holder.stamp too (claim rev 6.5 §2)
				return true
			}) {
				delete(w.admit[c.Holder.App].Reg, c.Op.ID)
			}
		} else if w.rec == nil {
			// authoritatively absent: cancel, retiring the incarnation and
			// remembering it, so a late record PUT can be removed
			inc, app, id := c.Holder.Inc, c.Holder.App, c.Op.ID
			if w.cas(c, func(n *claim) bool {
				if !sameOp(n.Op.ID, id) {
					return false
				}
				n.Holder, n.Op = nil, opState{}
				n.Regions, n.Boxes = nil, nil
				n.Retired = retire(n.Retired, inc)
				n.Cancelled = append(n.Cancelled, cancelledCreate{Inc: inc, Until: w.now + 4*horizonH})
				return true
			}) {
				delete(w.admit[app].Reg, id)
				w.logf("recovery: cancelled create %s inc %d", id, inc)
			}
		}
	case "deleting", "parking":
		if c.Holder == nil && c.Op.ParkSweep != "" {
			// a holderless sweep (claim rev 6.15 §6.2a) is the obligation
			// runners' to finish; the recovery pass only replays
			// FreeAfterPark for one whose chores are all acknowledged
			if c.oblFor(c.Op.ID) == 0 && w.now-c.Op.ReqAt >= 15 {
				id := c.Op.ID
				w.cas(c, func(n *claim) bool {
					if n.Holder != nil || n.Op.ID != id || n.oblFor(id) != 0 {
						return false
					}
					n.Op = opState{}
					return true
				})
			}
			return
		}
		if w.now-c.Op.ReqAt < 15 {
			return
		}
		w.deleteSteps(c)
	case "purging":
		if c.Holder == nil {
			return // a purge fence: its purge releases it (CC21)
		}
		if w.now-c.Op.ReqAt < 15 {
			return
		}
		if on("purge-journal") {
			w.resumePurges() // the purge's own phases, in order
			return
		}
		w.deleteSteps(c)
	case "landing":
		w.fixRecord()
		if !c.Op.Committed && w.now > c.Op.CutDeadline+10 && w.now > c.Op.FlipAt+10 {
			w.rollback(c, "landing deadline")
		}
	case "moving", "reverting", "restoring", "serving":
		w.fixRecord()
	}
}

// orphan: claim rev 5 §5.1 (CC17). A record whose incarnation is retired,
// or is not the claim's holder, is an orphan at any time; rev 4's
// cancelled_creates window is the mutation.
func (w *world) orphanOf(c *claim, inc int) bool {
	if c == nil {
		return false // no claim at all: a legacy record, adopted (§9), never an orphan
	}
	if c.Holder != nil && c.Holder.Inc == inc {
		return false
	}
	if c.Op.Kind == "creating" && c.Op.NewInc == inc {
		return false
	}
	if !on("orphan-no-deadline") {
		for _, x := range c.Cancelled {
			if x.Inc == inc && w.now <= x.Until {
				return true
			}
		}
		return false
	}
	return true
}

// ---------- the operator (calm phase only) ----------

func (w *world) operator() {
	c := w.claim
	if c == nil {
		return
	}
	// readmit a quarantined box once it has retired its old state (--readmit)
	for _, id := range c.Lost {
		if b := w.boxByID(id); b != nil && !b.Destroyed && b.Row == nil && b.DB == nil && b.Files == nil && len(b.Procs) == 0 {
			w.cas(c, func(n *claim) bool {
				var keep []int
				for _, x := range n.Lost {
					if x != id {
						keep = append(keep, x)
					}
				}
				n.Lost = keep
				return true
			})
			w.logf("operator readmitted box %d", id)
			return
		}
	}
	// a stuck relocation (§5.8): carlos ops host resolve (model ruling 13;
	// the spec names the command and leaves its effect open). Nothing
	// moves away from a live box without its stop evidence:
	//   - a live source still op.exec (starting, releasing, no release):
	//     the relocation is abandoned and the host serves on the source;
	//   - a live target op.exec without a release (its receipt lost after
	//     P opened, its release overdue): it is sent to probe-failed, where
	//     it runs its release sequence;
	//   - op.exec none, released, or lost: replayed from pinning with the
	//     replacement count reset, the exclusions cleared and the source
	//     eligible again (Resolved), an empty listing accepted
	if c.Holder != nil && c.Op.Kind == "relocating" && c.Op.Stuck != "" {
		op := c.Op
		released := op.Release != nil && op.Release.Box == op.Exec
		dead := w.boxByID(op.Exec)
		if dead != nil && !dead.Destroyed {
			dead = nil
		}
		if ex := w.boxByID(op.Exec); ex != nil && !ex.Destroyed && op.Phase == "probe-failed" && !released && !contains(c.Lost, ex.ID) && (w.c1Fallback() || on("reloc-release-needs-flush") && w.releaseUnrecoverable(ex, c)) {
			// with claim ask C1 open no failed target can release at all
			// (finding 4), so its release is unrecoverable by construction
			// a live target that cannot release (its replicator and its
			// files are gone and nothing recorded a flush, rev 9): no
			// release will ever come, so the operator stops the box, and
			// its declared loss (claim D2) lets the next resolve replay
			// the operation from pinning
			w.stop(ex)
			w.stats["operator stopped a target that cannot release"]++
			w.logf("operator stopped box %d: it cannot release relocation %s", ex.ID, op.ID)
			return
		}
		var what string
		if w.cas(c, func(n *claim) bool {
			if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "relocating" || n.Op.Stuck == "" {
				return false
			}
			if dead != nil && !contains(n.Lost, dead.ID) {
				// a stopped or terminated executor: the operator's declare-lost
				// (claim D2) puts it on lost, as a fence would, before the replay
				w.fenceBox(n, dead)
				released = false
			}
			switch {
			case op.Exec >= 0 && op.Exec == op.Src && !released && !contains(n.Lost, op.Exec) && (op.Phase == "starting" || op.Phase == "releasing"):
				src := op.Src
				n.Op = opState{Kind: "serving", Exec: src, Parts: []int{src}}
				what = "abandoned; the host serves on its source"
			case op.Exec >= 0 && !released && !contains(n.Lost, op.Exec):
				n.Op.Stuck, n.Op.Phase, n.Op.FailedAt, n.Op.Release = "", "probe-failed", w.now, nil
				what = "sent its live target to probe-failed"
			default:
				w.abandonPrepared(n)
				n.Op.Stuck, n.Op.Replacements, n.Op.Excluded, n.Op.Resolved = "", 0, nil, true
				n.Op.PinFailover = false // the replay enters pinning directly (rev 6)
				if released && op.Exec >= 0 && op.Exec != op.Src {
					// the target that just failed its probe is not the
					// operator's fresh choice (rev 4); a released source is
					// eligible again, as ruling 13 says (final review minor 4)
					n.Op.Excluded = []int{op.Exec}
				}
				n.Op.Phase, n.Op.Exec, n.Op.Tgt = "pinning", -1, -1
				n.Op.Pin, n.Op.Prepared, n.Op.Receipt = nil, "", nil
				n.Op.BaseKey, n.Op.BaseVer, n.Op.BaseTX = "", 0, 0
				n.Op.ProbeTries, n.Op.Attempts = 0, 0
				var parts []int
				for _, p := range n.Op.Parts {
					if !contains(n.Lost, p) && p != op.Exec {
						parts = append(parts, p)
					}
				}
				n.Op.Parts = parts
				what = "replayed from pinning"
			}
			return true
		}) {
			w.stats["operator resolved a stuck relocation"]++
			w.logf("operator resolved stuck relocation %s: %s", op.ID, what)
		}
		return
	}
	// a host marked failed by the retry policy (rev 7.11 §7.14): the operator
	// looks at it and clears the mark; the next attempt is a new session
	for _, b := range w.liveBoxes() {
		if b.hostFailed && w.now >= b.failedAt+horizonH/2 {
			b.hostFailed, b.failCounted, b.startFails, b.nextStart = false, false, 0, 0
			w.stats["failed hosts cleared"]++
			w.logf("operator cleared the failed host on box %d", b.ID)
			return
		}
	}
	// a destroyed box that owes only local chores: declare it lost (claim
	// §7.2 removes its obligations); compute fencing is the box being gone
	for _, o := range c.Obl {
		if o.Kind != "park-local" {
			continue
		}
		if b := w.boxByID(o.Box); b != nil && b.Destroyed {
			w.cas(c, func(n *claim) bool {
				var keep []obligation
				for _, x := range n.Obl {
					if !(x.Kind == "park-local" && x.Box == o.Box) {
						keep = append(keep, x)
					}
				}
				n.Obl = keep
				n.Lost = addUniq(n.Lost, o.Box)
				return true
			})
			w.logf("operator declared box %d lost (local chores only)", o.Box)
			return
		}
	}
	w.declareTargetLost(c)
}

// releaseUnrecoverable is the operator's check before it stops a live
// target that has not released (rev 10, Astra round 7): only when no
// release can ever come. That needs no usable flush record for this
// stop (rev 11: a record counts only with root's durable receipt, or while
// the controller that recorded it is still running), and either the
// session's replicator or the box's files directory gone for good
// (relocRelease refuses without either). Any other release, a slow
// acknowledgement or a recorded flush not yet journaled, is resumed: the
// resolve clears stuck and sends the target back to probe-failed. The
// mutation of operator-stop-unrecoverable-only stops every such target,
// as rev 9 did.
func (w *world) releaseUnrecoverable(b *box, c *claim) bool {
	if !on("operator-stop-unrecoverable-only") {
		return true
	}
	h := c.Holder
	if r := b.relRecord; r != nil && r.Op == c.Op.ID && r.Phase == c.Op.Phase && r.E.Session == h.Session && r.E.Gen == h.Gen &&
		(r.Receipt || r.Ctl == b.Epoch || !on("reloc-release-needs-receipt")) {
		return false // a recorded flush with root's receipt, or its own controller still alive to finish it (rev 11)
	}
	replicator := b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == h.Gen && b.DB.Epoch == b.Epoch && b.DB.Session == h.Session
	files := b.Files != nil && b.Files.Inc == h.Inc
	return !replicator || !files
}

// declareTargetLost is the declared loss of a landing's or a rollback's
// target that is gone (claim D2, moves §7.10): the operator's path, and,
// for a terminated target, the regional worker's automatic fence (claim
// D2 "Automatic termination", rev 4). It proceeds record first, so a loss
// racing the record's commit CAS follows whichever the record holds.
func (w *world) declareTargetLost(c *claim) {
	if c.Holder == nil {
		return
	}
	op := c.Op
	if (op.Kind != "landing" && op.Kind != "restoring") || op.Committed {
		return
	}
	t := w.boxByID(op.Tgt)
	if t == nil {
		return
	}
	owesEvidence := op.Exec == op.Tgt
	owesReturn := op.Kind == "restoring" && op.Exec == -1 && op.DirtyKnown && op.Dirty && op.Phase != "returned"
	if !owesEvidence && !owesReturn {
		return
	}
	ok := false
	if on("declare-lost-compute") {
		ok = t.Destroyed // cloud API: stopped or terminated
	} else {
		l := w.execLease[t.ID]
		ok = t.Destroyed || (l.Op == op.ID && w.now > l.Expires+lostMargin)
	}
	if !ok {
		return
	}
	if op.Kind == "landing" {
		// it proceeds as a rollback, record first (§7.8, §7.10); the
		// claim follows, and the loss is applied to the restoring claim on
		// the operator's next call
		if w.rec != nil && w.rec.MoveID == op.ID && w.rec.MoveState == "landing" {
			w.rollback(c, "declared lost")
		} else {
			w.followRecord(c)
		}
		return
	}
	if w.cas(c, func(n *claim) bool {
		if !sameOp(n.Op.ID, op.ID) || n.Op.Committed || n.Op.Kind != "restoring" {
			return false
		}
		n.Lost = addUniq(n.Lost, t.ID)
		n.Op.Exec, n.Op.DirtyKnown, n.Op.Dirty, n.Op.Phase, n.Op.LostBox, n.Op.LostAt = -1, true, false, "released", t.ID, w.now
		var keep []obligation
		for _, o := range n.Obl {
			if o.Kind == "park-local" && o.Box == t.ID {
				continue
			}
			keep = append(keep, o)
		}
		n.Obl = keep
		if owesReturn && n.Op.RollbackID != "" && on("lost-return-park") {
			// a return the lost target had begun: park its lane in the
			// source's region, after the lane's staging horizon (model 27)
			n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.SrcRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.RollbackID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc))
		}
		if owesEvidence { // a box that released already queued these
			// the lost target's parks cite the declaration's fencing bound
			// (claim rev 6.23 §2: a lost-box source uses the persisted
			// fenced_at), the staging lane its seal bound
			n.Obl = append(n.Obl,
				w.scoped(w.newObl(n, "park-superseded", op.TgtRegion, -1, c.Holder.Location, "", w.now+horizonH), w.now, c.Holder.Inc),
				w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", op.SealQuiet), stagingBound(op.SealQuiet), c.Holder.Inc))
			// the lost target's region is parked whole: the records whose
			// predecessors are there (a restarted probe's, whose dead
			// writer never published) are covered in this CAS (claim rev
			// 6.14 item 6; seed 1036 / 3,000)
			w.coverRegionPredecessors(n, op.TgtRegion, "lost-target release")
		}
		return true
	}) {
		// the lost target's acknowledged writes are declared by the source's
		// admission (writes_lost for the target's uncertified session, or a
		// sealed baseline), never exempted here (R10-6). Its registered
		// timers went with it and no return lane will carry them: the
		// declaration names them, and alerts (seed 4120 / 30,000)
		w.logf("operator declared box %d lost for %s", t.ID, op.ID)
		n := 0
		for i := range w.acks {
			x := &w.acks[i]
			if x.Box == t.ID && x.Inc == c.Holder.Inc && x.Kind == "timer" && !x.Exempt && !w.delivered[x.W] {
				x.Exempt = true
				n++
			}
		}
		if n > 0 {
			w.stats["timers declared lost"] += n
			w.alert("box %d declared lost for %s: %d acknowledged timers registered on it are lost with it", t.ID, op.ID, n)
		}
	}
}