package host

import (
	"context"
	"errors"
	"path/filepath"
	"reflect"
	"testing"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

const (
	mpHost   = "shop.example"
	mpOp     = "mv_policy"
	mpInc    = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
	mpTarget = "uk-box"
	mpSource = "ie-box"
)

// mpClaim is the claim after the flip: landing for mpOp at phase, the
// target executing, the holder at stamp 7.
func mpClaim(kind, phase string) *hostclaim.Claim {
	return &hostclaim.Claim{V: 1, Host: mpHost, Gen: 9,
		Holder: &hostclaim.Holder{Family: "instance", Account: "bab", App: "titogo", Incarnation: mpInc, Stamp: 7, Generation: "g-src", Session: 3, Location: "instances/" + mpHost + "/" + mpInc},
		Op: hostclaim.Op{Kind: kind, ID: mpOp, Phase: phase, Exec: mpTarget, Participants: []string{mpSource, mpTarget},
			Intent: map[string]string{"to": "uk", "source": mpSource, "target": mpTarget, "source_enrollment": "enr-ie"}}}
}

// mpSelection is the selection hostbox.WorkSource would capture for that
// claim on the target.
func mpSelection(c *hostclaim.Claim, action string) hostbox.Selection {
	return hostbox.Selection{Host: mpHost, Action: action, Box: mpTarget, Region: "uk",
		Captured: hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp},
		Kind:     c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase, Generation: c.Holder.Generation, Session: c.Holder.Session}
}

// mpPins are the inputs the move pinned at readiness.
func mpPins() *provision.MoveInputs {
	return &provision.MoveInputs{BundleETags: map[string]string{"default": "e-default", "environment": "e-env"}, OverlayETag: "e-overlay", Release: "v1.2.3"}
}

// mpRecord is the landing record the reader answers: this move, this
// incarnation, landing, with the pins.
func mpRecord() *provision.Record {
	return &provision.Record{Host: mpHost, Account: "bab", App: "titogo", Region: "uk",
		Move: &provision.MoveState{ID: mpOp, Incarnation: mpInc, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandConfigured, Inputs: mpPins()}}
}

// mpReader counts its reads and answers rec, or err.
type mpReader struct {
	reads int
	rec   *provision.Record
	err   error
	args  []string
}

func (r *mpReader) read(_ context.Context, acct, app, host string) (*provision.Record, error) {
	r.reads++
	r.args = []string{acct, app, host}
	if r.err != nil {
		return nil, r.err
	}
	return r.rec, nil
}

// mpPolicy is the policy over a reader answering the landing record.
func mpPolicy() (func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error, *mpReader) {
	r := &mpReader{rec: mpRecord()}
	return MovesClaimPolicy(r.read), r
}

// mpIdentity is the identity that fits the action: the pinned config and
// release for the held writes, nothing for the rest.
func mpIdentity(action string) hostwrite.Identity {
	switch action {
	case "held-config":
		return HeldConfigIdentity(mpPins().BundleETags, mpPins().OverlayETag)
	case "held-release":
		return hostwrite.Identity{Release: "v1.2.3"}
	}
	return hostwrite.Identity{}
}

var mpActions = []string{"held-config", "held-release", "held-timers", "restore", "spawn", "seed", "writer", "install"}

// Every action against every landing phase, including the probe's and the
// commit's: granted exactly where the table says, refused with the typed
// error everywhere else. The same cells through hostbox.Policy.Check with
// the registry hold the landing has.
func TestMovesClaimPolicyGrantsByPhase(t *testing.T) {
	granted := map[string]map[string]bool{
		provision.LandLocal:      {"held-timers": true},
		provision.LandConfigured: {"held-config": true, "held-release": true, "restore": true, "install": true},
		// The probe's rows (Plan 3 Task 5): its spawn needs the launch
		// identity, which mpIdentity does not give it (movesprobe_policy_test).
		provision.LandProbing: {"held-config": true, "held-release": true},
	}
	phases := append(provision.LandingOrder(), "committed")
	moves, reader := mpPolicy()
	policy := hostbox.Policy{Moves: moves, Held: func(context.Context, string) (string, error) { return mpOp, nil }}
	grants := 0
	for _, phase := range phases {
		for _, action := range mpActions {
			c := mpClaim(hostclaim.KindLanding, phase)
			s := mpSelection(c, action)
			err := moves(t.Context(), c, s, mpIdentity(action))
			want := granted[phase][action]
			if (err == nil) != want {
				t.Errorf("%s/%s: err %v, want granted %v", phase, action, err, want)
				continue
			}
			if err != nil {
				var r *MovesPolicyRefusal
				if !errors.As(err, &r) || !errors.Is(err, ErrMovesRefused) || r.Action != action || r.Phase != phase || r.Kind != hostclaim.KindLanding || r.Reason == "" {
					t.Errorf("%s/%s: not the typed refusal: %#v", phase, action, err)
				}
			}
			if chk := policy.Check(t.Context(), c, s, mpIdentity(action)); (chk == nil) != want {
				t.Errorf("%s/%s through Policy.Check: %v, want granted %v", phase, action, chk, want)
			}
			if want {
				grants++
			}
		}
	}
	if grants != 7 {
		t.Fatalf("%d cells granted, want 7", grants)
	}
	// The record is read for the held config and release only: twice per
	// pass over the table (the direct call and Policy.Check), at configured
	// and at probing.
	if reader.reads != 8 || !reflect.DeepEqual(reader.args, []string{"bab", "titogo", mpHost}) {
		t.Fatalf("record reads: %d %v", reader.reads, reader.args)
	}
}

