package hostlife

import (
	"crypto/sha256"
	"fmt"
	"testing"
)

// Astra's round 1 on the two-box plan (rev 4): a direct test for every
// guard the round found unproven, the phase matrix stopped between single
// actor transitions, and claim §11's additional witnesses that rev 3 left
// outside the model. Each test names the rule whose mutation makes it
// fail.

// stepUntil advances the world one actor at a time and stops the moment
// done holds: the tick's own work (completions, stalls, the two-box
// step), then each live box's worker, obligation runner, reconcile and
// passes, the executor's wake, then the recovery pass, with done checked
// after every single one. A fault injected after it returns lands
// exactly between two transitions.
func (w *world) stepUntil(t *testing.T, max int, what string, done func() bool) bool {
	t.Helper()
	if done() {
		return true
	}
	for i := 0; i < max; i++ {
		w.now++
		w.completeDue(false)
		if p := w.recPend; p != nil && p.at <= w.now {
			w.recPend = nil
			if w.rec == nil {
				w.rec = p.rec
			}
		}
		w.runStalled()
		if w.perRegion > 1 {
			w.twoBoxStep()
		}
		if done() {
			return true
		}
		for _, b := range w.liveBoxes() {
			for _, act := range []func(){
				func() { w.worker(b) }, func() { w.obligationRunner(b) }, func() { w.reconcile(b) },
				func() {
					if len(b.plans) == 0 {
						w.passPlan(b, "config")
						w.passPlan(b, "adopt")
					}
					w.passExec(b)
				},
				w.wakeExec,
			} {
				act()
				if len(w.violations) > 0 {
					return false
				}
				if done() {
					return true
				}
			}
		}
		w.recoveryPass()
		w.checkStep()
		if len(w.violations) > 0 {
			return false
		}
		if done() {
			return true
		}
	}
	w.violate("INV-3 (witness): %s not reached within %d steps: %s", what, max, w.state())
	return false
}

// atPhase: the relocation is exactly in phase p (not beyond it), as
// 1b-next names it: an initial failover at pinning with PinFailover set
// is in starting (rev 7, Astra round 4 finding 2).
func (w *world) atPhase(p string) func() bool {
	return func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && relocPhaseName(c.Op) == p && c.Op.Stuck == ""
	}
}

// relocPhaseName is the operation's phase as 1b-next names it: the
// model's pinning with op.PinFailover set is production's starting.
func relocPhaseName(op opState) string {
	if op.Phase == "pinning" && op.PinFailover {
		return "starting"
	}
	return op.Phase
}

// liveIn counts region r's live boxes.
func (w *world) liveIn(r int) int { return len(w.regionBoxes(r)) }

// ---------- finding 6: the termination matrix, exact ----------

// TestL18TerminationAtEveryExactPhase (L18 "termination during every
// relocation phase", rev 4): the relocation is stopped exactly in each
// phase of §5.8's table, probe-failed included, between two single
// actor transitions; the phase and the region's live topology are
// asserted immediately before the participant is terminated; no spare
// exists beforehand, and replacements come only from the modelled group
// after the loss. One-box region: the failover's target (the box the
// group launched) is terminated. Two-box region: the drain's source or
// target is.
func TestL18TerminationAtEveryExactPhase(t *testing.T) {
	phases := []string{"starting", "releasing", "pinning", "installing", "installed", "probing", "probe-failed", "probed"}
	for _, boxes := range []int{1, 2} {
		for _, who := range []string{"source", "target"} {
			for _, phase := range phases {
				if boxes == 1 && (who == "source" || phase == "releasing") {
					continue // a one-box region relocates only by failover, whose source is fenced at its start; its starting is PinFailover's wait (rev 7)
				}
				if phase == "probe-failed" && who == "source" {
					continue // the source has released before any probe; its loss is the pinning case
				}
				name := fmt.Sprintf("%s/%s@%s", map[int]string{1: "one-box region", 2: "two-box region"}[boxes], who, phase)
				t.Run(name, func(t *testing.T) {
					w := regionWorld(boxes)
					from := w.serveAndPublish(t)
					w.appWrite(from)
					w.timerOn(from)
					var target func() *box
					if boxes == 2 {
						w.drainTo(t, nil)
						tgt := w.boxByID(w.claim.Op.Tgt)
						tgt.FailApp = phase == "probe-failed"
						target = func() *box { return w.boxByID(w.claim.Op.Tgt) }
					} else {
						w.terminate(from)
						if !w.stepUntil(t, 400, "the failover started with its target", func() bool { c := w.claim; return c.Op.Kind == "relocating" && c.Op.Tgt >= 0 }) {
							w.mustHold(t)
						}
						if c := w.claim; relocPhaseName(c.Op) != "starting" {
							t.Fatalf("the failover did not start in starting (PinFailover unset): %s", w.state())
						}
						w.boxByID(w.claim.Op.Tgt).FailApp = phase == "probe-failed"
						target = func() *box { return w.boxByID(w.claim.Op.Tgt) }
					}
					if !w.stepUntil(t, 2000, "relocation exactly at "+phase, w.atPhase(phase)) {
						w.mustHold(t)
					}
					// immediately before the injection: the exact phase, and the
					// region's live topology (no spare)
					if c := w.claim; relocPhaseName(c.Op) != phase {
						t.Fatalf("not at %s: %s", phase, w.state())
					}
					if c := w.claim; phase == "pinning" && (c.Op.PinFailover || c.Op.Pin != nil && c.Op.Exec >= 0) {
						t.Fatalf("pinning is after PinFailover and before installing: %s", w.state())
					}
					if n := w.liveIn(0); n != boxes {
						t.Fatalf("region 0 has %d live boxes before the injection, want %d", n, boxes)
					}
					victim := from
					if who == "target" {
						victim = target()
					}
					if victim == nil || victim.Destroyed {
						t.Fatalf("no live %s at %s: %s", who, phase, w.state())
					}
					w.terminate(victim)
					w.servingOn(t, nil)
					w.finish(t)
					w.mustHold(t)
					if c := w.claim; c.Op.Exec == victim.ID {
						t.Fatalf("serving on the terminated %s", who)
					}
				})
			}
		}
	}
}

// ---------- finding 2: stuck refuses admission ----------

