package hostclaim

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

const (
	moveA = "mv_abcdefghijklmnopqrstuvwxyz"
	moveB = "mv_bcdefghijklmnopqrstuvwxyza"
)

// movingTestClaim is this file's fixture (delete_test.go already defines
// servingClaim(t, ctx, s) with another shape). It is a VALID claim: the
// holder's key, incarnation and location follow the grammar Validate
// enforces, and the fixture asserts Validate() so it can never drift from
// the real grammar.
func movingTestClaim(t *testing.T) *Claim {
	t.Helper()
	created := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
	inc, err := Incarnation("bab", "app", "a.go.example.com", created)
	if err != nil {
		t.Fatal(err)
	}
	c := &Claim{V: schemaVersion, Host: "a.go.example.com", Gen: 3,
		Holder:  &Holder{Family: "instance", Key: "apps/bab/app/instances/a.go.example.com.json", Account: "bab", App: "app", Incarnation: inc, CreatedAt: created, Location: "instances/a.go.example.com/" + inc, Stamp: 3},
		Op:      Op{Kind: KindServing, Exec: "ie-box"},
		Touched: Touched{Complete: true, Regions: []string{"ie"}, Boxes: []string{"ie-box"}}}
	if err := c.Validate(); err != nil {
		t.Fatalf("fixture must validate: %v", err)
	}
	return c
}

func TestBeginMoveFromServingOnly(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Kind != KindMoving || c.Op.ID != moveA || c.Op.Exec != "ie-box" || c.Op.By != "ops" || !c.Op.StartedAt.Equal(now) || len(c.Op.Participants) != 2 || c.Op.Participants[0] != "ie-box" || c.Op.Participants[1] != "uk-box" || c.Op.Intent["to"] != "uk" {
		t.Fatalf("op after BeginMove: %+v", c.Op)
	}
	// A plain BeginMove carries the move's own three keys and nothing else:
	// the admission and both enrollments are BeginAdmittedMove's.
	if want := map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}; !reflect.DeepEqual(c.Op.Intent, want) {
		t.Fatalf("intent after BeginMove = %v, want %v", c.Op.Intent, want)
	}
	if c.Op.Phase != PhaseMoving {
		t.Fatalf("phase after BeginMove: %q", c.Op.Phase)
	}
	if err := c.Validate(); err != nil {
		t.Fatalf("claim after BeginMove must validate: %v", err)
	}
	// K12: nothing is touched yet; the source adds the region before its
	// first staging round.
	if len(c.Touched.Regions) != 1 || len(c.Touched.Boxes) != 1 {
		t.Fatalf("touched grew at BeginMove: %+v", c.Touched)
	}
	// Not from serving.
	if err := BeginMove(moveB, "ops", "ie-box", "uk-box", "uk", now)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("second move: %v", err)
	}
	// exec must be the source.
	d := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "us-box", "uk-box", "uk", now)(d); !errors.Is(err, ErrWrongState) {
		t.Fatalf("source not exec: %v", err)
	}
	// A pending chore in the target region refuses (CC11).
	e := movingTestClaim(t)
	e.Obligations = []Obligation{{ID: "o1", Op: "x", Kind: "park-replica", Region: "uk", Retirement: "r", Owner: "region:uk"}}
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(e); !errors.Is(err, ErrPendingChore) {
		t.Fatalf("pending chore: %v", err)
	}
	// ... and in the target box.
	f := movingTestClaim(t)
	f.Obligations = []Obligation{{ID: "o1", Op: "x", Kind: "park-local", Box: "uk-box", Retirement: "r", Owner: "box:uk-box"}}
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(f); !errors.Is(err, ErrPendingChore) {
		t.Fatalf("pending box chore: %v", err)
	}
	// A claim with no holder is not movable.
	g := movingTestClaim(t)
	g.Holder = nil
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(g); !errors.Is(err, ErrWrongState) {
		t.Fatalf("no holder: %v", err)
	}
	// Missing inputs are malformed.
	for _, args := range [][4]string{{"", "ie-box", "uk-box", "uk"}, {moveA, "", "uk-box", "uk"}, {moveA, "ie-box", "", "uk"}, {moveA, "ie-box", "uk-box", ""}} {
		h := movingTestClaim(t)
		if err := BeginMove(args[0], "ops", args[1], args[2], args[3], now)(h); !errors.Is(err, ErrMalformed) {
			t.Fatalf("args %v: %v", args, err)
		}
		if h.Op.Kind != KindServing {
			t.Fatalf("a refused BeginMove changed the op: %+v", h.Op)
		}
	}
}

func TestTouchAppendsOnceUnderTheOp(t *testing.T) {
	c := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", time.Now())(c); err != nil {
		t.Fatal(err)
	}
	if err := Touch(moveA, []string{"uk"}, nil)(c); err != nil {
		t.Fatal(err)
	}
	if err := Touch(moveA, []string{"uk"}, []string{"uk-box"})(c); err != nil {
		t.Fatal(err)
	}
	if len(c.Touched.Regions) != 2 || len(c.Touched.Boxes) != 2 || c.Touched.Boxes[1] != "uk-box" {
		t.Fatalf("touched: %+v", c.Touched)
	}
	if err := Touch("mv_other0000000000000000000000", []string{"se"}, nil)(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("touch under another op: %v", err)
	}
	if len(c.Touched.Regions) != 2 {
		t.Fatalf("a refused Touch changed touched: %+v", c.Touched)
	}
	if err := Touch("", []string{"se"}, nil)(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("touch under no op: %v", err)
	}
	if err := RequireMoveOp(moveA, KindMoving, "")(c); err != nil {
		t.Fatal(err)
	}
	if err := RequireMoveOp(moveA, KindLanding, "")(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("wrong kind: %v", err)
	}
	if err := RequireMoveOp(moveA, KindMoving, PhaseSealed)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("wrong phase: %v", err)
	}
	if err := RequireMoveOp(moveB, KindMoving, "")(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("wrong op: %v", err)
	}
	c.Op.Phase = PhaseSealed
	if err := RequireMoveOp(moveA, KindMoving, PhaseSealed)(c); err != nil {
		t.Fatalf("sealed: %v", err)
	}
}

func TestReleaseMoveFromMovingNeedsNoChores(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	for _, phase := range []string{PhaseMoving} {
		c := movingTestClaim(t)
		if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(c); err != nil {
			t.Fatal(err)
		}
		c.Op.Phase = phase
		// A chore under this op blocks the release.
		c.Obligations = []Obligation{{ID: "o1", Op: moveA, Kind: "park-replica", Region: "uk", Retirement: "r", Owner: "region:uk"}}
		if err := ReleaseMove(moveA, "ie-box")(c); !errors.Is(err, ErrPendingChore) {
			t.Fatalf("phase %q with a chore: %v", phase, err)
		}
		if c.Op.Kind != KindMoving {
			t.Fatalf("a refused release changed the op: %+v", c.Op)
		}
		// A chore under another op does not.
		c.Obligations[0].Op = "someone-else"
		if err := ReleaseMove(moveB, "ie-box")(c); !errors.Is(err, ErrStaleOp) {
			t.Fatalf("release under another op: %v", err)
		}
		if err := ReleaseMove(moveA, "ie-box")(c); err != nil {
			t.Fatalf("phase %q: %v", phase, err)
		}
		if c.Op.Kind != KindServing || c.Op.ID != "" || c.Op.Exec != "ie-box" || c.Op.Journal["released_move"] != moveA || c.Op.Phase != "" || len(c.Op.Participants) != 0 {
			t.Fatalf("op after release: %+v", c.Op)
		}
		if err := c.Validate(); err != nil {
			t.Fatalf("claim after release must validate: %v", err)
		}
	}
	// Not from an aborting move (ruling K7: that is ReleaseAbort's, after
	// the lane park the abort queued).
	a := movingTestClaim(t)
	_ = BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(a)
	a.Op.Phase = PhaseAborting
	if err := ReleaseMove(moveA, "ie-box")(a); !errors.Is(err, ErrWrongState) {
		t.Fatalf("release from aborting: %v", err)
	}
	// Not from serving, and not from a sealed move.
	s := movingTestClaim(t)
	if err := ReleaseMove(moveA, "ie-box")(s); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("release from serving: %v", err)
	}
	m := movingTestClaim(t)
	_ = BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(m)
	m.Op.Phase = PhaseSealed
	if err := ReleaseMove(moveA, "ie-box")(m); !errors.Is(err, ErrWrongState) {
		t.Fatalf("release from sealed: %v", err)
	}
	// Another kind under the same id is not a move.
	l := movingTestClaim(t)
	_ = BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(l)
	l.Op.Kind = KindLanding
	if err := ReleaseMove(moveA, "ie-box")(l); !errors.Is(err, ErrWrongState) {
		t.Fatalf("release from landing: %v", err)
	}
	// The executor is named.
	e := movingTestClaim(t)
	_ = BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(e)
	if err := ReleaseMove(moveA, "")(e); !errors.Is(err, ErrMalformed) {
		t.Fatalf("release without exec: %v", err)
	}
}

func TestReleaseMoveLeavesEvidence(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	plain := movingTestClaim(t)
	if err := RequireReleasedMove(moveA)(plain); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("a plain serving claim satisfies the replay rule: %v", err)
	}
	c := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(c); err != nil {
		t.Fatal(err)
	}
	if err := RequireReleasedMove(moveA)(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("a moving claim satisfies the replay rule: %v", err)
	}
	if err := ReleaseMove(moveA, "ie-box")(c); err != nil {
		t.Fatal(err)
	}
	if err := RequireReleasedMove(moveA)(c); err != nil {
		t.Fatalf("evidence after release: %v", err)
	}
	if err := RequireReleasedMove(moveB)(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("another move's id: %v", err)
	}
	if err := RequireReleasedMove("")(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("empty id: %v", err)
	}
	// A non-serving kind carrying the journal is not evidence.
	c.Op.Kind, c.Op.ID = KindMoving, moveB
	if err := RequireReleasedMove(moveA)(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("moving claim with a stale journal: %v", err)
	}
}

// BeginMove and ReleaseMove own only the op: the holder (session, writer,
// admissions, generation and their writes_lost), touched, inventory,
// obligations of other ops and the rest are carried through unchanged.
func TestMoveRoundTripPreservesTheRestOfTheClaim(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	c := movingTestClaim(t)
	c.Holder.Generation = "gen-7"
	c.Holder.Session = 9
	c.Holder.WriterBox = "ie-box"
	c.Holder.Admissions = []AdmissionRecord{{Region: "ie", Session: 9, Box: "ie-box", Generation: "gen-7", Mode: "admitted", Coverage: "covered", Covered: true,
		WritesLost:           &WritesLost{Box: "ie-box", Generation: "gen-6", Session: 8, RemoteMaxTXID: 41},
		AdditionalWritesLost: []WritesLost{{Box: "ie-box", Generation: "gen-5", Session: 7, RemoteMaxTXID: 12}}}}
	c.Obligations = []Obligation{{ID: "o1", Op: "old", Kind: "park-replica", Region: "se", Retirement: "r", Owner: "region:se"}}
	c.Lost = []string{"se"}
	c.RetiredIncarnations = []string{"abc"}
	c.QuietAfter = map[string]time.Time{"se": now}
	c.Inventory = map[string]InventoryEntry{"ie-box": {Layout: "nested", Paths: map[string]uint64{"/data/x": 3}}}
	if err := c.Validate(); err != nil {
		t.Fatalf("fixture: %v", err)
	}
	before := mustJSON(t, c)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(c); err != nil {
		t.Fatal(err)
	}
	if err := ReleaseMove(moveA, "ie-box")(c); err != nil {
		t.Fatal(err)
	}
	// Only the op differs.
	var round Claim
	if err := json.Unmarshal(before, &round); err != nil {
		t.Fatal(err)
	}
	round.Op = c.Op
	if !reflect.DeepEqual(round, *c) {
		t.Fatalf("round trip changed more than the op:\nbefore %s\nafter  %s", before, mustJSON(t, c))
	}
	if err := c.Validate(); err != nil {
		t.Fatalf("after round trip: %v", err)
	}
}

func mustJSON(t *testing.T, c *Claim) []byte {
	t.Helper()
	b, err := json.Marshal(c)
	if err != nil {
		t.Fatal(err)
	}
	return b
}

const stoppedSource = "moves/bab/app/h/mv/stopped-source.json"