// The claim's role and the selection's shape: a non-move op, the other move
// kinds, a mismatched op, box or region, the source after the flip, a lost
// box, a stale capture, a mismatched kind or phase and a wrong identity are
// all refused, and nothing but the exact landing cell passes.
func TestMovesClaimPolicyRefusesOtherRoles(t *testing.T) {
	base := func() (*hostclaim.Claim, hostbox.Selection) {
		c := mpClaim(hostclaim.KindLanding, provision.LandLocal)
		return c, mpSelection(c, "held-timers")
	}
	cases := map[string]func(c *hostclaim.Claim, s *hostbox.Selection, id *hostwrite.Identity){
		"serving op": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: mpTarget}
			s.Kind, s.Op, s.Phase = c.Op.Kind, "", ""
		},
		"creating op": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Kind = hostclaim.KindCreating
			s.Kind = c.Op.Kind
		},
		"moving, the source executing": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Kind, c.Op.Phase, c.Op.Exec = hostclaim.KindMoving, hostclaim.PhaseSealed, mpSource
			s.Kind, s.Phase, s.Box, s.Region = c.Op.Kind, c.Op.Phase, mpSource, "ie"
		},
		"moving, the target asking": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Kind, c.Op.Phase = hostclaim.KindMoving, provision.LandLocal
			s.Kind = c.Op.Kind
		},
		"reverting": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Kind = hostclaim.KindReverting
			s.Kind = c.Op.Kind
		},
		"restoring": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Kind = hostclaim.KindRestoring
			s.Kind = c.Op.Kind
		},
		"another op selected": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Op = "mv_other" },
		"the claim under another op": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.ID = "mv_other"
		},
		"another host's claim": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) { c.Host = "other.example" },
		"the source box after the flip": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			s.Box, s.Region = mpSource, "ie"
		},
		"the source named executor": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Exec = mpSource
			s.Box, s.Region = mpSource, "ie"
		},
		"exec none": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) { c.Op.Exec = "" },
		"a third box executing": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Exec = "de-box"
			s.Box = "de-box"
		},
		"the target not the second participant": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Participants = []string{mpTarget, mpSource}
		},
		"the intent names another target": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Intent["target"] = "de-box"
		},
		"the intent names this box as source too": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Intent["source"] = mpTarget
		},
		"another region": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Region = "ie" },
		"no region":      func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Region = "" },
		"the box is lost": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {
			c.Lost = []string{mpTarget}
		},
		"a stale capture": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Captured.Stamp = 6 },
		"an empty capture": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			s.Captured = hostguard.Captured{}
		},
		"another incarnation captured": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			s.Captured.Incarnation = "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210"
		},
		"the claim moved on since the selection": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {
			c.Op.Phase = provision.LandConfigured
		},
		"the selection's kind is not the claim's": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			s.Kind = hostclaim.KindServing
		},
		"no holder": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) { c.Holder = nil },
		"no action": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Action = "" },
		"an ordinary action": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {
			s.Action = "timer-delivery"
		},
		"a config identity on the timers": func(_ *hostclaim.Claim, _ *hostbox.Selection, id *hostwrite.Identity) {
			id.Config = "sha256:abc"
		},
		"a release identity on the timers": func(_ *hostclaim.Claim, _ *hostbox.Selection, id *hostwrite.Identity) {
			id.Release = "v1"
		},
	}
	for name, mutate := range cases {
		t.Run(name, func(t *testing.T) {
			c, s := base()
			id := hostwrite.Identity{}
			mutate(c, &s, &id)
			moves, _ := mpPolicy()
			err := moves(t.Context(), c, s, id)
			if err == nil {
				t.Fatal("granted")
			}
			if !errors.Is(err, ErrMovesRefused) {
				t.Fatalf("not the typed refusal: %v", err)
			}
		})
	}
	t.Run("the base cell is granted", func(t *testing.T) {
		c, s := base()
		moves, _ := mpPolicy()
		if err := moves(t.Context(), c, s, hostwrite.Identity{}); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("held-config needs the pinned config identity and no release", func(t *testing.T) {
		c := mpClaim(hostclaim.KindLanding, provision.LandConfigured)
		s := mpSelection(c, "held-config")
		moves, _ := mpPolicy()
		pinned := mpIdentity("held-config").Config
		for _, id := range []hostwrite.Identity{{}, {Release: "v1.2.3"}, {Config: pinned, Release: "v1.2.3"}} {
			if err := moves(t.Context(), c, s, id); !errors.Is(err, ErrMovesRefused) {
				t.Fatalf("%+v: %v", id, err)
			}
		}
		if err := moves(t.Context(), c, s, hostwrite.Identity{Config: pinned}); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("held-release needs the pinned release and no config", func(t *testing.T) {
		c := mpClaim(hostclaim.KindLanding, provision.LandConfigured)
		s := mpSelection(c, "held-release")
		moves, _ := mpPolicy()
		pinned := mpIdentity("held-config").Config
		for _, id := range []hostwrite.Identity{{}, {Config: pinned}, {Config: pinned, Release: "v1.2.3"}} {
			if err := moves(t.Context(), c, s, id); !errors.Is(err, ErrMovesRefused) {
				t.Fatalf("%+v: %v", id, err)
			}
		}
		if err := moves(t.Context(), c, s, hostwrite.Identity{Release: "v1.2.3"}); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("a nil claim or a dead context", func(t *testing.T) {
		c, s := base()
		moves, _ := mpPolicy()
		if err := moves(t.Context(), nil, s, hostwrite.Identity{}); !errors.Is(err, ErrMovesRefused) {
			t.Fatal(err)
		}
		ctx, cancel := context.WithCancel(t.Context())
		cancel()
		if err := moves(ctx, c, s, hostwrite.Identity{}); !errors.Is(err, context.Canceled) {
			t.Fatal(err)
		}
		//lint:ignore SA1012 the nil context is the case under test
		if err := moves(nil, c, s, hostwrite.Identity{}); !errors.Is(err, ErrMovesRefused) { //nolint:staticcheck
			t.Fatal(err)
		}
	})
	t.Run("the refusal says what and why", func(t *testing.T) {
		c, s := base()
		s.Region = "ie"
		moves, _ := mpPolicy()
		err := moves(t.Context(), c, s, hostwrite.Identity{})
		var r *MovesPolicyRefusal
		if !errors.As(err, &r) || r.Host != mpHost || r.Op != mpOp || r.Action != "held-timers" || r.Box != mpTarget || r.Kind != hostclaim.KindLanding || r.Phase != provision.LandLocal || r.Reason == "" || err.Error() == "" {
			t.Fatalf("%#v", err)
		}
	})
}

// The held config and release are the move's pinned inputs (spec §9.1,
// §9.2): at configured the policy reads the record and grants a held-release
// identity equal to Inputs.Release and a held-config identity equal to the
// pinned etags' canonical form, and nothing else. Missing pins, a record
// that is not this landing, a read error or no reader refuse. The record is
// not read for any other action.
func TestMovesClaimPolicyBindsHeldInputsToThePins(t *testing.T) {
	at := func(action string) (*hostclaim.Claim, hostbox.Selection) {
		c := mpClaim(hostclaim.KindLanding, provision.LandConfigured)
		return c, mpSelection(c, action)
	}
	refused := func(t *testing.T, err error, reason string) {
		t.Helper()
		var r *MovesPolicyRefusal
		if err == nil || !errors.As(err, &r) || !errors.Is(err, ErrMovesRefused) {
			t.Fatalf("not the typed refusal: %v", err)
		}
		if reason != "" && r.Reason != reason {
			t.Fatalf("reason %q, want %q", r.Reason, reason)
		}
	}
	t.Run("the pinned identities are granted", func(t *testing.T) {
		moves, reader := mpPolicy()
		c, s := at("held-release")
		if err := moves(t.Context(), c, s, hostwrite.Identity{Release: "v1.2.3"}); err != nil {
			t.Fatal(err)
		}
		c, s = at("held-config")
		if err := moves(t.Context(), c, s, HeldConfigIdentity(map[string]string{"default": "e-default", "environment": "e-env"}, "e-overlay")); err != nil {
			t.Fatal(err)
		}
		if reader.reads != 2 {
			t.Fatalf("%d reads", reader.reads)
		}
	})
	t.Run("a mismatched release", func(t *testing.T) {
		moves, _ := mpPolicy()
		c, s := at("held-release")
		for _, v := range []string{"v1.2.4", "v1.2.3 ", "V1.2.3", "sha256:v1.2.3"} {
			refused(t, moves(t.Context(), c, s, hostwrite.Identity{Release: v}), "the release is not the pinned one")
		}
	})
	t.Run("a mismatched config", func(t *testing.T) {
		moves, _ := mpPolicy()
		c, s := at("held-config")
		for _, id := range []hostwrite.Identity{
			HeldConfigIdentity(map[string]string{"default": "e-other", "environment": "e-env"}, "e-overlay"),
			HeldConfigIdentity(map[string]string{"default": "e-default"}, "e-overlay"),
			HeldConfigIdentity(map[string]string{"default": "e-default", "environment": "e-env", "extra": "e"}, "e-overlay"),
			HeldConfigIdentity(map[string]string{"default": "e-default", "environment": "e-env"}, "e-other"),
			HeldConfigIdentity(map[string]string{"default": "e-default", "environment": "e-env"}, ""),
			{Config: "sha256:abc;default=true;environment=true;instance=true"},
			{Config: "etags:"},
		} {
			refused(t, moves(t.Context(), c, s, id), "the config is not the pinned one")
		}
	})
	t.Run("missing pins", func(t *testing.T) {
		for name, mutate := range map[string]func(*provision.Record){
			"no inputs":         func(r *provision.Record) { r.Move.Inputs = nil },
			"no release pinned": func(r *provision.Record) { r.Move.Inputs.Release = "" },
		} {
			t.Run(name, func(t *testing.T) {
				rec := mpRecord()
				mutate(rec)
				moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
				c, s := at("held-release")
				errRelease := moves(t.Context(), c, s, hostwrite.Identity{Release: "v1.2.3"})
				c, s = at("held-config")
				errConfig := moves(t.Context(), c, s, mpIdentity("held-config"))
				switch name {
				case "no release pinned":
					refused(t, errRelease, "no pinned release")
					if errConfig != nil {
						t.Fatalf("the config pins are intact: %v", errConfig)
					}
				default:
					refused(t, errRelease, "no pinned inputs")
					refused(t, errConfig, "no pinned inputs")
				}
			})
		}
	})
	// An app without a default bundle or an overlay pins empty etags at
	// readiness (movesheldenv.go): that is a configuration, with its own
	// identity, granted when the render names the same empty set. Missing
	// Move.Inputs is still refused.
	t.Run("an explicitly pinned empty configuration is granted", func(t *testing.T) {
		for name, pins := range map[string]*provision.MoveInputs{
			"nil etags":       {Release: "v1.2.3"},
			"empty map":       {BundleETags: map[string]string{}, Release: "v1.2.3"},
			"all-empty etags": {BundleETags: map[string]string{"default": "", "environment": ""}, OverlayETag: "", Release: "v1.2.3"},
		} {
			t.Run(name, func(t *testing.T) {
				rec := mpRecord()
				rec.Move.Inputs = pins
				moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
				c, s := at("held-config")
				empty := HeldConfigIdentity(nil, "")
				if empty.Config == "" {
					t.Fatal("the empty configuration has no identity")
				}
				if err := moves(t.Context(), c, s, empty); err != nil {
					t.Fatal(err)
				}
				if err := moves(t.Context(), c, s, HeldConfigIdentity(map[string]string{"default": "", "environment": ""}, "")); err != nil {
					t.Fatalf("all-empty etags are the empty configuration: %v", err)
				}
				refused(t, moves(t.Context(), c, s, HeldConfigIdentity(map[string]string{"default": "e-default"}, "")), "the config is not the pinned one")
				refused(t, moves(t.Context(), c, s, HeldConfigIdentity(nil, "e-overlay")), "the config is not the pinned one")
				refused(t, moves(t.Context(), c, s, hostwrite.Identity{}), "the identity does not fit the action")
				c, s = at("held-release")
				if err := moves(t.Context(), c, s, hostwrite.Identity{Release: "v1.2.3"}); err != nil {
					t.Fatal(err)
				}
			})
		}
	})
	// The encoding is structural: a value that contains the delimiters of
	// a flat encoding cannot collide with another set of etags.
	t.Run("delimiters in an etag do not collide", func(t *testing.T) {
		collisions := [][2]struct {
			etags   map[string]string
			overlay string
		}{
			{{map[string]string{"default": "a;environment=b"}, ""}, {map[string]string{"default": "a", "environment": "b"}, ""}},
			{{map[string]string{"default": "a"}, "x;environment=b"}, {map[string]string{"default": "a", "environment": "b"}, "x"}},
			{{map[string]string{"default": "a;overlay=o"}, ""}, {map[string]string{"default": "a"}, "o"}},
			{{map[string]string{"default=x": "a"}, ""}, {map[string]string{"default": "x=a"}, ""}},
			{{map[string]string{"default": `a","environment":"b`}, ""}, {map[string]string{"default": "a", "environment": "b"}, ""}},
		}
		for i, pair := range collisions {
			a := HeldConfigIdentity(pair[0].etags, pair[0].overlay)
			b := HeldConfigIdentity(pair[1].etags, pair[1].overlay)
			if a == b {
				t.Fatalf("pair %d collides: %q", i, a.Config)
			}
			rec := mpRecord()
			rec.Move.Inputs.BundleETags, rec.Move.Inputs.OverlayETag = pair[1].etags, pair[1].overlay
			moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
			c, s := at("held-config")
			refused(t, moves(t.Context(), c, s, a), "the config is not the pinned one")
			if err := moves(t.Context(), c, s, b); err != nil {
				t.Fatalf("pair %d: the pinned set refused: %v", i, err)
			}
		}
	})
	t.Run("a record read error", func(t *testing.T) {
		cause := errors.New("the store is down")
		moves := MovesClaimPolicy((&mpReader{err: cause}).read)
		for _, action := range []string{"held-release", "held-config"} {
			c, s := at(action)
			err := moves(t.Context(), c, s, mpIdentity(action))
			refused(t, err, "")
			if !errors.Is(err, cause) {
				t.Fatalf("the cause is not carried: %v", err)
			}
		}
	})
	t.Run("a record that is not this landing", func(t *testing.T) {
		for name, mutate := range map[string]func(*provision.Record) *provision.Record{
			"absent":              func(*provision.Record) *provision.Record { return nil },
			"no move":             func(r *provision.Record) *provision.Record { r.Move = nil; return r },
			"another move":        func(r *provision.Record) *provision.Record { r.Move.ID = "mv_other"; return r },
			"another incarnation": func(r *provision.Record) *provision.Record { r.Move.Incarnation = "0"; return r },
			"not landing":         func(r *provision.Record) *provision.Record { r.Move.State = provision.MoveRollingBack; return r },
			"another host":        func(r *provision.Record) *provision.Record { r.Host = "other.example"; return r },
		} {
			t.Run(name, func(t *testing.T) {
				moves := MovesClaimPolicy((&mpReader{rec: mutate(mpRecord())}).read)
				for _, action := range []string{"held-release", "held-config"} {
					c, s := at(action)
					refused(t, moves(t.Context(), c, s, mpIdentity(action)), "the record is not this landing")
				}
			})
		}
	})
	t.Run("a holder without account or app", func(t *testing.T) {
		moves, reader := mpPolicy()
		for _, action := range []string{"held-release", "held-config"} {
			c, s := at(action)
			c.Holder.App = ""
			refused(t, moves(t.Context(), c, s, mpIdentity(action)), "the holder names no account or app")
			c, s = at(action)
			c.Holder.Account = ""
			refused(t, moves(t.Context(), c, s, mpIdentity(action)), "the holder names no account or app")
		}
		if reader.reads != 0 {
			t.Fatalf("%d reads", reader.reads)
		}
	})
	t.Run("no reader", func(t *testing.T) {
		moves := MovesClaimPolicy(nil)
		for _, action := range []string{"held-release", "held-config"} {
			c, s := at(action)
			refused(t, moves(t.Context(), c, s, mpIdentity(action)), "no record reader")
		}
		// The other grants need no record.
		c := mpClaim(hostclaim.KindLanding, provision.LandLocal)
		if err := moves(t.Context(), c, mpSelection(c, "held-timers"), hostwrite.Identity{}); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("the record is read for the held config and release only", func(t *testing.T) {
		moves, reader := mpPolicy()
		for phase, action := range map[string]string{provision.LandLocal: "held-timers", provision.LandConfigured: "restore", provision.LandRestored: "install", provision.LandChecking: "spawn"} {
			c := mpClaim(hostclaim.KindLanding, phase)
			moves(t.Context(), c, mpSelection(c, action), hostwrite.Identity{})
		}
		if reader.reads != 0 {
			t.Fatalf("%d reads", reader.reads)
		}
	})
	t.Run("the canonical config identity", func(t *testing.T) {
		a := HeldConfigIdentity(map[string]string{"environment": "e-env", "default": "e-default", "instance": ""}, "e-overlay")
		b := HeldConfigIdentity(map[string]string{"default": "e-default", "environment": "e-env"}, "e-overlay")
		if a != b || a.Release != "" || a.Config == "" {
			t.Fatalf("%+v vs %+v", a, b)
		}
		// The empty configuration has one identity, whichever way it is spelt.
		e1, e2, e3 := HeldConfigIdentity(nil, ""), HeldConfigIdentity(map[string]string{}, ""), HeldConfigIdentity(map[string]string{"default": ""}, "")
		if e1.Config == "" || e1 != e2 || e2 != e3 || e1 == b {
			t.Fatalf("%+v %+v %+v", e1, e2, e3)
		}
		if HeldConfigIdentity(nil, "o") == e1 || HeldConfigIdentity(map[string]string{"default": "d"}, "") == e1 {
			t.Fatal("an overlay or an etag alone is the empty configuration")
		}
	})
}

// The held-timers import with CARLOS_HOST_CLAIM off is the registry's own
// ReplaceTimers: the rows land exactly as before, bookkeeping included,
// another host's rows untouched. With the flag on, the write runs inside the
// selected Work, or not at all.
func TestHeldTimersReplacerPassesThroughWithTheFlagOff(t *testing.T) {
	t.Parallel()
	open := func(t *testing.T) *registry.Registry {
		t.Helper()
		reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() { reg.Close() })
		for _, tm := range []registry.Timer{
			{Host: mpHost, Name: "old", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1},
			{Host: "other.example", Name: "other", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1},
		} {
			if err := reg.UpsertTimer(tm); err != nil {
				t.Fatal(err)
			}
		}
		return reg
	}
	rows := []registry.Timer{
		{Host: mpHost, Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, LastAt: 4000, LastStatus: "ok", LastMs: 12, Attempts: 2, CreatedAt: 10, DueAt: 4900, Registration: "reg-nightly"},
		{Host: mpHost, Name: "once-1", Kind: "once", Spec: "2026-10-06T00:00:00Z", Path: "/o", NextAt: 6000, CreatedAt: 11, DueAt: 6000, Registration: "reg-once"},
	}
	unchanged := func(t *testing.T, reg *registry.Registry) {
		t.Helper()
		got, _ := reg.TimersFor(mpHost)
		if len(got) != 1 || got[0].Name != "old" {
			t.Fatalf("the rows changed: %+v", got)
		}
	}
	t.Run("flag off: the registry's replacement, exactly", func(t *testing.T) {
		reg, want := open(t), open(t)
		if err := heldTimersReplacer(hostwrite.Config{}, reg)(t.Context(), mpHost, rows); err != nil {
			t.Fatal(err)
		}
		if err := want.ReplaceTimers(mpHost, rows); err != nil {
			t.Fatal(err)
		}
		got, _ := reg.TimersFor(mpHost)
		ref, _ := want.TimersFor(mpHost)
		if len(got) != 2 || !reflect.DeepEqual(got, ref) {
			t.Fatalf("through the replacer:\n%+v\nthrough the registry:\n%+v", got, ref)
		}
		if other, _ := reg.TimersFor("other.example"); len(other) != 1 {
			t.Fatalf("another host's rows: %+v", other)
		}
	})
	t.Run("flag off: a row for another host writes nothing", func(t *testing.T) {
		reg := open(t)
		bad := append([]registry.Timer{{Host: "other.example", Name: "x", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 1, CreatedAt: 1}}, rows...)
		if err := heldTimersReplacer(hostwrite.Config{}, reg)(t.Context(), mpHost, bad); err == nil {
			t.Fatal("accepted")
		}
		unchanged(t, reg)
	})
	t.Run("flag on, no selector: refused, nothing written", func(t *testing.T) {
		reg := open(t)
		if err := heldTimersReplacer(hostwrite.Config{Enabled: true}, reg)(t.Context(), mpHost, rows); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatal(err)
		}
		unchanged(t, reg)
	})
	t.Run("flag on: the write runs inside the selected held-timers Work", func(t *testing.T) {
		reg := open(t)
		sel := &mpSelector{}
		if err := heldTimersReplacer(hostwrite.Config{Enabled: true, Selector: sel}, reg)(t.Context(), mpHost, rows); err != nil {
			t.Fatal(err)
		}
		if sel.host != mpHost || sel.action != "held-timers" || sel.with != 1 || sel.identity != (hostwrite.Identity{}) {
			t.Fatalf("selection: %+v", sel)
		}
		if got, _ := reg.TimersFor(mpHost); len(got) != 2 {
			t.Fatalf("rows: %+v", got)
		}
	})
	t.Run("flag on: a refused selection writes nothing", func(t *testing.T) {
		reg := open(t)
		sel := &mpSelector{deny: errors.New("refused")}
		if err := heldTimersReplacer(hostwrite.Config{Enabled: true, Selector: sel}, reg)(t.Context(), mpHost, rows); err == nil || !errors.Is(err, sel.deny) {
			t.Fatal(err)
		}
		unchanged(t, reg)
	})
	t.Run("no registry: refused", func(t *testing.T) {
		if err := heldTimersReplacer(hostwrite.Config{}, nil)(t.Context(), mpHost, rows); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatal(err)
		}
	})
}

// mpSelector records the selection the replacer asks for and runs, or
// refuses, the write.
type mpSelector struct {
	host, action string
	with         int
	identity     hostwrite.Identity
	deny         error
}

func (s *mpSelector) Select(_ context.Context, host, action string) (hostwrite.Work, error) {
	s.host, s.action = host, action
	return s, nil
}
func (s *mpSelector) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
	s.with++
	s.identity = id
	if s.deny != nil {
		return s.deny
	}
	return fn()
}

// armPhase1 wires the landing's timers import through the replacer: with
// the flag off it is the registry's replacement, as before.
func TestArmPhase1WiresTheHeldTimersImport(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	var cfg moveCfg
	hib := activator.Config{Bucket: "data-uk", Region: "eu-west-2", Prefix: "instances"}
	armPhase1(&cfg, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
	if cfg.ReplaceTimers == nil {
		t.Fatal("no timers import")
	}
	rows := []registry.Timer{{Host: mpHost, Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, CreatedAt: 10, DueAt: 4900, Registration: "reg-nightly"}}
	if err := cfg.ReplaceTimers(t.Context(), mpHost, rows); err != nil {
		t.Fatal(err)
	}
	if got, _ := reg.TimersFor(mpHost); len(got) != 1 || got[0].Name != "nightly" || got[0].Registration != "reg-nightly" {
		t.Fatalf("rows: %+v", got)
	}
	// The flag on without a selector: the import refuses rather than
	// writing outside the claim.
	var on moveCfg
	armPhase1(&on, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{Enabled: true}, func(string) string { return "" })
	if err := on.ReplaceTimers(t.Context(), mpHost, nil); !errors.Is(err, hostwrite.ErrUnavailable) {
		t.Fatal(err)
	}
	if got, _ := reg.TimersFor(mpHost); len(got) != 1 {
		t.Fatalf("a refused import wrote: %+v", got)
	}
}

// The recovery fixtures (Plan 4 Task 3b, ruling J12): the source executing
// again under the move's op, its row held by the rollback id (restoring) or
// the move id (reverting).
const (
	rpRB    = "rb_abcdefghijklmnopqrstuvwxyz"
	rpStamp = 7 // the source installation's stamp S the flip journalled
)

// rpRecovery are the inputs the return rendered at configured: another
// bundle etag and another release than the move pinned.
func rpRecovery() *provision.MoveInputs {
	return &provision.MoveInputs{BundleETags: map[string]string{"default": "e-default-2", "environment": "e-env"}, OverlayETag: "e-overlay", Release: "v1.3.0"}
}

// rpRestoringClaim is the claim of a rollback after the return: restoring at
// phase, the source executing at S, the rollback id in the intent.
func rpRestoringClaim(phase string) *hostclaim.Claim {
	c := mpClaim(hostclaim.KindRestoring, phase)
	c.Op.Exec = mpSource
	c.Op.Intent["rollback"] = rpRB
	c.Op.Journal = map[string]string{hostclaim.JournalSourceStamp: "7", hostclaim.JournalSrcGen: "g-src"}
	c.Holder.Stamp = rpStamp
	return c
}

// rpRevertingClaim is the claim of a revert: reverting, the source
// executing, rev_phase in the journal and op.phase mirrored from probing on
// (J1); before it, the sealed moving phase the revert began from.
func rpRevertingClaim(revPhase string) *hostclaim.Claim {
	phase := hostclaim.PhaseSealed
	switch revPhase {
	case hostclaim.PhaseProbing, hostclaim.PhaseProbed, hostclaim.PhaseOpened:
		phase = revPhase
	}
	c := mpClaim(hostclaim.KindReverting, phase)
	c.Op.Exec = mpSource
	c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: revPhase, hostclaim.JournalEvidenceSource: "moves/ev"}
	return c
}

// rpSelection is the selection the source captures for that claim.
func rpSelection(c *hostclaim.Claim, action string) hostbox.Selection {
	s := mpSelection(c, action)
	s.Box, s.Region = mpSource, "ie"
	return s
}

// rpRestoringRecord is the dirty rollback with the source at phase; the
// recovery inputs are recorded from the CAS that ends configured on.
func rpRestoringRecord(phase string) *provision.Record {
	rec := mpRecord()
	rec.Move.State, rec.Move.LandingPhase = provision.MoveRollingBack, provision.LandRestored
	rec.Move.Rollback = &provision.MoveRollback{ID: rpRB, Dirty: true, TargetPhase: provision.RbTargetReturned, SourcePhase: phase}
	switch phase {
	case provision.RbSourceRestored, provision.RbSourceProbing, provision.RbSourceProbed, provision.RbSourceRecovered, provision.RbSourceOpened:
		rec.Move.Rollback.RecoveryInputs = rpRecovery()
	}
	return rec
}

// rpRevertingRecord is the revert's record: reverting, the region still the
// source's, with the inputs its probe recorded (MoveRecordRevertInputs).
func rpRevertingRecord() *provision.Record {
	rec := mpRecord()
	rec.Region, rec.Move.State, rec.Move.LandingPhase = "ie", provision.MoveReverting, ""
	rec.Move.RecoveryInputs = rpRecovery()
	return rec
}

// rpLaunch is the launch identity a probe's spawn carries (Plan 3 H4).
func rpLaunch(in *provision.MoveInputs) hostwrite.Identity {
	return hostwrite.Identity{Config: HeldConfigIdentity(in.BundleETags, in.OverlayETag).Config, Release: in.Release}
}

// rpIdentity fits the action against the recovery inputs.
func rpIdentity(action string) hostwrite.Identity {
	in := rpRecovery()
	switch action {
	case "held-config":
		return HeldConfigIdentity(in.BundleETags, in.OverlayETag)
	case "held-release":
		return hostwrite.Identity{Release: in.Release}
	case "spawn":
		return rpLaunch(in)
	}
	return hostwrite.Identity{}
}

// rpClaimPhase is the claim phase a rollback's source phase runs under: the
// claim stays at claimed through the source's install (Task 3) and mirrors
// probing, probed and opened (J1).
func rpClaimPhase(sourcePhase string) string {
	switch sourcePhase {
	case provision.RbSourceWaiting:
		return hostclaim.PhaseReleased
	case provision.RbSourceProbing, provision.RbSourceProbed, provision.RbSourceOpened:
		return sourcePhase
	case provision.RbSourceRecovered:
		return hostclaim.PhaseProbed
	}
	return hostclaim.PhaseClaimed
}

var rpSourcePhases = []string{
	provision.RbSourceWaiting, provision.RbSourceClaimed, provision.RbSourceRestoring, provision.RbSourceVerified,
	provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal,
	provision.RbSourceConfigured, provision.RbSourceRestored, provision.RbSourceProbing, provision.RbSourceProbed,
	provision.RbSourceRecovered, provision.RbSourceOpened,
}

func rpRefused(t *testing.T, err error, reason string) {
	t.Helper()
	var r *MovesPolicyRefusal
	if err == nil || !errors.As(err, &r) || !errors.Is(err, ErrMovesRefused) {
		t.Fatalf("not the typed refusal: %v", err)
	}
	if reason != "" && r.Reason != reason {
		t.Fatalf("reason %q, want %q", r.Reason, reason)
	}
}

// restoring, the source's return (J12): every action against every source
// phase of the rollback, under the claim phase that phase runs under. The
// dirty install's held writes are granted at the phases that write them
// (timers at local; config, release, restore and install at configured,
// whose closing CAS writes restored, G8 and the final review's finding 3),
// and the held config render, the release materialization and the spawn at
// probing (§7.7); nothing else, restored included. A record phase under the
// wrong claim phase is refused.
func TestMovesClaimPolicyRestoringRows(t *testing.T) {
	granted := map[string]map[string]bool{
		provision.RbSourceLocal:      {"held-timers": true},
		provision.RbSourceConfigured: {"held-config": true, "held-release": true, "restore": true, "install": true},
		provision.RbSourceProbing:    {"held-config": true, "held-release": true, "spawn": true},
	}
	grants := 0
	for _, phase := range rpSourcePhases {
		reader := &mpReader{rec: rpRestoringRecord(phase)}
		moves := MovesClaimPolicy(reader.read)
		for _, action := range mpActions {
			c := rpRestoringClaim(rpClaimPhase(phase))
			err := moves(t.Context(), c, rpSelection(c, action), rpIdentity(action))
			want := granted[phase][action]
			if (err == nil) != want {
				t.Errorf("%s/%s: err %v, want granted %v", phase, action, err, want)
				continue
			}
			if err != nil {
				var r *MovesPolicyRefusal
				if !errors.As(err, &r) || r.Kind != hostclaim.KindRestoring || r.Box != mpSource || r.Reason == "" {
					t.Errorf("%s/%s: not the typed refusal: %#v", phase, action, err)
				}
			} else {
				grants++
			}
		}
		if reader.reads == 0 || !reflect.DeepEqual(reader.args, []string{"bab", "titogo", mpHost}) {
			t.Errorf("%s: the rows are keyed by the record: %d reads %v", phase, reader.reads, reader.args)
		}
	}
	if grants != 8 {
		t.Fatalf("%d cells granted, want 8", grants)
	}
	t.Run("the wrong claim phase", func(t *testing.T) {
		for _, tc := range []struct{ record, claim, action string }{
			{provision.RbSourceLocal, hostclaim.PhaseProbing, "held-timers"},
			{provision.RbSourceConfigured, hostclaim.PhaseReleased, "held-config"},
			{provision.RbSourceConfigured, hostclaim.PhaseProbed, "install"},
			{provision.RbSourceProbing, hostclaim.PhaseClaimed, "spawn"},
			{provision.RbSourceProbing, hostclaim.PhaseProbed, "held-config"},
		} {
			moves := MovesClaimPolicy((&mpReader{rec: rpRestoringRecord(tc.record)}).read)
			c := rpRestoringClaim(tc.claim)
			rpRefused(t, moves(t.Context(), c, rpSelection(c, tc.action), rpIdentity(tc.action)), "")
		}
	})
}

// reverting, the source's pre-flip recovery (J12): the rows are keyed by the
// claim's rev_phase. Only probing grants: the held config render and the
// release materialization before each probe (§7.7), and the spawn. The same cells through hostbox.Policy.Check with the
// revert's hold, the move id.
func TestMovesClaimPolicyRevertingRows(t *testing.T) {
	reader := &mpReader{rec: rpRevertingRecord()}
	moves := MovesClaimPolicy(reader.read)
	policy := hostbox.Policy{Moves: moves, Held: func(context.Context, string) (string, error) { return mpOp, nil }}
	grants := 0
	for _, rev := range []string{hostclaim.RevPhaseEvidence, hostclaim.RevPhaseVerify, hostclaim.PhaseProbing, hostclaim.PhaseProbed, hostclaim.PhaseOpened} {
		for _, action := range mpActions {
			c := rpRevertingClaim(rev)
			s := rpSelection(c, action)
			err := moves(t.Context(), c, s, rpIdentity(action))
			want := rev == hostclaim.PhaseProbing && (action == "held-config" || action == "held-release" || action == "spawn")
			if (err == nil) != want {
				t.Errorf("%s/%s: err %v, want granted %v", rev, action, err, want)
				continue
			}
			if err != nil && !errors.Is(err, ErrMovesRefused) {
				t.Errorf("%s/%s: not the typed refusal: %v", rev, action, err)
			}
			if chk := policy.Check(t.Context(), c, s, rpIdentity(action)); (chk == nil) != want {
				t.Errorf("%s/%s through Policy.Check: %v, want granted %v", rev, action, chk, want)
			}
			if want {
				grants++
			}
		}
	}
	if grants != 3 {
		t.Fatalf("%d cells granted, want 3", grants)
	}
	t.Run("rev_phase and op.phase disagree", func(t *testing.T) {
		c := rpRevertingClaim(hostclaim.PhaseProbing)
		c.Op.Phase = hostclaim.PhaseSealed
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "spawn"), rpIdentity("spawn")), "")
		c = rpRevertingClaim(hostclaim.RevPhaseVerify)
		c.Op.Phase = hostclaim.PhaseProbing
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "spawn"), rpIdentity("spawn")), "")
	})
}

