package hostlife

import (
	"strings"
	"testing"
)

// twoBoxWitness is newWitness with a second enrolled box in every region.
func twoBoxWitness() *world {
	w := newWitness()
	w.perRegion = 2
	for r := 0; r < nRegions; r++ {
		w.addBox(r)
	}
	return w
}

// TestDrainEndToEnd: a serving host is drained from box 0 to box 3 in
// region 0 through every phase of §5.8's table; the host serves on box 3
// with every acknowledged write and timer, box 0's installation is
// retired by its park-local, and the first probe admission's coverage is
// released (L9): no write is declared lost; the first serving wake on box
// 3 is cold and records a writes_lost for the stopped probe session, which
// acknowledged nothing here (L18 joint round 2 finding 1).
func TestDrainEndToEnd(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.reach(t, 50, "the host runs on box 0", func() bool { return hasMain(from) }, w.actAll(), func() { w.wake(from) })
	w.appWrite(from)
	w.nextW++
	from.Timers = append(from.Timers, w.nextW)
	w.acks = append(w.acks, ack{W: w.nextW, Inc: 1, Box: from.ID, Session: w.claim.Holder.Session, At: w.now, Kind: "timer"})
	w.userRelocate()
	if c := w.claim; c.Op.Kind != "relocating" || c.Op.Src != from.ID || c.Op.Tgt != to.ID || c.Op.Phase != "starting" {
		t.Fatalf("the drain did not start: %s", w.state())
	}
	seen := map[string]bool{}
	w.reach(t, 1500, "serving on box 3", func() bool {
		seen[w.claim.Op.Phase] = true
		return w.claim.Op.Kind == "serving" && w.claim.Op.Exec == to.ID
	}, w.actAll(), w.recoveryPass, func() { w.wake(to) })
	for _, p := range []string{"starting", "releasing", "pinning", "installing", "installed", "probing", "probed"} {
		if !seen[p] {
			t.Errorf("phase %s never observed", p)
		}
	}
	if w.stats["admissions covered: released"] != 1 || w.stats["writes_lost declared"] > 1 || w.stats["writes declared lost"] != 0 {
		t.Errorf("released coverage: %d released, %d writes_lost covering %d writes (want 1, at most the first serving wake's, and 0)", w.stats["admissions covered: released"], w.stats["writes_lost declared"], w.stats["writes declared lost"])
	}
	w.finish(t)
	if len(w.violations) > 0 {
		t.Fatalf("%s", w.violations[0])
	}
	if from.DB != nil || from.Files != nil || from.Row != nil {
		t.Errorf("box 0 still holds local state after its park-local: db=%v files=%v row=%v", from.DB != nil, from.Files != nil, from.Row)
	}
}

// TestUnenrolledReferenceNeverChosen (§2, HA7; Review Focus 4): the
// executor is terminated; the region's other enrolled box holds no
// secrets key, so it may work but cannot take the host (R9); the
// replacement that launched is not yet enrolled. The failover must wait
// at pinning with no target until the replacement enrolls, never name
// the unenrolled reference (which could not be fenced if it died).
func TestUnenrolledReferenceNeverChosen(t *testing.T) {
	w := twoBoxWitness()
	from, peer := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.reach(t, 50, "the host runs on box 0", func() bool { return hasMain(from) }, w.actAll(), func() { w.wake(from) })
	w.published(t, from)
	peer.HasKey = false
	w.terminate(from)
	w.now += replaceDelay
	w.launchReplacements()
	nb := w.boxes[len(w.boxes)-1]
	if nb.Region != 0 || nb.Enrolled {
		t.Fatalf("no unenrolled replacement in region 0: %+v", nb)
	}
	nb.EnrollAt = w.now + 10*enrollDelay // its confirmation is slow
	// nb stays unenrolled and the peer has no key: no box is eligible, so
	// the failover does not start (Worker.Start's Eligible, rev 7) and no
	// unenrolled box is named
	w.drive(enrollDelay, func() bool { return false }, w.actAll(), w.recoveryPass)
	if c := w.claim; c.Op.Kind != "serving" || c.Op.Exec != from.ID || c.Op.Tgt == nb.ID {
		t.Fatalf("an unenrolled box was named, or the failover started with no eligible target: %s", w.state())
	}
	nb.EnrollAt = w.now - enrollDelay
	w.enrollStep()
	if !nb.Enrolled {
		t.Fatalf("the replacement did not enroll")
	}
	w.reach(t, 1500, "serving on the replacement", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec == nb.ID
	}, w.actAll(), w.recoveryPass, w.operator, func() { w.wake(nb) })
	w.finish(t)
	if len(w.violations) > 0 {
		t.Fatalf("%s", w.violations[0])
	}
}

// TestTwoWorkersOneChore (claim P6-D5; Review Focus 1): a push leaves a
// dead generation; both boxes of the region run its park-generation at
// once. The chore converges with no residue and no double declaration.
func TestTwoWorkersOneChore(t *testing.T) {
	w := twoBoxWitness()
	b := w.boxes[0]
	w.serve(b, 1, app0)
	w.userPush()
	w.reach(t, 800, "the push served and its predecessor's chore queued", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && len(c.Obl) > 0
	}, w.actAll(), w.recoveryPass, func() { w.wake(b) })
	w.finish(t)
	if len(w.violations) > 0 {
		t.Fatalf("%s", w.violations[0])
	}
	if r := w.residue(); len(r) > 0 {
		t.Fatalf("residue after both workers ran the chore: %v", r)
	}
}