// movingToSealed walks a fresh moving claim to the state the flip needs:
// source evidence journalled, exec none, phase sealed.
func movingToSealed(t *testing.T) *Claim {
	t.Helper()
	c := movingTestClaim(t)
	c.Holder.Generation = "gen-7" // the source's generation, which the flip must name
	for i, tr := range []func(*Claim) error{
		BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", time.Now()),
		ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource),
		SetJournal(moveA, map[string]string{JournalSeal: `{"n":7}`, JournalCutSession: "5"}),
		SetMovePhase(moveA, KindMoving, PhaseMoving, PhaseSealed),
	} {
		if err := tr(c); err != nil {
			t.Fatalf("step %d: %v", i, err)
		}
	}
	return c
}

func TestCutClaimCASesFollowK3AndK4(t *testing.T) {
	c := movingTestClaim(t)
	op := moveA
	if err := BeginMove(op, "ops", "ie-box", "uk-box", "uk", time.Now())(c); err != nil {
		t.Fatal(err)
	}
	// Only the executor gives execution up, and only citing evidence.
	if err := ExecToNone(op, "uk-box", stoppedSource, JournalEvidenceSource)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("non-exec to none: %v", err)
	}
	if err := ExecToNone(op, "ie-box", "", JournalEvidenceSource)(c); !errors.Is(err, ErrMalformed) {
		t.Fatalf("no evidence: %v", err)
	}
	if err := ExecToNone(op, "ie-box", stoppedSource, JournalSeal)(c); !errors.Is(err, ErrMalformed) {
		t.Fatalf("evidence under a non-evidence key: %v", err)
	}
	if err := ExecToNone(moveB, "ie-box", stoppedSource, JournalEvidenceSource)(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("another op: %v", err)
	}
	// The flip needs the seal phase and exec none.
	if err := Flip(op, "uk-box", 3, "")(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("flip before sealed: %v", err)
	}
	if err := ExecToNone(op, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil || c.Op.Exec != "" {
		t.Fatalf("to none: %v %+v", err, c.Op)
	}
	// A crash replay of the same give-up is idempotent; other evidence is not.
	if err := ExecToNone(op, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil {
		t.Fatalf("replay: %v", err)
	}
	if err := ExecToNone(op, "ie-box", "moves/other.json", JournalEvidenceSource)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("replay with other evidence: %v", err)
	}
	// With exec already none, an empty box is nobody: it must neither
	// replay nor replace the evidence.
	given := mustJSON(t, c)
	for _, ev := range []string{stoppedSource, "moves/other.json"} {
		if err := ExecToNone(op, "", ev, JournalEvidenceSource)(c); !errors.Is(err, ErrMalformed) {
			t.Fatalf("empty box with %q: %v", ev, err)
		}
		if got := mustJSON(t, c); string(got) != string(given) {
			t.Fatalf("empty box with %q mutated the claim: %s", ev, got)
		}
	}
	// Exec none and evidence, but not sealed: still no flip.
	if err := Flip(op, "uk-box", 3, "")(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("flip before the seal phase: %v", err)
	}
	if err := SetJournal(op, map[string]string{JournalSeal: `{"n":7}`, JournalCutSession: "5"})(c); err != nil || c.Op.Journal[JournalSeal] != `{"n":7}` {
		t.Fatalf("journal: %v", err)
	}
	if err := SetJournal(moveB, map[string]string{JournalSeal: "x"})(c); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("journal under another op: %v", err)
	}
	if err := SetMovePhase(op, KindLanding, PhaseMoving, PhaseSealed)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("phase under the wrong kind: %v", err)
	}
	if err := SetMovePhase(op, KindMoving, PhaseMoving, PhaseSealed)(c); err != nil || c.Op.Phase != PhaseSealed {
		t.Fatalf("phase: %v", err)
	}
	if err := SetMovePhase(op, KindMoving, PhaseMoving, PhaseSealed)(c); !errors.Is(err, ErrWrongState) {
		t.Fatal("phase replay")
	}
	// The flip's target must be the move's target.
	if err := Flip(op, "us-box", 3, "")(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("flip to a stranger: %v", err)
	}
	before := c.Holder.Stamp
	c.Gen = 9 // what Update will have set
	if err := Flip(op, "uk-box", before, "")(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Kind != KindLanding || c.Op.Phase != "checking" || c.Op.Exec != "uk-box" || c.Op.Journal[JournalSourceStamp] != "3" || c.Op.Journal[JournalSrcGen] != "" || c.Holder.Stamp != 10 {
		t.Fatalf("after flip: %+v stamp %d", c.Op, c.Holder.Stamp)
	}
	// holder.generation is untouched by the flip (section 7.5).
	if c.Holder.Generation != "" {
		t.Fatal("flip moved the generation")
	}
}

// Flip is written by either participant (G12): the target may write it
// after a source crash. A second write of the same flip is a no-op that
// does not restamp the holder; a different flip is refused.
func TestFlipIsIdempotentForEitherWriter(t *testing.T) {
	c := movingToSealed(t)
	c.Gen = 9
	// The first flip must name the source's own stamp and generation: a
	// wrong identity would be journalled as the rollback's return.
	sealed := mustJSON(t, c)
	for name, fn := range map[string]func(*Claim) error{
		"first flip, wrong stamp":      Flip(moveA, "uk-box", 4, "gen-7"),
		"first flip, zero stamp":       Flip(moveA, "uk-box", 0, "gen-7"),
		"first flip, wrong generation": Flip(moveA, "uk-box", 3, "gen-8"),
		"first flip, empty generation": Flip(moveA, "uk-box", 3, ""),
	} {
		if err := fn(c); !errors.Is(err, ErrWrongState) {
			t.Fatalf("%s: %v", name, err)
		}
		if got := mustJSON(t, c); string(got) != string(sealed) {
			t.Fatalf("%s mutated the claim:\nbefore %s\nafter  %s", name, sealed, got)
		}
	}
	if err := Flip(moveA, "uk-box", 3, "gen-7")(c); err != nil {
		t.Fatal(err)
	}
	c.Gen = 10 // the replay observes the claim the flip produced
	op := c.Op
	if err := Flip(moveA, "uk-box", 3, "gen-7")(c); err != nil {
		t.Fatalf("replay: %v", err)
	}
	if c.Holder.Stamp != 10 || !reflect.DeepEqual(c.Op, op) {
		t.Fatalf("replay moved the claim: %+v stamp %d", c.Op, c.Holder.Stamp)
	}
	// The replay is checked against the journal, not the holder: the holder
	// stamp is now F (10), not S (3).
	flipped := mustJSON(t, c)
	for name, fn := range map[string]func(*Claim) error{
		"another stamp":        Flip(moveA, "uk-box", 4, "gen-7"),
		"the new holder stamp": Flip(moveA, "uk-box", 10, "gen-7"),
		"another generation":   Flip(moveA, "uk-box", 3, "gen-8"),
		"an empty generation":  Flip(moveA, "uk-box", 3, ""),
		"another target":       Flip(moveA, "us-box", 3, "gen-7"),
		"another op":           Flip(moveB, "uk-box", 3, "gen-7"),
	} {
		if err := fn(c); err == nil {
			t.Fatalf("%s: a different flip was accepted", name)
		}
		if got := mustJSON(t, c); string(got) != string(flipped) {
			t.Fatalf("%s mutated the claim", name)
		}
	}
	// Once the landing has moved past checking, the replay is stale.
	c.Op.Phase = "committing"
	if err := Flip(moveA, "uk-box", 3, "gen-7")(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("replay after the landing moved on: %v", err)
	}
}

// The flip's holder.stamp is the gen the CAS produces: run it through the
// real Update and compare.
func TestFlipStampIsTheGenUpdateProduces(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	c := movingToSealed(t)
	key, err := Key(c.Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(ctx, key, mustJSON(t, c), ""); err != nil {
		t.Fatal(err)
	}
	o, err := Read(ctx, s, c.Host)
	if err != nil {
		t.Fatal(err)
	}
	o, err = Update(ctx, s, o, Flip(moveA, "uk-box", c.Holder.Stamp, "gen-7"))
	if err != nil {
		t.Fatal(err)
	}
	if o.Claim.Holder.Stamp != o.Claim.Gen || o.Claim.Gen != c.Gen+1 {
		t.Fatalf("stamp %d gen %d, want both %d", o.Claim.Holder.Stamp, o.Claim.Gen, c.Gen+1)
	}
	if o.Claim.Op.Journal[JournalSourceStamp] != "3" || o.Claim.Op.Journal[JournalSrcGen] != "gen-7" {
		t.Fatalf("journal: %v", o.Claim.Op.Journal)
	}
}

func TestRevertNeedsEvidenceAndReturnsExecToSource(t *testing.T) {
	c := movingTestClaim(t)
	op := moveA
	if err := BeginMove(op, "ops", "ie-box", "uk-box", "uk", time.Now())(c); err != nil {
		t.Fatal(err)
	}
	scope := ParkScope{Location: "moves/" + c.Host + "/" + op, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Now(), RequestLimit: 120 * time.Second}
	ids := 0
	newID := func() string { ids++; return fmt.Sprintf("id%d", ids) }
	if err := Revert(op, "ie-box", "uk", &scope, newID)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("revert while exec held: %v", err)
	}
	if err := ExecToNone(op, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil {
		t.Fatal(err)
	}
	if err := Revert(op, "uk-box", "uk", &scope, newID)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("revert to the target: %v", err)
	}
	if err := Revert(op, "ie-box", "se", &scope, newID)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("revert naming a region the move never targeted: %v", err)
	}
	if err := Revert(op, "ie-box", "uk", nil, newID)(c); !errors.Is(err, ErrMalformed) {
		t.Fatalf("revert with no staging scope: %v", err)
	}
	live := scope
	live.Location = c.Holder.Location
	if err := Revert(op, "ie-box", "uk", &live, newID)(c); !errors.Is(err, ErrMalformed) {
		t.Fatalf("revert whose park would reach the live location: %v", err)
	}
	// The scope is this move's own lane and this holder's incarnation: not
	// the host-wide staging prefix, not another op's lane, not another
	// installation's incarnation.
	hostWide := scope
	hostWide.Location = "moves/" + c.Host
	otherOp := scope
	otherOp.Location = "moves/" + c.Host + "/" + moveB
	otherInc := scope
	otherInc.Incarnation = strings.Repeat("ab", 32)
	preRevert := mustJSON(t, c)
	for name, sc := range map[string]ParkScope{"host-wide": hostWide, "another op's lane": otherOp, "another incarnation": otherInc} {
		sc := sc
		if err := Revert(op, "ie-box", "uk", &sc, newID)(c); !errors.Is(err, ErrMalformed) {
			t.Fatalf("revert with a %s scope: %v", name, err)
		}
		if got := mustJSON(t, c); string(got) != string(preRevert) {
			t.Fatalf("a refused revert (%s scope) mutated the claim", name)
		}
	}
	if ids != 0 {
		t.Fatalf("a refused revert allocated %d ids", ids)
	}
	if c.Op.Kind != KindMoving || len(c.Obligations) != 0 {
		t.Fatalf("a refused revert left changes: %+v", c.Op)
	}
	if err := Revert(op, "ie-box", "uk", &scope, newID)(c); err != nil || c.Op.Kind != KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[JournalRevPhase] != "evidence" {
		t.Fatalf("revert: %v %+v", err, c.Op)
	}
	if len(c.Obligations) != 1 || c.Obligations[0].Kind != "park-replica" || c.Obligations[0].Region != "uk" || c.Obligations[0].Park == nil ||
		c.Obligations[0].Owner != "region:uk" || c.Obligations[0].Op != op || c.Obligations[0].Park.Location != scope.Location {
		t.Fatalf("revert queued no target staging park: %+v", c.Obligations)
	}
	c.Gen = 9
	if err := c.Validate(); err != nil {
		t.Fatalf("claim after revert must validate: %v", err)
	}
}

// A revert keeps rev_phase if a replay already set it, and queues the
// staging park once even when one is already owed under the op.
func TestRevertKeepsRevPhaseAndQueuesTheParkOnce(t *testing.T) {
	c := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", time.Now())(c); err != nil {
		t.Fatal(err)
	}
	if err := ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil {
		t.Fatal(err)
	}
	scope := ParkScope{Location: "moves/" + c.Host + "/" + moveA, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Now(), RequestLimit: time.Minute}
	c.Obligations = []Obligation{{ID: "o1", Op: moveA, Kind: "park-replica", Region: "uk", Retirement: "r1", Owner: "region:uk", Park: &scope}}
	if err := SetJournal(moveA, map[string]string{JournalRevPhase: "staging"})(c); err != nil {
		t.Fatal(err)
	}
	if err := Revert(moveA, "ie-box", "uk", &scope, counter())(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Journal[JournalRevPhase] != "staging" || len(c.Obligations) != 1 {
		t.Fatalf("rev_phase %q, obligations %+v", c.Op.Journal[JournalRevPhase], c.Obligations)
	}
}