// M11 (Q11): a revert's held render, held release and spawn are bound to the
// inputs its probe recorded before rendering them (Move.RecoveryInputs,
// MoveRecordRevertInputs at probing), as a rollback's are bound to
// Rollback.RecoveryInputs: never to the move's pins, never by shape alone,
// and nothing before the record.
func TestMovesClaimPolicyRevertBindsRecordedInputs(t *testing.T) {
	recovered, pinned := rpRecovery(), mpPins()
	c := rpRevertingClaim(hostclaim.PhaseProbing)
	rec := rpRevertingRecord()
	moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
	try := func(action string, id hostwrite.Identity) error {
		return moves(t.Context(), c, rpSelection(c, action), id)
	}
	cfgOf := func(in *provision.MoveInputs) hostwrite.Identity {
		return HeldConfigIdentity(in.BundleETags, in.OverlayETag)
	}
	for action, id := range map[string]hostwrite.Identity{"held-config": cfgOf(recovered), "held-release": {Release: recovered.Release}, "spawn": rpLaunch(recovered)} {
		if err := try(action, id); err != nil {
			t.Fatalf("%s with the recorded inputs: %v", action, err)
		}
	}
	rpRefused(t, try("held-config", cfgOf(pinned)), "the config is not the recovery's")
	rpRefused(t, try("held-release", hostwrite.Identity{Release: pinned.Release}), "the release is not the recovery's")
	rpRefused(t, try("spawn", rpLaunch(pinned)), "the launch is not the recovery's")
	rpRefused(t, try("spawn", hostwrite.Identity{Config: cfgOf(recovered).Config, Release: pinned.Release}), "the launch is not the recovery's")

	// Nothing recorded: the probe records before it renders, so every row
	// refuses; the move's pins never stand in.
	rec.Move.RecoveryInputs = nil
	for action, id := range map[string]hostwrite.Identity{"held-config": cfgOf(pinned), "held-release": {Release: pinned.Release}, "spawn": rpLaunch(pinned)} {
		rpRefused(t, try(action, id), "no recorded recovery inputs")
	}
	rec.Move.RecoveryInputs = &provision.MoveInputs{BundleETags: recovered.BundleETags, OverlayETag: recovered.OverlayETag}
	rpRefused(t, try("held-release", hostwrite.Identity{Release: recovered.Release}), "no recorded recovery release")
	rpRefused(t, try("spawn", rpLaunch(recovered)), "no recorded recovery release")
}

