package hostlife

import "testing"

// ---------- L18 joint round findings 2 to 4: witnesses and direct tests ----------

// completedDrain drains the host from box 0 to box 3 with fault set on
// the target first, and drives until the claim serves on box 3 with its
// completion recorded (regionha_completion_test.go).
func (w *world) completedDrain(t *testing.T, fault func(to *box)) (from, to *box, ok bool) {
	w.twoBox()
	from, to = w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	if fault != nil {
		fault(to)
	}
	w.drainTo(t, to)
	w.extraBox(0, 30) // a target for a successor drain from box 3 (box 0 owes its park-local)
	ok = w.reach(t, 1500, "serving on box 3 with its completion recorded", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec == to.ID && c.Op.Completed != nil
	}, w.act(from, to), w.recoveryPass)
	return from, to, ok
}

// marked: the serving op's completion is marked discharged.
func (w *world) marked() bool {
	d := w.claim.Op.Completed
	return w.claim.Op.Kind == "serving" && d != nil && d.Discharged
}

// wProbeStopBeforeServing (reloc-probe-stop): a drain completes; the
// probe's session is stopped with root's receipt before the serving CAS,
// and the host wakes on box 3 afterwards. Completed over the running
// probe, its session outlives the kind change on a box whose
// participation is not discharged (INV-31).
func wProbeStopBeforeServing(t *testing.T, w *world) {
	from, to, ok := w.completedDrain(t, nil)
	if !ok {
		return
	}
	_ = from
	w.reach(t, 200, "the host runs on box 3 after the completion", func() bool { return hasMain(to) && w.marked() }, w.actAll(), w.recoveryPass, func() { w.wake(to) })
	w.finish(t)
}

// wProbeStopMeetsAnotherFence (reloc-probe-fence-owner, the stop): the
// probe has passed when another operation's fence closes the host on box
// 3. The probe stop refuses with nothing stopped, and ruling 67 makes the
// relocation stuck for an operator. Installed over that fence, the stop
// later lifts it as its own, and the serving wake starts the host under a
// fence its owner never lifted (INV-32).
func wProbeStopMeetsAnotherFence(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 probe passed", func() bool {
		c := w.claim
		return c.Op.Kind == "relocating" && c.Op.Phase == "probed" && c.Op.ProbeStop == nil
	}, w.act(from, to), w.recoveryPass) {
		return
	}
	to.actFence, to.foreignFence = "op-elsewhere", "op-elsewhere"
	w.reach(t, 3*releaseDeadline, "stuck with the probe unstopped", func() bool { return w.claim.Op.Stuck != "" }, w.act(from, to), w.recoveryPass, func() { w.wake(to) })
}

// wLiftAfterRestartMeetsAnotherFence (reloc-probe-fence-owner, the lift,
// a crash): the completion's discharge fails for a while; the target's
// agent restarts, which empties the activator's fences, and another
// operation fences the host. The completion pass then discharges,
// releases the hold and lifts only its own fence: the other one stays
// and the serving wake waits. Lifting any fence opens the host under it
// (INV-32).
func wLiftAfterRestartMeetsAnotherFence(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		return
	}
	w.crash(to)
	to.actFence, to.foreignFence = "op-elsewhere", "op-elsewhere"
	to.dischargeFails = false
	w.reach(t, 100, "the completion marked", w.marked, w.act(to), func() { w.wake(to) })
	w.drive(20, never, w.act(to), func() { w.wake(to) })
}

// wDischargeFailsHoldsTheRow (reloc-completion-discharge-first, refusal
// and crash): C10's discharge fails for thirty steps; the row stays held
// and the fence installed, so no serving wake moves the holder off the
// stopped probe session. Then the discharge succeeds, the agent crashes
// after the hold's release, and the next pass resumes from the claim:
// discharged again on positive evidence, marked, and the host serves.
// With the hold released and the fence lifted first, a request admits a
// writer before the discharge (INV-31).
func wDischargeFailsHoldsTheRow(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		return
	}
	w.drive(30, never, w.act(to), w.recoveryPass, func() { w.wake(to) })
	to.dischargeFails, to.crashAfter = false, "release"
	w.reach(t, 200, "the completion marked after a crash", w.marked, w.act(to), w.recoveryPass, func() { w.wake(to) })
	w.finish(t)
}