// The cut's transitions own only the op, the journal, the revert's staging
// park and (the flip) the holder's stamp; every other field is carried
// through unchanged.
func TestCutTransitionsPreserveTheRestOfTheClaim(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	fixture := func() *Claim {
		c := movingTestClaim(t)
		c.Holder.Generation = "gen-7"
		c.Holder.Session = 9
		c.Holder.WriterBox = "ie-box"
		c.Holder.Admissions = []AdmissionRecord{{Region: "ie", Session: 9, Box: "ie-box", Generation: "gen-7", Mode: "admitted", Coverage: "covered", Covered: true,
			WritesLost:           &WritesLost{Box: "ie-box", Generation: "gen-6", Session: 8, RemoteMaxTXID: 41},
			AdditionalWritesLost: []WritesLost{{Box: "ie-box", Generation: "gen-5", Session: 7, RemoteMaxTXID: 12}}}}
		c.Obligations = []Obligation{{ID: "o1", Op: "old", Kind: "park-replica", Region: "se", Retirement: "r", Owner: "region:se"}}
		c.Lost = []string{"se"}
		c.RetiredIncarnations = []string{"abc"}
		c.QuietAfter = map[string]time.Time{"se": now}
		c.Inventory = map[string]InventoryEntry{"ie-box": {Layout: "nested", Paths: map[string]uint64{"/data/x": 3}}}
		if err := c.Validate(); err != nil {
			t.Fatalf("fixture: %v", err)
		}
		return c
	}
	scope := ParkScope{Location: "moves/a.go.example.com/" + moveA, Incarnation: fixture().Holder.Incarnation, EvidenceAt: now, RequestLimit: time.Minute}
	begin := func() []func(*Claim) error {
		return []func(*Claim) error{
			BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now),
			ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource),
		}
	}
	cases := map[string]struct {
		steps   []func(*Claim) error
		restamp bool // the flip restamps the holder
		parks   int  // staging parks the transition queues
	}{
		"flip": {restamp: true, steps: append(begin(),
			SetJournal(moveA, map[string]string{JournalSeal: `{"n":7}`}),
			SetMovePhase(moveA, KindMoving, PhaseMoving, PhaseSealed),
			Flip(moveA, "uk-box", 3, "gen-7"))},
		"revert": {parks: 1, steps: append(begin(), Revert(moveA, "ie-box", "uk", &scope, counter()))},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			c := fixture()
			before := mustJSON(t, c)
			for i, tr := range tc.steps {
				if err := tr(c); err != nil {
					t.Fatalf("step %d: %v", i, err)
				}
			}
			if len(c.Obligations) != 1+tc.parks || !reflect.DeepEqual(c.Obligations[0], fixture().Obligations[0]) {
				t.Fatalf("obligations of other ops changed: %+v", c.Obligations)
			}
			var round Claim
			if err := json.Unmarshal(before, &round); err != nil {
				t.Fatal(err)
			}
			round.Op, round.Obligations = c.Op, c.Obligations
			if tc.restamp {
				round.Holder.Stamp = c.Holder.Stamp
			}
			if !reflect.DeepEqual(round, *c) {
				t.Fatalf("%s changed more than it owns:\nbefore %s\nafter  %s", name, before, mustJSON(t, c))
			}
			c.Gen++ // what Update would do
			if err := c.Validate(); err != nil {
				t.Fatalf("after %s: %v", name, err)
			}
		})
	}
}

// A revert replayed after a crash finds the claim already reverting with
// the same op, region, scope and rev_phase: nil, no new ids, one park.
func TestRevertReplayIsIdempotent(t *testing.T) {
	c := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", time.Now())(c); err != nil {
		t.Fatal(err)
	}
	if err := ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil {
		t.Fatal(err)
	}
	scope := ParkScope{Location: "moves/" + c.Host + "/" + moveA, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Now().UTC(), RequestLimit: time.Minute}
	ids := 0
	newID := func() string { ids++; return fmt.Sprintf("id%d", ids) }
	if err := Revert(moveA, "ie-box", "uk", &scope, newID)(c); err != nil {
		t.Fatal(err)
	}
	after, usedIDs := mustJSON(t, c), ids
	if err := Revert(moveA, "ie-box", "uk", &scope, newID)(c); err != nil {
		t.Fatalf("replay: %v", err)
	}
	if got := mustJSON(t, c); string(got) != string(after) || ids != usedIDs || len(c.Obligations) != 1 {
		t.Fatalf("replay changed the claim or allocated ids (%d, was %d): %s", ids, usedIDs, mustJSON(t, c))
	}
	// A replay that is not the same revert is refused, and changes nothing.
	later := scope
	later.EvidenceAt = scope.EvidenceAt.Add(time.Second)
	for name, fn := range map[string]func(*Claim) error{
		"another scope":  Revert(moveA, "ie-box", "uk", &later, newID),
		"another region": Revert(moveA, "ie-box", "se", &scope, newID),
		"another source": Revert(moveA, "uk-box", "uk", &scope, newID),
		"another op":     Revert(moveB, "ie-box", "uk", &scope, newID),
	} {
		if err := fn(c); err == nil {
			t.Fatalf("%s: accepted as a replay", name)
		}
		if got := mustJSON(t, c); string(got) != string(after) || ids != usedIDs {
			t.Fatalf("%s changed the claim or allocated ids", name)
		}
	}
	// Once the revert has moved past evidence, the replay is stale.
	c.Op.Journal[JournalRevPhase] = "opened"
	if err := Revert(moveA, "ie-box", "uk", &scope, newID)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("replay after the revert moved on: %v", err)
	}
}

// parkFixture is a valid landing-shaped claim for the park queue helpers:
// op moveA, a holder with an incarnation, and the scope helpers an op needs.
func parkFixture(t *testing.T) (*Claim, func(location string) ParkScope) {
	t.Helper()
	c := movingTestClaim(t)
	if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC))(c); err != nil {
		t.Fatal(err)
	}
	scope := func(location string) ParkScope {
		return ParkScope{Location: location, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Date(2026, 10, 6, 9, 5, 0, 0, time.UTC), RequestLimit: 2 * time.Minute}
	}
	return c, scope
}

// inState puts a claim in the op state a park helper's lifecycle check reads:
// the kind and phase, as the transitions that own them would leave it.
func inState(c *Claim, kind, phase string) {
	c.Op.Kind, c.Op.Phase = kind, phase
}

// The move's retirements ride the engine's existing kinds. A real store
// round trip: Validate accepts what the helper wrote and it reads back.
func TestLocationParkIsAParkReplicaTheStoreAccepts(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	o := servingClaim(t, ctx, s)
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	o, err := Update(ctx, s, o, BeginMove(moveA, "ops", "adi-knitting", "uk-box", "uk", now))
	if err != nil {
		t.Fatal(err)
	}
	inc := o.Claim.Holder.Incarnation
	scope := ParkScope{Location: "instances/" + testHost + "/" + inc, Incarnation: inc, EvidenceAt: now, RequestLimit: 2 * time.Minute}
	// The commit's claim state: the source's location is parked from here.
	if o, err = Update(ctx, s, o, func(c *Claim) error { inState(c, KindLanding, "committed"); return nil }); err != nil {
		t.Fatal(err)
	}
	o, err = Update(ctx, s, o, QueueLocationPark(moveA, "ie", "adi-knitting", scope, counter()))
	if err != nil {
		t.Fatal(err)
	}
	o, err = Read(ctx, s, testHost)
	if err != nil {
		t.Fatal(err)
	}
	if err := o.Claim.Validate(); err != nil {
		t.Fatal(err)
	}
	if len(o.Claim.Obligations) != 2 {
		t.Fatalf("obligations: %+v", o.Claim.Obligations)
	}
	r, l := o.Claim.Obligations[0], o.Claim.Obligations[1]
	if r.Kind != "park-replica" || r.Op != moveA || r.Region != "ie" || r.Owner != "region:ie" || r.Park == nil || !sameParkScope(*r.Park, scope) || r.Retirement == "" || r.ID == "" || r.ID == r.Retirement {
		t.Fatalf("replica park: %+v", r)
	}
	if l.Kind != "park-local" || l.Op != moveA || l.Box != "adi-knitting" || l.Owner != "box:adi-knitting" || l.Park != nil || l.Region != "" || l.Retirement == "" {
		t.Fatalf("local park: %+v", l)
	}
	// The same call again, across the store: nothing new, no ids spent.
	ids := 0
	again := func() string { ids++; return fmt.Sprintf("late_%d", ids) }
	if o, err = Update(ctx, s, o, QueueLocationPark(moveA, "ie", "adi-knitting", scope, again)); err != nil || ids != 0 || len(o.Claim.Obligations) != 2 {
		t.Fatalf("replay: ids=%d obligations=%d err=%v", ids, len(o.Claim.Obligations), err)
	}
	// Completing both by the existing chore completion ends them.
	for _, ob := range o.Claim.Obligations {
		if o, err = Update(ctx, s, o, CompleteObligation(ob.Op, ob.ID)); err != nil {
			t.Fatal(err)
		}
	}
	if len(o.Claim.Obligations) != 0 {
		t.Fatal("chores left")
	}
}