// TestStuckSourceRefusesAdmission (reloc-stuck-refuses-admission): three
// targets die in starting, so the exhausted fence leaves the live source
// op.exec with its ready row and the operation stuck. A request then
// reaches the source, whose app has stopped: no writer is admitted.
func TestStuckSourceRefusesAdmission(t *testing.T) {
	w := twoBoxWitness()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	w.running(t, from)
	b30, b31 := w.extraBox(0, 30), w.extraBox(0, 31)
	w.userRelocate()
	for _, x := range []*box{w.boxByID(w.claim.Op.Tgt), b30, b31, w.boxes[3]} {
		if x == nil || x.Destroyed || x.ID != w.claim.Op.Tgt {
			continue
		}
		w.terminate(x)
		w.reach(t, 60, "the dead target fenced", func() bool { return contains(w.claim.Lost, x.ID) }, w.act(from))
	}
	if c := w.claim; c.Op.Stuck == "" || c.Op.Exec != from.ID || c.Op.Phase != "starting" || from.Row == nil || from.Row.Status != "ready" {
		t.Fatalf("not stuck in starting with the live source op.exec and its row ready: %s", w.state())
	}
	from.Procs = nil // the app stopped; the next request wakes it
	before := w.stats["admissions: cold"] + w.stats["admissions: warm"]
	w.drive(30, never, func() { w.wake(from) })
	w.mustHold(t)
	if n := w.stats["admissions: cold"] + w.stats["admissions: warm"] - before; n != 0 || hasMain(from) {
		t.Fatalf("a stuck operation admitted %d writers on its source (running %v)", n, hasMain(from))
	}
}

// ---------- finding 3: the sole box is checked in an HA region ----------

// TestHASoleBoxIsCheckedLikeAnyOther (ha-sole-box-eligibility): in a
// one-box region with region HA on, the only box is refused when it is
// unenrolled, lacks the secrets key, runs an old executable, is on lost or
// owes a park-local for the host; in the one-box shape (region HA off)
// the region's box is chosen as today.
func TestHASoleBoxIsCheckedLikeAnyOther(t *testing.T) {
	for _, why := range []string{"unenrolled", "no key", "old executable", "lost", "owes a park-local"} {
		t.Run(why, func(t *testing.T) {
			w := regionWorld(1)
			b := w.boxes[0]
			w.serve(w.boxes[1], 1, app0) // the claim lives in region 1; region 0 is the create's
			switch why {
			case "unenrolled":
				b.Enrolled = false
			case "no key":
				b.HasKey = false
			case "old executable":
				b.Old = true
			case "lost":
				w.claim.Lost = addUniq(w.claim.Lost, b.ID)
			case "owes a park-local":
				w.claim.Obl = append(w.claim.Obl, obligation{ID: "pl", Op: "old", Kind: "park-local", Box: b.ID, Region: -1, Ret: "old"})
			}
			if got := w.chooseBox(0, app0); got != nil {
				t.Fatalf("region HA chose its sole box %d although it is %s", got.ID, why)
			}
		})
	}
	t.Run("one-box shape keeps today's pick", func(t *testing.T) {
		w := newWitness()
		w.boxes[0].Enrolled = false
		if got := w.chooseBox(0, app0); got != w.boxes[0] {
			t.Fatalf("the one-box shape did not pick the region's box: %v", got)
		}
	})
}

// ---------- finding 4: the acknowledgement barrier at every transition ----------

// TestReplacementWithoutAcknowledgementHoldsEachTransition
// (reloc-exec-ack-barrier): the target dies at installing and the spare
// that replaces it has not acknowledged. Driving the installation's steps
// without the spare's acknowledgement, the operation stays at installing
// (ToInstalled's requireAcks); once the spare acknowledges, it proceeds.
func TestReplacementWithoutAcknowledgementHoldsEachTransition(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	spare := w.extraBox(0, 30)
	if !w.reach(t, 1500, "installing", w.atPhase("installing"), w.act(from, to), w.recoveryPass) {
		w.mustHold(t)
	}
	w.terminate(to)
	w.reach(t, 60, "replaced at installing", func() bool { return w.claim.Op.Exec == spare.ID }, w.act(from))
	if contains(w.claim.Op.Acks, spare.ID) {
		t.Fatalf("the spare acknowledged before it ran")
	}
	// the spare's installation steps, without its acknowledgement
	spare.ExecUntil = w.localNow(spare) + execTTL
	for i := 0; i < 40; i++ {
		w.tick()
		w.relocInstalling(spare, w.claim.clone())
	}
	w.mustHold(t)
	if c := w.claim; c.Op.Phase != "installing" {
		t.Fatalf("left installing without the spare's acknowledgement: %s", w.state())
	}
	w.servingOn(t, spare)
	w.finish(t)
	w.mustHold(t)
}

// ---------- finding 5: deadlines and delays from the fence record ----------

// TestPinWaitsForEveryFenceObservation (reloc-pin-delay-every-fence): the
// source's release is journaled, then the target is terminated and
// replaced by a spare before any pin. The pin waits out the target's
// fence observation plus the delay, a release notwithstanding.
func TestPinWaitsForEveryFenceObservation(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	spare := w.extraBox(0, 30)
	if !w.stepUntil(t, 1500, "the source's release journaled", func() bool { c := w.claim; return c.Op.Release != nil && c.Op.Pin == nil }) {
		w.mustHold(t)
	}
	w.terminate(to)
	ok := w.stepUntil(t, 400, "the pin", func() bool {
		c := w.claim
		if c.Op.Pin != nil && len(c.Op.Observed) > 0 && w.now < latestObservation(c.Op)+failoverPinDelay {
			t.Fatalf("pinned at %d, before the target's observation %d plus %d", w.now, latestObservation(c.Op), failoverPinDelay)
		}
		return c.Op.Pin != nil
	})
	if !ok {
		w.mustHold(t)
	}
	if !contains(w.claim.Lost, to.ID) || w.claim.Op.Tgt != spare.ID {
		t.Fatalf("the target was not replaced before the pin: %s", w.state())
	}
	w.servingOn(t, spare)
	w.finish(t)
	w.mustHold(t)
}

// TestDeadTargetDeadlineFromObservation (reloc-deadline-from-observed-at):
// the drain's target dies at installing while no worker of the region can
// observe it (the source is frozen) for longer than the deadline, and the
// group launches no spare. The first observation writes the fence record;
// the operation waits a whole deadline from it before it is stuck.
func TestDeadTargetDeadlineFromObservation(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	w.oneBoxRegion[0] = true // no replacement launches: no spare
	if !w.reach(t, 1500, "installing", w.atPhase("installing"), w.act(from, to), w.recoveryPass) {
		w.mustHold(t)
	}
	w.terminate(to)
	from.Frozen = true
	w.drive(releaseDeadline+20, never, w.act(w.boxes[1], w.boxes[2]))
	from.Frozen = false
	w.stepUntil(t, 40, "the fence record", func() bool { return to.FenceObservedAt > 0 })
	seen := w.now
	w.drive(releaseDeadline-5, never, w.act(from), w.recoveryPass)
	if w.claim.Op.Stuck != "" {
		t.Fatalf("stuck %d steps after the observation at %d, before its deadline (terminated at %d)", w.now-seen, seen, to.TerminatedAt)
	}
	w.reach(t, 60, "stuck after the deadline", func() bool { return w.claim.Op.Stuck != "" }, w.act(from), w.recoveryPass)
	w.mustHold(t)
}

// ---------- finding 7: the five guards, each discriminated ----------

