package host

import (
	"context"
	"crypto/ed25519"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The revert on the source (moves spec §7.4 reverting; Plan 4 Task 6). The
// fixture is the cut harness driven to a revert, either from stopping (the
// stop evidence never held: the claim still moving under the source, no
// baselines on the record) or from stopped after execution was given up
// (the claim reverting, the baselines recorded). Its row follows the holds
// the cut and the revert write; the hold's status change, the open and the
// probe (Task 9, BLOCKED-ON-CLAIM C1) are faked; the target region's box
// (uk-box) runs the lane park through the real park engine over the same
// buckets with a fake lease table. Everything else is the real revert: the
// cut's evidence and giveUpExecution, revertClaim, hostclaim's and
// provision's own transitions.

type revertHarness struct {
	*cutHarness
	tdir     string
	tleases  *fakeParkLeases
	tw       *movesWorker
	probe    func(req recoveryProbeRequest) (recoveryProbeOutcome, error)
	probes   []recoveryProbeRequest
	restatus []string // status:op of each status change asked for
	opens    []string // the op of each open asked for
	events   []string
	rvAt     string
	rvDone   bool
	rvHits   []string
}

func newRevertHarness(t *testing.T, opts ...p1Option) *revertHarness {
	t.Helper()
	return newRevertHarnessOn(t, newCutHarness(t, opts...))
}

// newRevertHarnessOn is newRevertHarness over a cut harness the caller
// built (a landing harness's, whose source replica is a real one).
func newRevertHarnessOn(t *testing.T, ch *cutHarness) *revertHarness {
	t.Helper()
	h := &revertHarness{cutHarness: ch, tdir: t.TempDir(), tleases: &fakeParkLeases{}}
	// The row follows the cut's hold, as holdMoving keeps it: held moving
	// under the op, a row held by another op refused.
	h.cfg.HoldMoving = func(host, op string) error {
		h.holds = append(h.holds, host+"="+op)
		if h.route.HoldOp != "" && h.route.HoldOp != op {
			return fmt.Errorf("%s is held %s by %s", host, h.route.Status, h.route.HoldOp)
		}
		if h.route.HoldOp == "" {
			h.route.Status, h.route.HoldOp = "moving", op
		}
		return nil
	}
	return h
}

// fromStopping drives the cut to a revert at stopping: the evidence never
// held, the deadline passed, the record reverting with no baselines and the
// claim still moving under the source.
func (h *revertHarness) fromStopping(t *testing.T) {
	t.Helper()
	h.ready(t)
	h.ev.stopped = false
	h.cut(t)
	h.pass(t)
	h.clock.Advance(2 * time.Minute)
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveReverting || m.CutPhase != provision.CutStopping || m.Source != nil || m.Reason != provision.RevertStopEvidenceBudget {
		t.Fatalf("the fixture did not revert at stopping: %+v", m)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != h.srcBox {
		t.Fatalf("the fixture's claim: %+v", c.Op)
	}
	if h.route.Status != "moving" || h.route.HoldOp != p1ID {
		t.Fatalf("the fixture's row: %+v", h.route)
	}
}

// fromStopped drives the cut to a revert at stopped after execution was
// given up: the baselines recorded and the claim reverting at evidence.
func (h *revertHarness) fromStopped(t *testing.T) {
	t.Helper()
	h.ready(t)
	h.crashAt = "after-exec-none"
	h.cut(t)
	h.crashPass(t)
	h.clock.Advance(2 * time.Minute)
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveReverting || m.CutPhase != provision.CutStopped || m.Source == nil || m.Source.Files == nil {
		t.Fatalf("the fixture did not revert at stopped: %+v", m)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.RevPhaseEvidence {
		t.Fatalf("the fixture's claim: %+v", c.Op)
	}
}

// arm wires the revert's hooks on the source box.
func (h *revertHarness) arm(t *testing.T) {
	t.Helper()
	h.cfg.RestatusHold = func(host, status, op string, validate func() error) (bool, error) {
		h.restatus = append(h.restatus, status+":"+op)
		if validate != nil {
			if err := validate(); err != nil {
				return false, err
			}
		}
		if host != p1Host || h.route.HoldOp != op {
			return false, nil
		}
		h.route.Status = status
		return true, nil
	}
	h.cfg.ReleaseHeld = func(host, op string) (bool, error) {
		h.opens = append(h.opens, op)
		if op == "" || h.route.HoldOp != op {
			return false, nil
		}
		h.route.Status, h.route.HoldOp = "ready", ""
		return true, nil
	}
	h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
		h.events = append(h.events, ev.Event)
		return nil
	}
	h.cfg.recoveryProbe = func(ctx context.Context, _ release.Store, _ provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
		h.probes = append(h.probes, req)
		if h.probe != nil {
			return h.probe(req)
		}
		if err := req.Record(ctx, revertRendered()); err != nil {
			return recoveryProbeWaiting, err
		}
		return recoveryProbeAnswered, nil
	}
	h.cfg.rvFault = func(point string) error {
		h.rvHits = append(h.rvHits, point)
		if point == h.rvAt && !h.rvDone {
			h.rvDone = true
			return errCrash
		}
		return nil
	}
}

// revertRendered are the inputs the fake probe renders: not the pinned ones.
func revertRendered() provision.MoveInputs {
	return provision.MoveInputs{BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9", Release: "v7"}
}

// laneCfg is the target region's box, its lane park armed.
func (h *revertHarness) laneCfg() moveCfg {
	c := h.target()
	c.DataDir = h.tdir
	armParkFakes(&c, h.s, h.tleases, park.PlacementRegion(c.Region), roleTarget)
	return c
}

// tPass runs one pass as the target region's box.
func (h *revertHarness) tPass(t *testing.T) {
	t.Helper()
	if h.tw == nil {
		h.tw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	}
	h.tw.cfg = h.laneCfg()
	if err := movesPass(context.Background(), h.s, h.tw); err != nil {
		t.Fatal(err)
	}
}

func (h *revertHarness) verifications() int {
	n := 0
	for _, p := range h.rvHits {
		if p == "verify" {
			n++
		}
	}
	return n
}

func (h *revertHarness) revPhase(t *testing.T) string {
	t.Helper()
	return h.claim(t).Op.Journal[hostclaim.JournalRevPhase]
}

// laneChore is the lane's park the revert queued in the target region.
func (h *revertHarness) laneChore(t *testing.T) *hostclaim.Obligation {
	t.Helper()
	for _, o := range h.claim(t).Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Region == park.PlacementRegion(h.dst) && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) {
			return &o
		}
	}
	return nil
}

// laneKeys are the objects still in the move's lane in the target bucket.
func (h *revertHarness) laneKeys() []string {
	var out []string
	for k := range h.f.buckets[h.dstBucket] {
		if strings.HasPrefix(k, moves.ScopeLocation(p1Host, p1ID)+"/") {
			out = append(out, k)
		}
	}
	return out
}

// pastHorizon moves the clock past the lane park's acknowledgement horizon.
func (h *revertHarness) pastHorizon(t *testing.T, o *hostclaim.Obligation) {
	t.Helper()
	until := o.Park.EvidenceAt.Add(park.SignatureAge + o.Park.RequestLimit)
	if d := until.Sub(h.clock.now); d > 0 {
		h.clock.Advance(d + time.Second)
	}
}

// reopened asserts the source reopened: the claim reverting/opened under
// the source, the record still reverting with ReopenedAt, the row open.
func (h *revertHarness) reopened(t *testing.T) {
	t.Helper()
	m := h.move(t)
	if m == nil || m.State != provision.MoveReverting || m.ReopenedAt == nil {
		t.Fatalf("the source did not reopen: %+v", m)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseOpened || c.Op.Exec != h.srcBox || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseOpened {
		t.Fatalf("the claim after the reopen: %+v", c.Op)
	}
	if h.route.HoldOp != "" || h.route.Status != "ready" {
		t.Fatalf("the row is not open: %+v", h.route)
	}
}

// drive runs the source's and the target's passes, moving the clock past the
// lane park's horizon once the source has opened, until the record closes.
func (h *revertHarness) drive(t *testing.T) {
	t.Helper()
	for i := 0; i < 12; i++ {
		h.pass(t)
		h.tPass(t)
		if h.read(t, p1Host).Move == nil {
			return
		}
		if c := h.claim(t); c.Op.Kind == hostclaim.KindReverting && c.Op.Phase == hostclaim.PhaseOpened {
			if o := h.laneChore(t); o != nil {
				h.pastHorizon(t, o)
			}
		}
	}
	t.Fatalf("the revert never closed: %+v, claim %+v, hits %v", h.move(t), h.claim(t).Op, h.rvHits)
}

// closed asserts the end: the record at rest in ie with LastMove rolled_back,
// the claim serving under the source with this move recorded as released,
// the row open, the source's journal parked.
func (h *revertHarness) closed(t *testing.T) {
	t.Helper()
	rec := h.read(t, p1Host)
	if rec.Move != nil || rec.Region != h.src || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "rolled_back" {
		t.Fatalf("the record after the close: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != h.srcBox || hostclaim.RequireReleasedMove(p1ID)(c) != nil {
		t.Fatalf("the claim after the close: %+v", c.Op)
	}
	if h.route.HoldOp != "" || h.route.Status != "ready" {
		t.Fatalf("the row after the close: %+v", h.route)
	}
	if moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
		t.Fatal("the source's journal of the move is still live")
	}
	parked, _ := filepath.Glob(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, "*", "artifacts", ".moves", p1Host, p1ID))
	if len(parked) != 1 {
		t.Fatalf("the source's journal is not parked once: %v", parked)
	}
	for _, s := range h.restatus {
		if s != "restoring:"+p1ID {
			t.Fatalf("a status change other than restoring under the move id: %v", h.restatus)
		}
	}
}

// TestRevertFromStoppingComputesEvidenceFirst (§7.4 evidence, J7, J8): a
// revert that began at stopping first computes the stop evidence, and starts
// nothing until it holds: the row stays held moving under the move id and is
// fenced again, the claim stays moving under the source, no report is
// written and nothing is probed. Once the tree is proven stopped, the cut's
// giveUpExecution publishes the complete stopped-source.json and gives up
// execution citing it, revertClaim writes the claim half (reverting, rev_phase
// evidence, the lane's park queued), the row goes restoring under the move id,
// and rev_phase steps evidence to probing (no baselines, so no verify), op.phase
// following (J1).
func TestRevertFromStoppingComputesEvidenceFirst(t *testing.T) {
	t.Parallel()
	h := newRevertHarness(t)
	h.fromStopping(t)
	h.arm(t)
	h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
	fences, calls := h.fences, h.ev.calls
	h.pass(t)
	if h.ev.calls != calls+1 || h.fences != fences+1 {
		t.Fatalf("the evidence was not asked again behind a fresh fence: evidence %d->%d fences %d->%d", calls, h.ev.calls, fences, h.fences)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalEvidenceSource] != "" {
		t.Fatalf("the claim moved before the evidence held: %+v", c.Op)
	}
	if _, _, err := h.s.Get(context.Background(), h.stoppedKey()); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a stopped report before the evidence held: %v", err)
	}
	if len(h.probes) != 0 || len(h.restatus) != 0 || h.route.Status != "moving" || h.route.HoldOp != p1ID {
		t.Fatalf("something started before the evidence held: probes %d restatus %v row %+v", len(h.probes), h.restatus, h.route)
	}
	if m := h.move(t); m.State != provision.MoveReverting || m.Source != nil {
		t.Fatalf("the record: %+v", m)
	}

	// The tree is proven stopped; a crash right after the report leaves the
	// claim still moving under the source: nothing has started.
	h.ev.stopped = true
	h.crashAt, h.crashed = "after-report", false
	h.crashPass(t)
	if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
		t.Fatalf("the claim after the report alone: %+v", c.Op)
	}
	if len(h.probes) != 0 || len(h.restatus) != 0 {
		t.Fatalf("something started before execution was given up: probes %d restatus %v", len(h.probes), h.restatus)
	}
	h.freshWorker()
	h.pass(t)
	rep, err := readStoppedReport(context.Background(), h.s, h.stoppedKey())
	if err != nil || !rep.Complete() || rep.OpID != p1ID || rep.Role != "source" {
		t.Fatalf("the stopped report: %v %+v", err, rep)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalEvidenceSource] != h.stoppedKey() {
		t.Fatalf("the claim half: %+v", c.Op)
	}
	if c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Op.Phase != hostclaim.PhaseProbing {
		t.Fatalf("rev_phase %q op.phase %q, want probing for both (no baselines)", c.Op.Journal[hostclaim.JournalRevPhase], c.Op.Phase)
	}
	if o := h.laneChore(t); o == nil || o.Park.Incarnation != h.inc || o.Park.RequestLimit != cutRequestLimit || o.Owner != "region:uk" {
		t.Fatalf("the lane's park: %+v", o)
	}
	if h.verifications() != 0 {
		t.Fatalf("a revert from stopping verified %d times: no baseline was recorded", h.verifications())
	}
	if m := h.move(t); m.Source != nil {
		t.Fatalf("the revert recorded baselines: %+v", m.Source)
	}
	if h.route.Status != "restoring" || h.route.HoldOp != p1ID {
		t.Fatalf("the row: %+v, want restoring under the move id", h.route)
	}
	if len(h.probes) != 1 {
		t.Fatalf("%d probes, want 1", len(h.probes))
	}
	req := h.probes[0]
	if req.RollbackID != "" || !req.Clean || req.Generation != "" || req.SourceGeneration != "cur" || req.Coverage != "" || req.Attempts != recoveryProbeAttempts {
		t.Fatalf("the probe request: %+v", req)
	}

	// The probe answers: probed, reopened, the row opened, the claim opened.
	h.probe = nil
	h.pass(t)
	h.reopened(t)
	m := h.move(t)
	if m.RecoveryInputs == nil || m.RecoveryInputs.BundleETags["default"] != "e9" || m.Inputs == nil || m.Inputs.BundleETags["default"] != "e1" {
		t.Fatalf("recovery inputs %+v, move.inputs %+v", m.RecoveryInputs, m.Inputs)
	}
	if len(h.opens) != 1 || h.opens[0] != p1ID {
		t.Fatalf("the open: %v", h.opens)
	}
}