func TestStagingParkScopeIsTheLane(t *testing.T) {
	c, scope := parkFixture(t)
	lane := scope("moves/" + c.Host + "/" + moveA)
	if err := QueueStagingPark(moveA, "uk", lane, counter())(c); err != nil {
		t.Fatal(err)
	}
	if len(c.Obligations) != 1 {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	o := c.Obligations[0]
	if o.Kind != "park-replica" || o.Region != "uk" || o.Owner != "region:uk" || o.Op != moveA || o.Park.Location != "moves/"+c.Host+"/"+moveA {
		t.Fatalf("staging park: %+v", o)
	}
	c.Gen++
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	// Anything but this move's own lane is refused, and nothing is queued.
	for name, location := range map[string]string{
		"the live location":  "instances/" + c.Host + "/" + c.Holder.Incarnation,
		"the legacy prefix":  "instances/" + c.Host,
		"the staging prefix": "moves/" + c.Host,
		"another move":       "moves/" + c.Host + "/" + moveB,
		"another host":       "moves/other.example.com/" + moveA,
		"below the lane":     "moves/" + c.Host + "/" + moveA + "/x",
	} {
		d, _ := parkFixture(t)
		bad := scope(location)
		ids := 0
		if err := QueueStagingPark(moveA, "uk", bad, func() string { ids++; return "x" })(d); err == nil || len(d.Obligations) != 0 || ids != 0 {
			t.Errorf("%s: accepted (%v), obligations %d", name, err, len(d.Obligations))
		}
	}
	// Another holder's incarnation is refused.
	d, _ := parkFixture(t)
	foreign := scope("moves/" + d.Host + "/" + moveA)
	foreign.Incarnation = strings.Repeat("a", 64)
	if err := QueueStagingPark(moveA, "uk", foreign, counter())(d); !errors.Is(err, ErrMalformed) {
		t.Errorf("foreign incarnation: %v", err)
	}
	// A stale op is refused.
	if err := QueueStagingPark(moveB, "uk", lane, counter())(d); !errors.Is(err, ErrStaleOp) {
		t.Errorf("stale op: %v", err)
	}
}

// A revert already queued the lane's park: queueing it again is a no-op,
// and a different scope for the same lane is refused, not overwritten.
func TestStagingParkAfterRevertIsTheSameChore(t *testing.T) {
	c, scope := parkFixture(t)
	if err := ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil {
		t.Fatal(err)
	}
	lane := scope("moves/" + c.Host + "/" + moveA)
	if err := Revert(moveA, "ie-box", "uk", &lane, counter())(c); err != nil {
		t.Fatal(err)
	}
	ids := 0
	newID := func() string { ids++; return "z" }
	if err := QueueStagingPark(moveA, "uk", lane, newID)(c); err != nil || ids != 0 || len(c.Obligations) != 1 {
		t.Fatalf("same lane: %v ids=%d obligations=%d", err, ids, len(c.Obligations))
	}
	later := lane
	later.EvidenceAt = lane.EvidenceAt.Add(time.Second)
	if err := QueueStagingPark(moveA, "uk", later, newID)(c); !errors.Is(err, ErrWrongState) || ids != 0 || len(c.Obligations) != 1 {
		t.Fatalf("changed scope: %v ids=%d", err, ids)
	}
}

func TestLocationParkAndStagingParkCoexistInOneRegion(t *testing.T) {
	// A rolled-back target at cleaning owes the lane and the location in the
	// same region under one op.
	c, scope := parkFixture(t)
	inState(c, KindRestoring, "released")
	lane := scope("moves/" + c.Host + "/" + moveA)
	loc := scope(c.Holder.Location)
	n := counter()
	for i := 0; i < 2; i++ {
		if err := QueueStagingPark(moveA, "uk", lane, n)(c); err != nil {
			t.Fatal(err)
		}
		if err := QueueLocationPark(moveA, "uk", "", loc, n)(c); err != nil {
			t.Fatal(err)
		}
	}
	if len(c.Obligations) != 2 {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	c.Gen++
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	// The location's local chore is per box, not per region.
	if err := QueueLocationPark(moveA, "uk", "uk-box", loc, n)(c); err != nil || len(c.Obligations) != 3 {
		t.Fatalf("add the box: %v %+v", err, c.Obligations)
	}
	if err := QueueLocationPark(moveA, "uk", "uk-box", loc, n)(c); err != nil || len(c.Obligations) != 3 {
		t.Fatalf("box twice: %v %+v", err, c.Obligations)
	}
}

// A whole location is retired only where the region has stopped being
// authoritative: the source after the commit (landing/committed), and a
// rolled-back target (restoring, from its release). Every other state, and
// the other region in either state, is refused with nothing queued and no
// ids spent.
func TestLocationParkOnlyWhereTheRegionIsRetired(t *testing.T) {
	type tc struct {
		kind, phase, region, box string
		allowed                  bool
	}
	cases := map[string]tc{
		"source after the commit":                {KindLanding, "committed", "ie", "ie-box", true},
		"source after the commit, region only":   {KindLanding, "committed", "ie", "", true},
		"rolled-back target at cleaning":         {KindRestoring, "released", "uk", "uk-box", true},
		"rolled-back target after the source":    {KindRestoring, "claimed", "uk", "uk-box", true},
		"source before the cut":                  {KindMoving, PhaseMoving, "ie", "ie-box", false},
		"source sealed":                          {KindMoving, PhaseSealed, "ie", "ie-box", false},
		"source in the abort":                    {KindMoving, PhaseAborting, "ie", "ie-box", false},
		"source while the landing checks":        {KindLanding, "checking", "ie", "ie-box", false},
		"source while the landing probes":        {KindLanding, "probing", "ie", "ie-box", false},
		"source while reverting":                 {KindReverting, "evidence", "ie", "ie-box", false},
		"source after the rollback began":        {KindRestoring, "released", "ie", "ie-box", false},
		"target after the commit":                {KindLanding, "committed", "uk", "uk-box", false},
		"target before the commit":               {KindLanding, "probing", "uk", "uk-box", false},
		"target before its release":              {KindRestoring, "releasing", "uk", "uk-box", false},
		"target still serving":                   {KindServing, "", "uk", "uk-box", false},
		"the target's box from the source":       {KindLanding, "committed", "ie", "uk-box", false},
		"the source's box from a rolled-back uk": {KindRestoring, "released", "uk", "ie-box", false},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			c, scope := parkFixture(t)
			inState(c, tc.kind, tc.phase)
			ids := 0
			newID := func() string { ids++; return fmt.Sprintf("p%d", ids) }
			before := mustJSON(t, c)
			err := QueueLocationPark(moveA, tc.region, tc.box, scope(c.Holder.Location), newID)(c)
			if tc.allowed {
				if err != nil || len(c.Obligations) == 0 {
					t.Fatalf("refused: %v", err)
				}
				c.Gen++
				if err := c.Validate(); err != nil {
					t.Fatal(err)
				}
				return
			}
			if !errors.Is(err, ErrWrongState) || ids != 0 || string(mustJSON(t, c)) != string(before) {
				t.Fatalf("err=%v ids=%d, claim changed=%v", err, ids, string(mustJSON(t, c)) != string(before))
			}
		})
	}
}

// Location and residue parks name holder.location, whichever form the
// holder uses (spec §6.4): the legacy instances/<host> or the
// incarnation-scoped instances/<host>/<inc>, never the other one. A trailing
// slash is normalised away on both sides.
func TestParkLocationMustBeTheHoldersLocation(t *testing.T) {
	for _, form := range []string{"scoped", "legacy"} {
		t.Run(form, func(t *testing.T) {
			c, scope := parkFixture(t)
			scoped := "instances/" + c.Host + "/" + c.Holder.Incarnation
			legacy := "instances/" + c.Host
			own, other := scoped, legacy
			if form == "legacy" {
				own, other = legacy, scoped
			}
			c.Holder.Location = own
			inState(c, KindLanding, "committed")
			if err := c.Validate(); err != nil {
				t.Fatal(err)
			}
			ids := 0
			newID := func() string { ids++; return fmt.Sprintf("p%d", ids) }
			before := mustJSON(t, c)
			for name, fn := range map[string]func(*Claim) error{
				"location":    QueueLocationPark(moveA, "ie", "ie-box", scope(other), newID),
				"residue":     QueueResidueParks(moveA, "ie", scope(other), []string{"gen-1"}, newID),
				"slash/other": QueueLocationPark(moveA, "ie", "", scope(other+"/"), newID),
			} {
				if err := fn(c); !errors.Is(err, ErrMalformed) && !errors.Is(err, ErrWrongState) || ids != 0 || string(mustJSON(t, c)) != string(before) {
					t.Errorf("%s: err=%v ids=%d", name, err, ids)
				}
			}
			// The holder's own form is accepted, with or without a trailing slash,
			// and what is stored has none.
			if err := QueueLocationPark(moveA, "ie", "", scope(own+"/"), newID)(c); err != nil {
				t.Fatalf("own location with a slash: %v", err)
			}
			if err := QueueResidueParks(moveA, "ie", scope(own), []string{"gen-1"}, newID)(c); err != nil {
				t.Fatalf("own location residue: %v", err)
			}
			if len(c.Obligations) != 2 || c.Obligations[0].Park.Location != own || c.Obligations[1].Park.Location != own {
				t.Fatalf("stored: %+v", c.Obligations)
			}
		})
	}
}

func TestLocationParkRefusals(t *testing.T) {
	c, scope := parkFixture(t)
	inState(c, KindLanding, "committed")
	loc := scope(c.Holder.Location)
	lane := scope("moves/" + c.Host + "/" + moveA)
	bad := loc
	bad.RequestLimit = 0
	changed := loc
	changed.EvidenceAt = loc.EvidenceAt.Add(time.Hour)
	if err := QueueLocationPark(moveA, "ie", "", loc, counter())(c); err != nil {
		t.Fatal(err)
	}
	for name, fn := range map[string]func(*Claim) error{
		"a lane through the location door": QueueLocationPark(moveA, "ie", "", lane, counter()),
		"no region":                        QueueLocationPark(moveA, "", "", loc, counter()),
		"an invalid scope":                 QueueLocationPark(moveA, "ie", "", bad, counter()),
		"another scope, same chore":        QueueLocationPark(moveA, "ie", "", changed, counter()),
		"another op":                       QueueLocationPark(moveB, "ie", "", loc, counter()),
		"no id source":                     QueueLocationPark(moveA, "ie", "", loc, nil),
	} {
		before := mustJSON(t, c)
		if err := fn(c); err == nil {
			t.Errorf("%s: accepted", name)
		}
		if string(mustJSON(t, c)) != string(before) {
			t.Errorf("%s: changed the claim", name)
		}
	}
}