// wMarkLastAfterAFailedDischarge (reloc-completion-mark-last): the
// discharge fails, and a drain from box 3 is asked for meanwhile; with
// the mark last the record stays unmarked and the barrier refuses it
// until the discharge succeeds. Marked first, the barrier lets the drain
// name box 3 while its participation is still the completed op's
// (INV-31), and the discharge is never retried.
func wMarkLastAfterAFailedDischarge(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		return
	}
	w.drive(10, never, w.act(to), w.recoveryPass)
	w.userRelocate()
	w.drive(5, never, w.act(to), w.recoveryPass)
	to.dischargeFails = false
	if !w.reach(t, 200, "the completion marked", w.marked, w.act(to), w.recoveryPass, func() { w.wake(to) }) {
		return
	}
	w.running(t, to)
	w.userRelocate()
	w.servingOn(t, nil)
	w.finish(t)
}

// wFailedMarkRetriedFromTheClaim (reloc-completion-from-claim, a failed
// mark and a crash): the mark fails once after the hold's release and
// the lift; the next pass finds the unmarked completion from the claim
// record and marks it. A second drain completes with a crash after the
// lift, and is marked on the next pass too. Found from the row's hold,
// which the release removed, the completion is never marked, and nothing
// starts from the host again (INV-3).
func wFailedMarkRetriedFromTheClaim(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, func(to *box) { to.markFails = true })
	if !ok {
		return
	}
	if !w.reach(t, 200, "the completion marked after a failed mark", w.marked, w.act(to), w.recoveryPass, func() { w.wake(to) }) {
		return
	}
	w.running(t, to)
	w.userRelocate()
	back := w.boxByID(w.claim.Op.Tgt)
	if back == nil || w.claim.Op.Kind != "relocating" {
		return
	}
	back.crashAfter = "lift"
	w.servingOn(t, back)
	w.reach(t, 200, "the second completion marked after a crash", w.marked, w.actAll(), w.recoveryPass, w.wakeExec)
	w.finish(t)
}

// wSuccessorWaitsForTheDischarge (reloc-completion-barrier): while the
// completion's discharge fails, a drain, a move and a push from box 3
// are all refused; once it is discharged and marked the drain starts and
// completes. Without the barrier the drain names box 3 while its
// participation is still the completed op's (INV-31).
func wSuccessorWaitsForTheDischarge(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		return
	}
	w.userRelocate()
	w.moveTo(1)
	w.userPush()
	w.drive(5, never, w.act(to), w.recoveryPass)
	if w.claim.Op.Kind != "serving" {
		return // the mutation started one; INV-31 was recorded at the step
	}
	to.dischargeFails = false
	if !w.reach(t, 200, "the completion marked", w.marked, w.act(to), w.recoveryPass, func() { w.wake(to) }) {
		return
	}
	w.running(t, to)
	w.userRelocate()
	w.servingOn(t, nil)
	w.finish(t)
}

// wCompletionWithRegionHAOff (reloc-completion-claim-mode, ruling 82,
// a restart): the discharge fails, then region HA is turned off and the
// target's agent restarts; the completion-only pass still discharges,
// releases, lifts and marks, and the host serves on box 3. Run only with
// region HA on, the completion is never finished (INV-3).
func wCompletionWithRegionHAOff(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		return
	}
	w.haOff = true
	w.crash(to)
	to.dischargeFails = false
	w.reach(t, 200, "the completion finished with region HA off", func() bool { return w.marked() && hasMain(to) }, w.act(to), w.recoveryPass, func() { w.wake(to) })
	w.finish(t)
}

// wServingWakeHandsOver (reloc-completed-handover, N6): after the
// completion the first serving wake on box 3 hands over from the probe's
// receipt and is cold, as claimServingOwners.Select makes every serving
// wake (L18 joint round 2 finding 1): a fresh generation, a new
// installation stamp, a writes_lost record for the stopped probe session,
// and P retired by the park-generation that follows. Routed to main's
// stop, its serving-shaped fence replaces the probe's and the handover
// refuses for good (INV-3).
func wServingWakeHandsOver(t *testing.T, w *world) {
	_, to, ok := w.completedDrain(t, nil)
	if !ok {
		return
	}
	d, p, stamp := *w.claim.Op.Completed, w.claim.Holder.Gen, w.claim.Holder.Stamp
	if !w.reach(t, 200, "the host runs on box 3", func() bool { return hasMain(to) && to.open == nil }, w.actAll(), w.recoveryPass, func() { w.wake(to) }) {
		return
	}
	w.checkFirstWakeCold(t, d, p, stamp)
	w.reach(t, 600, "P retired by its park-generation", func() bool { return !w.hasFilesAnywhere(p) && len(w.claim.Obl) == 0 }, w.actAll(), w.recoveryPass)
	w.finish(t)
}