// TestRevertVerifyThenProbeNeverVerifiesAgain (§7.4 verify and probing;
// witness wRollbackJournal for the revert, INV-2): a revert from stopped
// verifies its files and timers against the stopped baselines, then journals
// probing before the first attempt. The probe's process writes (the files and
// the timers change) and the agent restarts: the next pass runs the probe
// only, never the verification, and the revert is not stuck.
func TestRevertVerifyThenProbeNeverVerifiesAgain(t *testing.T) {
	t.Parallel()
	h := newRevertHarness(t)
	h.fromStopped(t)
	h.arm(t)
	h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
	h.pass(t)
	if h.verifications() != 1 {
		t.Fatalf("verified %d times, want once", h.verifications())
	}
	c := h.claim(t)
	if c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Op.Phase != hostclaim.PhaseProbing {
		t.Fatalf("rev_phase %q op.phase %q, want probing", c.Op.Journal[hostclaim.JournalRevPhase], c.Op.Phase)
	}
	if len(h.probes) != 1 || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
		t.Fatalf("probes %d row %+v", len(h.probes), h.route)
	}
	// The probe's process writes, and the agent restarts.
	if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("written by the probe's process"), 0o644); err != nil {
		t.Fatal(err)
	}
	h.timers = append(h.timers, registry.Timer{Host: p1Host, Name: "added", Kind: "cron", Spec: "0 4 * * *", NextAt: 2000})
	h.freshWorker()
	h.pass(t)
	if h.verifications() != 1 || len(h.probes) != 2 {
		t.Fatalf("after the restart: verified %d times, %d probes; want 1 and 2", h.verifications(), len(h.probes))
	}
	if m := h.move(t); m.State != provision.MoveReverting || m.StuckReason != "" {
		t.Fatalf("the revert stopped on its own probe's writes: %+v", m)
	}
	h.probe = nil
	h.freshWorker()
	h.pass(t)
	h.reopened(t)
	if h.verifications() != 1 || len(h.probes) != 3 {
		t.Fatalf("verified %d times, %d probes; want 1 and 3", h.verifications(), len(h.probes))
	}
}