func TestResidueParksArePerGeneration(t *testing.T) {
	c, scope := parkFixture(t)
	loc := scope("instances/" + c.Host + "/" + c.Holder.Incarnation)
	n := counter()
	gens := []string{"gen-4", RootGeneration, "gen-9"}
	if err := QueueResidueParks(moveA, "uk", loc, gens, n)(c); err != nil {
		t.Fatal(err)
	}
	if len(c.Obligations) != 3 {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	seen := map[string]bool{}
	for _, o := range c.Obligations {
		if o.Kind != "park-generation" || o.Op != moveA || o.Region != "uk" || o.Owner != "region:uk" || o.Park == nil || !sameParkScope(*o.Park, loc) || o.Retirement == "" || o.Box != "" {
			t.Fatalf("residue park: %+v", o)
		}
		seen[o.Generation] = true
	}
	if !seen["gen-4"] || !seen["gen-9"] || !seen["@root"] {
		t.Fatalf("generations: %v", seen)
	}
	c.Gen++
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	// Each chore owns its scope: editing one does not touch another.
	c.Obligations[0].Park.RequestLimit = time.Hour
	if c.Obligations[1].Park.RequestLimit == time.Hour {
		t.Fatal("chores share one scope")
	}
	// A double call, a repeated generation in the list and a superset add
	// only what is missing, and spend ids only for that.
	ids := 0
	next := func() string { ids++; return fmt.Sprintf("n%d", ids) }
	loc.RequestLimit = 2 * time.Minute
	c.Obligations[0].Park.RequestLimit = 2 * time.Minute
	if err := QueueResidueParks(moveA, "uk", loc, append(gens, "gen-4", "gen-11", "gen-11"), next)(c); err != nil {
		t.Fatal(err)
	}
	if len(c.Obligations) != 4 || ids != 2 || c.Obligations[3].Generation != "gen-11" {
		t.Fatalf("superset: %d obligations, %d ids", len(c.Obligations), ids)
	}
	// No generations is nothing to do; an empty name is malformed.
	if err := QueueResidueParks(moveA, "uk", loc, nil, next)(c); err != nil || len(c.Obligations) != 4 {
		t.Fatalf("none: %v", err)
	}
	if err := QueueResidueParks(moveA, "uk", loc, []string{"gen-12", ""}, next)(c); !errors.Is(err, ErrMalformed) {
		t.Fatalf("empty generation: %v", err)
	}
	// The staging lane is not a residue location.
	if err := QueueResidueParks(moveA, "uk", scope("moves/"+c.Host+"/"+moveA), []string{"gen-1"}, next)(c); err == nil {
		t.Fatal("residue parked a lane")
	}
}

func TestLocalParkAndTheEarlierStampsPaths(t *testing.T) {
	c, _ := parkFixture(t)
	c.Inventory = map[string]InventoryEntry{"uk-box": {Layout: "nested", Paths: map[string]uint64{"/data/a": 4, "/data/b": 7}}}
	if !InventoryHasStamp(c, "uk-box", 4) || InventoryHasStamp(c, "uk-box", 5) || InventoryHasStamp(c, "ie-box", 4) || InventoryHasStamp(c, "uk-box", 0) {
		t.Fatal("InventoryHasStamp")
	}
	ids := 0
	newID := func() string { ids++; return fmt.Sprintf("l%d", ids) }
	if err := QueueLocalPark(moveA, "uk-box", newID)(c); err != nil {
		t.Fatal(err)
	}
	if err := QueueLocalPark(moveA, "uk-box", newID)(c); err != nil || ids != 2 || len(c.Obligations) != 1 {
		t.Fatalf("double call: %v ids=%d obligations=%d", err, ids, len(c.Obligations))
	}
	o := c.Obligations[0]
	if o.Kind != "park-local" || o.Box != "uk-box" || o.Owner != "box:uk-box" || o.Op != moveA || o.Region != "" || o.Park != nil {
		t.Fatalf("local park: %+v", o)
	}
	c.Gen++
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	// The replay with S' paths still there must not complete the chore.
	if err := CompleteLocalPark(moveA, "uk-box", 4)(c); !errors.Is(err, ErrWrongState) || len(c.Obligations) != 1 {
		t.Fatalf("complete with paths left: %v", err)
	}
	// With no S' paths the chore is completed under the op; a second time is a no-op.
	c.Inventory["uk-box"] = InventoryEntry{Layout: "nested", Paths: map[string]uint64{"/data/b": 7}}
	if err := CompleteLocalPark(moveA, "uk-box", 4)(c); err != nil || len(c.Obligations) != 0 {
		t.Fatalf("complete: %v %+v", err, c.Obligations)
	}
	if err := CompleteLocalPark(moveA, "uk-box", 4)(c); err != nil {
		t.Fatalf("replay: %v", err)
	}
	// It is the op's own chore only.
	if err := QueueLocalPark(moveA, "uk-box", newID)(c); err != nil {
		t.Fatal(err)
	}
	if err := CompleteLocalPark(moveB, "uk-box", 4)(c); !errors.Is(err, ErrStaleOp) || len(c.Obligations) != 1 {
		t.Fatalf("another op: %v", err)
	}
	if err := QueueLocalPark(moveA, "", newID)(c); err == nil {
		t.Fatal("a local park with no box")
	}
}

// The park queue helpers own only the obligations they append. Every
// other claim field, and every other op's chore, is carried through.
func TestParkQueueHelpersPreserveTheRestOfTheClaim(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	build := func() *Claim {
		c := movingTestClaim(t)
		c.Holder.Generation = "gen-7"
		c.Holder.Session = 9
		c.Holder.WriterBox = "ie-box"
		c.Holder.Admissions = []AdmissionRecord{{Region: "ie", Session: 9, Box: "ie-box", Generation: "gen-7", Mode: "admitted", Coverage: "covered", Covered: true,
			WritesLost:           &WritesLost{Box: "ie-box", Generation: "gen-6", Session: 8, RemoteMaxTXID: 41},
			AdditionalWritesLost: []WritesLost{{Box: "ie-box", Generation: "gen-5", Session: 7, RemoteMaxTXID: 12}}}}
		c.Obligations = []Obligation{{ID: "o1", Op: "old", Kind: "park-replica", Region: "se", Retirement: "r", Owner: "region:se"}}
		c.Lost = []string{"se"}
		c.RetiredIncarnations = []string{"abc"}
		c.QuietAfter = map[string]time.Time{"se": now}
		c.Touched = Touched{Complete: true, Regions: []string{"ie", "uk"}, Boxes: []string{"ie-box", "uk-box"}}
		c.Inventory = map[string]InventoryEntry{"ie-box": {Layout: "nested", Paths: map[string]uint64{"/data/x": 3}}}
		if err := BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", now)(c); err != nil {
			t.Fatal(err)
		}
		inState(c, KindLanding, "committed")
		if err := c.Validate(); err != nil {
			t.Fatalf("fixture: %v", err)
		}
		return c
	}
	base := build()
	scope := func(loc string) ParkScope {
		return ParkScope{Location: loc, Incarnation: base.Holder.Incarnation, EvidenceAt: now, RequestLimit: time.Minute}
	}
	loc := scope(base.Holder.Location)
	lane := scope("moves/" + base.Host + "/" + moveA)
	cases := map[string]struct {
		fn   func(*Claim) error
		want int
	}{
		"location+box": {QueueLocationPark(moveA, "ie", "ie-box", loc, counter()), 2},
		"staging":      {QueueStagingPark(moveA, "uk", lane, counter()), 1},
		"local":        {QueueLocalPark(moveA, "uk-box", counter()), 1},
		"residue":      {QueueResidueParks(moveA, "uk", loc, []string{"gen-1", RootGeneration}, counter()), 2},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			c := build()
			before := mustJSON(t, c)
			if err := tc.fn(c); err != nil {
				t.Fatal(err)
			}
			if len(c.Obligations) != 1+tc.want || !reflect.DeepEqual(c.Obligations[0], build().Obligations[0]) {
				t.Fatalf("obligations: %+v", c.Obligations)
			}
			var round Claim
			if err := json.Unmarshal(before, &round); err != nil {
				t.Fatal(err)
			}
			round.Obligations = c.Obligations
			if !reflect.DeepEqual(round, *c) {
				t.Fatalf("changed more than the obligations:\nbefore %s\nafter  %s", before, mustJSON(t, c))
			}
			c.Gen++
			if err := c.Validate(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

// probedLanding drives the claim to landing/probed the way the landing does:
// the flip, then every record phase mirrored through SetMovePhase. The
// target's probe admission is on the holder, its predecessor in region ie.
func probedLanding(t *testing.T) *Claim {
	t.Helper()
	c := movingToSealed(t)
	c.Gen = 7
	if err := Flip(moveA, "uk-box", 3, "gen-7")(c); err != nil {
		t.Fatal(err)
	}
	for _, step := range [][2]string{{"checking", "parking"}, {"parking", "building"}, {"building", "published"}, {"published", "local"}, {"local", "configured"}, {"configured", "restored"}, {"restored", "probing"}, {"probing", "probed"}} {
		if err := SetMovePhase(moveA, KindLanding, step[0], step[1])(c); err != nil {
			t.Fatal(err)
		}
	}
	c.Holder.Session = 9
	c.Holder.Admissions = []AdmissionRecord{{Region: "uk", PredecessorRegion: "ie", Session: 9, Box: "uk-box", Generation: moveA, Predecessor: "gen-7", Mode: "cold", Coverage: "sealed"}}
	return c
}

func commitScopes(c *Claim) (src, stg ParkScope) {
	at := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
	src = ParkScope{Location: c.Holder.Location, Incarnation: c.Holder.Incarnation, EvidenceAt: at, RequestLimit: 120 * time.Second}
	stg = ParkScope{Location: "moves/" + c.Host + "/" + moveA, Incarnation: c.Holder.Incarnation, EvidenceAt: at, RequestLimit: 120 * time.Second}
	return src, stg
}

func obligationKinds(c *Claim) map[string]int {
	kinds := map[string]int{}
	for _, o := range c.Obligations {
		kinds[o.Kind+"/"+o.Region+o.Box]++
	}
	return kinds
}

// wCommitCoverage, wCommitReplayed, H3, H5(b).
func TestCommitQueuesObligationsAndCoversTheSourceRegion(t *testing.T) {
	for _, from := range []string{"probed"} {
		c := probedLanding(t)
		c.Op.Phase = from
		// The source box holds paths of the stay the move leaves: stamp S = 3.
		c.Inventory = map[string]InventoryEntry{"ie-box": {Layout: "flat", Paths: map[string]uint64{"/data/a": 3}}}
		stamp, session, gen := c.Holder.Stamp, c.Holder.Session, c.Holder.Generation
		ids := 0
		newID := func() string { ids++; return fmt.Sprintf("id%d", ids) }
		src, stg := commitScopes(c)
		if err := Commit(moveA, "ie", "ie-box", "uk", src, stg, newID)(c); err != nil {
			t.Fatalf("from %s: %v", from, err)
		}
		if c.Op.Kind != KindLanding || c.Op.Phase != PhaseCommitted || c.Op.Exec != "uk-box" {
			t.Fatalf("after commit: %+v", c.Op)
		}
		// H3: the holder's stamp, session and generation are the flip's, untouched.
		if c.Holder.Stamp != stamp || c.Holder.Session != session || c.Holder.Generation != gen {
			t.Fatalf("commit touched the holder: %+v", c.Holder)
		}
		k := obligationKinds(c)
		if k["park-replica/ie"] != 1 || k["park-local/ie-box"] != 1 || k["park-replica/uk"] != 1 || len(c.Obligations) != 3 {
			t.Fatalf("obligations: %+v", c.Obligations)
		}
		for _, o := range c.Obligations {
			if o.Kind == "park-replica" && o.Region == "ie" && (o.Park == nil || o.Park.Location != c.Holder.Location || o.Owner != "region:ie") {
				t.Fatalf("source park: %+v", o)
			}
			if o.Kind == "park-replica" && o.Region == "uk" && (o.Park == nil || o.Park.Location != stg.Location) {
				t.Fatalf("staging park: %+v", o)
			}
		}
		if !c.Holder.Admissions[0].Covered {
			t.Fatal("the source-region predecessor is not covered (wCommitCoverage)")
		}
		if err := c.Validate(); err != nil {
			t.Fatalf("the committed claim must validate: %v", err)
		}
		// Replay: nothing doubles and no ids are minted (wCommitReplayed).
		before, usedIDs := mustJSON(t, c), ids
		if err := Commit(moveA, "ie", "ie-box", "uk", src, stg, newID)(c); err != nil || ids != usedIDs || string(mustJSON(t, c)) != string(before) {
			t.Fatalf("replay: %v ids %d->%d", err, usedIDs, ids)
		}
	}
}

// H5(b): the source's park-local is queued only when the source box lists
// paths of the journalled source stamp S.
func TestCommitQueuesTheSourceParkLocalOnlyForTheSourceStamp(t *testing.T) {
	for name, inv := range map[string]map[string]InventoryEntry{
		"no inventory":         nil,
		"other stamp only":     {"ie-box": {Layout: "flat", Paths: map[string]uint64{"/data/a": 2}}},
		"stamp on another box": {"uk-box": {Layout: "flat", Paths: map[string]uint64{"/data/a": 3}}},
	} {
		c := probedLanding(t)
		c.Inventory = inv
		ids := 0
		newID := func() string { ids++; return fmt.Sprintf("id%d", ids) }
		src, stg := commitScopes(c)
		if err := Commit(moveA, "ie", "ie-box", "uk", src, stg, newID)(c); err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		k := obligationKinds(c)
		if k["park-replica/ie"] != 1 || k["park-replica/uk"] != 1 || len(c.Obligations) != 2 {
			t.Fatalf("%s: obligations %+v", name, c.Obligations)
		}
		if !c.Holder.Admissions[0].Covered {
			t.Fatalf("%s: the source region is not covered", name)
		}
	}
}

// wCommitInterrupted: a commit that is refused leaves the claim as it was.
func TestCommitRefusesAnythingButAProbingTargetLanding(t *testing.T) {
	newID := func() string { return "x" }
	type commitFn = func(c *Claim, src, stg ParkScope) func(*Claim) error
	std := func(op, region, box string) commitFn {
		return func(c *Claim, src, stg ParkScope) func(*Claim) error {
			return Commit(op, region, box, "uk", src, stg, newID)
		}
	}
	for name, tc := range map[string]struct {
		mutate func(c *Claim, src, stg *ParkScope)
		fn     commitFn
		want   error
	}{
		"serving claim":                            {func(c *Claim, _, _ *ParkScope) { c.Op = Op{Kind: KindServing, Exec: "ie-box"} }, std(moveA, "ie", "ie-box"), ErrStaleOp},
		"another op":                               {func(*Claim, *ParkScope, *ParkScope) {}, std(moveB, "ie", "ie-box"), ErrStaleOp},
		"moving":                                   {func(c *Claim, _, _ *ParkScope) { c.Op.Kind, c.Op.Phase, c.Op.Exec = KindMoving, PhaseSealed, "" }, std(moveA, "ie", "ie-box"), ErrWrongState},
		"landing before probing":                   {func(c *Claim, _, _ *ParkScope) { c.Op.Phase = "restored" }, std(moveA, "ie", "ie-box"), ErrWrongState},
		"exec not the target":                      {func(c *Claim, _, _ *ParkScope) { c.Op.Exec = "ie-box" }, std(moveA, "ie", "ie-box"), ErrWrongState},
		"exec none":                                {func(c *Claim, _, _ *ParkScope) { c.Op.Exec = "" }, std(moveA, "ie", "ie-box"), ErrWrongState},
		"source box is not the source":             {func(*Claim, *ParkScope, *ParkScope) {}, std(moveA, "ie", "uk-box"), ErrWrongState},
		"source region is the target":              {func(*Claim, *ParkScope, *ParkScope) {}, std(moveA, "uk", "ie-box"), ErrWrongState},
		"source park is not the holder's location": {func(_ *Claim, s, _ *ParkScope) { s.Location = "instances/a.go.example.com/other" }, std(moveA, "ie", "ie-box"), ErrMalformed},
		"staging park is another lane":             {func(c *Claim, _, g *ParkScope) { g.Location = "moves/" + c.Host + "/" + moveB }, std(moveA, "ie", "ie-box"), ErrMalformed},
	} {
		c := probedLanding(t)
		c.Inventory = map[string]InventoryEntry{"ie-box": {Layout: "flat", Paths: map[string]uint64{"/data/a": 3}}}
		src, stg := commitScopes(c)
		tc.mutate(c, &src, &stg)
		before := mustJSON(t, c)
		if err := tc.fn(c, src, stg)(c); !errors.Is(err, tc.want) {
			t.Fatalf("%s: got %v, want %v", name, err, tc.want)
		}
		if after := mustJSON(t, c); string(after) != string(before) {
			t.Fatalf("%s mutated the claim:\nbefore %s\nafter  %s", name, before, after)
		}
	}
}

// ---- Plan 4 Task 2: the rollback and the return (§7.8, K4, K8) ----

const (
	rbA           = "rb_abcdefghijklmnopqrstuvwxyz"
	rbB           = "rb_bcdefghijklmnopqrstuvwxyza"
	stoppedTarget = "moves/bab/app/h/rb/stopped-target.json"
)

// targetWroteLanding is probedLanding with the target's probe admitted as
// the writer: holder.generation and writer_box are the target's, and the
// last admission record is the target's writer (its predecessor the
// source's generation in ie). The flip at gen 7 made the target's
// installation stamp F = 8.
func targetWroteLanding(t *testing.T) *Claim {
	t.Helper()
	c := probedLanding(t)
	c.Holder.Generation, c.Holder.WriterBox = moveA, "uk-box"
	if c.Holder.Stamp != 8 {
		t.Fatalf("fixture: the flip's stamp is %d, want 8", c.Holder.Stamp)
	}
	return c
}

// rolledBack is targetWroteLanding after the rollback's claim half.
func rolledBack(t *testing.T) *Claim {
	t.Helper()
	c := targetWroteLanding(t)
	if err := RollBack(moveA, rbA)(c); err != nil {
		t.Fatal(err)
	}
	return c
}

func releaseScope(c *Claim) ParkScope {
	return ParkScope{Location: c.Holder.Location, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Date(2026, 10, 6, 11, 0, 0, 0, time.UTC), RequestLimit: 2 * time.Minute}
}

// released is rolledBack after the target's release, with no park-local
// (the target listed no paths of F).
func released(t *testing.T) *Claim {
	t.Helper()
	c := rolledBack(t)
	if err := TargetReleased(moveA, "uk-box", "uk", stoppedTarget, releaseScope(c), counter())(c); err != nil {
		t.Fatal(err)
	}
	return c
}

func validAfterWrite(t *testing.T, c *Claim) {
	t.Helper()
	d := *c
	d.Gen++
	if err := d.Validate(); err != nil {
		t.Fatalf("the claim after the write must validate: %v", err)
	}
}

// The commit and the rollback race on the claim as on the record: whichever
// writes first, the other is refused and leaves the claim as it found it.
func TestRollBackRacesTheCommitOnTheClaim(t *testing.T) {
	newID := counter()
	// The commit first.
	c := targetWroteLanding(t)
	src, stg := commitScopes(c)
	if err := Commit(moveA, "ie", "ie-box", "uk", src, stg, newID)(c); err != nil {
		t.Fatal(err)
	}
	before := mustJSON(t, c)
	if err := RollBack(moveA, rbA)(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("rollback after the commit: %v", err)
	}
	// The rollback first.
	d := targetWroteLanding(t)
	if err := RollBack(moveA, rbA)(d); err != nil {
		t.Fatal(err)
	}
	if d.Op.Kind != KindRestoring || d.Op.Phase != PhaseReleasing || d.Op.Exec != "uk-box" || d.Op.Intent["rollback"] != rbA ||
		d.Op.ID != moveA || len(d.Op.Participants) != 2 {
		t.Fatalf("op after the rollback: %+v", d.Op)
	}
	// J15: the target installation's stamp F is journalled for its release.
	if d.Op.Journal[JournalTargetStamp] != "8" {
		t.Fatalf("target stamp journal %q", d.Op.Journal[JournalTargetStamp])
	}
	if d.Holder.Stamp != 8 || d.Holder.Generation != moveA {
		t.Fatalf("the rollback touched the holder: %+v", d.Holder)
	}
	validAfterWrite(t, d)
	before = mustJSON(t, d)
	if err := Commit(moveA, "ie", "ie-box", "uk", src, stg, newID)(d); !errors.Is(err, ErrWrongState) || string(mustJSON(t, d)) != string(before) {
		t.Fatalf("commit after the rollback: %v", err)
	}
	// A replay of the same rollback is a no-op; another rollback id is refused.
	if err := RollBack(moveA, rbA)(d); err != nil || string(mustJSON(t, d)) != string(before) {
		t.Fatalf("replay: %v", err)
	}
	if err := RollBack(moveA, rbB)(d); !errors.Is(err, ErrWrongState) || string(mustJSON(t, d)) != string(before) {
		t.Fatalf("another rollback id: %v", err)
	}
	// Every landing phase before the commit rolls back.
	for _, phase := range []string{"checking", "parking", "building", "published", "local", "configured", "restored", "probing", "probed"} {
		e := targetWroteLanding(t)
		e.Op.Phase = phase
		if err := RollBack(moveA, rbA)(e); err != nil || e.Op.Kind != KindRestoring || e.Op.Phase != PhaseReleasing {
			t.Fatalf("from landing/%s: %v %+v", phase, err, e.Op)
		}
	}
	// Refusals leave the claim alone.
	for name, tc := range map[string]struct {
		mutate func(*Claim)
		op, rb string
		want   error
	}{
		"another op":          {func(*Claim) {}, moveB, rbA, ErrStaleOp},
		"a move id":           {func(*Claim) {}, moveA, moveB, ErrMalformed},
		"no id":               {func(*Claim) {}, moveA, "", ErrMalformed},
		"short id":            {func(*Claim) {}, moveA, "rb_abc", ErrMalformed},
		"serving":             {func(c *Claim) { c.Op = Op{Kind: KindServing, Exec: "uk-box"} }, moveA, rbA, ErrStaleOp},
		"moving, sealed":      {func(c *Claim) { c.Op.Kind, c.Op.Phase, c.Op.Exec = KindMoving, PhaseSealed, "" }, moveA, rbA, ErrWrongState},
		"reverting":           {func(c *Claim) { c.Op.Kind, c.Op.Phase, c.Op.Exec = KindReverting, "", "ie-box" }, moveA, rbA, ErrWrongState},
		"landing, no holder":  {func(c *Claim) { c.Holder = nil }, moveA, rbA, ErrWrongState},
		"landing, committed":  {func(c *Claim) { c.Op.Phase = PhaseCommitted }, moveA, rbA, ErrWrongState},
		"restoring under rbB": {func(c *Claim) { c.Op.Kind, c.Op.Phase, c.Op.Intent["rollback"] = KindRestoring, PhaseReleasing, rbB }, moveA, rbA, ErrWrongState},
	} {
		e := targetWroteLanding(t)
		tc.mutate(e)
		before := mustJSON(t, e)
		if err := RollBack(tc.op, tc.rb)(e); !errors.Is(err, tc.want) || string(mustJSON(t, e)) != string(before) {
			t.Fatalf("%s: got %v, want %v", name, err, tc.want)
		}
	}
}

// J16: a record already at landing whose claim is still moving/sealed (the
// flip's claim half not yet written) rolls back through RollBackFlipped,
// which writes the flip first, in the same write.
func TestRollBackFlippedWritesTheFlipFirst(t *testing.T) {
	c := movingToSealed(t)
	c.Gen = 7
	before := mustJSON(t, c)
	if err := RollBack(moveA, rbA)(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("plain rollback over moving/sealed: %v", err)
	}
	// The flip half refuses (the source's stamp is wrong): nothing is written.
	if err := RollBackFlipped(moveA, rbA, "uk-box", 2, "gen-7")(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("a refused flip half: %v\nbefore %s\nafter  %s", err, before, mustJSON(t, c))
	}
	if err := RollBackFlipped(moveA, "", "uk-box", 3, "gen-7")(c); !errors.Is(err, ErrMalformed) || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("a malformed rollback id: %v", err)
	}
	if err := RollBackFlipped(moveA, rbA, "uk-box", 3, "gen-7")(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Kind != KindRestoring || c.Op.Phase != PhaseReleasing || c.Op.Exec != "uk-box" || c.Op.Intent["rollback"] != rbA {
		t.Fatalf("op: %+v", c.Op)
	}
	if c.Holder.Stamp != 8 || c.Op.Journal[JournalTargetStamp] != "8" || c.Op.Journal[JournalSourceStamp] != "3" || c.Op.Journal[JournalSrcGen] != "gen-7" {
		t.Fatalf("stamps: holder %d journal %v", c.Holder.Stamp, c.Op.Journal)
	}
	validAfterWrite(t, c)
	// Over an already flipped (landing) claim it is RollBack, and a replay is a no-op.
	d := targetWroteLanding(t)
	if err := RollBackFlipped(moveA, rbA, "uk-box", 3, "gen-7")(d); err != nil || d.Op.Kind != KindRestoring {
		t.Fatalf("over landing: %v %+v", err, d.Op)
	}
	after := mustJSON(t, d)
	if err := RollBackFlipped(moveA, rbA, "uk-box", 3, "gen-7")(d); err != nil || string(mustJSON(t, d)) != string(after) {
		t.Fatalf("replay: %v", err)
	}
}

// §7.8 released, claim rev 6.14 item 6 (R12-3), J15.
func TestTargetReleasedCoversTheTargetRegion(t *testing.T) {
	build := func(inventory map[string]InventoryEntry) *Claim {
		c := targetWroteLanding(t)
		// A restarted probe: its record's predecessor is the dead probe's
		// generation, in the target's region.
		c.Holder.Admissions = append(c.Holder.Admissions, AdmissionRecord{Region: "uk", PredecessorRegion: "uk", Session: 10, Box: "uk-box", Generation: "gen-r2", Predecessor: moveA, Mode: "cold"})
		c.Holder.Session, c.Holder.Generation = 10, "gen-r2"
		c.Inventory = inventory
		if err := RollBack(moveA, rbA)(c); err != nil {
			t.Fatal(err)
		}
		return c
	}
	withF := map[string]InventoryEntry{"uk-box": {Layout: "nested", Paths: map[string]uint64{"/data/uk": 8}}}
	c := build(withF)
	ids := 0
	newID := func() string { ids++; return fmt.Sprintf("t%d", ids) }
	holder := *c.Holder
	if err := TargetReleased(moveA, "uk-box", "uk", stoppedTarget, releaseScope(c), newID)(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Kind != KindRestoring || c.Op.Phase != PhaseReleased || c.Op.Exec != "" || c.Op.Journal[JournalEvidenceTarget] != stoppedTarget {
		t.Fatalf("op: %+v", c.Op)
	}
	k := obligationKinds(c)
	if k["park-replica/uk"] != 1 || k["park-local/uk-box"] != 1 || len(c.Obligations) != 2 {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	for _, o := range c.Obligations {
		if o.Op != moveA || (o.Kind == "park-replica" && (o.Park == nil || o.Park.Location != c.Holder.Location || o.Owner != "region:uk")) {
			t.Fatalf("obligation: %+v", o)
		}
	}
	if c.Holder.Admissions[0].Covered || !c.Holder.Admissions[1].Covered {
		t.Fatalf("coverage: ie-predecessor %v, uk-predecessor %v", c.Holder.Admissions[0].Covered, c.Holder.Admissions[1].Covered)
	}
	if c.Holder.Stamp != holder.Stamp || c.Holder.Session != holder.Session || c.Holder.Generation != holder.Generation || c.Holder.WriterBox != holder.WriterBox {
		t.Fatalf("the release touched the holder: %+v", c.Holder)
	}
	validAfterWrite(t, c)
	// A replay queues nothing and mints no ids, even once a chore is done.
	used, before := ids, mustJSON(t, c)
	if err := TargetReleased(moveA, "uk-box", "uk", stoppedTarget, releaseScope(c), newID)(c); err != nil || ids != used || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("replay: %v ids %d->%d", err, used, ids)
	}
	c.Obligations = c.Obligations[:1]
	before = mustJSON(t, c)
	if err := TargetReleased(moveA, "uk-box", "uk", stoppedTarget, releaseScope(c), newID)(c); err != nil || ids != used || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("replay after a completed chore: %v ids %d->%d", err, used, ids)
	}
	// J15: no paths of F on the target, no park-local.
	for name, inv := range map[string]map[string]InventoryEntry{
		"no inventory":     nil,
		"another stamp":    {"uk-box": {Layout: "nested", Paths: map[string]uint64{"/data/uk": 3}}},
		"F on another box": {"ie-box": {Layout: "nested", Paths: map[string]uint64{"/data/ie": 8}}},
	} {
		d := build(inv)
		if err := TargetReleased(moveA, "uk-box", "uk", stoppedTarget, releaseScope(d), counter())(d); err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		if k := obligationKinds(d); k["park-replica/uk"] != 1 || len(d.Obligations) != 1 {
			t.Fatalf("%s: obligations %+v", name, d.Obligations)
		}
	}
	// Refusals leave the claim as found.
	for name, tc := range map[string]struct {
		mutate              func(*Claim)
		box, region, ev, op string
		want                error
	}{
		"another op":             {func(*Claim) {}, "uk-box", "uk", stoppedTarget, moveB, ErrStaleOp},
		"not rolled back":        {func(c *Claim) { c.Op.Kind, c.Op.Phase = KindLanding, "probed" }, "uk-box", "uk", stoppedTarget, moveA, ErrWrongState},
		"the source's box":       {func(*Claim) {}, "ie-box", "uk", stoppedTarget, moveA, ErrWrongState},
		"the source's region":    {func(*Claim) {}, "uk-box", "ie", stoppedTarget, moveA, ErrWrongState},
		"no evidence":            {func(*Claim) {}, "uk-box", "uk", "", moveA, ErrMalformed},
		"exec is not the target": {func(c *Claim) { c.Op.Exec = "ie-box" }, "uk-box", "uk", stoppedTarget, moveA, ErrWrongState},
		"released, other evidence": {func(c *Claim) {
			c.Op.Phase, c.Op.Exec, c.Op.Journal[JournalEvidenceTarget] = PhaseReleased, "", "other"
		}, "uk-box", "uk", stoppedTarget, moveA, ErrWrongState},
	} {
		d := build(withF)
		tc.mutate(d)
		before := mustJSON(t, d)
		if err := TargetReleased(tc.op, tc.box, tc.region, tc.ev, releaseScope(d), counter())(d); !errors.Is(err, tc.want) || string(mustJSON(t, d)) != string(before) {
			t.Fatalf("%s: got %v, want %v", name, err, tc.want)
		}
	}
	// A location park that is not the holder's location refuses the whole
	// write: exec, phase and coverage stay as found.
	d := build(withF)
	before = mustJSON(t, d)
	bad := releaseScope(d)
	bad.Location = "instances/" + d.Host
	if err := TargetReleased(moveA, "uk-box", "uk", stoppedTarget, bad, counter())(d); !errors.Is(err, ErrMalformed) || string(mustJSON(t, d)) != string(before) {
		t.Fatalf("bad scope: %v\nbefore %s\nafter  %s", err, before, mustJSON(t, d))
	}
}

// Review Focus 3 (wCleanReturnIdentity, INV-16, R11-4): the clean return
// names the source's generation and stamp again and leaves the admissions,
// whose last record is the target's writer, for the re-admission to name.
func TestReturnToSourceCleanNamesSrcGenAndKeepsAdmissions(t *testing.T) {
	c := released(t)
	admissions := append([]AdmissionRecord(nil), c.Holder.Admissions...)
	session, writer := c.Holder.Session, c.Holder.WriterBox
	if err := ReturnToSource(moveA, "ie-box", true, ReturnFence{})(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Kind != KindRestoring || c.Op.Phase != PhaseClaimed || c.Op.Exec != "ie-box" {
		t.Fatalf("op: %+v", c.Op)
	}
	if c.Holder.Generation != "gen-7" || c.Holder.Stamp != 3 {
		t.Fatalf("holder generation %q stamp %d, want gen-7 and 3", c.Holder.Generation, c.Holder.Stamp)
	}
	if !reflect.DeepEqual(c.Holder.Admissions, admissions) || c.Holder.Session != session || c.Holder.WriterBox != writer {
		t.Fatalf("the return touched the admissions: %+v", c.Holder)
	}
	if last := c.Holder.Admissions[len(c.Holder.Admissions)-1]; last.Box != "uk-box" || last.Generation != moveA {
		t.Fatalf("the last admission is not the target's writer: %+v", last)
	}
	validAfterWrite(t, c)
	// A replay is a no-op.
	before := mustJSON(t, c)
	if err := ReturnToSource(moveA, "ie-box", true, ReturnFence{})(c); err != nil || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("replay: %v", err)
	}
	// Refusals.
	for name, tc := range map[string]struct {
		claim  func() *Claim
		op     string
		source string
		want   error
	}{
		"another op":         {func() *Claim { return released(t) }, moveB, "ie-box", ErrStaleOp},
		"the target":         {func() *Claim { return released(t) }, moveA, "uk-box", ErrWrongState},
		"no source":          {func() *Claim { return released(t) }, moveA, "", ErrMalformed},
		"before the release": {func() *Claim { return rolledBack(t) }, moveA, "ie-box", ErrWrongState},
		"landing":            {func() *Claim { return targetWroteLanding(t) }, moveA, "ie-box", ErrWrongState},
		"no evidence": {func() *Claim {
			c := released(t)
			delete(c.Op.Journal, JournalEvidenceTarget)
			return c
		}, moveA, "ie-box", ErrWrongState},
		"no source stamp": {func() *Claim {
			c := released(t)
			delete(c.Op.Journal, JournalSourceStamp)
			return c
		}, moveA, "ie-box", ErrWrongState},
		"no source gen": {func() *Claim {
			c := released(t)
			delete(c.Op.Journal, JournalSrcGen)
			return c
		}, moveA, "ie-box", ErrWrongState},
		"exec still a box": {func() *Claim {
			c := released(t)
			c.Op.Exec = "uk-box"
			return c
		}, moveA, "ie-box", ErrWrongState},
		"claimed by the other": {func() *Claim {
			c := released(t)
			c.Op.Phase, c.Op.Exec = PhaseClaimed, "uk-box"
			return c
		}, moveA, "ie-box", ErrWrongState},
		"probing": {func() *Claim {
			c := released(t)
			c.Op.Phase, c.Op.Exec = PhaseProbing, "ie-box"
			return c
		}, moveA, "ie-box", ErrWrongState},
	} {
		c := tc.claim()
		before := mustJSON(t, c)
		if err := ReturnToSource(tc.op, tc.source, true, ReturnFence{})(c); !errors.Is(err, tc.want) || string(mustJSON(t, c)) != string(before) {
			t.Fatalf("%s: got %v, want %v", name, err, tc.want)
		}
	}
}

// A dirty return names the source's stamp again and leaves holder.generation
// at the target's last admission: the source's probe names the return
// generation at its cold admission (§7.8, §19.1 table).
func TestReturnToSourceDirtyLeavesGeneration(t *testing.T) {
	c := released(t)
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Phase != PhaseClaimed || c.Op.Exec != "ie-box" || c.Holder.Stamp != 3 || c.Holder.Generation != moveA {
		t.Fatalf("op %+v holder %+v", c.Op, c.Holder)
	}
	// src_gen may be the origin (""): the clean return then names it.
	d := released(t)
	d.Op.Journal[JournalSrcGen] = ""
	if err := ReturnToSource(moveA, "ie-box", true, ReturnFence{})(d); err != nil || d.Holder.Generation != "" {
		t.Fatalf("origin generation: %v %q", err, d.Holder.Generation)
	}
}

// declaredLost is rolledBack (restoring/releasing, exec the target) after
// claim Plan 8's declare-lost transition for the target.
func declaredLost(t *testing.T) *Claim {
	t.Helper()
	c := rolledBack(t)
	in := LostDeclaration{LossID: "loss_1", ExpectedClaimGen: c.Gen, ExpectedClaimETag: `"etag"`,
		HolderIncarnation: c.Holder.Incarnation, HolderStamp: c.Holder.Stamp, HolderSession: c.Holder.Session,
		OpID: moveA, OpKind: KindRestoring, OpPhase: PhaseReleasing, Box: "uk-box", Enrollment: "enroll-uk",
		QuarantineRevision: 1, FencedAt: time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC),
		Consequences: LossConsequences{RevokeExecution: true, RemoveParticipation: true, RecoveryIntent: []byte(`{"kind":"lost-target"}`)}}
	if err := declareLostTransition(in, testLossOwner{})(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Exec != "none" || !reflect.DeepEqual(c.Op.Participants, []string{"ie-box"}) || c.Op.Journal["loss.loss_1"] == "" || c.Op.Journal[JournalEvidenceTarget] != "" || c.Op.Phase != PhaseReleasing {
		t.Fatalf("fixture: %+v", c.Op)
	}
	return c
}

// §7.10: an operator's declaration stands in for the target's evidence. The
// claim is as Plan 8's transition leaves it (exec "none", the target out of
// participants, loss.<id> journalled, no evidence_target), stepped to
// released by the resolve door (J16) before the source claims.
func TestReturnToSourceAfterADeclaredLoss(t *testing.T) {
	c := declaredLost(t)
	before := mustJSON(t, c)
	// J16: the resolve door steps the claim to released first.
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
		t.Fatalf("at releasing: %v", err)
	}
	c.Op.Phase = PhaseReleased
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Exec != "ie-box" || c.Op.Phase != PhaseClaimed || c.Holder.Stamp != 3 {
		t.Fatalf("op %+v stamp %d", c.Op, c.Holder.Stamp)
	}
	validAfterWrite(t, c)
	// The same without the loss entry is refused.
	d := declaredLost(t)
	d.Op.Phase = PhaseReleased
	delete(d.Op.Journal, "loss.loss_1")
	before = mustJSON(t, d)
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(d); !errors.Is(err, ErrWrongState) || string(mustJSON(t, d)) != string(before) {
		t.Fatalf("without the loss entry: %v", err)
	}
	// A loss entry the claim never recorded, or one that left the target a
	// participant, is not the target's declaration.
	e := declaredLost(t)
	e.Op.Phase = PhaseReleased
	e.LossRefs = nil
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(e); !errors.Is(err, ErrWrongState) {
		t.Fatalf("an unrecorded loss: %v", err)
	}
	f := declaredLost(t)
	f.Op.Phase = PhaseReleased
	f.Op.Participants = []string{"ie-box", "uk-box"}
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(f); !errors.Is(err, ErrWrongState) {
		t.Fatalf("the target still a participant: %v", err)
	}
}

// J5: a termination fence for the target, verified by the caller before
// Update (VerifyReturnFence), authorises the return when the closure finds
// the same fence.<target> entry.
func TestReturnToSourceWithAVerifiedFence(t *testing.T) {
	const ref = `{"enrollment":"0123456789abcdef0123456789abcdef","key":"control/box-fences/0123456789abcdef0123456789abcdef.json","sha256":"00"}`
	fenced := func() *Claim {
		c := rolledBack(t)
		c.Op.Phase, c.Op.Exec = PhaseReleased, "none"
		c.Op.Journal["fence.uk-box"] = ref
		return c
	}
	good := ReturnFence{op: moveA, box: "uk-box", ref: ref}
	c := fenced()
	if err := ReturnToSource(moveA, "ie-box", true, good)(c); err != nil {
		t.Fatal(err)
	}
	if c.Op.Exec != "ie-box" || c.Op.Phase != PhaseClaimed || c.Holder.Generation != "gen-7" {
		t.Fatalf("op %+v holder %+v", c.Op, c.Holder)
	}
	for name, f := range map[string]ReturnFence{
		"unverified":            {},
		"another op":            {op: moveB, box: "uk-box", ref: ref},
		"the source's fence":    {op: moveA, box: "ie-box", ref: ref},
		"another journal value": {op: moveA, box: "uk-box", ref: ref + " "},
	} {
		d := fenced()
		before := mustJSON(t, d)
		if err := ReturnToSource(moveA, "ie-box", true, f)(d); !errors.Is(err, ErrWrongState) || string(mustJSON(t, d)) != string(before) {
			t.Fatalf("%s: %v", name, err)
		}
	}
	// VerifyReturnFence verifies before it vouches: a journal entry that does
	// not decode, or no store, is ErrFence.
	op := fenced().Op
	if _, err := VerifyReturnFence(context.Background(), nil, nil, op, "uk-box"); !errors.Is(err, ErrFence) {
		t.Fatalf("no store: %v", err)
	}
	op.Journal["fence.uk-box"] = "not json"
	if _, err := VerifyReturnFence(context.Background(), release.NewDirStore(t.TempDir()), nil, op, "uk-box"); !errors.Is(err, ErrFence) {
		t.Fatalf("bad journal: %v", err)
	}
}

// K8 and J6: the release to serving from each terminal move state, once no
// chore under the op remains. The executor named is the one executing.
func TestReleaseMoveWaitsForObligations(t *testing.T) {
	states := map[string]struct {
		kind, phase, exec string
	}{
		"committed landing":         {KindLanding, PhaseCommitted, "uk-box"},
		"opened rollback":           {KindRestoring, PhaseOpened, "ie-box"},
		"opened revert":             {KindReverting, PhaseOpened, "ie-box"},
		"the door, before the seal": {KindMoving, PhaseMoving, "ie-box"},
	}
	for name, st := range states {
		c := targetWroteLanding(t)
		c.Op.Kind, c.Op.Phase, c.Op.Exec = st.kind, st.phase, st.exec
		c.Obligations = []Obligation{{ID: "o1", Op: moveA, Kind: "park-replica", Region: "ie", Retirement: "r", Owner: "region:ie"}}
		before := mustJSON(t, c)
		if err := ReleaseMove(moveA, st.exec)(c); !errors.Is(err, ErrPendingChore) || string(mustJSON(t, c)) != string(before) {
			t.Fatalf("%s with a chore: %v", name, err)
		}
		c.Obligations[0].Op = "another-op"
		if err := ReleaseMove(moveA, st.exec)(c); err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		if c.Op.Kind != KindServing || c.Op.Exec != st.exec || c.Op.ID != "" || c.Op.Phase != "" || len(c.Op.Participants) != 0 || c.Op.Journal["released_move"] != moveA || len(c.Op.Journal) != 1 {
			t.Fatalf("%s: op after release %+v", name, c.Op)
		}
		if err := RequireReleasedMove(moveA)(c); err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		validAfterWrite(t, c)
	}
	// The new states name their executor exactly.
	for name, st := range states {
		if st.kind == KindMoving {
			continue
		}
		c := targetWroteLanding(t)
		c.Op.Kind, c.Op.Phase, c.Op.Exec = st.kind, st.phase, st.exec
		other := "ie-box"
		if st.exec == "ie-box" {
			other = "uk-box"
		}
		if err := ReleaseMove(moveA, other)(c); !errors.Is(err, ErrWrongState) {
			t.Fatalf("%s released to %s: %v", name, other, err)
		}
	}
	// Not before those states: a revert's phase is read from op.phase (J1),
	// not rev_phase.
	for name, st := range map[string]struct{ kind, phase string }{
		"landing probed":      {KindLanding, "probed"},
		"restoring probed":    {KindRestoring, PhaseProbed},
		"restoring claimed":   {KindRestoring, PhaseClaimed},
		"reverting probed":    {KindReverting, PhaseProbed},
		"reverting, rev only": {KindReverting, PhaseSealed},
		"moving sealed":       {KindMoving, PhaseSealed},
		"moving aborting":     {KindMoving, PhaseAborting},
	} {
		c := targetWroteLanding(t)
		c.Op.Kind, c.Op.Phase, c.Op.Exec = st.kind, st.phase, "ie-box"
		if st.phase == PhaseSealed && st.kind == KindReverting {
			setJournal(c, JournalRevPhase, PhaseOpened)
		}
		if err := ReleaseMove(moveA, "ie-box")(c); !errors.Is(err, ErrWrongState) {
			t.Fatalf("%s: %v", name, err)
		}
	}
}

// revertingClaim is a revert as Plan 2's Revert leaves it: from the cut's
// stopping (phase "", no files baseline) or from sealed (a baseline).
func revertingClaim(t *testing.T, sealed bool) *Claim {
	t.Helper()
	var c *Claim
	if sealed {
		c = movingToSealed(t)
	} else {
		c = movingTestClaim(t)
		for _, tr := range []func(*Claim) error{
			BeginMove(moveA, "ops", "ie-box", "uk-box", "uk", time.Now()),
			ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource),
		} {
			if err := tr(c); err != nil {
				t.Fatal(err)
			}
		}
	}
	lane := ParkScope{Location: "moves/" + c.Host + "/" + moveA, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC), RequestLimit: time.Minute}
	if err := Revert(moveA, "ie-box", "uk", &lane, counter())(c); err != nil {
		t.Fatal(err)
	}
	return c
}