// checkFirstWakeCold asserts the first serving wake after completion d of
// prepared generation p at stamp: cold, into a fresh generation, at a new
// stamp, declaring the stopped probe session's writes_lost.
func (w *world) checkFirstWakeCold(t *testing.T, d completedEntry, p string, stamp int) {
	t.Helper()
	h := w.claim.Holder
	l := lastAdmission(h)
	switch {
	case h.Gen == p || h.Gen == "":
		t.Errorf("the first serving wake continued P %q (generation %q)", p, h.Gen)
	case h.Stamp == stamp:
		t.Errorf("the first serving wake kept the installation stamp %d", stamp)
	case l == nil || l.Mode != "cold" || l.Box != d.Box || l.Gen != h.Gen:
		t.Errorf("the first serving wake was not a cold admission on box %d: %+v", d.Box, l)
	case l.Lost == nil || l.Lost.Session != d.Session || l.Lost.Box != d.Box || l.Lost.Gen != p:
		t.Errorf("the first serving wake did not declare the stopped probe session %d's writes_lost: %+v", d.Session, l.Lost)
	}
}

// hasFilesAnywhere: some region's bucket holds an object of generation g
// of the holder's location.
func (w *world) hasFilesAnywhere(g string) bool {
	for _, b := range w.boxes {
		if w.hasFiles(b, w.claim.Holder.Location, g) {
			return true
		}
	}
	return false
}

// probeWriteThenServe drains box 0 to box 3; once the probe passed, the
// probe session acknowledges a write its replicator has not shipped; then
// the relocation completes and the first serving wake on box 3 runs. It
// returns the write, the completion and the P and stamp it completed with.
func (w *world) probeWriteThenServe(t *testing.T) (wr int, d completedEntry, p string, stamp int, ok bool) {
	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 probe passed", func() bool { c := w.claim; return c.Op.Phase == "probed" && c.Op.ProbeStop == nil && hasMain(to) }, w.act(from, to), w.recoveryPass) {
		return 0, d, "", 0, false
	}
	wr = w.appWrite(to) // acknowledged by the probe session, not yet replicated
	if !w.reach(t, 200, "serving on box 3", func() bool { return w.claim.Op.Kind == "serving" && w.claim.Op.Completed != nil }, w.act(from, to), w.recoveryPass) {
		return 0, d, "", 0, false
	}
	d, p, stamp = *w.claim.Op.Completed, w.claim.Holder.Gen, w.claim.Holder.Stamp
	ok = w.reach(t, 200, "the host runs on box 3", func() bool { return hasMain(to) && to.open == nil }, w.actAll(), w.recoveryPass, func() { w.wake(to) })
	return wr, d, p, stamp, ok
}

// wProbeStopFlushes (reloc-probe-stop-flush, ruling 87): the probe
// session acknowledges a write the replicator has not shipped; the probe
// stop flushes it before root's stop, so the first serving wake, cold,
// declares a writes_lost that names none of the probe session's writes.
// Stopped with nothing flushed, that wake declares the write lost
// (INV-34).
func wProbeStopFlushes(t *testing.T, w *world) {
	if _, _, _, _, ok := w.probeWriteThenServe(t); !ok {
		return
	}
	w.finish(t)
}

// TestFirstServingWakeIsCold (L18 joint round 2 finding 1, ruling 87):
// the first serving wake after the completion is cold (a fresh
// generation, a new stamp, a writes_lost for the stopped probe session),
// and since the probe stop flushed, the write the probe session
// acknowledged is in the new generation and no write is declared lost.
func TestFirstServingWakeIsCold(t *testing.T) {
	w := newWitness()
	wr, d, p, stamp, ok := w.probeWriteThenServe(t)
	if !ok {
		w.mustHold(t)
	}
	w.mustHold(t)
	w.checkFirstWakeCold(t, d, p, stamp)
	to := w.boxByID(d.Box)
	for _, a := range w.acks {
		if a.W == wr && a.Exempt {
			t.Fatalf("the probe session's write %d was declared lost: %s", wr, w.state())
		}
	}
	if to.DB == nil || !contains(to.DB.Log, wr) || w.stats["writes declared lost"] != 0 {
		t.Fatalf("the probe session's write %d is not in the first serving wake's database (%d writes declared lost): %s", wr, w.stats["writes declared lost"], w.state())
	}
	w.finish(t)
	w.mustHold(t)
}