// TestRevertVerifyMismatchIsStuck (INV-4): a revert from stopped whose files
// or timers no longer match the stopped baselines stops for an operator with
// the baseline-mismatch sentence before anything is probed; rev_phase stays
// verify, the row stays held restoring.
func TestRevertVerifyMismatchIsStuck(t *testing.T) {
	t.Parallel()
	for _, what := range []string{"files", "timers"} {
		t.Run(what, func(t *testing.T) {
			h := newRevertHarness(t)
			h.fromStopped(t)
			h.arm(t)
			if what == "files" {
				if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil {
					t.Fatal(err)
				}
			} else {
				h.timers = nil
			}
			h.pass(t)
			m := h.move(t)
			if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Reason != provision.RevertCutDeadline {
				t.Fatalf("not stuck with the mismatch: %+v", m)
			}
			if h.revPhase(t) != hostclaim.RevPhaseVerify || len(h.probes) != 0 {
				t.Fatalf("rev_phase %q, %d probes", h.revPhase(t), len(h.probes))
			}
			if h.route.Status != "restoring" || h.route.HoldOp != p1ID {
				t.Fatalf("the row: %+v", h.route)
			}
			h.pass(t)
			if h.verifications() != 1 || h.move(t).State != provision.MoveStuck {
				t.Fatalf("a stuck revert ran again: %d verifications", h.verifications())
			}
		})
	}
}

// retry is the operator's move resolve --retry: provision.MoveRetry through
// the record writer, as the console door writes it (Task 7).
func (h *revertHarness) retry(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetry(g)); err != nil {
		t.Fatal(err)
	}
	if m := h.move(t); m.State != provision.MoveReverting || m.StuckReason != "" {
		t.Fatalf("after the retry: %+v", m)
	}
}