// §7.4 and J1, J8: rev_phase runs evidence, verify, probing, probed, opened;
// verify only with the files baseline the record keeps from stopped, and
// straight to probing without it. op.phase mirrors probing, probed and
// opened in the same write.
func TestRevPhaseOrder(t *testing.T) {
	for _, tc := range []struct {
		name     string
		sealed   bool
		baseline bool
		steps    [][2]string
		pre      string
	}{
		{"from stopped, with a baseline", true, true, [][2]string{{"evidence", "verify"}, {"verify", "probing"}, {"probing", "probed"}, {"probed", "opened"}}, PhaseSealed},
		{"from stopping, no baseline", false, false, [][2]string{{"evidence", "probing"}, {"probing", "probed"}, {"probed", "opened"}}, PhaseMoving},
	} {
		t.Run(tc.name, func(t *testing.T) {
			c := revertingClaim(t, tc.sealed)
			if c.Op.Journal[JournalRevPhase] != "evidence" || c.Op.Phase != tc.pre {
				t.Fatalf("fixture: %+v", c.Op)
			}
			// The other entry is refused.
			wrong := "probing"
			if !tc.baseline {
				wrong = "verify"
			}
			before := mustJSON(t, c)
			if err := SetRevPhase(moveA, "evidence", wrong, tc.baseline)(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
				t.Fatalf("evidence to %s: %v", wrong, err)
			}
			for _, step := range tc.steps {
				if err := SetRevPhase(moveA, step[0], step[1], tc.baseline)(c); err != nil {
					t.Fatalf("%s to %s: %v", step[0], step[1], err)
				}
				if c.Op.Journal[JournalRevPhase] != step[1] {
					t.Fatalf("rev_phase %q after %v", c.Op.Journal[JournalRevPhase], step)
				}
				want := tc.pre
				if step[1] != "verify" {
					want = step[1]
				}
				if c.Op.Phase != want || c.Op.Kind != KindReverting || c.Op.Exec != "ie-box" {
					t.Fatalf("op after %v: %+v", step, c.Op)
				}
				validAfterWrite(t, c)
				// A replay finds the phase moved.
				before := mustJSON(t, c)
				if err := SetRevPhase(moveA, step[0], step[1], tc.baseline)(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
					t.Fatalf("replay of %v: %v", step, err)
				}
			}
			// opened is the last phase; ReleaseMove takes it from there.
			if err := SetRevPhase(moveA, "opened", "probing", tc.baseline)(c); !errors.Is(err, ErrWrongState) {
				t.Fatalf("after opened: %v", err)
			}
		})
	}
	// Skips, other kinds, other ops and a phase that disagrees with rev_phase.
	c := revertingClaim(t, true)
	for name, tc := range map[string]struct {
		op, from, to string
		want         error
	}{
		"skip verify to probed": {moveA, "evidence", "probed", ErrWrongState},
		"skip to opened":        {moveA, "evidence", "opened", ErrWrongState},
		"from the wrong phase":  {moveA, "verify", "probing", ErrWrongState},
		"backwards":             {moveA, "evidence", "evidence", ErrWrongState},
		"an unknown phase":      {moveA, "evidence", "stuck", ErrWrongState},
		"another op":            {moveB, "evidence", "verify", ErrStaleOp},
	} {
		before := mustJSON(t, c)
		if err := SetRevPhase(tc.op, tc.from, tc.to, true)(c); !errors.Is(err, tc.want) || string(mustJSON(t, c)) != string(before) {
			t.Fatalf("%s: got %v, want %v", name, err, tc.want)
		}
	}
	d := revertingClaim(t, true)
	d.Op.Phase = PhaseProbed // disagrees with rev_phase evidence
	if err := SetRevPhase(moveA, "evidence", "verify", true)(d); !errors.Is(err, ErrWrongState) {
		t.Fatalf("op.phase ahead of rev_phase: %v", err)
	}
	e := rolledBack(t)
	setJournal(e, JournalRevPhase, "evidence")
	if err := SetRevPhase(moveA, "evidence", "verify", true)(e); !errors.Is(err, ErrWrongState) {
		t.Fatalf("a restoring op: %v", err)
	}
}