// TestProbeStopRefusedAfterARestartBeforeTheRecord (reloc-probe-stop-flush,
// ruling 87): the probe passed, its flush is held, and the target's agent
// restarts before the positions are recorded. The new agent has no
// replicator for the probe session, so the probe stop refuses; nothing is
// journaled, and ruling 67's deadline makes the relocation stuck, as a
// release would be. The mutation stops at once, and this test fails.
func TestProbeStopRefusedAfterARestartBeforeTheRecord(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	w.reach(t, 1500, "the probe passed", func() bool { c := w.claim; return c.Op.Phase == "probed" && c.Op.ProbeStop == nil && hasMain(to) }, w.act(from, to), w.recoveryPass)
	w.appWrite(to)
	loc := w.claim.Holder.Location
	w.holdKey(loc)
	w.drive(3, never, w.act(from, to))
	if to.probeRecord != nil || w.claim.Op.ProbeStop != nil {
		t.Fatalf("the probe stop recorded or journaled with its flush held: %s", w.state())
	}
	w.crash(to)
	w.hold = nil
	w.completeHeld(loc)
	w.reach(t, 3*releaseDeadline, "stuck at the probe stop's deadline", func() bool { return w.claim.Op.Stuck != "" }, w.act(from, to), w.recoveryPass)
	w.mustHold(t)
	if c := w.claim; c.Op.Phase != "probed" || c.Op.ProbeStop != nil || to.probeRecord != nil || w.stats["probe stop: no replicator or files directory and no record, refused"] == 0 {
		t.Fatalf("the probe stop did not refuse until the deadline: %s", w.state())
	}
}

// TestNoAdmissionBetweenStopAndFence (reloc-admits-probe-only, L18 joint
// round 2 finding 3): the source's release stops its writers and closes
// the session's launch journal, and its flush is held, so root's
// admission fence (Collect) has not come. A request reaches the source,
// then its agent restarts and another request reaches it: no writer is
// admitted in the interval, because relocatingAdmission admits only the
// probe. The mutation admits the source's wake, and this test fails.
func TestNoAdmissionBetweenStopAndFence(t *testing.T) {
	w := twoBoxWitness()
	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 is held: no positions, no Collect
	w.drainTo(t, to)
	w.reach(t, 200, "the source stopped, its launch journal closed", func() bool {
		return w.claim.Op.Phase == "releasing" && from.launchClosedOp == w.claim.Op.ID && !hasMain(from)
	}, w.act(from, to), w.recoveryPass)
	session, op := w.claim.Holder.Session, w.claim.Op.ID
	for _, restart := range []bool{false, true} {
		if restart {
			w.crash(from)
		}
		w.drive(5, never, func() { w.wake(from) }, w.act(from, to))
		if from.relFence == op {
			t.Fatalf("root's admission fence came before the flush")
		}
		if hasMain(from) || w.claim.Holder.Session != session || from.open != nil {
			t.Fatalf("a wake admitted a writer between the stop and the fence (restart %v): %s", restart, w.state())
		}
	}
	if w.stats["admission refused: relocating admits only the probe"] == 0 {
		t.Fatalf("no wake reached the admission: %v", w.stats)
	}
	w.mustHold(t)
	w.hold = nil
	w.completeHeld(loc)
	w.servingOn(t, nil)
	w.finish(t)
	w.mustHold(t)
}

// wDelayedSourceHoldsItsRow (reloc-completed-catchup-hold, finding 3):
// the source releases and its admission fence closes, then its
// reconcile is delayed until the target has completed. Its first
// reconcile after that holds the row for the completed op, never
// leaving it ready, until its park-local retires it. Without the
// catch-up hold the row its release left ready stays ready (INV-33).
func wDelayedSourceHoldsItsRow(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
	}, func() { w.worker(from) }, w.act(to), w.recoveryPass) {
		return
	}
	if !w.reach(t, 1500, "serving on box 3, the source not reconciled", func() bool {
		c := w.claim
		return c.Op.Kind == "serving" && c.Op.Exec == to.ID
	}, func() { w.worker(from) }, w.act(to), w.recoveryPass) {
		return
	}
	w.reconcile(from)
	w.finish(t)
}