// TestRetryResumesARevertsRevPhase (Task 7's box side for a revert, J20): a
// stuck revert resumes its rev_phase. Stuck at verify, each retry verifies
// again (never skipped), unrepaired it is stuck again, repaired it goes on.
// Stuck at probing after fifteen minutes of failed probes, a retry probes
// again with a fresh fifteen minutes and never verifies again.
func TestRetryResumesARevertsRevPhase(t *testing.T) {
	t.Parallel()
	t.Run("verify", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		h.arm(t)
		upload := filepath.Join(h.cfg.DataDir, p1Host, "upload.txt")
		if err := os.WriteFile(upload, []byte("changed while stopped"), 0o644); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveStuck {
			t.Fatalf("not stuck: %+v", m)
		}
		h.retry(t)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || h.verifications() != 2 || len(h.probes) != 0 {
			t.Fatalf("an unrepaired retry: %+v, %d verifications, %d probes", m, h.verifications(), len(h.probes))
		}
		if err := os.WriteFile(upload, []byte("hello"), 0o644); err != nil {
			t.Fatal(err)
		}
		h.retry(t)
		h.pass(t)
		h.reopened(t)
		if h.verifications() != 3 || len(h.probes) != 1 {
			t.Fatalf("%d verifications, %d probes; want 3 and 1", h.verifications(), len(h.probes))
		}
		if got := h.move(t).Reason; got != provision.RevertCutDeadline {
			t.Fatalf("the retry changed the revert's cause: %q", got)
		}
	})
	t.Run("probing", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		h.arm(t)
		h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
		h.pass(t)
		h.clock.Advance(10 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveReverting {
			t.Fatalf("stuck before fifteen minutes: %+v", m)
		}
		h.clock.Advance(6 * time.Minute)
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
			t.Fatalf("not stuck after fifteen minutes of failed probes: %+v", m)
		}
		if h.revPhase(t) != hostclaim.PhaseProbing || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
			t.Fatalf("rev_phase %q row %+v", h.revPhase(t), h.route)
		}
		probes := len(h.probes)
		h.pass(t)
		if len(h.probes) != probes {
			t.Fatal("a stuck revert probed")
		}
		// The retry gives the probe fifteen minutes again.
		h.retry(t)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveReverting || len(h.probes) != probes+1 {
			t.Fatalf("after the retry: %+v, %d probes", m, len(h.probes))
		}
		h.clock.Advance(14 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveReverting {
			t.Fatalf("the retry's budget did not start again: %+v", m)
		}
		h.clock.Advance(2 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
			t.Fatalf("not stuck again: %+v", m)
		}
		h.retry(t)
		h.probe = nil
		h.pass(t)
		h.reopened(t)
		if h.verifications() != 1 {
			t.Fatalf("verified %d times across the retries, want once", h.verifications())
		}
	})
}

// TestRevertOpensAndReleases (§7.4 opened, §7.11): after the reopen the
// source waits for the lane park the revert queued in the target region. The
// target region's box archives the lane under parked/<inc>/<host>/<ret>/moves/
// through the park engine; the obligation is acknowledged only after its
// horizon; and only then the claim goes to serving under the source and the
// record closes with LastMove rolled_back, the offline time counted from the
// cut to the reopen, and the source's journal parked.
func TestRevertOpensAndReleases(t *testing.T) {
	t.Parallel()
	h := newRevertHarness(t)
	h.fromStopped(t)
	h.arm(t)
	h.pass(t)
	h.reopened(t)
	o := h.laneChore(t)
	if o == nil {
		t.Fatal("no lane park queued")
	}
	lane := h.laneKeys()
	if len(lane) == 0 {
		t.Fatal("the fixture has no lane objects")
	}
	h.pass(t)
	h.reopened(t)
	if h.laneChore(t) == nil || len(h.events) == 0 || h.events[len(h.events)-1] == "rolled_back" {
		t.Fatalf("the source closed before the lane was parked: %v", h.events)
	}
	// The target region's box archives the lane; the horizon has not passed.
	h.tPass(t)
	if left := h.laneKeys(); len(left) != 0 {
		t.Fatalf("the lane still holds %v", left)
	}
	prefix := "parked/" + o.Park.Incarnation + "/" + p1Host + "/" + o.Retirement + "/moves/"
	archived := 0
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			archived++
		}
	}
	if archived < len(lane) {
		t.Fatalf("%d archived under %s, the lane had %d", archived, prefix, len(lane))
	}
	if h.laneChore(t) == nil {
		t.Fatal("the lane park was acknowledged before its horizon")
	}
	h.pass(t)
	h.reopened(t)
	// Past the horizon the target acknowledges, and the source closes.
	h.pastHorizon(t, o)
	h.tPass(t)
	if h.laneChore(t) != nil {
		t.Fatal("the lane park was not acknowledged after its horizon")
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseOpened {
		t.Fatalf("the claim before the source's pass: %+v", c.Op)
	}
	reopenedAt, cutAt := *h.move(t).ReopenedAt, *h.move(t).CutAt
	h.pass(t)
	h.closed(t)
	if got, want := h.read(t, p1Host).LastMove.OfflineS, reopenedAt.Sub(cutAt).Seconds(); got != want || got <= 0 {
		t.Fatalf("offline %v, want %v", got, want)
	}
	if n := countString(h.events, "rolled_back"); n != 1 {
		t.Fatalf("the rolled_back event %d times: %v", n, h.events)
	}
	// Another pass on either box changes nothing.
	before := h.claim(t)
	h.pass(t)
	h.tPass(t)
	if after := h.claim(t); after.Gen != before.Gen {
		t.Fatalf("a pass after the close wrote the claim (%d to %d)", before.Gen, after.Gen)
	}
}

func countString(xs []string, s string) int {
	n := 0
	for _, x := range xs {
		if x == s {
			n++
		}
	}
	return n
}

// Home part 2, Task 4 (found by TestMoveIntoHomePreFlipRevert): a real
// recovery probe wake admits a writer, so the claim the revert's step read
// before the probe is stale when the probe answers. The probed step reads
// the claim again; with the step's own read it lost the CAS every pass, and
// each pass's probe admitted one more fresh generation until the attempts
// were spent. Here, on the default named move, the probe runs through the
// production adapter and the claim runtime (Admit, Prepare, Restore,
// Reserve, Materialize) with the production recovery owners, and admits
// once; the source then reopens in the same pass.
func TestRevertProbedAfterARealAdmission(t *testing.T) {
	h := newRevertHarnessOn(t, newLandingHarness(t, realLaunchEnv).cutHarness)
	h.fromStopped(t)
	h.arm(t)
	rec := revertRecovery(t, h)
	h.pass(t)
	h.reopened(t)
	if rec.wakes != 1 || rec.materialized.Claim == nil {
		t.Fatalf("the revert probe's starts: %d, materialized %v", rec.wakes, rec.materialized.Claim != nil)
	}
	h.pass(t)
	h.reopened(t)
	if rec.wakes != 1 {
		t.Fatalf("a pass after the reopen probed again: %d starts", rec.wakes)
	}
}