// Only the claim op's executor, the move's source, in the source's region,
// for this rollback's record: the target (the cleaning's ReplaceTimers
// included, which is a plain registry call and never a policy row), a third
// box, a lost source and every record that is not this recovery are refused.
func TestMovesClaimPolicyRecoveryRefusesOtherRoles(t *testing.T) {
	type cell struct {
		claim  func() *hostclaim.Claim
		record func() *provision.Record
		action string
	}
	restoring := func(sourcePhase, action string) cell {
		return cell{func() *hostclaim.Claim { return rpRestoringClaim(rpClaimPhase(sourcePhase)) },
			func() *provision.Record { return rpRestoringRecord(sourcePhase) }, action}
	}
	reverting := cell{func() *hostclaim.Claim { return rpRevertingClaim(hostclaim.PhaseProbing) }, rpRevertingRecord, "spawn"}
	bases := map[string]cell{
		"restoring local":      restoring(provision.RbSourceLocal, "held-timers"),
		"restoring configured": restoring(provision.RbSourceConfigured, "held-config"),
		"restoring install":    restoring(provision.RbSourceConfigured, "install"),
		"restoring probing":    restoring(provision.RbSourceProbing, "spawn"),
		"reverting probing":    reverting,
	}
	type mutation func(c *hostclaim.Claim, s *hostbox.Selection, rec *provision.Record)
	cases := map[string]mutation{
		"the target asking, the source executing": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *provision.Record) {
			s.Box, s.Region = mpTarget, "uk"
		},
		"the target executing and asking": func(c *hostclaim.Claim, s *hostbox.Selection, _ *provision.Record) {
			c.Op.Exec = mpTarget
			s.Box, s.Region = mpTarget, "uk"
		},
		"the target executing, the source asking": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) {
			c.Op.Exec = mpTarget
		},
		"exec none": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) { c.Op.Exec = "" },
		"a third box executing and asking": func(c *hostclaim.Claim, s *hostbox.Selection, _ *provision.Record) {
			c.Op.Exec, s.Box = "de-box", "de-box"
		},
		"the source not the first participant": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) {
			c.Op.Participants = []string{mpTarget, mpSource}
		},
		"no participants": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) { c.Op.Participants = nil },
		"the intent names another source": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) {
			c.Op.Intent["source"] = "de-box"
		},
		"the intent names this box as target too": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) {
			c.Op.Intent["target"] = mpSource
		},
		"the source is lost":  func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) { c.Lost = []string{mpSource} },
		"the target's region": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *provision.Record) { s.Region = "uk" },
		"a third region":      func(_ *hostclaim.Claim, s *hostbox.Selection, _ *provision.Record) { s.Region = "de" },
		"the record names another source region": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.From = "de"
		},
		"a stale capture": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *provision.Record) { s.Captured.Stamp = 6 },
		"another move's record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.ID = "mv_other"
		},
		"another incarnation's record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.Incarnation = "0"
		},
		"another host's record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) { rec.Host = "other.example" },
		"no move":               func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) { rec.Move = nil },
		"a stuck record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.State = provision.MoveStuck
		},
		"a landing record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.State = provision.MoveLanding
		},
	}
	for baseName, base := range bases {
		for name, mutate := range cases {
			t.Run(baseName+"/"+name, func(t *testing.T) {
				c, rec := base.claim(), base.record()
				s := rpSelection(c, base.action)
				mutate(c, &s, rec)
				moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
				rpRefused(t, moves(t.Context(), c, s, rpIdentity(base.action)), "")
			})
		}
		t.Run(baseName+"/the base cell is granted", func(t *testing.T) {
			c := base.claim()
			moves := MovesClaimPolicy((&mpReader{rec: base.record()}).read)
			if err := moves(t.Context(), c, rpSelection(c, base.action), rpIdentity(base.action)); err != nil {
				t.Fatal(err)
			}
		})
	}
	restoringOnly := map[string]mutation{
		"another rollback's record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.Rollback.ID = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"
		},
		"the claim names no rollback": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *provision.Record) {
			delete(c.Op.Intent, "rollback")
		},
		"no rollback journal": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) { rec.Move.Rollback = nil },
		"a reverting record": func(_ *hostclaim.Claim, _ *hostbox.Selection, rec *provision.Record) {
			rec.Move.State = provision.MoveReverting
		},
	}
	for name, mutate := range restoringOnly {
		for _, phase := range []string{provision.RbSourceLocal, provision.RbSourceProbing} {
			t.Run("restoring "+phase+"/"+name, func(t *testing.T) {
				base := restoring(phase, map[string]string{provision.RbSourceLocal: "held-timers", provision.RbSourceProbing: "spawn"}[phase])
				c, rec := base.claim(), base.record()
				s := rpSelection(c, base.action)
				mutate(c, &s, rec)
				moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
				rpRefused(t, moves(t.Context(), c, s, rpIdentity(base.action)), "")
			})
		}
	}
	t.Run("a reverting claim over a rolling_back record", func(t *testing.T) {
		c := rpRevertingClaim(hostclaim.PhaseProbing)
		moves := MovesClaimPolicy((&mpReader{rec: rpRestoringRecord(provision.RbSourceProbing)}).read)
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "spawn"), rpIdentity("spawn")), "")
	})
	t.Run("the dirty install rows need a dirty rollback", func(t *testing.T) {
		for _, tc := range []struct{ phase, action string }{
			{provision.RbSourceLocal, "held-timers"}, {provision.RbSourceConfigured, "held-config"},
			{provision.RbSourceConfigured, "restore"}, {provision.RbSourceConfigured, "install"},
		} {
			phase, action := tc.phase, tc.action
			rec := rpRestoringRecord(phase)
			rec.Move.Rollback.Dirty = false
			c := rpRestoringClaim(hostclaim.PhaseClaimed)
			moves := MovesClaimPolicy((&mpReader{rec: rec}).read)
			rpRefused(t, moves(t.Context(), c, rpSelection(c, action), rpIdentity(action)), "")
		}
	})
	t.Run("the install runs at the source's pre-move stamp", func(t *testing.T) {
		for _, mutate := range []func(*hostclaim.Claim){
			func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalSourceStamp] = "8" },
			func(c *hostclaim.Claim) { delete(c.Op.Journal, hostclaim.JournalSourceStamp) },
		} {
			c := rpRestoringClaim(hostclaim.PhaseClaimed)
			mutate(c)
			moves := MovesClaimPolicy((&mpReader{rec: rpRestoringRecord(provision.RbSourceLocal)}).read)
			rpRefused(t, moves(t.Context(), c, rpSelection(c, "held-timers"), hostwrite.Identity{}), "")
		}
		// The probe's cold admission issues a new stamp (J14): probing is not
		// bound to S.
		c := rpRestoringClaim(hostclaim.PhaseProbing)
		c.Holder.Stamp = 12
		s := rpSelection(c, "spawn")
		moves := MovesClaimPolicy((&mpReader{rec: rpRestoringRecord(provision.RbSourceProbing)}).read)
		if err := moves(t.Context(), c, s, rpIdentity("spawn")); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("the target's cleaning timers are no policy row", func(t *testing.T) {
		for _, phase := range []string{hostclaim.PhaseReleasing, hostclaim.PhaseReleased, hostclaim.PhaseClaimed, hostclaim.PhaseProbing, hostclaim.PhaseOpened} {
			for _, exec := range []string{mpTarget, "", mpSource} {
				c := rpRestoringClaim(phase)
				c.Op.Exec = exec
				s := rpSelection(c, "held-timers")
				s.Box, s.Region = mpTarget, "uk"
				moves := MovesClaimPolicy((&mpReader{rec: rpRestoringRecord(provision.RbSourceLocal)}).read)
				rpRefused(t, moves(t.Context(), c, s, hostwrite.Identity{}), "")
			}
		}
	})
	t.Run("no reader, a read error, no account", func(t *testing.T) {
		for _, b := range []cell{restoring(provision.RbSourceLocal, "held-timers"), reverting} {
			c := b.claim()
			rpRefused(t, MovesClaimPolicy(nil)(t.Context(), c, rpSelection(c, b.action), rpIdentity(b.action)), "no record reader")
			cause := errors.New("the store is down")
			err := MovesClaimPolicy((&mpReader{err: cause}).read)(t.Context(), c, rpSelection(c, b.action), rpIdentity(b.action))
			rpRefused(t, err, "the record cannot be read")
			if !errors.Is(err, cause) {
				t.Fatalf("the cause is not carried: %v", err)
			}
			c.Holder.Account = ""
			rpRefused(t, MovesClaimPolicy((&mpReader{rec: b.record()}).read)(t.Context(), c, rpSelection(c, b.action), rpIdentity(b.action)), "the holder names no account or app")
		}
	})
	t.Run("a role refusal reads no record", func(t *testing.T) {
		reader := &mpReader{rec: rpRestoringRecord(provision.RbSourceLocal)}
		c := rpRestoringClaim(hostclaim.PhaseClaimed)
		s := rpSelection(c, "held-timers")
		s.Box, s.Region = mpTarget, "uk"
		rpRefused(t, MovesClaimPolicy(reader.read)(t.Context(), c, s, hostwrite.Identity{}), "")
		s = rpSelection(c, "held-timers")
		s.Region = "uk"
		rpRefused(t, MovesClaimPolicy(reader.read)(t.Context(), c, s, hostwrite.Identity{}), "this region is the move's target region")
		c.Lost = []string{mpSource}
		rpRefused(t, MovesClaimPolicy(reader.read)(t.Context(), c, rpSelection(c, "held-timers"), hostwrite.Identity{}), "this box is declared lost")
		if reader.reads != 0 {
			t.Fatalf("%d reads", reader.reads)
		}
	})
}