// TestOnlyTheExecutorWritesP (reloc-exec-only-installs): at installing,
// only the source (not op.exec) runs: nothing is written under P and no
// local install is made on it.
func TestOnlyTheExecutorWritesP(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	if !w.stepUntil(t, 1500, "installing", w.atPhase("installing")) {
		w.mustHold(t)
	}
	p := w.claim.Op.Prepared
	for _, k := range w.buckets[0].keys(dataKey(w.claim.Holder.Location, p, "")) {
		w.buckets[0].objs[k] = nil // whatever the target began is set aside: only the source acts now
		delete(w.buckets[0].objs, k)
	}
	w.drive(40, never, w.act(from))
	if ks := w.buckets[0].keys(dataKey(w.claim.Holder.Location, p, "")); len(ks) > 0 || from.receipt != nil || from.Files != nil && from.Files.Gen == p {
		t.Fatalf("the source, not op.exec, wrote P or installed it: %v", ks)
	}
}

// TestInstallAppliesThePinnedManifest (files-pin-exact): a newer manifest
// of the dead executor lands between the pin and the install; the install
// applies the pinned version, never the newest.
func TestInstallAppliesThePinnedManifest(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	loc := w.claim.Holder.Location
	late := w.timerOn(from)
	w.holdKey(loc + "files/")
	w.publishFiles(from) // the newer manifest, carrying the timer, held
	w.hold = nil
	w.terminate(from)
	if !w.stepUntil(t, 400, "the pin journaled", func() bool { c := w.claim; return c.Op.Pin != nil }) {
		w.mustHold(t)
	}
	if n := w.completeHeld(loc + "files/"); n == 0 {
		t.Fatalf("no held manifest to land")
	}
	if !w.servingOn(t, to) {
		w.mustHold(t)
	}
	if contains(to.Timers, late) {
		t.Fatalf("the install applied the manifest that landed after the pin")
	}
}

// TestFailoverPinWaitsOutTheDelay (reloc-failover-pin-delay): no pin
// before the fence's observed_at plus the delay.
func TestFailoverPinWaitsOutTheDelay(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.terminate(from)
	ok := w.stepUntil(t, 400, "the pin", func() bool {
		c := w.claim
		if c.Op.Pin != nil && w.now < c.Op.FenceAt+failoverPinDelay {
			t.Fatalf("pinned at %d, before the fence's observation %d plus %d", w.now, c.Op.FenceAt, failoverPinDelay)
		}
		return c.Op.Kind == "relocating" && c.Op.Pin != nil
	})
	if !ok {
		w.mustHold(t)
	}
	w.servingOn(t, to)
	w.finish(t)
	w.mustHold(t)
}

// TestFailoverPinWaitsForTheAcknowledgement (reloc-failover-ack-before-pin;
// rev 5, Astra round 2 finding 1; 1b-next rev 19 PinFailover): the
// executor's newer files manifest is still in flight when it is
// terminated. The target is named, but its controller does not run, so
// it has not acknowledged; the pin waits past the delay, the manifest
// lands, and only then does the target acknowledge. The pin names the
// newer manifest. The mutation pins before the acknowledgement, and the
// pin misses it.
func TestFailoverPinWaitsForTheAcknowledgement(t *testing.T) {
	w := twoBoxWitness()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	spare := w.extraBox(0, 30) // a second live box of the region: the worker while the target's controller is down
	c := w.claim
	loc, gen := c.Holder.Location, c.Holder.Gen
	from.Files.Env++ // a newer files state: the admin token rotated
	newer := manifestKey(loc, gen, from.pubSeq+1)
	w.holdKey(newer)
	w.publishFiles(from)
	w.hold = nil
	if from.pubSeq == 0 || manifestKey(loc, gen, from.pubSeq) != newer {
		t.Fatalf("the newer manifest was not issued: %s", w.state())
	}
	w.terminate(from)
	if !w.stepUntil(t, 400, "the failover's target named", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && (c.Op.Tgt >= 0 || c.Op.Pin != nil)
	}) {
		w.mustHold(t)
	}
	if w.claim.Op.Pin != nil {
		t.Fatalf("pinned before a target was named and acknowledged: %s", w.state())
	}
	tgt := w.boxByID(w.claim.Op.Tgt)
	worker := spare
	if tgt == spare {
		worker = w.boxes[3]
	}
	w.drive(failoverPinDelay+20, never, w.act(worker)) // the target's controller does not run
	if contains(w.claim.Op.Acks, tgt.ID) {
		t.Fatalf("the target acknowledged without running")
	}
	if p := w.claim.Op.Pin; p != nil {
		t.Fatalf("pinned %s before the target acknowledged: %s", p.ManKey, w.state())
	}
	if n := w.completeHeld(newer); n != 1 {
		t.Fatalf("held manifest PUTs completed: %d, want 1", n)
	}
	if !w.reach(t, 400, "the pin", func() bool { return w.claim.Op.Pin != nil }, w.act(worker, tgt)) {
		w.mustHold(t)
	}
	if p := w.claim.Op.Pin; p.ManKey != newer {
		t.Fatalf("the pin names %s, want the manifest that landed while the target had not acknowledged, %s", p.ManKey, newer)
	}
	w.servingOn(t, tgt)
	w.finish(t)
	w.mustHold(t)
}

// convertedMove serves the host on box 0 of a two-box witness world with
// a spare box 30 in region 0, starts a move to region 1, and terminates
// the source before the flip, so the regional worker converts the move
// to a relocation at pinning with no target (moves §7.10). before runs
// before the move starts.
func (w *world) convertedMove(t *testing.T, before func()) *box {
	src := w.boxes[0]
	w.serve(src, 1, app0)
	w.running(t, src)
	w.published(t, src)
	if before != nil {
		before()
	}
	w.moveTo(1)
	if w.claim.Op.Kind != "moving" {
		t.Fatalf("no move: %s", w.state())
	}
	tgt := w.boxByID(w.claim.Op.Tgt)
	w.reach(t, 100, "staging begins", func() bool {
		c := w.claim
		return c.Op.Kind == "moving" && c.Op.Phase == "preparing" && contains(c.Regions, tgt.Region)
	}, w.act(src, tgt))
	w.terminate(src)
	if !w.stepUntil(t, 400, "the move converted", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.FenceBox == src.ID
	}) {
		w.mustHold(t)
	}
	if c := w.claim; c.Op.PinFailover || c.Op.Pin != nil {
		t.Fatalf("not a converted move at pinning without a pin: %s", w.state())
	}
	return src
}