// TestDelayedSourceHoldsItsRow is the witness's schedule with its
// assertions: the source's row is still ready when the target completes
// (the release no longer holds it, ruling 28), held for the completed op
// after the source's next reconcile, and gone once its park-local ran.
func TestDelayedSourceHoldsItsRow(t *testing.T) {
	w := newWitness()
	w.twoBox()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	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
	}, func() { w.worker(from) }, w.act(to), w.recoveryPass)
	w.reach(t, 1500, "serving on box 3", func() bool { return w.claim.Op.Kind == "serving" && w.claim.Op.Exec == to.ID }, func() { w.worker(from) }, w.act(to), w.recoveryPass)
	w.mustHold(t)
	if from.Row == nil || from.Row.Status != "ready" {
		t.Fatalf("the delayed source's row is not the ready row its release left: %+v", from.Row)
	}
	op := w.claim.Op.Completed.Op
	w.reconcile(from)
	w.mustHold(t)
	if from.Row == nil || from.Row.Status != "relocating" || from.Row.HoldOp != op {
		t.Fatalf("the catch-up hold did not hold the source's row for %s: %+v", op, from.Row)
	}
	w.finish(t)
	w.mustHold(t)
	if from.Row != nil {
		t.Fatalf("the source's row outlived its park-local: %+v", from.Row)
	}
}

// TestCatchUpHoldResidualAfterASuccessor is the residual scenario ruling
// 85 names and leaves open, modelled as the built code has it and not
// hidden: the source's reconcile is delayed, the target completes and is
// discharged and marked, a successor (a drain from the target, or a move
// to region 1) replaces the serving op, which needs only the target's
// discharge, and only then does the source reconcile. The successor took
// the completion record with it, so the catch-up hold no longer applies,
// and the source's row stays ready from then until its park-local
// retires it. No invariant fails: the gap is availability, not a second
// writer (the source's wake is refused by the exec check), so this is a
// scenario test that asserts the gap is there, not a mutation witness.
func TestCatchUpHoldResidualAfterASuccessor(t *testing.T) {
	for _, successor := range []string{"relocation", "move"} {
		t.Run(successor, func(t *testing.T) {
			w := newWitness()
			w.twoBox()
			from, to := w.boxes[0], w.boxes[3]
			w.serve(from, 1, app0)
			w.running(t, from)
			w.drainTo(t, to)
			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
			}, func() { w.worker(from) }, w.act(to), w.recoveryPass)
			w.reach(t, 1500, "the completion marked, the source not reconciled", w.marked, func() { w.worker(from) }, w.act(to), w.recoveryPass, func() { w.wake(to) })
			w.running(t, to)
			// the region's chores the completion queued run on box 3; only
			// box 0's park-local stays owed (box 0's agent is the slow one)
			w.reach(t, 600, "only the source's park-local owed", func() bool {
				for _, o := range w.claim.Obl {
					if !(o.Kind == "park-local" && o.Box == from.ID) {
						return false
					}
				}
				return true
			}, w.act(to), w.recoveryPass, func() { w.wake(to) })
			w.mustHold(t)
			if from.Row == nil || from.Row.Status != "ready" {
				t.Fatalf("the delayed source's row is not the ready row its release left: %+v", from.Row)
			}
			want := "relocating"
			if successor == "relocation" {
				w.extraBox(0, 30)
				w.userRelocate()
			} else {
				w.moveTo(1)
				want = "moving"
			}
			if w.claim.Op.Kind != want {
				t.Fatalf("the successor %s did not start: %s", successor, w.state())
			}
			w.reconcile(from)
			w.wake(from)
			w.mustHold(t)
			if from.Row == nil || from.Row.Status != "ready" || hasMain(from) {
				t.Fatalf("the residual is not as ruling 85 describes it (row %+v, running %v)", from.Row, hasMain(from))
			}
			t.Logf("residual: box %d's row is ready under %s %s, with its park-local still owed", from.ID, w.claim.Op.Kind, w.claim.Op.ID)
			w.finish(t)
			w.mustHold(t)
			if from.Row != nil {
				t.Fatalf("the source's row outlived its park-local: %+v", from.Row)
			}
		})
	}
}