// The census condition on that re-read (census-ruling-home-part2): the
// fresh claim is taken only when it differs from the step's own read by the
// probe wake's own writes (its admission of a new generation for this box's
// writer and its restore's re-tag of this box's database). Here the real
// probe answers as above, and one foreign write lands after it and before
// the probed CAS: a change to the op, to the holder or its session, or to a
// claim field the probe never writes. The step refuses it as a conflict,
// the claim stays at probing with the foreign change in place, and the row
// stays held: the step is left to its next pass, as the stale CAS left it.
func TestRevertProbedRefusesAForeignWrite(t *testing.T) {
	for _, e := range []struct {
		name string
		edit func(*hostclaim.Claim)
	}{
		{"op target", func(c *hostclaim.Claim) { c.Op.Intent["target"] = "other-box" }},
		{"op journal", func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalSrcGen] = "other" }},
		{"holder session", func(c *hostclaim.Claim) {
			c.Holder.Session++
			c.Holder.Admissions[len(c.Holder.Admissions)-1].Session++
		}},
		{"holder writer box", func(c *hostclaim.Claim) { c.Holder.WriterBox = "uk-box" }},
		{"quiet key", func(c *hostclaim.Claim) {
			c.QuietAfter = map[string]time.Time{"uk-box": p1Created}
		}},
		{"chore", func(c *hostclaim.Claim) {
			o := c.Obligations[0]
			o.ID = "0123456789abcdef0123456789abcdef"
			c.Obligations = append(c.Obligations, o)
		}},
		{"another box's inventory", func(c *hostclaim.Claim) {
			c.Inventory["uk-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1}}
		}},
	} {
		t.Run(e.name, func(t *testing.T) {
			h := newRevertHarnessOn(t, newLandingHarness(t, realLaunchEnv).cutHarness)
			h.fromStopped(t)
			h.arm(t)
			rec := revertRecovery(t, h)
			probe, wrote := h.cfg.recoveryProbe, false
			var foreign *hostclaim.Claim
			var before, own hostclaim.Observed
			read := func() hostclaim.Observed {
				o, err := hostclaim.Read(context.Background(), h.s, p1Host)
				if err != nil {
					t.Fatal(err)
				}
				return o
			}
			h.cfg.recoveryProbe = func(ctx context.Context, s release.Store, r provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
				before = read()
				out, err := probe(ctx, s, r, req)
				if out == recoveryProbeAnswered && !wrote {
					wrote = true
					own = read()
					rewriteP1Claim(t, h.s, e.edit)
					foreign = h.claim(t)
				}
				return out, err
			}
			err := h.w().revertSource(context.Background(), h.s, h.read(t, p1Host), nil)
			if !wrote || rec.wakes != 1 {
				t.Fatalf("the probe did not answer once: wrote %v, %d starts", wrote, rec.wakes)
			}
			if !errors.Is(err, hostclaim.ErrConflict) {
				t.Fatalf("the revert's probed took a foreign write: %v", err)
			}
			// The same change in place of the probe's own last write (the
			// gen alone cannot tell them apart) is refused too, and the
			// probe's own writes alone pass.
			box := h.w().cfg.Box
			if err := revertProbeWrites(box, h.cfg.DataDir, before, own); err != nil || own.Claim.Gen != before.Claim.Gen+2 {
				t.Fatalf("the probe's own writes (gen %d to %d): %v", before.Claim.Gen, own.Claim.Gen, err)
			}
			swapped, err := revertClaimCopy(own.Claim)
			if err != nil {
				t.Fatal(err)
			}
			e.edit(swapped)
			if err := revertProbeWrites(box, h.cfg.DataDir, before, hostclaim.Observed{Host: own.Host, ETag: own.ETag + "-foreign", Claim: swapped}); !errors.Is(err, hostclaim.ErrConflict) {
				t.Fatalf("a foreign change at the probe's own gen was taken: %v", err)
			}
			c := h.claim(t)
			if c.Op.Phase != hostclaim.PhaseProbing || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Gen != foreign.Gen {
				t.Fatalf("the claim after the refusal: gen %d (foreign %d), %+v", c.Gen, foreign.Gen, c.Op)
			}
			if m := h.move(t); m.ReopenedAt != nil || h.route.HoldOp != p1ID {
				t.Fatalf("the refusal reopened: %+v, row %+v", m, h.route)
			}
		})
	}
}

// TestRevertProbedRefusesAnUnreplayableAdmission pins today's stall, so it
// is explicit and not silent (amadan #300). revertProbeWrites replays the
// probe's admission with hostclaim.AdmitWriter, which cannot vouch for a
// loss-bound admission (AdmitWriterWithLoss) or a fence-bound one
// (AdmitWriterWithFencedTargetLoss): their verified flags are hostclaim's
// own and are not exposed, since exposing them would be a forging surface.
// So the re-read refuses such an admission with hostclaim.ErrConflict and
// the step stays at probing, as it did before the re-read existed. The later
// fix is a hostclaim-owned function that re-verifies those admissions (#300).
//
// The recovery fixtures produce both kinds only on a return (a restoring
// claim); revertProbeWrites reads no op kind, so the real admissions made
// through the claim's writer are checked against it directly, beside a plain
// admission over the same fixture that does replay. The revert step itself
// is then driven with a probe whose one claim write is shaped as a
// loss-bound admission.
func TestRevertProbedRefusesAnUnreplayableAdmission(t *testing.T) {
	admit := func(t *testing.T, f *rcFixture, pub ed25519.PublicKey) (hostclaim.Observed, hostclaim.Observed) {
		t.Helper()
		before := f.read()
		in := f.selectAdmission(before)
		after, err := f.lossWriter(pub).Admit(t.Context(), before, in.Admission, in.Baseline)
		if err != nil {
			t.Fatal(err)
		}
		if after.Claim.Gen != before.Claim.Gen+1 {
			t.Fatalf("the admission wrote gen %d over %d", after.Claim.Gen, before.Claim.Gen)
		}
		return before, after
	}
	t.Run("a plain admission replays", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		before, after := admit(t, f, nil)
		if err := revertProbeWrites(f.srcBox, f.dataDir, before, after); err != nil {
			t.Fatalf("the plain admission was refused: %v", err)
		}
	})
	t.Run("loss-bound", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		pub := f.declare(false, []hostclaim.WriteSource{rcTargetSource("t2", 25), rcTargetSource(poOp, 20), rcCutSource()}, func(c *hostclaim.Claim) {
			c.Holder.Session = 25
			c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
		})
		before, after := admit(t, f, pub)
		if a := after.Claim.Holder.Admissions[len(after.Claim.Holder.Admissions)-1]; a.LossID != rcLossID {
			t.Fatalf("not a loss-bound admission: %+v", a)
		}
		if err := revertProbeWrites(f.srcBox, f.dataDir, before, after); !errors.Is(err, hostclaim.ErrConflict) || !strings.Contains(err.Error(), "amadan #300") {
			t.Fatalf("a loss-bound admission was replayed: %v", err)
		}
	})
	t.Run("fence-bound", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		record := f.fence()
		before, after := admit(t, f, nil)
		if after.Claim.Op.Journal["fence_admitted."+record.ID] == "" {
			t.Fatalf("not a fence-bound admission: %+v", after.Claim.Op.Journal)
		}
		if err := revertProbeWrites(f.srcBox, f.dataDir, before, after); !errors.Is(err, hostclaim.ErrConflict) || !strings.Contains(err.Error(), "amadan #300") {
			t.Fatalf("a fence-bound admission was replayed: %v", err)
		}
	})
	t.Run("the revert step stays at probing", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		h.arm(t)
		var admitted *hostclaim.Claim
		h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) {
			if admitted == nil {
				rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
					hd := c.Holder
					session := c.Gen + 1
					hd.Admissions = append(hd.Admissions, hostclaim.AdmissionRecord{Session: session, Box: h.srcBox, Generation: "lost-gen", Predecessor: hd.Generation, Mode: "cold", Region: "ie", PredecessorRegion: "ie", LossID: rcLossID})
					hd.Session, hd.Stamp, hd.Generation, hd.WriterBox = session, session, "lost-gen", h.srcBox
					c.Op.Journal["loss_admitted."+rcLossID] = strconv.FormatUint(session, 10)
				})
				admitted = h.claim(t)
			}
			return recoveryProbeAnswered, nil
		}
		err := h.w().revertSource(context.Background(), h.s, h.read(t, p1Host), nil)
		if admitted == nil {
			t.Fatal("the probe never ran")
		}
		if !errors.Is(err, hostclaim.ErrConflict) || !strings.Contains(err.Error(), "amadan #300") {
			t.Fatalf("the revert's probed took a loss-bound admission: %v", err)
		}
		c := h.claim(t)
		if c.Gen != admitted.Gen || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
			t.Fatalf("the claim after the refusal: gen %d (admitted %d), %+v", c.Gen, admitted.Gen, c.Op)
		}
		if m := h.move(t); m.ReopenedAt != nil || h.route.HoldOp != p1ID {
			t.Fatalf("the refusal reopened: %+v, row %+v", m, h.route)
		}
	})
}