// A recovery renders the current inputs and records them as
// Rollback.RecoveryInputs; it is never bound to Move.Inputs (spec §7.7,
// J2-R2-5). The held config and release at configured are that render
// (nothing is recorded until the CAS that ends configured); from then on
// every held render and the spawn name the recorded recovery inputs. A
// revert has no recorded recovery inputs: its render and spawn are bound by
// shape and never to Move.Inputs. The same difference on a landing is
// refused.
func TestMovesClaimPolicyBindsRecoveryInputs(t *testing.T) {
	pinned, recovered := mpPins(), rpRecovery()
	try := func(phase, action string, rec *provision.Record, id hostwrite.Identity) error {
		c := rpRestoringClaim(rpClaimPhase(phase))
		return MovesClaimPolicy((&mpReader{rec: rec}).read)(t.Context(), c, rpSelection(c, action), id)
	}
	cfgOf := func(in *provision.MoveInputs) hostwrite.Identity {
		return HeldConfigIdentity(in.BundleETags, in.OverlayETag)
	}
	t.Run("configured: the render that records them", func(t *testing.T) {
		rec := rpRestoringRecord(provision.RbSourceConfigured)
		for _, in := range []*provision.MoveInputs{recovered, pinned} {
			if err := try(provision.RbSourceConfigured, "held-config", rec, cfgOf(in)); err != nil {
				t.Fatalf("%+v: %v", in, err)
			}
			if err := try(provision.RbSourceConfigured, "held-release", rec, hostwrite.Identity{Release: in.Release}); err != nil {
				t.Fatalf("%+v: %v", in, err)
			}
		}
		rpRefused(t, try(provision.RbSourceConfigured, "held-config", rec, hostwrite.Identity{Config: cfgOf(recovered).Config, Release: "v1.3.0"}), "the identity does not fit the action")
		rpRefused(t, try(provision.RbSourceConfigured, "held-release", rec, hostwrite.Identity{}), "the identity does not fit the action")
		// A replay of configured that finds them recorded is bound to them.
		rec.Move.Rollback.RecoveryInputs = rpRecovery()
		if err := try(provision.RbSourceConfigured, "held-config", rec, cfgOf(recovered)); err != nil {
			t.Fatal(err)
		}
		rpRefused(t, try(provision.RbSourceConfigured, "held-config", rec, cfgOf(pinned)), "the config is not the recovery's")
		rpRefused(t, try(provision.RbSourceConfigured, "held-release", rec, hostwrite.Identity{Release: pinned.Release}), "the release is not the recovery's")
	})
	t.Run("probing, dirty: the recorded recovery inputs, not the pins", func(t *testing.T) {
		rec := rpRestoringRecord(provision.RbSourceProbing)
		if err := try(provision.RbSourceProbing, "held-config", rec, cfgOf(recovered)); err != nil {
			t.Fatal(err)
		}
		if err := try(provision.RbSourceProbing, "spawn", rec, rpLaunch(recovered)); err != nil {
			t.Fatal(err)
		}
		rpRefused(t, try(provision.RbSourceProbing, "held-config", rec, cfgOf(pinned)), "the config is not the recovery's")
		rpRefused(t, try(provision.RbSourceProbing, "spawn", rec, rpLaunch(pinned)), "the launch is not the recovery's")
		rpRefused(t, try(provision.RbSourceProbing, "spawn", rec, hostwrite.Identity{Config: cfgOf(recovered).Config, Release: pinned.Release}), "the launch is not the recovery's")
		for _, id := range []hostwrite.Identity{{}, {Config: cfgOf(recovered).Config}, {Release: recovered.Release}} {
			rpRefused(t, try(provision.RbSourceProbing, "spawn", rec, id), "the identity does not fit the action")
		}
		rec.Move.Rollback.RecoveryInputs = nil
		rpRefused(t, try(provision.RbSourceProbing, "held-config", rec, cfgOf(recovered)), "no recorded recovery inputs")
		rpRefused(t, try(provision.RbSourceProbing, "spawn", rec, rpLaunch(recovered)), "no recorded recovery inputs")
		rec.Move.Rollback.RecoveryInputs = &provision.MoveInputs{BundleETags: recovered.BundleETags}
		rpRefused(t, try(provision.RbSourceProbing, "spawn", rec, rpLaunch(recovered)), "no recorded recovery release")
	})
	t.Run("probing, clean: the render before the record, the spawn after it", func(t *testing.T) {
		rec := rpRestoringRecord(provision.RbSourceProbing)
		rec.Move.Rollback.Dirty, rec.Move.Rollback.RecoveryInputs = false, nil
		if err := try(provision.RbSourceProbing, "held-config", rec, cfgOf(recovered)); err != nil {
			t.Fatal(err)
		}
		rpRefused(t, try(provision.RbSourceProbing, "spawn", rec, rpLaunch(recovered)), "no recorded recovery inputs")
		rec.Move.Rollback.RecoveryInputs = rpRecovery()
		if err := try(provision.RbSourceProbing, "spawn", rec, rpLaunch(recovered)); err != nil {
			t.Fatal(err)
		}
		rpRefused(t, try(provision.RbSourceProbing, "held-config", rec, cfgOf(pinned)), "the config is not the recovery's")
	})
	t.Run("reverting: the recorded inputs, never the pins (M11)", func(t *testing.T) {
		moves := MovesClaimPolicy((&mpReader{rec: rpRevertingRecord()}).read)
		c := rpRevertingClaim(hostclaim.PhaseProbing)
		if err := moves(t.Context(), c, rpSelection(c, "held-config"), cfgOf(recovered)); err != nil {
			t.Fatal(err)
		}
		if err := moves(t.Context(), c, rpSelection(c, "spawn"), rpLaunch(recovered)); err != nil {
			t.Fatal(err)
		}
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "held-config"), cfgOf(pinned)), "the config is not the recovery's")
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "spawn"), rpLaunch(pinned)), "the launch is not the recovery's")
		for _, id := range []hostwrite.Identity{{}, {Config: cfgOf(recovered).Config}, {Release: "v1.3.0"}} {
			rpRefused(t, moves(t.Context(), c, rpSelection(c, "spawn"), id), "the identity does not fit the action")
		}
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "held-config"), rpLaunch(recovered)), "the identity does not fit the action")
		if err := moves(t.Context(), c, rpSelection(c, "held-release"), hostwrite.Identity{Release: recovered.Release}); err != nil {
			t.Fatal(err)
		}
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "held-release"), hostwrite.Identity{Release: pinned.Release}), "the release is not the recovery's")
		rpRefused(t, moves(t.Context(), c, rpSelection(c, "held-release"), rpLaunch(recovered)), "the identity does not fit the action")
	})
	t.Run("a landing: the same difference is refused", func(t *testing.T) {
		moves, _ := mpPolicy()
		c := mpClaim(hostclaim.KindLanding, provision.LandConfigured)
		rpRefused(t, moves(t.Context(), c, mpSelection(c, "held-config"), cfgOf(recovered)), "the config is not the pinned one")
		rpRefused(t, moves(t.Context(), c, mpSelection(c, "held-release"), hostwrite.Identity{Release: recovered.Release}), "the release is not the pinned one")
		// A landing record carrying a rollback's recovery inputs still binds
		// to the pins.
		rec := mpRecord()
		rec.Move.Rollback = &provision.MoveRollback{ID: rpRB, RecoveryInputs: rpRecovery()}
		moves = MovesClaimPolicy((&mpReader{rec: rec}).read)
		rpRefused(t, moves(t.Context(), c, mpSelection(c, "held-config"), cfgOf(recovered)), "the config is not the pinned one")
	})
}