// TestFailedTargetCannotReleaseWithoutC1 (reloc-c1-failed-release-refused,
// finding 4): with claim ask C1 open, a drain's target that fails its
// probe cannot release in probe-failed: nothing is journaled, the
// operation is stuck at the release deadline, and the operator's resolve
// stops the target and replays, so the host serves again. The mutation
// releases as if C1 had landed, and this test fails.
func TestFailedTargetCannotReleaseWithoutC1(t *testing.T) {
	w := twoBoxWitness()
	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.extraBox(0, 30)
	w.reach(t, 1500, "probe-failed", func() bool { return w.claim.Op.Phase == "probe-failed" }, w.act(from, to), w.recoveryPass)
	w.reach(t, 3*releaseDeadline, "stuck in probe-failed", func() bool { return w.claim.Op.Stuck != "" }, w.act(from, to), w.recoveryPass)
	w.mustHold(t)
	if c := w.claim; c.Op.Phase != "probe-failed" || c.Op.Release != nil || c.Op.Exec != to.ID || w.stats["release journaled: failed target"] != 0 {
		t.Fatalf("the failed target released without claim ask C1: %s", w.state())
	}
	if w.stats["release refused: no failed-target obligation (C1 open)"] == 0 {
		t.Fatalf("the release was never refused: %v", w.stats)
	}
	w.servingOn(t, nil)
	if w.stats["operator stopped a target that cannot release"] == 0 {
		t.Fatalf("the operator did not stop the target: %v", w.stats)
	}
	w.finish(t)
	w.mustHold(t)
}

// TestLaterAttemptRefusedWithoutC11 (reloc-c11-later-attempt-refused,
// finding 4): with claim ask C11 open, the target's first attempt opens P
// and its agent crashes before the probe passes; the next attempt is
// refused before it is counted, and the operation is probe-failed with
// the count unchanged and no second admission. The mutation counts and
// makes it, and this test fails.
func TestLaterAttemptRefusedWithoutC11(t *testing.T) {
	w := twoBoxWitness()
	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) {
		w.mustHold(t)
	}
	attempts, gen := w.claim.Op.Attempts, w.claim.Holder.Gen
	w.crash(to)
	w.reach(t, 200, "probe-failed or a second attempt", func() bool {
		c := w.claim
		return c.Op.Phase == "probe-failed" || c.Op.Attempts != attempts || c.Holder.Gen != gen
	}, w.act(from, to), w.recoveryPass)
	w.mustHold(t)
	if c := w.claim; c.Op.Phase != "probe-failed" || c.Op.Attempts != attempts || c.Holder.Gen != gen {
		t.Fatalf("a later attempt was counted or made without claim ask C11 (attempts %d -> %d): %s", attempts, c.Op.Attempts, w.state())
	}
	if w.stats["probe-failed: a later attempt root cannot launch (C11 open)"] != 1 {
		t.Fatalf("the refusal was not recorded once: %v", w.stats)
	}
	w.servingOn(t, nil)
	w.finish(t)
	w.mustHold(t)
}

// TestCompletionBarrierExemptions (reloc-completion-barrier, ruling 77's
// exemptions): while box 3's completion is unmarked, a drain is refused;
// box 3 is then terminated, and the failover that cites its fence starts
// over the unmarked record and serves elsewhere.
func TestCompletionBarrierExemptions(t *testing.T) {
	w := newWitness()
	_, to, ok := w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		w.mustHold(t)
	}
	w.userRelocate()
	if w.claim.Op.Kind != "serving" || w.stats["start refused: a completed relocation is not discharged"] == 0 {
		t.Fatalf("the drain was not refused by the barrier: %s", w.state())
	}
	w.terminate(to)
	w.reach(t, 200, "the failover started over the unmarked record", func() bool { return w.claim.Op.Kind == "relocating" }, w.actAll(), w.recoveryPass)
	w.servingOn(t, nil)
	w.finish(t)
	w.mustHold(t)
}