// TestConvertedMovePinsBeforeAcknowledgement (rev 6, Astra round 3
// finding 1; rev 7, round 4 finding 1): a converted move enters pinning
// directly, as in 1b-next, and Worker.pinning names its target first,
// then selects and journals the pin, and checks the acknowledgements only
// for ToInstalling. The named target's controller does not run while the
// region's other box works: the pin is journaled without the target's
// acknowledgement, and installing waits for it.
func TestConvertedMovePinsBeforeAcknowledgement(t *testing.T) {
	w := twoBoxWitness()
	spare := w.extraBox(0, 30)
	w.convertedMove(t, nil)
	if !w.stepUntil(t, 400, "the target named", func() bool { c := w.claim; return c.Op.Tgt >= 0 || c.Op.Pin != nil }) {
		w.mustHold(t)
	}
	c := w.claim
	if c.Op.Pin != nil {
		t.Fatalf("pinned before a target was named: %s", w.state())
	}
	tgt := w.boxByID(c.Op.Tgt)
	worker := spare
	if tgt == spare {
		worker = w.boxes[3]
	}
	w.drive(failoverPinDelay+20, never, w.act(worker)) // the target's controller does not run
	c = w.claim
	if contains(c.Op.Acks, tgt.ID) {
		t.Fatalf("the target acknowledged without running")
	}
	if c.Op.Pin == nil {
		t.Fatalf("the converted move waited for an acknowledgement before its pin: %s", w.state())
	}
	if c.Op.Phase != "pinning" || c.Op.Exec >= 0 {
		t.Fatalf("installing without the target's acknowledgement: %s", w.state())
	}
	w.servingOn(t, tgt)
	w.finish(t)
	w.mustHold(t)
}

// TestConvertedMoveWaitsForATargetBeforeItsPin (reloc-name-before-pin;
// rev 7, Astra round 4 finding 1): the source's newer files manifest is
// in flight when the move starts and when the source is terminated. No
// box of the region is eligible (the spare has no key), so the converted
// move names no target and selects no pin; the manifest lands; the
// spare becomes eligible, is named, and the pin names the newer
// manifest. The mutation pins first, without the manifest.
func TestConvertedMoveWaitsForATargetBeforeItsPin(t *testing.T) {
	w := twoBoxWitness()
	peer := w.boxes[3]
	var newer string
	w.convertedMove(t, func() {
		src, c := w.boxes[0], w.claim
		src.Files.Env++ // a newer files state: the admin token rotated
		newer = manifestKey(c.Holder.Location, c.Holder.Gen, src.pubSeq+1)
		w.holdKey(newer)
		w.publishFiles(src)
		w.hold = nil
		peer.HasKey = false // the region's other box may work, never take the host
	})
	if newer == "" || w.boxes[0].pubSeq == 0 {
		t.Fatalf("the newer manifest was not issued")
	}
	w.drive(failoverPinDelay+20, never, w.act(peer))
	if c := w.claim; c.Op.Pin != nil || c.Op.Tgt >= 0 {
		t.Fatalf("pinned or named with no eligible box: %s", w.state())
	}
	if n := w.completeHeld(newer); n != 1 {
		t.Fatalf("held manifest PUTs completed: %d, want 1", n)
	}
	peer.HasKey = true
	if !w.reach(t, 400, "the pin", func() bool { return w.claim.Op.Pin != nil }, w.act(peer)) {
		w.mustHold(t)
	}
	if p := w.claim.Op.Pin; p.ManKey != newer {
		t.Fatalf("the pin names %s, want the manifest that landed while no box was eligible, %s", p.ManKey, newer)
	}
	w.servingOn(t, peer)
	w.finish(t)
	w.mustHold(t)
}

// convertedMoveTargetStopped converts a move (convertedMove: the source,
// box 0, is terminated before the flip) and stops the move's own target,
// the one participant the conversion left, before it acknowledges; it
// returns that box. Stopped is not terminated: nothing fences it, and
// it never acknowledges the converted operation.
func (w *world) convertedMoveTargetStopped(t *testing.T) *box {
	w.convertedMove(t, nil)
	c := w.claim
	if len(c.Op.Parts) != 1 || len(c.Op.Acks) != 0 || c.Op.Tgt >= 0 {
		t.Fatalf("not a converted move whose one participant is the move's target: %s", w.state())
	}
	mt := w.boxByID(c.Op.Parts[0])
	if mt == nil || mt.Region == c.Op.SrcRegion {
		t.Fatalf("the converted move's participant %d is not the move's target in another region: %s", c.Op.Parts[0], w.state())
	}
	w.stop(mt)
	return mt
}

// TestConvertedMoveDropsTheMoveTargetFromTheBarrier
// (reloc-name-drops-move-target; final review Important 1, 1b-next
// 151261fe): the move's source is terminated before the flip, and the
// move's own target in region 1, the one participant the conversion
// left, is stopped before it acknowledges. Naming the same-region target
// takes the move's target off the participants in the same CAS, and the
// relocation reaches installing and serves while that box stays
// stopped. The mutation keeps it on the barrier: the relocation waits at
// pinning for ever.
func TestConvertedMoveDropsTheMoveTargetFromTheBarrier(t *testing.T) {
	w := twoBoxWitness()
	mt := w.convertedMoveTargetStopped(t)
	if !w.reach(t, 400, "the target named", func() bool { return w.claim.Op.Tgt >= 0 }, w.actAll(), w.recoveryPass) {
		w.mustHold(t)
	}
	if c := w.claim; contains(c.Op.Parts, mt.ID) || len(c.Op.Parts) != 1 || c.Op.Parts[0] != c.Op.Tgt {
		t.Fatalf("the move's target %d is still on the barrier: parts=%v target=%d", mt.ID, c.Op.Parts, c.Op.Tgt)
	}
	if !w.relocPhase(t, "installing", w.actAll(), w.recoveryPass) {
		w.mustHold(t)
	}
	if !mt.Destroyed {
		t.Fatalf("the move's target started again")
	}
	w.servingOn(t, nil)
	w.finish(t)
	w.mustHold(t)
}

// TestFailoverStartWaitsForAdmission (reloc-failover-start-guard; rev
// 7, Astra round 4 finding 3): with the app's admission closed, a
// terminated serving executor's failover does not start (Worker.Start's
// Admission.Register); once it reopens, it starts and serves on the
// region's other box.
func TestFailoverStartWaitsForAdmission(t *testing.T) {
	w := twoBoxWitness()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	ad := w.admit[app0]
	ad.Closed = true
	w.terminate(from)
	w.drive(60, never, w.actAll(), w.recoveryPass)
	if c := w.claim; c.Op.Kind != "serving" || contains(c.Lost, from.ID) {
		t.Fatalf("a failover started while the app's admission was closed: %s", w.state())
	}
	ad.Closed = false
	if to := w.failedOver(t, from); to == nil || to.Region != from.Region {
		t.Fatalf("no failover in the region: %s", w.state()) // the peer, or the group's replacement, whichever choose() prefers
	}
	w.finish(t)
	w.mustHold(t)
}