// TestLateFileUploads (§5.6, §5.8; Review Focus 5): the executor
// registers a timer, its manifest PUT is held by the witness, and it is
// terminated. The failover pins the previous manifest, the install
// applies exactly that version, the held PUT lands after the pin, and
// the timer it carried is declared, never silently dropped; the late
// manifest is parked with generation_from.
func TestLateFileUploads(t *testing.T) {
	w := twoBoxWitness()
	w.perRegion = 2
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.reach(t, 50, "the host runs on box 0", func() bool { return hasMain(from) }, w.actAll(), func() { w.wake(from) })
	w.publishFiles(from) // the first manifest lands
	w.completeDue(true)
	w.nextW++
	late := w.nextW
	from.Timers = append(from.Timers, late)
	w.acks = append(w.acks, ack{W: late, Inc: 1, Box: from.ID, Session: w.claim.Holder.Session, At: w.now, Kind: "timer"})
	w.holdKey(w.claim.Holder.Location + "files/")
	w.publishFiles(from) // the second manifest, held
	w.hold = nil
	w.terminate(from)
	w.reach(t, 1500, "serving on box 3", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec == to.ID
	}, w.actAll(), w.recoveryPass, w.operator, func() { w.wake(to) })
	if contains(to.Timers, late) {
		t.Fatalf("the install applied a manifest that had not landed at the pin")
	}
	declared := false
	for _, a := range w.acks {
		if a.W == late && a.Exempt {
			declared = true
		}
	}
	if !declared {
		t.Fatalf("the timer the late manifest carried was not declared with the fence")
	}
	if n := w.completeHeld(w.claim.Holder.Location + "files/"); n != 1 {
		t.Fatalf("held manifest PUTs completed: %d, want 1", n)
	}
	w.finish(t)
	if len(w.violations) > 0 {
		t.Fatalf("%s", w.violations[0])
	}
	if r := w.residue(); len(r) > 0 {
		t.Fatalf("the late manifest was never parked: %v", r)
	}
}

// ---------- Task 5: the scenarios ----------

// twoBox turns a witness world into the two-box shape.
func (w *world) twoBox() {
	w.perRegion = 2
	for r := 0; r < nRegions; r++ {
		w.addBox(r)
	}
}

// regionWorld is a witness world with region HA on (the two-box shape's
// rules, its publisher and its group) whose region 0 holds boxes boxes:
// 2 is a two-box region (boxes 0 and 3); 1 is a one-box region (box 0
// alone, kept at one box by its group, H5's minimum being per region).
// Regions 1 and 2 have two boxes either way.
func regionWorld(boxes int) *world {
	w := newWitness()
	w.perRegion = 2
	for r := 0; r < nRegions; r++ {
		if r == 0 && boxes == 1 {
			w.oneBoxRegion[0] = true
			continue
		}
		w.addBox(r)
	}
	return w
}

// bothShapes runs an L18 case in a one-box and in a two-box region.
func bothShapes(t *testing.T, f func(t *testing.T, w *world, boxes int)) {
	for _, n := range []int{1, 2} {
		n := n
		t.Run(map[int]string{1: "one-box region", 2: "two-box region"}[n], func(t *testing.T) {
			w := regionWorld(n)
			f(t, w, n)
			if len(w.violations) > 0 {
				dump(t, w)
				t.Fatalf("%s", w.violations[0])
			}
		})
	}
}

// peer is region 0's second box, nil in a one-box region.
func (w *world) peer() *box {
	for _, b := range w.boxes {
		if b.ID == nRegions && !b.Destroyed {
			return b
		}
	}
	return nil
}

// extraBox enrolls another box in region r with the given id (a spare
// for the nested scenarios).
func (w *world) extraBox(r, id int) *box {
	b := &box{ID: id, Region: r, HasKey: true, Enrolled: true, DeleteOnTermination: true}
	b.forget()
	w.boxes = append(w.boxes, b)
	return b
}

// running drives the serving host until it runs on b.
func (w *world) running(t *testing.T, b *box) {
	w.reach(t, 60, "the host runs", func() bool { return hasMain(b) }, w.actAll(), func() { w.wake(b) })
}

// drainTo starts a drain of the serving host and fails the test unless
// it named to (any target when to is nil).
func (w *world) drainTo(t *testing.T, to *box) {
	w.userRelocate()
	if c := w.claim; c.Op.Kind != "relocating" || to != nil && c.Op.Tgt != to.ID {
		t.Fatalf("witness: the drain did not start (to %v): %s", to, w.state())
	}
}

// relocPhase drives until the relocation is in phase p (or beyond it, by
// the table's order), with the actors given.
func (w *world) relocPhase(t *testing.T, p string, actors ...func()) bool {
	order := []string{"starting", "releasing", "pinning", "installing", "installed", "probing", "probed", "opened"}
	at := func(x string) int {
		for i, o := range order {
			if o == x {
				return i
			}
		}
		return -1
	}
	return w.reach(t, 1500, "relocation at "+p, func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && at(c.Op.Phase) >= at(p) || c.Op.Kind == "serving" && p == "serving"
	}, actors...)
}

// servingOn drives until the host serves on b (on any box, when b is
// nil), with every actor the at-rest model has.
func (w *world) servingOn(t *testing.T, b *box) bool {
	return w.reach(t, 3000, "serving on the new executor", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && (b == nil && c.Op.Exec >= 0 || b != nil && c.Op.Exec == b.ID)
	}, w.actAll(), w.recoveryPass, w.operator, w.wakeExec)
}

// wakeExec wakes the executor, whichever box it is.
func (w *world) wakeExec() {
	if e := w.execBox(w.claim); e != nil {
		w.wake(e)
	}
}

// timerOn registers an acknowledged timer on b's running host.
func (w *world) timerOn(b *box) int {
	w.nextW++
	b.Timers = append(b.Timers, w.nextW)
	sess, gen := w.sessionOf(b)
	w.acks = append(w.acks, ack{W: w.nextW, Inc: w.claim.Holder.Inc, Box: b.ID, Session: sess, Gen: gen, At: w.now, Kind: "timer"})
	return w.nextW
}

// spareID is an id for an extra box of region 0 whose tieHash beats box
// 0's, so choose() prefers it to the source.
func spareID() int {
	for id := 40; ; id++ {
		if tieHash(id) > tieHash(0) {
			return id
		}
	}
}