// TestProbeStopRefusesAnotherFence (reloc-probe-fence-owner as ruling 87
// amends ruling 75): with another owner's fence on the host, the probe
// stop stops the writers and flushes and records the positions, then
// refuses: root is not asked (no receipt), no stop is journaled, and the
// other owner's fence is left as it is.
func TestProbeStopRefusesAnotherFence(t *testing.T) {
	w := twoBoxWitness()
	from, to := w.boxes[0], w.boxes[3]
	w.serve(from, 1, app0)
	w.running(t, from)
	w.drainTo(t, to)
	w.reach(t, 1500, "the probe passed", func() bool { c := w.claim; return c.Op.Phase == "probed" && c.Op.ProbeStop == nil && hasMain(to) }, w.act(from, to), w.recoveryPass)
	to.actFence = "op-elsewhere"
	w.drive(20, never, w.act(from, to))
	w.mustHold(t)
	c := w.claim
	if c.Op.ProbeStop != nil || to.actFence != "op-elsewhere" || to.StopEvidence[probeKey(c.Op.ID)] {
		t.Fatalf("the refused probe stop asked root or touched the fence (stop %+v, fence %q): %s", c.Op.ProbeStop, to.actFence, w.state())
	}
	if hasMain(to) || to.probeRecord == nil || to.probeRecord.Op != c.Op.ID {
		t.Fatalf("the probe's writers were not stopped and flushed before the refusal (running %v, record %+v)", hasMain(to), to.probeRecord)
	}
}

// purgeOverATerminatedCompletionBox is seed 2224's schedule (3,000, two
// boxes): the completion's discharge fails, a purge closes the app's
// admission, and the completion box is terminated before its discharge.
// The failover that would cite its fence is refused by the closed
// admission (Admission.Register), so only the purge can move.
func (w *world) purgeOverATerminatedCompletionBox(t *testing.T) (to *box, ok bool) {
	_, to, ok = w.completedDrain(t, func(to *box) { to.dischargeFails = true })
	if !ok {
		return to, false
	}
	w.userPurge()
	if !w.reach(t, 100, "the purge closed the app's admission", func() bool { return w.admit[app0].Closed }, w.runConsole) {
		return to, false
	}
	w.terminate(to)
	return to, true
}

// wPurgeOverAFencedCompletionBox (reloc-completion-fenced-exempt, ruling
// 86): once a worker has written the dead completion box's fence record,
// the purge verifies it and starts over the unmarked completion, and the
// host is freed with no operator. Without the exemption the purge is
// refused by the barrier and the failover by the purge's closed
// admission, for ever (INV-3).
func wPurgeOverAFencedCompletionBox(t *testing.T, w *world) {
	if _, ok := w.purgeOverATerminatedCompletionBox(t); !ok {
		return
	}
	w.reach(t, 2000, "the purge freed the host", func() bool { c := w.claim; return c == nil || c.Holder == nil }, w.actAll(), w.runConsole, w.recoveryPass)
	w.finish(t)
}

// TestTerminatedCompletionBoxBlocksAPurge was the gap the model found at
// seed 2224, closed by ruling 86 (1b-next ca282378): the purge refused by
// the barrier while the completion box's fence is not yet recorded, then
// started over it once the fence verifies, and the host freed with no
// operator's declare-lost (the model's operator has none since).
func TestTerminatedCompletionBoxBlocksAPurge(t *testing.T) {
	w := newWitness()
	to, ok := w.purgeOverATerminatedCompletionBox(t)
	if !ok {
		w.mustHold(t)
	}
	w.drive(60, never, w.runConsole) // the purge's start, before any worker observed the termination
	if c := w.claim; c.Op.Kind != "serving" || to.FenceObservedAt != 0 || w.stats["start refused: a completed relocation is not discharged"] == 0 {
		t.Fatalf("the purge was not refused before the fence record: fence observed at %d, %s", to.FenceObservedAt, w.state())
	}
	w.reach(t, 2000, "the purge freed the host", func() bool { c := w.claim; return c == nil || c.Holder == nil }, w.actAll(), w.runConsole, w.recoveryPass)
	w.mustHold(t)
	if to.FenceObservedAt == 0 || w.stats["start: the completion box's termination fence verified"] == 0 || w.stats["failover start: app admission closed"] == 0 {
		t.Fatalf("not ruling 86's path: fence observed at %d, %v", to.FenceObservedAt, w.stats)
	}
	if contains(w.claim.Lost, to.ID) && w.claim.Holder != nil {
		t.Fatalf("the box was declared lost: %s", w.state())
	}
}