// TestFailoverStartWaitsForParkReplica (reloc-failover-start-guard; rev
// 7, Astra round 4 finding 3): a park-replica chore is outstanding in the
// region when the serving executor is terminated. RelocationStartGuard
// refuses the failover's start until the chore is done; the source's own
// park-local (the fenced source) would not have.
func TestFailoverStartWaitsForParkReplica(t *testing.T) {
	w := twoBoxWitness()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.claim.Obl = append(w.claim.Obl, w.scoped(obligation{ID: "pr", Op: "old-op", Kind: "park-replica", Region: 0, Box: -1, Prefix: "moves/" + host + "/", Quiet: w.now + 20}, w.now, 1))
	w.claim.Obl = append(w.claim.Obl, obligation{ID: "pl", Op: "old-op", Kind: "park-local", Box: from.ID, Region: -1, Ret: "old-op"})
	from.DeleteOnTermination = false // the fence keeps the park-local open
	w.terminate(from)
	if !w.stepUntil(t, 400, "the failover's start", func() bool {
		c := w.claim
		if c.Op.Kind == "relocating" {
			for _, o := range c.Obl {
				if o.Kind == "park-replica" && o.Region == 0 {
					t.Fatalf("the failover started with a park-replica outstanding in its region: %s", w.state())
				}
			}
			return true
		}
		return false
	}) {
		w.mustHold(t)
	}
	if !w.owesLocal(w.claim, from) {
		t.Fatalf("the fenced source's park-local was not kept open: %s", w.state())
	}
	if to := w.failedOver(t, from); to == nil || to.Region != from.Region {
		t.Fatalf("no failover in the region: %s", w.state())
	}
	w.finish(t)
	w.mustHold(t)
}

// TestFailoverTargetLossAroundPinFailover (rev 7, Astra round 4 finding
// 2): an initial failover's target is terminated while PinFailover is
// pending (production's starting) and, in a second run, after it has
// cleared and before installing (pinning). In a two-box region with a
// spare, the target is replaced in the same phase and the host serves.
func TestFailoverTargetLossAroundPinFailover(t *testing.T) {
	for _, phase := range []string{"starting", "pinning"} {
		t.Run(phase, func(t *testing.T) {
			w := twoBoxWitness()
			from := w.boxes[0]
			w.extraBox(0, 30)
			w.serve(from, 1, app0)
			w.running(t, from)
			w.published(t, from)
			w.terminate(from)
			if !w.stepUntil(t, 400, "the failover exactly at "+phase, w.atPhase(phase)) {
				w.mustHold(t)
			}
			c := w.claim
			if (phase == "starting") != c.Op.PinFailover || c.Op.Exec >= 0 || c.Op.Tgt < 0 {
				t.Fatalf("not at %s: %s", phase, w.state())
			}
			victim := w.boxByID(c.Op.Tgt)
			w.terminate(victim)
			if !w.stepUntil(t, 400, "the target replaced", func() bool {
				c := w.claim
				return contains(c.Lost, victim.ID) && c.Op.Tgt >= 0 && c.Op.Tgt != victim.ID
			}) {
				w.mustHold(t)
			}
			if got := relocPhaseName(w.claim.Op); got != phase {
				t.Fatalf("the replacement left %s for %s: %s", phase, got, w.state())
			}
			w.servingOn(t, nil)
			w.finish(t)
			w.mustHold(t)
		})
	}
}

// TestPendingTargetFenceHoldsThePhase (reloc-fence-pending-blocks; rev
// 8, Astra round 5): a drain's target has acknowledged and is terminated,
// once in starting and once in pinning before a pin; the region's group
// launches no replacement, so the fence cannot be consumed until its
// deadline. Until then the source's worker makes no regional transition:
// starting does not go to releasing, and pinning selects no pin. Under
// the mutation the worker goes on into the phase switch and this test
// fails. Past the deadline the fence is consumed and the operation is
// stuck, as FenceTargetWithoutSpare makes it.
func TestPendingTargetFenceHoldsThePhase(t *testing.T) {
	for _, phase := range []string{"starting", "pinning"} {
		t.Run(phase, func(t *testing.T) {
			w := twoBoxWitness()
			from, to := w.boxes[0], w.boxes[3]
			w.serve(from, 1, app0)
			w.running(t, from)
			w.published(t, from)
			w.oneBoxRegion[0] = true // no replacement launches: no spare
			w.drainTo(t, to)
			if phase == "starting" {
				// the target's controller acknowledges (AcknowledgeExecution);
				// the source has not yet, so the operation stays in starting
				to.ExecUntil = w.localNow(to) + execTTL
				w.recordAck(to, w.claim.clone())
			} else if !w.stepUntil(t, 1500, "the target acknowledged at "+phase, func() bool {
				c := w.claim
				return contains(c.Op.Acks, to.ID) && w.atPhase(phase)() && c.Op.Pin == nil
			}) {
				w.mustHold(t)
			}
			if c := w.claim; !contains(c.Op.Acks, to.ID) || !w.atPhase(phase)() || c.Op.Pin != nil {
				t.Fatalf("not at %s with the target acknowledged: %s", phase, w.state())
			}
			w.oneBoxRegion[0] = false
			w.terminate(to)
			w.oneBoxRegion[0] = true
			for i := 0; i < releaseDeadline-10; i++ {
				w.tick(w.act(from))
				c := w.claim
				if relocPhaseName(c.Op) != phase || c.Op.Pin != nil {
					t.Fatalf("the phase moved on while the target's fence was pending (step %d): %s", i, w.state())
				}
				if contains(c.Lost, to.ID) {
					t.Fatalf("the dead target was fenced with no spare before its deadline: %s", w.state())
				}
			}
			w.mustHold(t)
			if w.stats["relocation: phase held, a participant's fence is pending"] == 0 {
				t.Fatalf("no worker step held the phase")
			}
			if !w.reach(t, 100, "the fence consumed into stuck", func() bool { c := w.claim; return contains(c.Lost, to.ID) && c.Op.Stuck != "" }, w.act(from)) {
				w.mustHold(t)
			}
		})
	}
}

// TestReleaseRefusedAfterACrashBeforeTheFlush (reloc-release-needs-flush;
// rev 9, Astra round 6): a drain's source is in releasing, its last write
// not yet uploaded, when its agent crashes: the session's replicator is
// gone and nothing recorded a flush for this stop. No release is
// journaled and no pin selected; at the release deadline the operation
// is stuck, and the operator's resolve returns the host to its source.
// The mutation journals the remote position with Flushed false.
func TestReleaseRefusedAfterACrashBeforeTheFlush(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	w.appWrite(from)
	w.holdKey(w.claim.Holder.Location) // the flush's upload does not land
	if !w.stepUntil(t, 1500, "the source waiting for its flush", func() bool {
		return w.claim.Op.Phase == "releasing" && w.stats["release: waits for the flush"] > 0
	}) {
		w.mustHold(t)
	}
	w.crash(from)
	w.hold = nil
	for i := 0; i < releaseDeadline+20; i++ {
		w.tick(w.act(from, to))
		c := w.claim
		if c.Op.Release != nil || c.Op.Pin != nil {
			t.Fatalf("a release or a pin without the source's flush (step %d): %s", i, w.state())
		}
		if c.Op.Stuck != "" {
			break
		}
	}
	if c := w.claim; c.Op.Stuck == "" || c.Op.Phase != "releasing" {
		t.Fatalf("not stuck in releasing at the release deadline: %s", w.state())
	}
	w.finish(t)
	w.mustHold(t)
	if c := w.claim; c.Op.Kind != "serving" || c.Op.Exec != from.ID {
		t.Fatalf("the operator's resolve did not return the host to its source: %s", w.state())
	}
}