func never() bool { return false }

// published drives until a manifest of b's host has landed (the files
// publisher runs in every two-box tick): a failover of a host none of
// whose manifests landed is finding F2's, not the scenario's.
func (w *world) published(t *testing.T, b *box) bool {
	return w.reach(t, 60, "the host's first manifest landed", func() bool {
		_, _, _, _, m := w.newestManifest(b.Region, w.claim.Holder.Location, w.claim.Holder.Gen)
		return m != nil
	}, w.act(b), func() { w.completeDue(true) })
}

// ---------- the witnesses (one per safety guard) ----------

func wRelocReleaseFirst(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	// only the target's worker runs while the source should be releasing:
	// with the rule it waits for the release; without it, it pins and
	// installs while the source still runs the host
	w.drive(12, never, w.act(to))
	w.servingOn(t, to)
	w.finish(t)
}

func wRelocInstallReceipt(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	third := w.extraBox(0, 30)
	if !w.reach(t, 1500, "the target's receipt", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Phase == "installed" && c.Op.Receipt != nil && c.Op.Receipt.Box == to.ID
	}, w.act(from, to), w.recoveryPass) {
		return
	}
	w.terminate(to)
	w.servingOn(t, third)
	w.finish(t)
}

// probeOpened drives until the target's first attempt has opened P and
// is still starting (L18 finding 2: once its app answers, the probe has
// passed and is stopped, never restarted).
func (w *world) probeOpened(t *testing.T, from, to *box) bool {
	w.openYield = func(b *box) bool { return b == to } // the attempt pauses after its admission, before replication opens
	defer func() { w.openYield = nil }()
	return w.reach(t, 1500, "P opened by an attempt still starting", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Phase == "probing" && preparedOpened(c) && to.open != nil
	}, w.act(from, to), w.recoveryPass)
}

// wRelocNoReinstall, wRelocVerifyReceipt and wRelocPreparedOpenedJournal
// make a later attempt after P's opening, which root cannot launch while
// claim ask C11 is open (finding 4): they show the rules C11 makes live.
func wRelocNoReinstall(t *testing.T, w *world) {
	w.twoBox()
	w.claimC11 = true
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	if !w.probeOpened(t, from, to) {
		return
	}
	w.timerOn(to)
	w.crash(to)
	w.servingOn(t, to)
	w.finish(t)
}

func wRelocReleasedCoverage(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.appWrite(from)
	loc := w.claim.Holder.Location
	w.holdKey(loc) // every upload of the executor is held
	w.replicate(from)
	w.hold = nil
	w.terminate(from)
	w.servingOn(t, to)
	w.completeHeld(loc) // the dead box's upload lands late, after the pin
	w.finish(t)
}

func wRelocSourceLoss(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.appWrite(from)
	loc := w.claim.Holder.Location
	w.holdKey(loc) // the flush and the final manifest are held: no release
	w.drainTo(t, to)
	w.relocPhase(t, "releasing", w.act(from, to), w.recoveryPass)
	w.drive(20, never, w.act(from, to))
	if w.claim.Op.Release != nil {
		t.Fatalf("witness: the release was journaled while its uploads were held")
	}
	w.hold = nil
	w.terminate(from)
	w.servingOn(t, to)
	w.completeHeld(loc)
	w.finish(t)
}

func wRelocRowCases(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	if !w.reach(t, 1500, "the source released", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Release != nil && c.Op.Release.Box == from.ID && !hasMain(from)
	}, w.act(from, to), w.recoveryPass) {
		return
	}
	// a paused duplicate of the source's earlier open lands now: its row
	// is ready again while its admission fence is closed. A request
	// reaches the box through its stale edges file, then reconcile decides
	// the row (in that order each step, so the request also meets the row
	// in the step after the completion, before the catch-up hold)
	from.Row.Status, from.Row.HoldOp = "ready", ""
	w.drive(80, never, func() {
		w.wake(from)
		w.reconcile(from)
	}, w.act(to), w.recoveryPass, func() { w.wake(to) })
	w.servingOn(t, to)
	w.finish(t)
}

func wFenceDischargeQualified(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	from.DeleteOnTermination = false
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	w.servingOn(t, to)
	if !w.owesLocal(w.claim, from) {
		t.Fatalf("witness: the former executor owes no park-local: %s", w.state())
	}
	w.terminate(from)
	w.reach(t, 60, "the former executor fenced", func() bool { return contains(w.claim.Lost, from.ID) }, w.act(to))
	w.finish(t)
}

func wFenceDeclaresTimers(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.publishFiles(from)
	w.completeDue(true)
	w.holdKey(w.claim.Holder.Location + "files/") // the manifest that would carry the timer never lands
	w.timerOn(from)
	w.publishFiles(from)
	w.hold = nil
	w.terminate(from)
	w.servingOn(t, to)
	w.completeHeld(w.claim.Holder.Location + "files/") // it lands after the pin: a version nothing reads
	w.finish(t)
}

func wRelocEnrolledOnly(t *testing.T, w *world) {
	w.twoBox()
	from, peer := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.published(t, from)
	peer.HasKey = false
	w.terminate(from)
	w.reach(t, 100, "the executor fenced", func() bool { return contains(w.claim.Lost, from.ID) }, w.act(peer), w.recoveryPass)
	// the replacement launches unenrolled; under the mutation it is named
	// before it enrolls (INV-27)
	w.servingOn(t, nil)
	w.finish(t)
}