// TestRevertReplays: a crash after any of the revert's side effects, then a
// restarted agent, still reaches the close, with the evidence asked once it
// is journalled, the hold only ever restoring under the move id, and the
// verification never run again once probing is journalled.
func TestRevertReplays(t *testing.T) {
	t.Parallel()
	type entry struct {
		name   string
		from   func(*revertHarness, *testing.T)
		cut    []string // the cut's fault points (giveUpExecution)
		points []string // the revert's
	}
	for _, e := range []entry{
		{"from stopping", (*revertHarness).fromStopping, []string{"after-checkpoint", "after-report", "after-exec-none"},
			[]string{"after-revert-evidence", "after-revert-claim", "after-restatus", "after-rev-probing", "after-rev-probed", "after-reopened", "after-row-opened", "after-rev-opened", "after-release-move", "after-close"}},
		{"from stopped", (*revertHarness).fromStopped, nil,
			[]string{"after-restatus", "verify", "after-rev-verify", "after-rev-probing", "after-rev-probed", "after-reopened", "after-row-opened", "after-rev-opened", "after-release-move", "after-close"}},
	} {
		for _, point := range append(append([]string{}, e.cut...), e.points...) {
			t.Run(e.name+"/"+point, func(t *testing.T) {
				h := newRevertHarness(t)
				e.from(h, t)
				if e.name == "from stopping" {
					h.ev.stopped = true
				}
				h.arm(t)
				isCut := false
				for _, p := range e.cut {
					isCut = isCut || p == point
				}
				// The fixture's own crash (fromStopped) has used the cut's
				// seam: each case watches only its own.
				crashed := func() bool { return h.rvDone }
				if isCut {
					h.crashAt, h.crashed = point, false
					crashed = func() bool { return h.crashed }
				} else {
					h.rvAt = point
				}
				for i := 0; i < 8 && !crashed(); i++ {
					h.pass(t)
					h.tPass(t)
					if c := h.claim(t); c.Op.Kind == hostclaim.KindReverting && c.Op.Phase == hostclaim.PhaseOpened {
						if o := h.laneChore(t); o != nil {
							h.pastHorizon(t, o)
						}
					}
				}
				if !crashed() {
					t.Fatalf("never reached %s (revert %v, cut %v)", point, h.rvHits, h.hits)
				}
				h.freshWorker()
				h.drive(t)
				h.closed(t)
				switch {
				case e.name == "from stopping" && h.verifications() != 0:
					t.Fatalf("a revert from stopping verified %d times", h.verifications())
				case e.name == "from stopped" && point == "verify" && h.verifications() != 2:
					t.Fatalf("verified %d times after a crash before the verification, want 2 (the crashed one counted)", h.verifications())
				case e.name == "from stopped" && point != "verify" && h.verifications() != 1:
					t.Fatalf("verified %d times, want once", h.verifications())
				}
				if len(h.probes) == 0 {
					t.Fatal("never probed")
				}
			})
		}
	}
}

// TestRevertJournalSweptAfterTheClose (the local journal sweep, Tasks 4 and
// 5): a crash after the close and before the source's journal is parked
// leaves it to the sweep, which parks it once no record carries the move; a
// journal whose revert is still in flight is left.
func TestRevertJournalSweptAfterTheClose(t *testing.T) {
	t.Parallel()
	h := newRevertHarness(t)
	h.fromStopped(t)
	h.arm(t)
	h.pass(t)
	h.reopened(t)
	// In flight: the sweep leaves the marked journal.
	h.w().sweepFinalJournals(context.Background(), h.s, []provision.Record{h.read(t, p1Host)})
	if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
		t.Fatal("the sweep parked a journal whose revert is in flight")
	}
	h.rvAt = "after-close"
	o := h.laneChore(t)
	h.tPass(t)
	h.pastHorizon(t, o)
	h.tPass(t)
	h.pass(t)
	if !h.rvDone {
		t.Fatalf("never reached the close: %v", h.rvHits)
	}
	if h.read(t, p1Host).Move != nil {
		t.Fatal("the close did not land")
	}
	if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
		t.Fatal("the crash after the close parked the journal")
	}
	h.freshWorker()
	h.pass(t)
	h.closed(t)
}