// J1: the source's recovery phases after claimed are mirrored onto the claim
// op in order: probing (before the first recovery attempt), probed, opened.
func TestRestoringPhaseOrder(t *testing.T) {
	c := released(t)
	if err := SetRestoringPhase(moveA, PhaseReleased, PhaseClaimed)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("released to claimed is ReturnToSource's: %v", err)
	}
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(c); err != nil {
		t.Fatal(err)
	}
	for _, step := range [][2]string{{PhaseClaimed, PhaseProbing}, {PhaseProbing, PhaseProbed}, {PhaseProbed, PhaseOpened}} {
		before := mustJSON(t, c)
		if err := SetRestoringPhase(moveA, step[0], "nonsense")(c); !errors.Is(err, ErrWrongState) || string(mustJSON(t, c)) != string(before) {
			t.Fatalf("from %s to nonsense: %v", step[0], err)
		}
		if err := SetRestoringPhase(moveA, step[0], step[1])(c); err != nil {
			t.Fatalf("%v: %v", step, err)
		}
		if c.Op.Phase != step[1] || c.Op.Exec != "ie-box" || c.Op.Kind != KindRestoring {
			t.Fatalf("op after %v: %+v", step, c.Op)
		}
		if err := SetRestoringPhase(moveA, step[0], step[1])(c); !errors.Is(err, ErrWrongState) {
			t.Fatalf("replay %v: %v", step, err)
		}
	}
	d := released(t)
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(d); err != nil {
		t.Fatal(err)
	}
	if err := SetRestoringPhase(moveA, PhaseClaimed, PhaseProbed)(d); !errors.Is(err, ErrWrongState) {
		t.Fatalf("skip: %v", err)
	}
	if err := SetRestoringPhase(moveB, PhaseClaimed, PhaseProbing)(d); !errors.Is(err, ErrStaleOp) {
		t.Fatalf("another op: %v", err)
	}
	// The source executes from claimed on; a claim whose exec is not the
	// source does not step.
	d.Op.Exec = "uk-box"
	if err := SetRestoringPhase(moveA, PhaseClaimed, PhaseProbing)(d); !errors.Is(err, ErrWrongState) {
		t.Fatalf("exec not the source: %v", err)
	}
}