func wFilesLiveSet(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.timerOn(from)
	if !w.reach(t, 30, "the origin's manifest landed", func() bool {
		_, _, _, _, m := w.newestManifest(0, w.claim.Holder.Location, w.claim.Holder.Gen)
		return m != nil
	}, w.actAll(), func() { w.completeDue(true) }) {
		return
	}
	// the agent restarts: the next wake is a cold admission into a fresh
	// generation, which publishes nothing (nothing changed), and the
	// origin's park-generation runs once the new marker lands
	w.crash(from)
	if !w.reach(t, 600, "a fresh generation serving and the origin parked", func() bool {
		c := w.claim
		if c.Holder.Gen == "" || !hasMain(from) || from.open != nil {
			return false
		}
		for _, o := range c.Obl {
			if o.Kind == "park-generation" {
				return false
			}
		}
		return true
	}, w.actAll(), w.recoveryPass, func() { w.wake(from) }) {
		return
	}
	w.terminate(from)
	w.servingOn(t, to)
	w.finish(t)
}

// rev 2: 1b-next rev 16's guards

func wRelocPinClaimGen(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	third := w.extraBox(0, 30)
	// the first round, observed by a pinner after the source's release and
	// before any pin; it stalls before its journal CAS
	if !w.reach(t, 1500, "the source's release journaled", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Release != nil && c.Op.Pin == nil
	}, w.act(from, to)) {
		return
	}
	stale := w.claim.clone()
	if !w.reach(t, 1500, "the target's probe writes into P", func() bool { return preparedOpened(w.claim) && hasMain(to) }, w.act(from, to), w.recoveryPass, func() { w.wake(to) }) {
		return
	}
	w.appWrite(to)
	w.replicate(to)
	w.completeDue(true) // the target's write is in P's remote before it dies
	w.terminate(to)
	if !w.reach(t, 100, "the nested pinning", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Phase == "pinning" && c.Op.Pin == nil && contains(c.Lost, to.ID)
	}, w.act(third)) {
		return
	}
	w.now += failoverPinDelay
	w.relocPinning(third, stale) // the stalled pinner resumes with the first round's selection
	w.servingOn(t, third)
	w.finish(t)
}

func wRelocReleaseDeadline(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.appWrite(from)
	loc := w.claim.Holder.Location
	w.holdKey(loc) // the source's flush never completes
	w.drainTo(t, to)
	w.reach(t, 3*releaseDeadline+200, "the host serving again", func() bool {
		c := w.claim
		return c.Op.Kind == "serving"
	}, w.actAll(), w.recoveryPass, w.operator)
	w.hold = nil
	w.completeHeld(loc)
	w.finish(t)
}

func wRelocEnsureLocal(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	if !w.reach(t, 1500, "the target's receipt journaled", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Receipt != nil && c.Op.Receipt.Box == to.ID
	}, w.act(from, to), w.recoveryPass) {
		return
	}
	// the target freezes past its execution lease: its relocation
	// controller is lost, and the scoped cleanup retires the unopened
	// local install when it runs again
	to.Frozen = true
	w.drive(execTTL+2, never, w.act(from))
	to.Frozen = false
	w.servingOn(t, to)
	w.finish(t)
}

func wRelocVerifyReceipt(t *testing.T, w *world) {
	w.twoBox()
	w.claimC11 = true
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	third := w.extraBox(0, 30)
	if !w.probeOpened(t, from, to) {
		return
	}
	// the volume holding the host's directory is replaced and the agent
	// restarts: the files, the database and the install journal are gone
	w.crash(to)
	to.Files, to.DB, to.receipt = nil, nil, nil
	w.servingOn(t, nil)
	_ = third
	w.finish(t)
}

func wRelocPreparedOpenedJournal(t *testing.T, w *world) {
	w.twoBox()
	w.claimC11 = true
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	if !w.probeOpened(t, from, to) {
		return
	}
	w.timerOn(to)
	w.crash(to) // the next attempt admits a fresh generation from P's
	w.openYield = func(b *box) bool { return b == to }
	reached := w.reach(t, 400, "the second attempt admitted and still starting", func() bool {
		c := w.claim
		return c.Op.Phase == "probing" && c.Holder.Gen != c.Op.Prepared && to.open != nil
	}, w.act(from, to), w.recoveryPass)
	w.openYield = nil
	if !reached {
		return
	}
	// holder.admissions is trimmed as AdmitWriter trims it past eight:
	// P's admission is no longer listed
	w.cas(w.claim, func(n *claim) bool {
		var keep []admissionRecord
		for _, e := range n.Holder.Admissions {
			if e.Gen != n.Op.Prepared {
				keep = append(keep, e)
			}
		}
		n.Holder.Admissions = keep
		return true
	})
	w.crash(to)
	w.servingOn(t, to)
	w.finish(t)
}

func wFenceReassignsRegional(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	// a dead generation of the location, with an object in region 0, whose
	// park-generation chore carries the legacy owner form box:0
	loc := w.claim.Holder.Location
	k := ltxKey(loc, "old", 0, 1, 1)
	w.buckets[0].objs[k] = []*version{{ID: w.newVer(), B: &blob{Kind: "ltx", Level: 0, Min: 1, Max: 1, Inc: 1}}}
	id := w.newID("ob")
	w.cas(w.claim, func(n *claim) bool {
		n.Obl = append(n.Obl, w.scoped(obligation{ID: id, Op: "old-op", Kind: "park-generation", Region: 0, Box: -1, Prefix: loc, Ret: "old", Quiet: w.now, Owner: boxOwner(from.ID)}, w.now, 1))
		return true
	})
	w.terminate(from)
	w.servingOn(t, to)
	w.reach(t, 600, "the dead box's regional chore done", func() bool { return !hasObl(w.claim.Obl, id) }, w.actAll(), w.recoveryPass)
	w.finish(t)
}

func wRelocReplaceNotOwing(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	// box 30 owes a park-local for the host: stale artifacts of an earlier
	// stay under stamp 7
	spare := w.extraBox(0, 30)
	spare.DB = &localDB{Inc: 1, Loc: w.claim.Holder.Location, State: map[string]int{}, Stamp: 7, Epoch: spare.Epoch}
	spare.install(1, 7)
	w.cas(w.claim, func(n *claim) bool {
		n.Obl = append(n.Obl, obligation{ID: w.newID("ob"), Op: "stay", Kind: "park-local", Box: spare.ID, Region: -1, Ret: "stay"})
		return true
	})
	if !w.relocPhase(t, "installing", w.act(from, to), w.recoveryPass) {
		return
	}
	w.terminate(to)
	w.servingOn(t, nil)
	w.finish(t)
}