// TestReleaseReplaysTheRecordedFlush (reloc-release-needs-flush; rev 9,
// Astra round 6; rev 11, round 8): the source's release sequence
// completes, records its flushed positions (ReleasePositions), and root
// persists the stop receipt; its agent then crashes before the release
// is journaled. The restarted agent has no replicator for the session,
// but the record and the receipt are durable: the release is replayed
// with the recorded positions, Flushed, and the drain completes. The
// mutation of reloc-release-needs-flush records nothing and journals the
// remote position with Flushed false.
func TestReleaseReplaysTheRecordedFlush(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	w.appWrite(from)
	if !w.stepUntil(t, 1500, "releasing", w.atPhase("releasing")) {
		w.mustHold(t)
	}
	var r *releaseEntry
	for i := 0; i < 60 && r == nil; i++ {
		r = w.relocRelease(from, w.claim.clone()) // steps 1 to 3, the record and the receipt, not yet journaled
		w.tick()
	}
	if r == nil || !r.Flushed {
		t.Fatalf("the release sequence did not complete: %v %s", r, w.state())
	}
	w.crash(from)
	if !w.reach(t, 60, "the release journaled", func() bool { return w.claim.Op.Release != nil }, w.act(from, to)) {
		w.mustHold(t)
	}
	got := w.claim.Op.Release
	if !got.Flushed || got.FinalTX != r.FinalTX || got.ManifestSeq != r.ManifestSeq {
		t.Fatalf("the journaled release %+v is not the recorded one %+v", got, r)
	}
	w.servingOn(t, to)
	w.finish(t)
	w.mustHold(t)
}

// probeFailedTarget drains the host from box 0 to box 3, whose app
// fails its probe, with a spare box 30 for the nested relocation, and
// stops once the operation is in probe-failed with box 3 op.exec and no
// release journaled.
func (w *world) probeFailedTarget(t *testing.T) (from, to *box) {
	w.claimC1 = true // these tests are of the failed target's release, which claim ask C1 makes possible (finding 4)
	from, to = w.boxes[0], w.boxes[3]
	to.FailApp = true
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	w.extraBox(0, 30)
	if !w.stepUntil(t, 1500, "probe-failed on the target", func() bool {
		c := w.claim
		return c.Op.Phase == "probe-failed" && c.Op.Exec == to.ID && c.Op.Release == nil
	}) {
		w.mustHold(t)
	}
	return from, to
}

// resolveOnly is the operator's resolve of a stuck relocation and
// nothing else (the destroyed-local declaration is left out).
func (w *world) resolveOnly() {
	if c := w.claim; c != nil && c.Op.Kind == "relocating" && c.Op.Stuck != "" {
		w.operator()
	}
}

// TestOperatorResumesASlowRelease (operator-stop-unrecoverable-only; rev
// 10, Astra round 7 finding 2): the failed target's final manifest PUT is
// not acknowledged before the release deadline, so the operation is
// stuck. Its replicator and files are intact: the operator's resolve
// resumes the release rather than stopping the box; once the PUT lands
// the release is journaled and the nested relocation serves. The
// mutation stops the box.
func TestOperatorResumesASlowRelease(t *testing.T) {
	w := twoBoxWitness()
	_, to := w.probeFailedTarget(t)
	h := w.claim.Holder
	held := dataKey(h.Location, h.Gen, "files/")
	w.holdKey(held)
	if !w.reach(t, 2*releaseDeadline, "stuck at the release deadline", func() bool { return w.claim.Op.Stuck != "" }, w.actAll(), w.recoveryPass) {
		w.mustHold(t)
	}
	if c := w.claim; c.Op.Phase != "probe-failed" || c.Op.Release != nil {
		t.Fatalf("not stuck in probe-failed without a release: %s", w.state())
	}
	w.resolveOnly()
	if c := w.claim; to.Destroyed || c.Op.Stuck != "" || c.Op.Phase != "probe-failed" || c.Op.Exec != to.ID {
		t.Fatalf("the resolve did not resume the target's release (stopped %v): %s", to.Destroyed, w.state())
	}
	w.hold = nil
	w.completeHeld(held)
	w.servingOn(t, nil)
	if w.stats["operator stopped a target that cannot release"] != 0 || w.stats["release journaled: failed target"] == 0 {
		t.Fatalf("the target was stopped, or never released: %v", w.stats)
	}
	w.finish(t)
	w.mustHold(t)
}

// TestOperatorResumesARecordedFlush (operator-stop-unrecoverable-only;
// rev 10, Astra round 7 finding 2): the failed target completes its
// release sequence, records it and gets root's receipt (rev 11), then its
// agent restarts (the session's replicator is gone) before the release is
// journaled, and the deadline passes. The recorded flush is a release waiting to be
// journaled: the resolve resumes it, the target replays the record, and
// the nested relocation serves. The mutation stops the box.
func TestOperatorResumesARecordedFlush(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.probeFailedTarget(t)
	var r *releaseEntry
	for i := 0; i < 60 && r == nil; i++ {
		r = w.relocRelease(to, w.claim.clone()) // steps 1 to 3, the record and the receipt, not journaled
		w.tick()
	}
	if r == nil {
		t.Fatalf("the target's release sequence did not complete: %s", w.state())
	}
	w.crash(to)
	spare := w.boxByID(30)
	if !w.reach(t, 2*releaseDeadline, "stuck at the release deadline", func() bool { return w.claim.Op.Stuck != "" }, w.act(from, spare), w.recoveryPass) {
		w.mustHold(t)
	}
	w.resolveOnly()
	if c := w.claim; to.Destroyed || c.Op.Stuck != "" || c.Op.Phase != "probe-failed" {
		t.Fatalf("the resolve did not resume the recorded release (stopped %v): %s", to.Destroyed, w.state())
	}
	w.servingOn(t, nil)
	if w.stats["operator stopped a target that cannot release"] != 0 || w.stats["release: replayed from the recorded positions"] == 0 {
		t.Fatalf("the target was stopped, or its record not replayed: %v", w.stats)
	}
	w.finish(t)
	w.mustHold(t)
}