// End to end (Task 3b fix round 1): the source's return writes through
// hostbox.Policy.Check with its row held by the rollback id, as the J3
// hand-over leaves it, and MovesClaimPolicy as Policy.Moves. Policy.Check
// takes the hold's owner from the claim's intent.rollback; the move id's
// hold, another rollback's, the target asking and the wrong phase refuse.
func TestMovesClaimPolicyRestoringThroughPolicyCheck(t *testing.T) {
	through := func(hold string, rec *provision.Record, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
		p := hostbox.Policy{Moves: MovesClaimPolicy((&mpReader{rec: rec}).read), Held: func(context.Context, string) (string, error) { return hold, nil }}
		return p.Check(t.Context(), c, s, id)
	}
	for _, tc := range []struct{ phase, action string }{
		{provision.RbSourceLocal, "held-timers"},
		{provision.RbSourceConfigured, "held-config"},
		{provision.RbSourceConfigured, "restore"},
		{provision.RbSourceConfigured, "install"},
		{provision.RbSourceProbing, "spawn"},
	} {
		phase, action := tc.phase, tc.action
		c := rpRestoringClaim(rpClaimPhase(phase))
		s := rpSelection(c, action)
		if err := through(rpRB, rpRestoringRecord(phase), c, s, rpIdentity(action)); err != nil {
			t.Errorf("%s/%s held by the rollback: %v", phase, action, err)
		}
		for _, hold := range []string{mpOp, "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"} {
			if err := through(hold, rpRestoringRecord(phase), c, s, rpIdentity(action)); !errors.Is(err, hostbox.ErrChanged) {
				t.Errorf("%s/%s held by %s: %v", phase, action, hold, err)
			}
		}
	}
	c := rpRestoringClaim(hostclaim.PhaseClaimed)
	s := rpSelection(c, "held-timers")
	s.Box, s.Region = mpTarget, "uk"
	if err := through(rpRB, rpRestoringRecord(provision.RbSourceLocal), c, s, hostwrite.Identity{}); err == nil {
		t.Fatal("the target's timers granted under the rollback's hold")
	}
	s = rpSelection(c, "held-timers")
	if err := through(rpRB, rpRestoringRecord(provision.RbSourceConfigured), c, s, hostwrite.Identity{}); !errors.Is(err, ErrMovesRefused) {
		t.Fatalf("held-timers at configured: %v", err)
	}
	// restored is written by the CAS that ends configured (G8): the install
	// is done, and nothing is restored or installed there.
	for _, action := range []string{"restore", "install"} {
		s = rpSelection(c, action)
		if err := through(rpRB, rpRestoringRecord(provision.RbSourceRestored), c, s, hostwrite.Identity{}); !errors.Is(err, ErrMovesRefused) {
			t.Fatalf("%s at restored: %v", action, err)
		}
	}
}