func wRelocExhaustedFence(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	b30, b31 := w.extraBox(0, 30), w.extraBox(0, 31)
	w.userRelocate()
	if w.claim.Op.Kind != "relocating" {
		t.Fatalf("witness: no drain")
	}
	// three targets die in starting, the live source op.exec throughout
	for _, x := range []*box{w.boxByID(w.claim.Op.Tgt), b30, b31, to} {
		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 w.claim.Op.Stuck == "" {
		t.Fatalf("witness: the exhausted target did not make the operation stuck: %s", w.state())
	}
	w.extraBox(0, spareID()) // a spare the operator's replay may choose
	w.servingOn(t, nil)
	w.finish(t)
}

func wFilesPreparedSeqZero(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	for i := 0; i < 3; i++ { // the source publishes three manifests: its sequence reaches 3
		w.timerOn(from)
		w.publishFiles(from)
		w.completeDue(true)
	}
	w.drainTo(t, to)
	third := w.extraBox(0, 30)
	if !w.reach(t, 1500, "the probe runs on the target", func() bool { return preparedOpened(w.claim) && hasMain(to) }, w.act(from, to), w.recoveryPass, func() { w.wake(to) }) {
		return
	}
	w.timerOn(to) // the probe session's timer, published at P's next sequence
	w.publishFiles(to)
	w.completeDue(true)
	w.terminate(to)
	w.servingOn(t, third)
	w.finish(t)
}

func wRelocPinEvidencedEmpty(t *testing.T, w *world) {
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	_ = to
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	// the listing finds nothing although the writes are durable (ruling
	// 15: a listing that finds nothing is no evidence); the model hides
	// the objects for a while, then the listing shows them again
	bk := w.buckets[0]
	hidden := map[string][]*version{}
	for _, k := range bk.keys(w.claim.Holder.Location) {
		if isDataKey(k) {
			hidden[k] = bk.objs[k]
			delete(bk.objs, k)
		}
	}
	w.terminate(from)
	w.drive(60, never, w.act(to), w.recoveryPass)
	for k, vs := range hidden {
		bk.objs[k] = vs
	}
	w.servingOn(t, nil)
	w.finish(t)
}

// rev 3: 1b-next rev 17 to 19's fixes for findings F1 to F3, and L21

// wL21Window (files-pin-needs-key): the L21 window, explicit. The cold
// admission's pre-start manifest copy has landed without instance.key
// (§5.6 copies the manifest before the app starts); the app minted the
// key after it started, wrote rows sealed with it and replicated them;
// the publisher's manifest carrying the key has not landed when the box
// is terminated. Recovery fails safe: the pin refuses, the operation is
// stuck, and no database is installed without its key. Without the rule
// the keyless manifest is pinned and the install mints a fresh key over
// the sealed rows (INV-4).
func wL21Window(t *testing.T, w *world) {
	w.twoBox()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	h := w.claim.Holder
	w.issued++
	pre := &blob{Kind: "files", Inc: h.Inc, Gen: h.Gen, Seq: 1, Base: w.issued} // the pre-start copy: no files, no key
	k := manifestKey(h.Location, h.Gen, 1)
	w.buckets[0].objs[k] = []*version{{ID: w.newVer(), B: pre}}
	from.pubGen, from.pubSeq = h.Gen, 1
	w.holdKey(h.Location + "files/") // the manifest with the key never lands
	w.running(t, from)
	w.appWrite(from)
	w.replicate(from)
	w.drive(30, never, w.act(from)) // the rows' uploads land; the manifest stays held
	w.terminate(from)
	w.drive(400, func() bool { return len(w.violations) > 0 || w.claim.Op.Stuck != "" }, w.actAll(), w.recoveryPass, w.wakeExec)
	if len(w.violations) == 0 {
		if c := w.claim; c.Op.Stuck == "" || c.Op.Prepared != "" {
			w.violate("INV-3 (witness): the L21 failover neither stopped nor installed: %s", w.state())
		}
	}
}

// wNoSpareBound (reloc-no-spare-bound, F1): a drain's target fails its
// probe in a two-box region; with the bound the operation is stuck and
// the operator returns the host to the source; without it, it waits.
func wNoSpareBound(t *testing.T, w *world) {
	w.twoBox()
	w.claimC1 = true // the failed target's release (claim ask C1, finding 4)
	from, to := w.boxes[0], w.boxes[3]
	to.FailApp = true
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	w.servingOn(t, from)
	w.finish(t)
}

// wDeadTargetBound (reloc-dead-target-bound, rev 18 finding 2): the
// drain's target is terminated at installing and the group launches no
// replacement (region 0 kept at one box: a capacity shortage), so no
// spare exists. With the bound the fence is consumed ReleaseDeadline
// later, the operation is stuck, and the operator's replay serves the
// host on the source; without it nothing moves.
func wDeadTargetBound(t *testing.T, w *world) {
	w.twoBox()
	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
	if !w.relocPhase(t, "installing", w.act(from, to), w.recoveryPass) {
		return
	}
	w.terminate(to)
	w.servingOn(t, from)
	w.finish(t)
}

// rev 4: Astra's round 1, claim §11's witnesses

// wMoveUnchangedFile (files-objects-region-bound; claim §11 "move
// landing, source cleanup, then target failure with an unchanged file"):
// the host moves to region 1 carrying a file unchanged since region 0
// published it; the landing's prepared manifest and the target's
// publications name an object in region 1's bucket; the source region is
// cleaned up; the target is terminated and the failover in region 1 pins
// that object. Without the region binding the manifests keep naming
// region 0's object, which no failover in region 1 can pin (INV-3).
func wMoveUnchangedFile(t *testing.T, w *world) {
	w.twoBox()
	src := w.boxes[0]
	w.serve(src, 1, app0)
	w.running(t, src)
	w.published(t, src)
	w.moveTo(1)
	if !w.servingOn(t, nil) {
		return
	}
	e := w.execBox(w.claim)
	if e.Region != 1 {
		w.violate("witness: the move did not land in region 1: %s", w.state())
		return
	}
	w.reach(t, 1200, "the source region cleaned up", func() bool { return len(w.claim.Obl) == 0 }, w.actAll(), w.recoveryPass)
	w.published(t, e)
	w.terminate(e)
	w.reach(t, 3000, "serving again after the target's failover", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec >= 0 && c.Op.Exec != e.ID
	}, w.actAll(), w.recoveryPass, w.operator, w.wakeExec)
	w.finish(t)
}