// TestStoppedTargetKeepsItsLocalCleanup (fence-discharge-qualified,
// operator-stop-unrecoverable-only; rev 10, Astra round 7 findings 1 and
// 2): the failed target loses its database and files with its agent, so
// its release is unrecoverable; the operator stops the box and the next
// resolve declares it lost and replays. The box's volume would be deleted
// on termination, but it was stopped, not terminated: its disk is kept,
// so when the recovery completes its park-local is still owed (claim §7.3
// D5, H19). The mutation of fence-discharge-qualified discharges it.
func TestStoppedTargetKeepsItsLocalCleanup(t *testing.T) {
	w := twoBoxWitness()
	_, to := w.probeFailedTarget(t)
	to.DeleteOnTermination = true
	w.crash(to)
	to.Files, to.DB, to.receipt = nil, nil, nil
	if !w.reach(t, 3000, "serving after the operator's stop and replay", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec >= 0 && c.Op.Exec != to.ID
	}, w.actAll(), w.recoveryPass, w.wakeExec, w.resolveOnly) {
		w.mustHold(t)
	}
	if !to.Destroyed || to.Terminated || w.stats["operator stopped a target that cannot release"] == 0 {
		t.Fatalf("the target was not stopped by the operator: %s", w.state())
	}
	if !w.owesLocal(w.claim, to) {
		t.Fatalf("the stopped target's local cleanup was discharged: %s", w.state())
	}
	w.finish(t) // the operator's destroyed-local declaration, an explicit attestation, retires it
	w.mustHold(t)
}

// recordWithoutReceipt runs b's release sequence until its flushed
// positions are recorded, and stops before the producers drain and root
// persists the receipt.
func (w *world) recordWithoutReceipt(t *testing.T, b *box) {
	for i := 0; i < 60 && b.relRecord == nil; i++ {
		w.relocRelease(b, w.claim.clone())
		w.tick()
	}
	if r := b.relRecord; r == nil || r.Receipt {
		t.Fatalf("no record without a receipt on box %d: %+v %s", b.ID, r, w.state())
	}
}

// TestReleaseRefusedAfterACrashBeforeTheReceipt (reloc-release-needs-receipt;
// rev 11, Astra round 8): the source records its flushed positions, and
// its agent crashes before root persists the stop receipt. The positions
// are not proof of the stop: the restarted agent's release is refused
// (RecoveryReleaseDrain), no release or pin appears, the deadline makes
// the operation stuck, and the resolve returns the host to its source.
// TestReleaseReplaysTheRecordedFlush is the other side: the crash after
// the receipt, replayed. The mutation replays the positions alone.
func TestReleaseRefusedAfterACrashBeforeTheReceipt(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	w.appWrite(from)
	if !w.stepUntil(t, 1500, "releasing", w.atPhase("releasing")) {
		w.mustHold(t)
	}
	w.recordWithoutReceipt(t, from)
	w.crash(from)
	for i := 0; i < releaseDeadline+20; i++ {
		w.tick(w.act(from, to))
		c := w.claim
		if c.Op.Release != nil || c.Op.Pin != nil {
			t.Fatalf("a release or a pin from recorded positions with no receipt (step %d): %s", i, w.state())
		}
		if c.Op.Stuck != "" {
			break
		}
	}
	if c := w.claim; c.Op.Stuck == "" || c.Op.Phase != "releasing" {
		t.Fatalf("not stuck in releasing at the release deadline: %s", w.state())
	}
	w.finish(t)
	w.mustHold(t)
	if c := w.claim; c.Op.Kind != "serving" || c.Op.Exec != from.ID {
		t.Fatalf("the operator's resolve did not return the host to its source: %s", w.state())
	}
}

// TestOperatorStopsATargetWithNoReceipt (reloc-release-needs-receipt,
// operator-stop-unrecoverable-only; rev 11, Astra round 8): the failed
// target records its flushed positions and its agent restarts before root
// persists the receipt. Its release can never come, so the operator stops
// it, and the next resolve replays the operation; the host serves
// elsewhere. The mutation treats the positions as recoverable and resumes
// a release that replays them.
func TestOperatorStopsATargetWithNoReceipt(t *testing.T) {
	w := twoBoxWitness()
	_, to := w.probeFailedTarget(t)
	w.recordWithoutReceipt(t, to)
	w.crash(to)
	if !w.reach(t, 3000, "serving after the operator's stop and replay", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec >= 0 && c.Op.Exec != to.ID
	}, w.actAll(), w.recoveryPass, w.wakeExec, w.resolveOnly) {
		w.mustHold(t)
	}
	if !to.Destroyed || w.stats["operator stopped a target that cannot release"] == 0 || w.stats["release journaled: failed target"] != 0 {
		t.Fatalf("the target was not stopped, or released from positions with no receipt: %v %s", w.stats, w.state())
	}
	w.finish(t)
	w.mustHold(t)
}

// TestPinNeverShortOfTheRelease (reloc-pin-reaches-release): the listing
// lags behind the journaled release (its newest file is not yet visible);
// no pin is journaled short of the release's final_txid.
func TestPinNeverShortOfTheRelease(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.appWrite(from)
	w.drainTo(t, to)
	if !w.stepUntil(t, 1500, "the release journaled", func() bool { c := w.claim; return c.Op.Release != nil && c.Op.Pin == nil }) {
		w.mustHold(t)
	}
	c := w.claim
	bk := w.buckets[0]
	var hidden string
	var vs []*version
	for _, k := range bk.keys(dataKey(c.Holder.Location, c.Op.Release.Gen, "")) {
		if l, _, max, ok := ltxLevel(k); ok && l == 0 && max == c.Op.Release.FinalTX {
			hidden, vs = k, bk.objs[k]
			delete(bk.objs, k)
		}
	}
	if hidden == "" {
		t.Fatalf("no level-0 file at the release's final_txid %d", c.Op.Release.FinalTX)
	}
	w.drive(20, never, w.act(from, to))
	if p := w.claim.Op.Pin; p != nil && p.TX < c.Op.Release.FinalTX {
		t.Fatalf("pinned at TXID %d, short of the release's %d", p.TX, c.Op.Release.FinalTX)
	}
	bk.objs[hidden] = vs
	w.servingOn(t, to)
	w.finish(t)
	w.mustHold(t)
}

// ---------- finding 8: claim §11's additional witnesses ----------

// TestCrashAtEachOpenBoundary (claim §11 "Crash between prepared
// admission, marker publication and replication opening"): the target's
// agent crashes exactly at each boundary of P's first admitted session:
// admitted with the app starting (its launch environment held unready),
// admitted with generation.json's PUT in flight (held), generation.json
// published and replication not yet open (rev 5, Astra round 2: the
// witness yield openYield stops the start between the two), and
// replication open; in a one-box and a two-box region.
func TestCrashAtEachOpenBoundary(t *testing.T) {
	for _, stage := range []string{"app", "marker", "published", "open"} {
		t.Run(stage, func(t *testing.T) {
			bothShapes(t, func(t *testing.T, w *world, boxes int) {
				from := w.serveAndPublish(t)
				w.timerOn(from)
				loc := w.claim.Holder.Location
				if boxes == 2 {
					w.drainTo(t, nil)
				} else {
					w.terminate(from)
				}
				if stage == "marker" {
					w.holdKey(genMarkerKey(loc))
				}
				yielded := false
				if stage == "published" {
					w.openYield = func(b *box) bool {
						if yielded || b.open == nil || !b.open.Prepared {
							return false
						}
						yielded = true
						return true
					}
				}
				var to *box
				hit := w.stepUntil(t, 2000, "P's open at "+stage, func() bool {
					c := w.claim
					if c.Op.Kind != "relocating" || c.Op.Tgt < 0 {
						return false
					}
					to = w.boxByID(c.Op.Tgt)
					if stage == "app" && c.Op.Phase == "probing" && to.envUnreadyFor == 0 {
						to.envUnreadyFor = 20 // the launch environment is not ready yet
					}
					if stage == "open" {
						return preparedOpened(c) && to.open == nil && hasMain(to)
					}
					if stage == "published" {
						if !yielded || to.open == nil || to.open.stage != "open" {
							return false
						}
						if g, _ := w.bucketGen(to, loc); g != to.open.Gen {
							t.Fatalf("yielded before generation.json named %q", to.open.Gen)
						}
						return true
					}
					return to.open != nil && to.open.Prepared && to.open.stage == stage
				})
				if !hit {
					return
				}
				w.crash(to)
				to.envUnreadyFor = 0
				w.hold, w.openYield = nil, nil
				w.servingOn(t, nil)
				w.finish(t)
			})
		})
	}
}