// The release is materialized before every recovery probe (§7.7; Task 3b
// fix round 2), through hostbox.Policy.Check with each recovery's own hold:
// held-release at probing is granted to the executing source, bound to the
// recorded recovery inputs on a return and on a revert (M11), and
// refused at another phase, to another box, and for a release that is not
// the recorded one.
func TestMovesClaimPolicyRecoveryReleaseAtProbing(t *testing.T) {
	through := func(hold string, rec *provision.Record, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
		p := hostbox.Policy{Moves: MovesClaimPolicy((&mpReader{rec: rec}).read), Held: func(context.Context, string) (string, error) { return hold, nil }}
		return p.Check(t.Context(), c, s, id)
	}
	recovered, pinned := rpRecovery(), mpPins()
	t.Run("restoring", func(t *testing.T) {
		c := rpRestoringClaim(hostclaim.PhaseProbing)
		s := rpSelection(c, "held-release")
		rec := rpRestoringRecord(provision.RbSourceProbing)
		if err := through(rpRB, rec, c, s, hostwrite.Identity{Release: recovered.Release}); err != nil {
			t.Fatalf("the executor at probing: %v", err)
		}
		for _, v := range []string{pinned.Release, "v1.3.1", ""} {
			if err := through(rpRB, rec, c, s, hostwrite.Identity{Release: v}); err == nil {
				t.Fatalf("release %q granted against the recorded %q", v, recovered.Release)
			}
		}
		rpRefused(t, through(rpRB, rec, c, s, hostwrite.Identity{Release: pinned.Release}), "the release is not the recovery's")
		// The wrong phase: the record past probing, or the claim behind it.
		for _, tc := range []struct{ record, claim string }{
			{provision.RbSourceProbed, hostclaim.PhaseProbed},
			{provision.RbSourceProbing, hostclaim.PhaseClaimed},
			{provision.RbSourceRestored, hostclaim.PhaseClaimed},
		} {
			c := rpRestoringClaim(tc.claim)
			if err := through(rpRB, rpRestoringRecord(tc.record), c, rpSelection(c, "held-release"), hostwrite.Identity{Release: recovered.Release}); err == nil {
				t.Fatalf("record %s, claim %s: granted", tc.record, tc.claim)
			}
		}
		// The wrong box: the target asking, or a third box executing.
		t2 := rpSelection(c, "held-release")
		t2.Box, t2.Region = mpTarget, "uk"
		if err := through(rpRB, rec, c, t2, hostwrite.Identity{Release: recovered.Release}); err == nil {
			t.Fatal("the target granted")
		}
		c3 := rpRestoringClaim(hostclaim.PhaseProbing)
		c3.Op.Exec = "de-box"
		s3 := rpSelection(c3, "held-release")
		s3.Box = "de-box"
		if err := through(rpRB, rec, c3, s3, hostwrite.Identity{Release: recovered.Release}); err == nil {
			t.Fatal("a third box granted")
		}
		// A clean return materializes before its inputs are recorded; a dirty
		// one needs them.
		clean := rpRestoringRecord(provision.RbSourceProbing)
		clean.Move.Rollback.Dirty, clean.Move.Rollback.RecoveryInputs = false, nil
		if err := through(rpRB, clean, c, s, hostwrite.Identity{Release: "v9"}); err != nil {
			t.Fatalf("a clean return before the record: %v", err)
		}
		dirty := rpRestoringRecord(provision.RbSourceProbing)
		dirty.Move.Rollback.RecoveryInputs = nil
		rpRefused(t, through(rpRB, dirty, c, s, hostwrite.Identity{Release: recovered.Release}), "no recorded recovery inputs")
	})
	t.Run("reverting", func(t *testing.T) {
		c := rpRevertingClaim(hostclaim.PhaseProbing)
		s := rpSelection(c, "held-release")
		rec := rpRevertingRecord()
		if err := through(mpOp, rec, c, s, hostwrite.Identity{Release: recovered.Release}); err != nil {
			t.Fatalf("the executor at probing: %v", err)
		}
		rpRefused(t, through(mpOp, rec, c, s, hostwrite.Identity{Release: pinned.Release}), "the release is not the recovery's")
		rpRefused(t, through(mpOp, rec, c, s, hostwrite.Identity{Release: recovered.Release, Config: "x"}), "the identity does not fit the action")
		for _, rev := range []string{hostclaim.RevPhaseEvidence, hostclaim.RevPhaseVerify, hostclaim.PhaseProbed, hostclaim.PhaseOpened} {
			c := rpRevertingClaim(rev)
			if err := through(mpOp, rec, c, rpSelection(c, "held-release"), hostwrite.Identity{Release: recovered.Release}); err == nil {
				t.Fatalf("rev_phase %s: granted", rev)
			}
		}
		t2 := rpSelection(c, "held-release")
		t2.Box, t2.Region = mpTarget, "uk"
		if err := through(mpOp, rec, c, t2, hostwrite.Identity{Release: recovered.Release}); err == nil {
			t.Fatal("the target granted")
		}
		// A revert's row is held by the move id, never a rollback's.
		if err := through(rpRB, rec, c, s, hostwrite.Identity{Release: recovered.Release}); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("a rollback hold on a revert: %v", err)
		}
	})
}