// TestDirtyReturnParksPreMoveFiles (moves a5e8c13f, L18 finding 1; D6):
// a move to region 1 rolls back dirty after its target wrote. While the
// target releases, the source region keeps the files live set; from the
// source's return CAS the newest admission (the target's) decides, so the
// return's install park retires the pre-move generation's manifest in the
// source region rather than keeping it for the at-rest sweep. INV-30 does
// not count it, the return serves on the source, and every invariant
// holds to quiescence.
func TestDirtyReturnParksPreMoveFiles(t *testing.T) {
	w := newWitness()
	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
	mk, _, _, _, m := w.newestManifest(src.Region, loc, w.claim.Holder.Gen)
	if m == nil {
		t.Fatalf("no pre-move manifest: %s", w.state())
	}
	w.moveTo(1)
	tgt := w.boxByID(w.claim.Op.Tgt)
	w.reach(t, 600, "the target's probe runs, before the commit", func() bool {
		c := w.claim
		return c.Op.Kind == "landing" && !c.Op.Committed && hasMain(tgt) && w.rec.MoveState == "landing"
	}, func() { w.worker(src); w.reconcile(src); w.worker(tgt); w.reconcile(tgt); w.runConsole() })
	w.mustHold(t)
	w.appWrite(tgt) // the rollback is dirty
	w.drive(w.claim.Op.CutDeadline+1-w.now, never, w.act(sib))
	w.reach(t, 1500, "the dirty return's install park done", func() bool {
		c := w.claim
		return c.Op.Kind == "restoring" && c.Op.Dirty && c.Op.SrcPhase != "waiting" && c.Op.SrcPhase != "claimed" && c.Op.SrcPhase != "install-park" || c.Op.Kind == "serving"
	}, w.actAll(), w.recoveryPass, w.runConsole)
	w.mustHold(t)
	if w.stats["rollback-dirty"] == 0 && w.claim.Op.Kind != "restoring" {
		t.Fatalf("the rollback was not dirty: %s", w.state())
	}
	if v := w.buckets[src.Region].current(mk); v != nil {
		t.Fatalf("the dirty return's install park kept the pre-move manifest %s (sweep owed %v): %s", mk, v.SweepOwed, w.state())
	}
	w.reach(t, 1500, "the dirty return serving on the source", func() bool { c := w.claim; return c.Op.Kind == "serving" && c.Op.Exec == src.ID }, w.actAll(), w.recoveryPass, w.runConsole, func() { w.wake(src) })
	w.finish(t)
	w.mustHold(t)
}

// wDirtyReturnWithoutAdmission (dirty-return-needs-admission, Astra joint
// round 4 item 2): the source's last write is not yet replicated at the
// cut, so the seal does not cover its replica (src_covered false), and the
// move's target stalls before its probe is admitted, so the landing rolls
// back. The return is dirty (the seal is returned) although the target
// never admitted: holder.generation is still the pre-move one, the source
// keeps the files home, INV-30 counts the source region, and the return's
// install parks nothing (movesinstall.go's targetAdmitted). Without the
// admission check the install park retires the manifest the source's next
// install would pin (INV-30).
func wDirtyReturnWithoutAdmission(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)
	w.appWrite(src) // not replicated before the cut: the seal is all that holds it
	loc := w.claim.Holder.Location
	w.holdKey(loc) // the source's uploads stay in flight through the seal
	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)
	if !w.reach(t, 600, "the flip, before the target's probe", func() bool { return w.claim.Op.Kind == "landing" }, func() { w.worker(src); w.reconcile(src); w.worker(tgt); w.reconcile(tgt); w.runConsole() }) {
		return
	}
	w.hold = nil
	w.completeHeld(loc)
	if targetAdmitted(w.claim) || w.claim.Op.SrcCovered {
		t.Fatalf("witness: not the scenario (target admitted %v, src_covered %v): %s", targetAdmitted(w.claim), w.claim.Op.SrcCovered, w.state())
	}
	// the target stalls: the landing's deadline passes with no probe
	w.drive(w.claim.Op.CutDeadline+12-w.now, never, w.act(sib), w.runConsole, w.recoveryPass, w.operator)
	if !w.reach(t, 1500, "the dirty return's install past its park", func() bool {
		c := w.claim
		return c.Op.Kind == "restoring" && c.Op.Dirty && c.Op.SrcPhase != "waiting" && c.Op.SrcPhase != "claimed" && c.Op.SrcPhase != "install-park" || c.Op.Kind == "serving"
	}, w.actAll(), w.recoveryPass, w.runConsole) {
		return
	}
	w.reach(t, 1500, "the return serving on the source", func() bool { c := w.claim; return c.Op.Kind == "serving" && c.Op.Exec == src.ID }, w.actAll(), w.recoveryPass, w.runConsole, func() { w.wake(src) })
	w.finish(t)
}