// TestRevertUnarmedInProduction (J18): production wires no recovery probe
// until Task 9 and no park engine leases until Plan 5, so the revert's
// recovery and the target's lane park are unarmed there. The claim half
// still follows the record (Plan 2's revertClaim); nothing else moves.
func TestRevertUnarmedInProduction(t *testing.T) {
	t.Parallel()
	if revertSourceArmed(moveCfg{Box: "ie-box", DataDir: "/d", RestatusHold: func(string, string, string, func() error) (bool, error) { return false, nil },
		ReleaseHeld: func(string, string) (bool, error) { return false, nil }, Timers: func(string) ([]registry.Timer, error) { return nil, nil },
		FilesDir: func(string) (string, string) { return "", "" }}) {
		t.Fatal("the revert is armed without a recovery probe")
	}
	if revertLaneArmed(moveCfg{Box: "uk-box", Region: "uk", OwnBucket: "data-uk", Versioned: func(context.Context, string, string) (lease.VersionedStore, error) { return nil, nil }}) {
		t.Fatal("the lane park is armed without the park engine's leases")
	}
	t.Run("from stopping", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopping(t)
		h.ev.stopped = true
		h.pass(t)
		if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
			t.Fatalf("an unarmed revert gave up execution: %+v", c.Op)
		}
	})
	t.Run("from stopped", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		lane := len(h.laneKeys())
		h.pass(t)
		c := h.claim(t)
		if c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.RevPhaseEvidence || h.route.Status != "moving" {
			t.Fatalf("an unarmed revert went on: %+v row %+v", c.Op, h.route)
		}
		tc := h.target()
		tc.DataDir = h.tdir
		h.tw = &movesWorker{cfg: tc, mirrors: map[string]lease.MirrorStore{}}
		if err := movesPass(context.Background(), h.s, h.tw); err != nil {
			t.Fatal(err)
		}
		if len(h.laneKeys()) != lane || len(h.tleases.acquired) != 0 {
			t.Fatal("an unarmed target parked the lane")
		}
	})
	t.Run("the claim half", func(t *testing.T) {
		// The record reverts and the claim half is lost: the unarmed pass
		// still writes it, as Plan 2's reverting arm did.
		h := newRevertHarness(t)
		h.ready(t)
		h.crashAt = "after-exec-none"
		h.cut(t)
		h.crashPass(t)
		g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
		if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRevert(g, provision.RevertCutDeadline)); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting {
			t.Fatalf("the claim half: %+v", c.Op)
		}
	})
}

// A reverting record's target region box has neither the row nor a journal
// of the move (the revert never flipped), so the participant rule answers
// it as the target on the state alone: its one piece of work is the lane park
// the claim owes its region. Any other state still needs the row or the
// journal.
func TestRevertingTargetIsAParticipant(t *testing.T) {
	cfg := moveCfg{Region: "uk", DataDir: t.TempDir(), OwnsRoute: func(string) bool { return false }}
	rec := func(state string) provision.Record {
		return provision.Record{Host: p1Host, Region: "ie", Move: &provision.MoveState{ID: p1ID, State: state, From: "ie", To: "uk"}}
	}
	if got := movesParticipantRecord(cfg, rec(provision.MoveReverting)); got != roleTarget {
		t.Fatalf("a reverting record's target: %q", got)
	}
	for _, state := range []string{provision.MoveCommitted, provision.MoveRollingBack, provision.MoveStuck, provision.MoveCutting} {
		if got := movesParticipantRecord(cfg, rec(state)); got != roleNone {
			t.Fatalf("%s with no row and no journal: %q", state, got)
		}
	}
	cfg.Region = "fr"
	if got := movesParticipantRecord(cfg, rec(provision.MoveReverting)); got != roleNone {
		t.Fatalf("a third region: %q", got)
	}
}

// restatusHold (production's RestatusHold): an op's own hold changes status
// under the host's hold lock, after the caller's validation, touching the
// stage-requested marker only when the change landed. A replay answers true
// and touches nothing; a row held by another op, an unheld row and a missing
// row answer false and are left; a validation error moves nothing.
func TestRestatusHoldIsConditional(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	for _, rt := range []registry.Route{
		{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true},
		{Host: "b.go.example.com", Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	var written []string
	restatus := restatusHold(reg, func(host string) { written = append(written, host) })
	if ok, err := restatus(p1Host, "restoring", p1ID, nil); ok || err != nil {
		t.Fatalf("an unheld row: %v %v", ok, err)
	}
	if err := holdMoving(reg, nil)(p1Host, p1ID); err != nil {
		t.Fatal(err)
	}
	refused := errors.New("the record moved on")
	if ok, err := restatus(p1Host, "restoring", p1ID, func() error { return refused }); ok || !errors.Is(err, refused) {
		t.Fatalf("a refused validation: %v %v", ok, err)
	}
	if rt, _ := reg.Get(p1Host); rt.Status != "moving" || len(written) != 0 {
		t.Fatalf("a refused validation moved the row: %+v, marker %d", rt, len(written))
	}
	for i := 0; i < 2; i++ {
		if ok, err := restatus(p1Host, "restoring", p1ID, func() error { return nil }); !ok || err != nil {
			t.Fatalf("restatus %d: %v %v", i, ok, err)
		}
	}
	if rt, _ := reg.Get(p1Host); rt.Status != "restoring" || rt.HoldOp != p1ID || len(written) != 1 {
		t.Fatalf("row %+v; marker touched %d times", rt, len(written))
	}
	if ok, err := restatus(p1Host, "restoring", p1ID2, nil); ok || err != nil {
		t.Fatalf("another op's hold: %v %v", ok, err)
	}
	if err := holdMoving(reg, nil)("b.go.example.com", p1ID2); err != nil {
		t.Fatal(err)
	}
	if ok, err := restatus("b.go.example.com", "restoring", p1ID, nil); ok || err != nil {
		t.Fatalf("a row held by another op: %v %v", ok, err)
	}
	if rt, _ := reg.Get("b.go.example.com"); rt.Status != "moving" || rt.HoldOp != p1ID2 {
		t.Fatalf("another op's row moved: %+v", rt)
	}
	if ok, err := restatus("absent.example.com", "restoring", p1ID, nil); ok || err != nil {
		t.Fatalf("a missing row: %v %v", ok, err)
	}
	// Production wires it, and leaves the revert and the lane park unarmed.
	var prod moveCfg
	hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}
	armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
	if prod.RestatusHold == nil || prod.ReleaseHeld == nil {
		t.Fatal("production does not wire the revert's row hooks")
	}
	prod.Box, prod.Region, prod.OwnBucket = "ie-box", "ie", "data-ie"
	if revertSourceArmed(prod) || revertLaneArmed(prod) {
		t.Fatal("production armed the revert")
	}
}

// A revert from stopping whose files directory cannot be archived (a special
// file) is not refused again: nothing compares a files digest on this path
// (§7.4 evidence), so giveUpExecution publishes the report without one,
// records no baselines, and the revert goes on.
func TestRevertFromStoppingWithUnarchivableFiles(t *testing.T) {
	t.Parallel()
	h := newRevertHarness(t)
	h.fromStopping(t)
	if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil {
		t.Fatal(err)
	}
	h.arm(t)
	h.ev.stopped = true
	h.pass(t)
	h.reopened(t)
	rep, err := readStoppedReport(context.Background(), h.s, h.stoppedKey())
	if err != nil || !rep.Complete() || rep.FilesPlain != "" {
		t.Fatalf("the stopped report: %v %+v", err, rep)
	}
	if m := h.move(t); m.Source != nil || m.Reason != provision.RevertStopEvidenceBudget {
		t.Fatalf("the record: %+v", m)
	}
	if h.verifications() != 0 || len(h.probes) != 1 {
		t.Fatalf("%d verifications, %d probes", h.verifications(), len(h.probes))
	}
}

// TestRevertProbeAttemptsExhaustedIsStuck (§7.7; Task 6 fix round 1): a
// revert's probe has the recovery probe's nine attempts as well as the
// fifteen minutes. A probe that uses them all stops the revert for an
// operator with the recovery probe's sentence, long before fifteen minutes;
// rev_phase stays probing and the row held. A retry probes again with nine
// attempts and never verifies again.
func TestRevertProbeAttemptsExhaustedIsStuck(t *testing.T) {
	t.Parallel()
	h := newRevertHarness(t)
	h.fromStopped(t)
	h.arm(t)
	h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeExhausted, nil }
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts || m.Reason != provision.RevertCutDeadline {
		t.Fatalf("not stuck on the exhausted probe: %+v", m)
	}
	if len(h.probes) != 1 || h.probes[0].Attempts != recoveryProbeAttempts {
		t.Fatalf("probes %+v, want one with %d attempts", h.probes, recoveryProbeAttempts)
	}
	if h.revPhase(t) != hostclaim.PhaseProbing || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
		t.Fatalf("rev_phase %q row %+v", h.revPhase(t), h.route)
	}
	h.retry(t)
	h.probe = nil
	h.pass(t)
	h.reopened(t)
	if len(h.probes) != 2 || h.probes[1].Attempts != recoveryProbeAttempts || h.verifications() != 1 {
		t.Fatalf("after the retry: probes %+v, %d verifications", h.probes, h.verifications())
	}
}