// §7.8 probed (dirty): the return lane moves/<host>/<rb id> is parked in the
// source's region, queued once, and the release to serving waits for it.
func TestQueueReturnLaneParkAtProbed(t *testing.T) {
	c := released(t)
	c.Obligations = nil // the target's chores are done
	if err := ReturnToSource(moveA, "ie-box", false, ReturnFence{})(c); err != nil {
		t.Fatal(err)
	}
	lane := releaseScope(c)
	lane.Location = "moves/" + c.Host + "/" + rbA
	ids := 0
	newID := func() string { ids++; return fmt.Sprintf("l%d", ids) }
	for _, phase := range []string{PhaseClaimed, PhaseProbing} {
		c.Op.Phase = phase
		if err := QueueReturnLanePark(moveA, "ie", lane, newID)(c); !errors.Is(err, ErrWrongState) || ids != 0 {
			t.Fatalf("at %s: %v", phase, err)
		}
	}
	c.Op.Phase = PhaseProbed
	for name, tc := range map[string]struct {
		region   string
		location string
		want     error
	}{
		"the target's region": {"uk", lane.Location, ErrWrongState},
		"the move's lane":     {"ie", "moves/" + c.Host + "/" + moveA, ErrMalformed},
		"the live location":   {"ie", c.Holder.Location, ErrMalformed},
		"the staging prefix":  {"ie", "moves/" + c.Host, ErrMalformed},
	} {
		s := lane
		s.Location = tc.location
		before := mustJSON(t, c)
		if err := QueueReturnLanePark(moveA, tc.region, s, newID)(c); !errors.Is(err, tc.want) || ids != 0 || string(mustJSON(t, c)) != string(before) {
			t.Fatalf("%s: %v", name, err)
		}
	}
	if err := QueueReturnLanePark(moveA, "ie", lane, newID)(c); err != nil {
		t.Fatal(err)
	}
	if k := obligationKinds(c); k["park-replica/ie"] != 1 || len(c.Obligations) != 1 || c.Obligations[0].Park.Location != lane.Location || c.Obligations[0].Owner != "region:ie" {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	validAfterWrite(t, c)
	used := ids
	if err := QueueReturnLanePark(moveA, "ie", lane, newID)(c); err != nil || ids != used || len(c.Obligations) != 1 {
		t.Fatalf("replay: %v", err)
	}
	other := lane
	other.RequestLimit = time.Hour
	if err := QueueReturnLanePark(moveA, "ie", other, newID)(c); !errors.Is(err, ErrWrongState) {
		t.Fatalf("another scope: %v", err)
	}
	// The release waits for the lane park.
	if err := SetRestoringPhase(moveA, PhaseProbed, PhaseOpened)(c); err != nil {
		t.Fatal(err)
	}
	if err := ReleaseMove(moveA, "ie-box")(c); !errors.Is(err, ErrPendingChore) {
		t.Fatalf("release before the lane park: %v", err)
	}
	// Once the park engine completes the chore, a replay at probed does not
	// queue it again.
	c.Obligations = nil
	c.Op.Phase = PhaseProbed
	if err := QueueReturnLanePark(moveA, "ie", lane, newID)(c); err != nil || len(c.Obligations) != 0 || ids != used {
		t.Fatalf("replay after completion: %v %+v", err, c.Obligations)
	}
	c.Op.Phase = PhaseOpened
	if err := ReleaseMove(moveA, "ie-box")(c); err != nil {
		t.Fatal(err)
	}
}