// wOriginsAfterTrim (files-origins-referenced; 1b-next rev 12 finding 1,
// R11-1): a file published in the origin generation never changes while
// ten cold admissions follow, so holder.admissions is trimmed past it
// while the live manifest still names the origin's object. The failover
// pins that object; authorized from the trimmed admissions alone, the pin
// is refused for ever (INV-3).
func wOriginsAfterTrim(t *testing.T, w *world) {
	w.twoBox()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	for i := 0; i < 10; i++ {
		w.crash(from)
		if !w.reach(t, 400, "a fresh cold admission serving", func() bool {
			return hasMain(from) && from.open == nil && len(w.claim.Obl) == 0
		}, w.actAll(), w.recoveryPass, func() { w.wake(from) }) {
			return
		}
	}
	for _, e := range w.claim.Holder.Admissions {
		if e.Gen == "" || e.Predecessor.Gen == "" {
			w.violate("witness: the origin generation is still in holder.admissions: %s", w.state())
			return
		}
	}
	w.terminate(from)
	w.servingOn(t, w.boxes[3])
	w.finish(t)
}

// wClearAfterCleanup (reloc-clear-after-cleanup; claim §11 "a failed
// cleanup leaving a controller record that blocks executable updates"):
// the drain completes; the former source's participation is over, but its
// scoped cleanup fails, so its controller record stays and root refuses
// an executable update; once the cleanup succeeds the record is cleared
// and the update runs. Without the rule the record is cleared after the
// failed cleanup and the update runs over it (INV-29).
func wClearAfterCleanup(t *testing.T, w *world) {
	w.twoBox()
	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)
	from.cleanupFails = true
	if !w.servingOn(t, to) {
		return
	}
	w.drive(20, never, w.act(from))
	if w.updateExecutable(from) {
		if on("reloc-clear-after-cleanup") {
			w.violate("witness: an update ran while the controller record existed")
		}
		return
	}
	from.cleanupFails = false
	w.reach(t, 100, "the record cleared", func() bool { return from.execRecord == "" }, w.act(from))
	if !w.updateExecutable(from) {
		w.violate("INV-3 (witness): the update stayed refused after a successful cleanup")
	}
	w.finish(t)
}

// wCommittedLandingTargetLost (fence-converts-committed-landing; claim
// §11 "automatic target loss racing the record commit CAS", finding F5):
// the record commits the landing, and the target is terminated before
// the claim follows; the fence makes the committed landing a failover in
// region 1, which serves; without it nothing ever finishes it (INV-3).
func wCommittedLandingTargetLost(t *testing.T, w *world) {
	w.twoBox()
	src := w.boxes[0]
	w.serve(src, 1, app0)
	w.running(t, src)
	w.published(t, src)
	w.moveTo(1)
	tgt := w.boxByID(w.claim.Op.Tgt)
	if !w.stepUntil(t, 1500, "the record committed, the claim not yet", func() bool {
		c := w.claim
		return w.rec != nil && w.rec.MoveState == "committed" && c.Op.Kind == "landing" && !c.Op.Committed
	}) {
		return
	}
	w.terminate(tgt)
	w.servingOn(t, nil)
	w.finish(t)
}

// wPushKeepsLiveFiles (files-live-set-moves-parks; finding F4): a push
// parks the whole location before installing its generation; the files
// live set survives it, so a later failover pins the live manifest's
// object. Without the rule the push archives them, and the failover can
// never pin the host's files (INV-3).
func wPushKeepsLiveFiles(t *testing.T, w *world) {
	w.twoBox()
	from := w.boxes[0]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.published(t, from)
	w.userPush()
	if !w.reach(t, 800, "the push served", func() bool { c := w.claim; return c.Op.Kind == "serving" && c.Holder.Gen != "" && len(c.Obl) == 0 }, w.actAll(), w.recoveryPass, func() { w.wake(from) }) {
		return
	}
	w.drive(10, never, w.act(from))
	w.completeDue(true)
	w.terminate(from)
	w.servingOn(t, w.boxes[3])
	w.finish(t)
}