// TestAutomaticTargetLossRacesRecordCommit (claim §11 "Automatic target
// loss racing the record commit CAS"): a move's target is terminated
// just before, and just after, its record commit CAS; the regional
// worker's automatic fence takes the operator's record-first path, so the
// claim follows whichever the record holds: a rollback to the source
// before the commit, a committed landing (then a failover in the target's
// region) after it.
func TestAutomaticTargetLossRacesRecordCommit(t *testing.T) {
	for _, when := range []string{"before the commit", "after the commit"} {
		t.Run(when, func(t *testing.T) {
			w := twoBoxWitness()
			src := w.boxes[0]
			w.serve(src, 1, app0)
			w.running(t, src)
			w.published(t, src)
			w.moveTo(1)
			if w.claim.Op.Kind != "moving" {
				t.Fatalf("no move: %s", w.state())
			}
			tgt := w.boxByID(w.claim.Op.Tgt)
			ok := w.stepUntil(t, 1500, "the commit boundary", func() bool {
				c := w.claim
				if when == "before the commit" {
					return c.Op.Kind == "landing" && c.Op.LandPhase == "probed" && w.rec != nil && w.rec.MoveState == "landing" && w.now >= c.Op.ProbedAt+15
				}
				return w.rec != nil && w.rec.MoveState == "committed" && c.Op.Kind == "landing" && !c.Op.Committed
			})
			if !ok {
				w.mustHold(t)
			}
			landing := w.claim.Op.ID
			w.terminate(tgt)
			if when == "after the commit" {
				// the failover keeps the landing's op id, as hostclaim
				// ConvertCommittedLanding does: the move's admission
				// registration is drained unless the claim carries it
				if !w.stepUntil(t, 400, "the committed landing converted", func() bool { return w.claim.Op.Kind == "relocating" }) {
					w.mustHold(t)
				}
				if w.claim.Op.ID != landing {
					t.Fatalf("the committed landing %s converted to relocation %s, want the landing's op id kept: %s", landing, w.claim.Op.ID, w.state())
				}
			}
			w.servingOn(t, nil)
			w.finish(t)
			w.mustHold(t)
			c := w.claim
			if when == "before the commit" && (w.rec.Region != src.Region || w.execBox(c).Region != src.Region) {
				t.Fatalf("a loss before the commit did not return the host to the source's region: %s", w.state())
			}
			if when == "after the commit" && (w.rec.Region != 1 || w.execBox(c).Region != 1) {
				t.Fatalf("a loss after the commit did not keep the host in the target's region: %s", w.state())
			}
		})
	}
}

// ---------- R18-2: the tie-break follows the host ----------

// TestTieHashFollowsTheHost (tie-hash-host; 1b-next R18-2): on equal
// loads two hostnames may choose different boxes; the model's tie-break
// hashes each host's own name.
func TestTieHashFollowsTheHost(t *testing.T) {
	w := regionWorld(2)
	a, b := w.boxes[0], w.peer()
	sum := func(host string, id int) uint32 { // claim P5-D2, computed here independently of the model
		s := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%d", host, id)))
		return uint32(s[0])<<24 | uint32(s[1])<<16 | uint32(s[2])<<8 | uint32(s[3])
	}
	wantH, wantH2 := a, a
	if sum("h", b.ID) > sum("h", a.ID) {
		wantH = b
	}
	if sum("h2", b.ID) > sum("h2", a.ID) {
		wantH2 = b
	}
	if wantH == wantH2 {
		t.Skip("hosts h and h2 choose the same box among boxes 0 and 3")
	}
	if got := w.chooseBoxFor("h", 0, app0); got != wantH {
		t.Fatalf("host h chose %v, want %d", got, wantH.ID)
	}
	if got := w.chooseBoxFor("h2", 0, app0); got != wantH2 {
		t.Fatalf("host h2 chose %v, want %d", got, wantH2.ID)
	}
}

// ---------- the final review's minor 4: a released source stays eligible ----------

// TestOperatorReplayKeepsAReleasedSourceEligible (model ruling 13; final
// review minor 4): a drain's source journals its release in releasing,
// still op.exec, and the target is then terminated with no spare (the
// group launches none). The dead target's bound makes the operation
// stuck with the released source as op.exec. The operator's replay
// makes the source eligible again: it excludes only a failed target,
// never the source, so the host serves on the source. Excluding the
// released source left the replay waiting at pinning for a spare that
// never comes.
func TestOperatorReplayKeepsAReleasedSourceEligible(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.drainTo(t, to)
	w.oneBoxRegion[0] = true // no replacement launches: no spare
	if !w.stepUntil(t, 400, "the source's release journaled", func() bool {
		c := w.claim
		return c.Op.Release != nil && c.Op.Release.Box == from.ID
	}) {
		w.mustHold(t)
	}
	if c := w.claim; c.Op.Phase != "releasing" || c.Op.Exec != from.ID {
		t.Fatalf("not released in releasing with the source op.exec: %s", w.state())
	}
	w.terminate(to)
	if !w.reach(t, 3*releaseDeadline, "stuck", func() bool { return w.claim.Op.Stuck != "" }, w.act(from), w.recoveryPass) {
		w.mustHold(t)
	}
	if c := w.claim; c.Op.Exec != from.ID || c.Op.Src != from.ID || c.Op.Release == nil || c.Op.Release.Box != from.ID {
		t.Fatalf("not stuck with the released source as op.exec: %s", w.state())
	}
	w.resolveOnly()
	if c := w.claim; c.Op.Phase != "pinning" || c.Op.Stuck != "" || !c.Op.Resolved {
		t.Fatalf("the resolve did not replay from pinning: %s", w.state())
	}
	if c := w.claim; contains(c.Op.Excluded, from.ID) {
		t.Fatalf("the replay excludes the released source %d: excluded=%v", from.ID, c.Op.Excluded)
	}
	w.servingOn(t, from)
	w.finish(t)
	w.mustHold(t)
}