// failStuckStore fails the record write that would mark the move stuck, as a
// store that refuses or loses the request does.
type failStuckStore struct {
	release.Store
	fail bool
}

func (f *failStuckStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if f.fail && key == provision.RecordKey("bab", "app", p1Host) && strings.Contains(string(body), `"state":"stuck"`) {
		return "", errors.New("injected: the record write failed")
	}
	return f.Store.Put(ctx, key, body, ifMatch)
}

// TestRevertProbeExhaustionSurvivesACrash (Task 6 fix round 1): once the
// revert's probe is exhausted (its attempts, or its fifteen minutes) the
// exhaustion is journalled before the stuck mark. A crash between the two, or
// a stuck mark the store refused, leaves the record reverting; the next pass
// marks it stuck without probing again and never grants a fresh window. Only
// a retry (the record back from stuck through MoveRetry) starts one.
func TestRevertProbeExhaustionSurvivesACrash(t *testing.T) {
	t.Parallel()
	for _, reason := range []string{provision.StuckRecoveryProbeAttempts, provision.StuckRevertProbeBudget} {
		for _, how := range []string{"crash", "failed record write"} {
			t.Run(reason+"/"+how, func(t *testing.T) {
				h := newRevertHarness(t)
				h.fromStopped(t)
				h.arm(t)
				exhausted := reason == provision.StuckRecoveryProbeAttempts
				// The probe records what it renders before each attempt, so
				// the record moves in the very pass the window ends.
				h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
					if err := req.Record(context.Background(), revertRendered()); err != nil {
						return recoveryProbeWaiting, err
					}
					if exhausted {
						return recoveryProbeExhausted, nil
					}
					return recoveryProbeWaiting, nil
				}
				if !exhausted {
					h.pass(t) // the window opens
					h.clock.Advance(16 * time.Minute)
				}
				probes := len(h.probes)
				fs := &failStuckStore{Store: h.s}
				if how == "crash" {
					h.rvAt = "after-probe-exhausted"
				} else {
					fs.fail = true
				}
				if err := movesPass(context.Background(), fs, h.w()); err != nil {
					t.Fatal(err)
				}
				if how == "crash" && !h.rvDone {
					t.Fatalf("never reached after-probe-exhausted: %v", h.rvHits)
				}
				if m := h.move(t); m.State != provision.MoveReverting {
					t.Fatalf("the fixture's stuck mark landed: %+v", m)
				}
				if len(h.probes) != probes+1 {
					t.Fatalf("%d probes in the exhausting pass, want 1", len(h.probes)-probes)
				}
				// The next pass, on a restarted agent with the clock where it
				// was: stuck, with no probe and no fresh window.
				h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeAnswered, nil }
				h.freshWorker()
				h.pass(t)
				m := h.move(t)
				if m.State != provision.MoveStuck || m.StuckReason != reason {
					t.Fatalf("the next pass did not mark it stuck: %+v", m)
				}
				if len(h.probes) != probes+1 {
					t.Fatalf("the next pass probed again (%d probes)", len(h.probes)-probes)
				}
				// A retry starts a new window, and its probe answers.
				h.retry(t)
				h.pass(t)
				h.reopened(t)
				if h.verifications() != 1 {
					t.Fatalf("verified %d times, want once", h.verifications())
				}
			})
		}
	}
	// A retry's new window is a whole fifteen minutes, even with the earlier
	// exhaustion still journalled when the stuck mark landed.
	t.Run("the retry's window", func(t *testing.T) {
		h := newRevertHarness(t)
		h.fromStopped(t)
		h.arm(t)
		h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
		h.pass(t)
		h.clock.Advance(16 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
			t.Fatalf("not stuck: %+v", m)
		}
		h.retry(t)
		h.freshWorker()
		h.pass(t)
		h.clock.Advance(14 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveReverting {
			t.Fatalf("the retry's window was not fresh: %+v", m)
		}
		h.clock.Advance(2 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveStuck {
			t.Fatalf("the retry's window never ended: %+v", m)
		}
	})
}