// wLateManifestAfterMove (files-live-region; joint round 1 finding 1,
// claim §6.2a): the source in region 0 changes a file; the first PUT of
// its new object is held, the publisher's retry lands, and the manifest
// naming that object is held too. The host moves to region 1, which
// admits a generation whose chain (genParent) leads back to the
// source's; the source region is cleaned up, and only then do both held
// PUTs land in region 0. Region 0 is no longer authoritative, so the late
// manifest and the late object are residue there, and the at-rest sweep
// retires both. Read without its region, the files live set finds the
// late manifest as the chain's newest in region 0 and the object as the
// one it names; they protect each other, and the sweep never retires
// them (INV-10).
func wLateManifestAfterMove(t *testing.T, w *world) {
	w.twoBox()
	src := w.boxes[0]
	w.serve(src, 1, app0)
	w.running(t, src)
	w.published(t, src)
	loc := w.claim.Holder.Location
	src.Files.Env++ // a changed file: its object first, then the manifest naming it
	obj := objectKey(loc, w.claim.Holder.Gen, src.Files)
	objHeld := false
	w.hold = func(r *request) bool {
		if r.key == obj && !objHeld {
			objHeld = true // the object's first PUT; the publisher's retry lands
			return true
		}
		return r.region == src.Region && isManifestKey(r.key) && strings.HasPrefix(r.key, loc)
	}
	held := func() (objs, mans int) {
		for _, r := range w.pending {
			if r.region == src.Region && r.doneAt == 1<<30 {
				if r.key == obj {
					objs++
				} else if isManifestKey(r.key) {
					mans++
				}
			}
		}
		return objs, mans
	}
	if !w.reach(t, 3*uploadTimeout, "the object's first PUT and the manifest naming it held", func() bool {
		o, m := held()
		return o == 1 && m == 1
	}) {
		return
	}
	w.hold = nil
	if src.Files.Obj != obj {
		t.Fatalf("witness: the held manifest does not name the held object %s: %+v", obj, src.Files)
	}
	w.moveTo(1)
	if w.claim.Op.Kind != "moving" {
		t.Fatalf("witness: the move did not start: %s", w.state())
	}
	if !w.servingOn(t, nil) {
		return
	}
	if e := w.execBox(w.claim); e == nil || e.Region != 1 {
		w.violate("witness: the move did not land in region 1: %s", w.state())
		return
	}
	if !w.reach(t, 1200, "the source region cleaned up", func() bool { return len(w.claim.Obl) == 0 }, w.actAll(), w.recoveryPass) {
		return
	}
	if n := w.completeHeld(loc + "files/"); n != 2 {
		t.Fatalf("witness: held object and manifest PUTs completed: %d, want 2", n)
	}
	w.finish(t)
}

// wMoveStoppedKeepsLiveFiles (files-home-move-phases; joint round 1
// review I1, moves §5.1's authoritative-region table): the source in
// region 0 serves with a keyed manifest in the origin generation. A move
// to region 1 starts, and the source's agent restarts inside it: its cold
// admission into a fresh generation publishes the marker, which queues
// the origin's park-generation in region 0 (PublishGeneration, legal
// inside an operation a writer runs in), and publishes no manifest of its
// own (nothing changed), so the origin's manifest stays the chain's
// newest. The source stops (op.exec none) and only then does the
// sibling, box 3, run the park-generation. The source is terminated and
// the fence converts the move to a failover in region 0. Region 0 is the
// authoritative region before the flip, so the files live set kept the
// manifest and the failover pins it and serves; with the files home read
// from op.exec alone it was archived, and the failover finds no manifest
// carrying instance.key: stuck for ever (INV-3).
func wMoveStoppedKeepsLiveFiles(t *testing.T, w *world) {
	w.twoBox()
	src, sib := w.boxes[0], w.boxes[3]
	w.serve(src, 1, app0)
	w.running(t, src)
	w.published(t, src)
	loc := w.claim.Holder.Location
	if _, _, _, _, m := w.newestManifest(0, loc, w.claim.Holder.Gen); !manifestHasKey(m) {
		t.Fatalf("witness: the origin's manifest carries no instance.key: %s", w.state())
	}
	w.moveTo(1)
	if w.claim.Op.Kind != "moving" {
		t.Fatalf("witness: the move did not start: %s", w.state())
	}
	tgt := w.boxByID(w.claim.Op.Tgt)
	w.crash(src)
	parkQueued := func() bool {
		for _, o := range w.claim.Obl {
			if o.Kind == "park-generation" && o.Region == src.Region && o.Ret == "@root" {
				return true
			}
		}
		return false
	}
	if !w.reach(t, 200, "a fresh generation inside the move and the origin's park-generation queued", func() bool {
		return w.claim.Holder.Gen != "" && parkQueued()
	}, func() { w.worker(src); w.reconcile(src); w.wake(src) }) {
		return
	}
	if !w.reach(t, 400, "the source stopped", func() bool {
		c := w.claim
		return c.Op.Kind == "moving" && c.Op.Exec == -1
	}, func() { w.worker(src); w.reconcile(src); w.worker(tgt); w.reconcile(tgt) }) {
		return
	}
	if !w.reach(t, 400, "the sibling ran the origin's park-generation while the move was stopped", func() bool { return !parkQueued() }, w.act(sib)) {
		return
	}
	if c := w.claim; c.Op.Kind != "moving" {
		t.Fatalf("witness: the move left its pre-flip phases before the park: %s", w.state())
	}
	w.terminate(src)
	w.servingOn(t, nil)
	w.finish(t)
}

// wRollbackKeepsSourceFiles (files-home-move-phases; Astra joint round
// 2): the stopped-move witness's setup, the source's restart inside a move
// to region 1 queuing the origin's park-generation in region 0 while the
// origin holds the keyed manifest, but the park waits until the move has
// flipped and the target's probe is admitted (holder.generation is the
// target's), before the commit. The sibling runs it then. The landing
// misses its cut deadline and rolls back cleanly: the return names the
// source's generation again in region 0, and the source serves there with
// its files unchanged, so the publisher has nothing to republish. The
// source is then terminated. Until the commit a rollback is possible, so
// region 0 stayed the files home and the failover there pins the keyed
// manifest and serves; with region 0 home only until the probe's
// admission (dfab457c) or never (the mutation) the park archived it, and
// the failover finds no manifest carrying instance.key: stuck for ever
// (INV-3).
func wRollbackKeepsSourceFiles(t *testing.T, w *world) {
	w.twoBox()
	src, sib := w.boxes[0], w.boxes[3]
	w.serve(src, 1, app0)
	w.running(t, src)
	w.published(t, src)
	loc := w.claim.Holder.Location
	published := w.stats["files manifests published"]
	if _, _, _, _, m := w.newestManifest(0, loc, w.claim.Holder.Gen); !manifestHasKey(m) {
		t.Fatalf("witness: the origin's manifest carries no instance.key: %s", w.state())
	}
	w.moveTo(1)
	if w.claim.Op.Kind != "moving" {
		t.Fatalf("witness: the move did not start: %s", w.state())
	}
	tgt := w.boxByID(w.claim.Op.Tgt)
	w.crash(src)
	parkQueued := func() bool {
		for _, o := range w.claim.Obl {
			if o.Kind == "park-generation" && o.Region == src.Region && o.Ret == "@root" {
				return true
			}
		}
		return false
	}
	if !w.reach(t, 200, "a fresh generation inside the move and the origin's park-generation queued", func() bool {
		return w.claim.Holder.Gen != "" && parkQueued()
	}, func() { w.worker(src); w.reconcile(src); w.wake(src) }) {
		return
	}
	srcGen := w.claim.Holder.Gen
	if !w.reach(t, 400, "the target's probe admitted, before the commit", func() bool {
		c := w.claim
		return c.Op.Kind == "landing" && !c.Op.Committed && c.Holder.Gen != srcGen && w.rec.MoveState == "landing"
	}, func() { w.worker(src); w.reconcile(src); w.worker(tgt); w.reconcile(tgt); w.runConsole() }) {
		return
	}
	if !w.reach(t, 400, "the sibling ran the origin's park-generation before the commit", func() bool { return !parkQueued() }, w.act(sib)) {
		return
	}
	if c := w.claim; c.Op.Kind != "landing" || c.Op.Committed {
		t.Fatalf("witness: the landing committed before the park: %s", w.state())
	}
	// the target's agent stalls past the cut deadline; it rolls back when it
	// next acts, having written nothing
	deadline := w.claim.Op.CutDeadline
	w.drive(deadline+1-w.now, never, w.act(sib))
	if !w.reach(t, 1500, "the clean return serving on the source", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec == src.ID // the return named srcGen; the re-admission moved to a fresh generation from it
	}, w.actAll(), w.recoveryPass, w.runConsole, func() { w.wake(src) }) {
		return
	}
	if w.stats["rollback-clean"] == 0 {
		t.Fatalf("witness: the rollback was not clean: %s", w.state())
	}
	w.drive(20, never, w.act(src))
	w.completeDue(true)
	if n := w.stats["files manifests published"]; n != published {
		t.Fatalf("witness: the publisher republished (%d manifests, %d before the move): the files are unchanged", n, published)
	}
	if _, _, _, _, m := w.newestManifest(src.Region, loc, w.claim.Holder.Gen); m == nil && on("files-home-move-phases") {
		t.Fatalf("witness: no manifest in the source region after the return: %s", w.state())
	}
	w.terminate(src)
	w.reach(t, 3000, "serving again after the source's failover", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec >= 0 && c.Op.Exec != src.ID
	}, w.actAll(), w.recoveryPass, w.operator, w.wakeExec)
	w.finish(t)
}

// wParkEvidenceEveryBox (park-evidence-every-box; final review Important
// 2, the shape of two-box seed 819): the host serves on box 0 with an
// acknowledged write its replicator has not shipped, and a delete
// begins. Region 0's park-replica is a regional chore, and box 3, the
// sibling, runs it while box 0's agent does nothing (its replicator, a
// process of its own, still runs). With the rule box 3 waits for box 0's
// stop evidence, which comes only when box 0's agent acts, and the park
// covers the write; without it box 3 parks region 0 on its own evidence,
// box 0's replicator then ships the write into the parked region, and the
// host is freed with it as residue (INV-10).
func wParkEvidenceEveryBox(t *testing.T, w *world) {
	w.twoBox()
	writer, sib := w.boxes[0], w.boxes[3]
	w.serve(writer, 1, app0)
	w.running(t, writer)
	w.appWrite(writer)
	w.userDelete()
	if w.claim.Op.Kind != "deleting" {
		t.Fatalf("witness: the delete did not begin: %s", w.state())
	}
	owed := func() bool {
		for _, o := range w.claim.Obl {
			if o.Kind == "park-replica" && o.Region == writer.Region && o.Prefix == w.claim.Holder.Location {
				return true
			}
		}
		return false
	}
	if !w.reach(t, 50, "region 0's park-replica queued", owed, w.runConsole) {
		return
	}
	w.drive(6*horizonH, func() bool { return !owed() }, w.runConsole, w.act(sib), w.recoveryPass)
	w.logf("witness: box 0's replicator ships its queue")
	w.replicate(writer)
	w.finish(t)
}

// TestFilesWitnessesStuckWithoutINV30 keeps the liveness half of the files
// witnesses (joint round 2). INV-30 now catches each of their mutations
// at the archive, so TestWitnesses names INV-30; with INV-30 off, the
// same scenario under the same mutation must still end in the failure it
// was written for, the failover that can never pin the host's files
// (INV-3), and must hold with the rule in place.
func TestFilesWitnessesStuckWithoutINV30(t *testing.T) {
	saved := mutated
	defer func() { mutated = saved }()
	n := 0
	for _, x := range witnesses {
		if x.want != "INV-30" || x.rule == "dirty-return-needs-admission" {
			// dirty-return-needs-admission's loss has no liveness consequence
			// in its scenario (the dirty return serves from the returned
			// seal): INV-30, the independent preservation invariant, is its
			// only oracle (Astra joint round 4)
			continue
		}
		n++
		for _, m := range []string{"", x.needs} {
			mutated = m
			w := newWitness()
			w.noPinWatch = true
			x.run(t, w)
			switch {
			case m == "" && len(w.violations) > 0:
				t.Errorf("witness %s (INV-30 off) violated with its rule in place: %s", x.rule, w.violations[0])
			case m != "" && (len(w.violations) == 0 || !namedFailure(w.violations[0], "INV-3")):
				dump(t, w)
				t.Errorf("witness %s under %q with INV-30 off: want INV-3, got %v", x.rule, m, w.violations)
			}
		}
	}
	if n != 4 {
		t.Fatalf("files witnesses naming INV-30: %d, want 4 (files-live-set, files-live-set-moves-parks, files-home-move-phases twice)", n)
	}
}