package host

import (
	"context"
	"encoding/json"
	"errors"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

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

// The moves park providers (Plan 5 Task 9a): the view, the claim's lease set
// over moves' scopes, the wake-source gate (Q1), the registration and the
// arming from the claim's MoveParks.

const (
	vwHost = "vw.example"
	vwAcct = "acct"
	vwApp  = "app"
	vwMove = "mv_vwvwvwvwvwvwvwvwvwvwvwvwvw"
	vwRb   = "rb_vwvwvwvwvwvwvwvwvwvwvwvwvw"
)

var vwCreated = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)

// vwFixture is one host's claim observation, its record and its data
// bucket, for the view.
type vwFixture struct {
	inc, location string
	rec           *provision.Record
	bucket        *leasetest.VersionedBucket
	reads, lists  int
}

func newVwFixture(t *testing.T) *vwFixture {
	t.Helper()
	inc, err := hostclaim.Incarnation(vwAcct, vwApp, vwHost, vwCreated)
	if err != nil {
		t.Fatal(err)
	}
	f := &vwFixture{inc: inc, location: "instances/" + vwHost + "/" + inc, bucket: leasetest.NewVersionedBucket()}
	f.bucket.AddBucket("data")
	f.rec = &provision.Record{Schema: 1, Host: vwHost, App: vwApp, Account: vwAcct, CreatedAt: vwCreated, Region: "ie"}
	return f
}

func (f *vwFixture) records() MovesRecordReader {
	return func(_ context.Context, acct, app, host string) (*provision.Record, error) {
		f.reads++
		if acct != vwAcct || app != vwApp || host != vwHost || f.rec == nil {
			return nil, errors.New("no such record")
		}
		cp := *f.rec
		return &cp, nil
	}
}

// countingReader counts the view's listings, to show a non-authoritative
// region reads nothing.
type countingReader struct {
	*leasetest.VersionedBucket
	f *vwFixture
}

func (r countingReader) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
	r.f.lists++
	return r.VersionedBucket.ListVersions(ctx, bucket, prefix)
}

func (f *vwFixture) view(region string) func(context.Context, hostclaim.Observed) (park.View, error) {
	return movesParkView(f.records(), countingReader{f.bucket, f}, "data", region)
}

// holder is the fixture's holder at generation gen with admissions, each
// "generation@region".
func (f *vwFixture) holder(gen string, admissions ...string) *hostclaim.Holder {
	h := &hostclaim.Holder{Family: "instance", Key: provision.RecordKey(vwAcct, vwApp, vwHost), Account: vwAcct, App: vwApp,
		Incarnation: f.inc, CreatedAt: vwCreated, Location: f.location, Generation: gen, Stamp: 3, Session: 3}
	for i, a := range admissions {
		g, r, _ := strings.Cut(a, "@")
		h.Admissions = append(h.Admissions, hostclaim.AdmissionRecord{Session: uint64(i + 1), Box: r + "-box", Generation: g, Mode: "cold", Region: r})
	}
	return h
}

func vwObs(c *hostclaim.Claim) hostclaim.Observed {
	c.Host = vwHost
	return hostclaim.Observed{Host: vwHost, ETag: "etag-1", Claim: c}
}

func (f *vwFixture) move(state string) {
	f.rec.Move = &provision.MoveState{ID: vwMove, Incarnation: f.inc, From: "ie", To: "uk", State: state}
}

// marker writes generation.json at the location naming gen, pinning a
// snapshot of it, and returns the marker's version and the snapshot's.
func (f *vwFixture) marker(t *testing.T, gen, inc string) (markerVersion, snapshot string) {
	t.Helper()
	snapKey := f.location + "/g-" + gen + "/0009/0000000000000001-0000000000000005.ltx"
	snapshot = f.bucket.Put("data", snapKey, []byte("snapshot"))
	body, err := json.Marshal(map[string]any{"op": gen, "incarnation": inc, "host": vwHost, "box": "uk-box", "installed_at": vwCreated,
		"manifest_sha256": "x", "snapshot_key": snapKey, "snapshot_version": snapshot, "snapshot_txid": 5, "snapshot_crc64": 1})
	if err != nil {
		t.Fatal(err)
	}
	return f.bucket.Put("data", f.location+"/generation.json", body), snapshot
}

// TestMoveParkViewFollowsTheAuthoritativeRegionTable: one row per claim
// state of the spec's table (§5.1), seen from a region that is neither, so
// only the authoritative region is asked of the view, plus the refusals.
func TestMoveParkViewFollowsTheAuthoritativeRegionTable(t *testing.T) {
	type row struct {
		name  string
		claim func(f *vwFixture) *hostclaim.Claim
		rec   func(f *vwFixture)
		want  string // "" wants a refusal
	}
	op := func(kind, phase string) hostclaim.Op {
		return hostclaim.Op{Kind: kind, ID: vwMove, Phase: phase, Exec: "box"}
	}
	rows := []row{
		{"free", func(f *vwFixture) *hostclaim.Claim { return &hostclaim.Claim{} }, nil, movesNoLiveRegion},
		{"creating", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(""), Op: hostclaim.Op{Kind: hostclaim.KindCreating, ID: "op"}}
		}, nil, movesNoLiveRegion},
		{"deleting", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindDeleting, ID: "op"}}
		}, nil, movesNoLiveRegion},
		{"parking", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindParking, ID: "op"}}
		}, nil, movesNoLiveRegion},
		{"purging", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindPurging, ID: "op"}}
		}, nil, movesNoLiveRegion},
		{"serving: the admission's region", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "old@ie", "cur@uk"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "uk-box"}}
		}, nil, "uk"},
		{"serving: a legacy holder takes the record's region", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(""), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}
		}, nil, "ie"},
		{"serving: no admission and no record region", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(""), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}
		}, func(f *vwFixture) { f.rec.Region = "" }, ""},
		{"pushing", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@uk"), Op: hostclaim.Op{Kind: hostclaim.KindPushing, ID: "push"}}
		}, nil, "uk"},
		{"moving before the flip", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: op(hostclaim.KindMoving, hostclaim.PhaseMoving)}
		}, func(f *vwFixture) { f.move(provision.MoveCutting) }, "ie"},
		{"reverting", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: op(hostclaim.KindReverting, "")}
		}, func(f *vwFixture) { f.move(provision.MoveReverting) }, "ie"},
		{"landing before the target's probe is admitted", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: op(hostclaim.KindLanding, provision.LandBuilding)}
		}, func(f *vwFixture) { f.move(provision.MoveLanding) }, "ie"},
		{"landing from the target's probe's admission", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(vwMove, "cur@ie", vwMove+"@uk"), Op: op(hostclaim.KindLanding, provision.LandProbing)}
		}, func(f *vwFixture) { f.move(provision.MoveLanding) }, "uk"},
		{"landing committed", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("fresh2", "cur@ie", vwMove+"@uk", "fresh2@uk"), Op: op(hostclaim.KindLanding, hostclaim.PhaseCommitted)}
		}, func(f *vwFixture) { f.move(provision.MoveCommitted) }, "uk"},
		{"landing committed with no admission in the target", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: op(hostclaim.KindLanding, hostclaim.PhaseCommitted)}
		}, func(f *vwFixture) { f.move(provision.MoveCommitted) }, ""},
		{"landing closed record first (M4)", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(vwMove, "cur@ie", vwMove+"@uk"), Op: op(hostclaim.KindLanding, hostclaim.PhaseCommitted)}
		}, func(f *vwFixture) {
			f.rec.LastMove = &provision.MoveOutcome{ID: vwMove, Outcome: "done", At: vwCreated}
		}, "uk"},
		{"restoring, clean, from the return CAS", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie", vwMove+"@uk"), Op: op(hostclaim.KindRestoring, hostclaim.PhaseClaimed)}
		}, func(f *vwFixture) { f.move(provision.MoveRollingBack) }, "ie"},
		{"restoring, dirty, before the source's probe", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(vwMove, "cur@ie", vwMove+"@uk"), Op: op(hostclaim.KindRestoring, hostclaim.PhaseClaimed)}
		}, func(f *vwFixture) { f.move(provision.MoveRollingBack) }, "uk"},
		{"restoring, dirty, from the source's probe's admission", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(vwRb, "cur@ie", vwMove+"@uk", vwRb+"@ie"), Op: op(hostclaim.KindRestoring, hostclaim.PhaseProbing)}
		}, func(f *vwFixture) { f.move(provision.MoveRollingBack) }, "ie"},
		{"restoring before any admission of the generation", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder(""), Op: op(hostclaim.KindRestoring, hostclaim.PhaseReleasing)}
		}, func(f *vwFixture) { f.move(provision.MoveRollingBack) }, "ie"},
		{"moving with the generation admitted in the target", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@uk"), Op: op(hostclaim.KindMoving, hostclaim.PhaseMoving)}
		}, func(f *vwFixture) { f.move(provision.MoveCutting) }, ""},
		{"a move op the record does not carry", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: op(hostclaim.KindLanding, provision.LandBuilding)}
		}, nil, ""},
		{"placing", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindPlacing, ID: "op"}}
		}, nil, ""},
		{"relocating", func(f *vwFixture) *hostclaim.Claim {
			return &hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: "relocating", ID: "op"}}
		}, nil, ""},
	}
	for _, r := range rows {
		t.Run(r.name, func(t *testing.T) {
			f := newVwFixture(t)
			if r.rec != nil {
				r.rec(f)
			}
			obs := vwObs(r.claim(f))
			v, err := f.view("us")(t.Context(), obs)
			if r.want == "" {
				if err == nil || !errors.Is(err, park.ErrUnsafe) {
					t.Fatalf("view %+v, %v; want a refusal", v, err)
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if v.AuthoritativeRegion != r.want || v.Region != "us" || v.ClaimETag != obs.ETag {
				t.Fatalf("view %+v; want %s authoritative, seen from us", v, r.want)
			}
			if v.MarkerKnown || f.lists != 0 {
				t.Fatalf("a non-authoritative region read the marker (%d listings)", f.lists)
			}
			if obs.Claim.Holder != nil && v.GenerationOf == nil {
				t.Fatal("no classifier for the holder's location")
			}
		})
	}
}

// TestMoveParkViewReadsTheMarkerByVersion: in the authoritative region the
// view reads generation.json by the version the listing names and pins the
// snapshot it names (the installed root); absent is the origin; a marker of
// another incarnation refuses.
func TestMoveParkViewReadsTheMarkerByVersion(t *testing.T) {
	serving := func(f *vwFixture) hostclaim.Observed {
		return vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@uk"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "uk-box"}})
	}
	t.Run("present", func(t *testing.T) {
		f := newVwFixture(t)
		mv, snap := f.marker(t, "cur", f.inc)
		v, err := f.view("uk")(t.Context(), serving(f))
		if err != nil {
			t.Fatal(err)
		}
		root := park.Ref{Key: f.location + "/g-cur/0009/0000000000000001-0000000000000005.ltx", VersionID: snap}
		if !v.MarkerKnown || v.MarkerGeneration != "cur" || v.Marker != (park.Ref{Key: f.location + "/generation.json", VersionID: mv}) || !v.Pinned[root] || len(v.Pinned) != 1 {
			t.Fatalf("view %+v", v)
		}
		if g, ok := v.GenerationOf(root); !ok || g != "cur" {
			t.Fatalf("the root classifies as %q %v", g, ok)
		}
	})
	t.Run("absent is the origin", func(t *testing.T) {
		f := newVwFixture(t)
		v, err := f.view("uk")(t.Context(), serving(f))
		if err != nil || !v.MarkerKnown || v.MarkerGeneration != "" || v.Marker != (park.Ref{}) || len(v.Pinned) != 0 {
			t.Fatalf("view %+v, %v", v, err)
		}
	})
	t.Run("another incarnation's marker", func(t *testing.T) {
		f := newVwFixture(t)
		f.marker(t, "cur", strings.Repeat("e", 64))
		if v, err := f.view("uk")(t.Context(), serving(f)); !errors.Is(err, park.ErrUnsafe) {
			t.Fatalf("view %+v, %v", v, err)
		}
	})
	t.Run("a deleted marker", func(t *testing.T) {
		f := newVwFixture(t)
		f.marker(t, "cur", f.inc)
		f.bucket.BareDelete("data", f.location+"/generation.json")
		if v, err := f.view("uk")(t.Context(), serving(f)); !errors.Is(err, park.ErrUnsafe) {
			t.Fatalf("view %+v, %v", v, err)
		}
	})
}

// TestMovesScopesResolveThroughTheClaimsLeaseSet (M6): the claim's lease set
// (hostbox.ReplicaParkLeaseSet, the resolver of the MoveParks engine's
// AggregateLeases) answers every scope a moves park names with that scope's
// own lock, and acquired through the real AggregateLeases it takes exactly
// that lock. The scopes are moves' own: the holder location, the residue
// park's, the lane and the op's lane.
func TestMovesScopesResolveThroughTheClaimsLeaseSet(t *testing.T) {
	inc := strings.Repeat("a", 64)
	holder := hostclaim.Holder{Location: "instances/" + vwHost + "/" + inc}
	scopes := []string{
		"instances/" + vwHost,
		strings.TrimSuffix(holder.Location, "/"),
		moves.ResidueScope(holder, vwCreated).Location,
		"moves/" + vwHost,
		moves.ScopeLocation(vwHost, vwMove),
	}
	for _, scope := range scopes {
		names, err := hostbox.ReplicaParkLeaseSet(t.Context(), "data-ie", scope)
		if err != nil || len(names) != 1 || names[0] != scope {
			t.Fatalf("%s: %v %v", scope, names, err)
		}
		leases := &fakeParkLeases{}
		held, err := hostbox.AggregateLeases{Leases: leases, Resolve: hostbox.ReplicaParkLeaseSet}.Acquire(t.Context(), "data-ie", scope)
		if err != nil {
			t.Fatalf("%s: %v", scope, err)
		}
		if got := leases.holding(); len(got) != 1 || got[0] != "data-ie/"+scope {
			t.Fatalf("%s holds %v", scope, got)
		}
		held.Close()
	}
	for _, scope := range []string{"", "instances", "instances/" + vwHost + "/g-cur", "instances/" + vwHost + "/" + strings.ToUpper(inc),
		"moves/" + vwHost + "/x/y", "parked/" + vwHost, "instances/../" + vwHost} {
		if names, err := hostbox.ReplicaParkLeaseSet(t.Context(), "data-ie", scope); err == nil {
			t.Fatalf("%q answered %v", scope, names)
		}
	}
}

// withMovesWakeSource empties the claim's move-source slots and the moves
// gate's own record for the test, restoring all three after it. register
// then registers a stand-in source through registerMovesWakeSource, which
// opens the gate (movesWakeSourceRegistered).
func withMovesWakeSource(t *testing.T, register bool) {
	t.Helper()
	claimMoveSources.Lock()
	oldSource, oldParks := claimMoveSources.Source, claimMoveSources.Parks
	claimMoveSources.Source, claimMoveSources.Parks = nil, nil
	claimMoveSources.Unlock()
	movesWakeSource.Lock()
	oldReg := movesWakeSource.registered
	movesWakeSource.registered = nil
	movesWakeSource.Unlock()
	t.Cleanup(func() {
		claimMoveSources.Lock()
		claimMoveSources.Source, claimMoveSources.Parks = oldSource, oldParks
		claimMoveSources.Unlock()
		movesWakeSource.Lock()
		movesWakeSource.registered = oldReg
		movesWakeSource.Unlock()
	})
	if register {
		if err := registerMovesWakeSource(&parkRegistrationWake{}); err != nil {
			t.Fatal(err)
		}
	}
}

// renewTable is a lease table behind hostbox.RenewingLeases' Open: one
// holder per bucket/scope, a second open of a held one refused as
// lease.Acquire refuses it. Everything above Open (RenewingLeases,
// AggregateLeases, AcquireHeld, LeaseScope) is the claim's own.
type renewTable struct {
	mu    sync.Mutex
	held  map[string]bool
	opens []string
}

type renewHeld struct {
	t    *renewTable
	key  string
	lost chan struct{}
}

func (r *renewTable) open(_ context.Context, bucket, scope, _ string) (hostbox.Renewable, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	key := bucket + "/" + scope
	if r.held[key] {
		return nil, &lease.HeldError{Owner: "another holder"}
	}
	if r.held == nil {
		r.held = map[string]bool{}
	}
	r.held[key] = true
	r.opens = append(r.opens, key)
	return &renewHeld{t: r, key: key, lost: make(chan struct{})}, nil
}

func (r *renewTable) holding() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	var out []string
	for k, on := range r.held {
		if on {
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out
}

func (h *renewHeld) Err() error {
	h.t.mu.Lock()
	defer h.t.mu.Unlock()
	if !h.t.held[h.key] {
		return errors.New("renew table: released")
	}
	return nil
}
func (h *renewHeld) Lost() <-chan struct{} { return h.lost }
func (h *renewHeld) Release(context.Context) error {
	h.t.mu.Lock()
	defer h.t.mu.Unlock()
	delete(h.t.held, h.key)
	return nil
}

// leases is the claim's park leases over the table, as Build assembles
// them: RenewingLeases under AggregateLeases with the replica park lease set.
func (r *renewTable) leases() hostbox.AggregateLeases {
	return hostbox.AggregateLeases{Leases: hostbox.RenewingLeases{Region: "eu-west-1", Open: r.open}, Resolve: hostbox.ReplicaParkLeaseSet}
}

// ownLease is the build's own-location lease over the table, as
// armPhase1's OwnLease takes it (RenewingLeases.Acquire).
func (r *renewTable) ownLease(ctx context.Context, bucket, location string) (park.HeldLease, error) {
	return hostbox.RenewingLeases{Region: "eu-west-1", Open: r.open}.Acquire(ctx, bucket, location)
}

// TestArmMovesPassRefusesAMismatchedBucketOrRegion (M17): the arms take the
// claim's engine only when its bucket and region are this box's own; any
// refusal sets none of the hooks.
func TestArmMovesPassRefusesAMismatchedBucketOrRegion(t *testing.T) {
	withMovesWakeSource(t, true)
	view := func(context.Context, hostclaim.Observed) (park.View, error) { return park.View{}, nil }
	auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }
	providers := func(bucket, region string) ClaimProviders {
		return ClaimProviders{moveParks: &claimMoveParks{
			local:  claimMoveLocalAuthorizer(func(context.Context) error { return nil }, auth),
			engine: park.Engine{Bucket: bucket, Region: region, Leases: (&renewTable{}).leases(), View: view, Authorize: auth},
			leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}}
	}
	unarmed := func(t *testing.T, c moveCfg) {
		t.Helper()
		if c.ParkLeases != nil || c.ParkView != nil || c.ParkAuthorize != nil || c.ParkRunHeld != nil || c.ParkLocalAuthorize != nil {
			t.Fatal("a refused arming set a hook")
		}
	}
	for name, p := range map[string]ClaimProviders{"another bucket": providers("data-uk", "ie"), "another region": providers("data-ie", "uk")} {
		t.Run(name, func(t *testing.T) {
			c := moveCfg{OwnBucket: "data-ie", Region: "ie"}
			if err := armMovesParks(t.Context(), &c, p); !errors.Is(err, hostbox.ErrChanged) {
				t.Fatalf("armed: %v", err)
			}
			unarmed(t, c)
		})
	}
	t.Run("no MoveParks", func(t *testing.T) {
		c := moveCfg{OwnBucket: "data-ie", Region: "ie"}
		if err := armMovesParks(t.Context(), &c, ClaimProviders{}); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("armed: %v", err)
		}
		unarmed(t, c)
	})
	t.Run("this box's own engine", func(t *testing.T) {
		c := moveCfg{OwnBucket: "data-ie", Region: "ie"}
		if err := armMovesParks(t.Context(), &c, providers("data-ie", "ie")); err != nil {
			t.Fatal(err)
		}
		if c.ParkLeases == nil || c.ParkView == nil || c.ParkAuthorize == nil || c.ParkRunHeld == nil || c.ParkLocalAuthorize == nil {
			t.Fatalf("hooks unset: %+v", c)
		}
	})
}

// TestArmMovesParksTakesRunHeldAndTheLocalAuthorizerFromTheClaim (M5, M9):
// the arm's RunHeld is park.Engine.RunHeld over the claim's own leases, so a
// residue park borrows the build's own-location lease through the claim's
// AggregateLeases (AcquireHeld, LeaseScope) where Engine.Run would wait on
// it; and its local authorizer is the claim's MoveParks one, which asks the
// claim's identity check and owner and refuses anything but a park-local. A
// claim engine whose leases cannot lend a held lease does not arm.
func TestArmMovesParksTakesRunHeldAndTheLocalAuthorizerFromTheClaim(t *testing.T) {
	withMovesWakeSource(t, true)
	ctx := t.Context()
	inc := strings.Repeat("a", 64)
	loc := "instances/" + vwHost + "/" + inc
	now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
	claims := release.NewDirStore(t.TempDir())
	if _, err := hostclaim.Update(ctx, claims, hostclaim.Observed{Host: vwHost, Absent: true}, func(c *hostclaim.Claim) error {
		c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op", Intent: map[string]string{"park_sweep": "v1"}}
		c.Obligations = []hostclaim.Obligation{{ID: "chore", Op: "op", Kind: "park-replica", Region: "ie", Owner: "region:ie", Retirement: "ret",
			Park: &hostclaim.ParkScope{Location: loc, Incarnation: inc, EvidenceAt: now.Add(-time.Hour), RequestLimit: hostbox.RequestLimit}}}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	data := leasetest.NewVersionedBucket()
	data.AddBucket("data-ie")
	data.Put("data-ie", loc+"/db", []byte("residue"))
	table := &renewTable{}
	view := func(_ context.Context, o hostclaim.Observed) (park.View, error) {
		return park.View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: movesNoLiveRegion}, nil
	}
	authorized := 0
	auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { authorized++; return nil }
	owned, identity := 0, error(nil)
	owner := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { owned++; return nil }
	providers := func(leases park.Leases) ClaimProviders {
		return ClaimProviders{moveParks: &claimMoveParks{
			local:  claimMoveLocalAuthorizer(func(context.Context) error { return identity }, owner),
			engine: park.Engine{Bucket: "data-ie", Region: "ie", Leases: leases, View: view, Authorize: auth},
			leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}}
	}

	t.Run("leases that cannot lend a held lease", func(t *testing.T) {
		c := moveCfg{OwnBucket: "data-ie", Region: "ie"}
		if err := armMovesParks(ctx, &c, providers(&fakeParkLeases{})); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("armed over leases with no AcquireHeld: %v", err)
		}
		if parkHooksArmed(c) || c.ParkRunHeld != nil || c.ParkLocalAuthorize != nil {
			t.Fatal("a refused arming set a hook")
		}
	})

	c := moveCfg{OwnBucket: "data-ie", Region: "ie"}
	if err := armMovesParks(ctx, &c, providers(table.leases())); err != nil {
		t.Fatal(err)
	}
	own, err := table.ownLease(ctx, c.OwnBucket, loc)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(own.Close)
	// The engine as an install builds it (installRun.engine): this box's
	// store and bucket, the armed leases, view and authorizer.
	eng := park.Engine{Store: data, Claims: claims, Bucket: c.OwnBucket, Region: c.parkRegion(), Leases: c.ParkLeases,
		Now: func() time.Time { return now }, NewID: func() string { return "next" }, View: c.ParkView, Authorize: c.ParkAuthorize}
	w := park.Work{Host: vwHost, ID: "chore", Op: "op"}
	if err := eng.Run(ctx, w); err == nil {
		t.Fatal("Engine.Run took the location's lock while the build holds it")
	}
	if err := c.ParkRunHeld(ctx, eng, w, own); err != nil && !errors.Is(err, park.ErrPending) {
		t.Fatal(err)
	}
	if versions, err := data.ListVersions(ctx, "data-ie", loc+"/"); err != nil || len(versions) != 0 {
		t.Fatalf("the residue is still there: %v %v", versions, err)
	}
	if len(table.opens) != 1 || authorized == 0 {
		t.Fatalf("lease opens %v, authorized %d: the park did not borrow the build's lease under the claim's authorizer", table.opens, authorized)
	}
	if err := own.Check(ctx); err != nil {
		t.Fatalf("the run released the build's own lease: %v", err)
	}

	local := hostclaim.Obligation{ID: "local", Op: "op", Kind: "park-local", Box: "ie-box"}
	if err := c.ParkLocalAuthorize(ctx, &hostclaim.Claim{Host: vwHost}, local); err != nil || owned != 1 {
		t.Fatalf("the claim's owner was not asked: %v (%d)", err, owned)
	}
	if err := c.ParkLocalAuthorize(ctx, &hostclaim.Claim{Host: vwHost}, hostclaim.Obligation{ID: "chore", Op: "op", Kind: "park-replica"}); err == nil || owned != 1 {
		t.Fatalf("a park-replica passed the local authorizer: %v", err)
	}
	identity = hostbox.ErrChanged
	if err := c.ParkLocalAuthorize(ctx, &hostclaim.Claim{Host: vwHost}, local); !errors.Is(err, hostbox.ErrChanged) || owned != 1 {
		t.Fatalf("the local authorizer skipped the claim's identity check: %v", err)
	}
}

// TestMovesParksRefuseUntilTheWakeSourceRegisters (Q1): one gate,
// movesWakeSourceRegistered, stands between the moves parks and production.
// Until this box's moves source is registered with the claim
// (registerMovesWakeSource, RegisterProductionClaimMoveSources), the
// registered view refuses with a named dependency, reading nothing, so the
// claim's sweep queues and runs nothing through it, and armMovesParks sets no
// hook. Registering the park sources alone, or another source, does not
// open it. Task 5's evidence is registered as it is, reachable as the
// sweep verifier.
func TestMovesParksRefuseUntilTheWakeSourceRegisters(t *testing.T) {
	const contract = "moves wake owners and C8 files manifest provider (Plan 5 Task 9b)"
	f := newVwFixture(t)
	evidence := movesRetirementEvidence{Records: f.records(), Now: time.Now}
	obs := vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}})
	auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }
	providers := ClaimProviders{moveParks: &claimMoveParks{
		local:  claimMoveLocalAuthorizer(func(context.Context) error { return nil }, auth),
		engine: park.Engine{Bucket: "data", Region: "ie", Leases: (&renewTable{}).leases(), View: f.view("ie"), Authorize: auth},
		leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}}
	shut := func(t *testing.T, src movesParkSources) {
		t.Helper()
		var dep *ClaimDependencyError
		if _, err := src.View(t.Context(), obs); !errors.As(err, &dep) || !strings.Contains(err.Error(), contract) {
			t.Fatalf("the view answered before the wake source registered: %v", err)
		}
		if f.reads != 0 || f.lists != 0 {
			t.Fatal("a refused view read the record or the bucket")
		}
		c := moveCfg{OwnBucket: "data", Region: "ie"}
		if err := armMovesParks(t.Context(), &c, providers); !errors.As(err, &dep) || !strings.Contains(err.Error(), contract) {
			t.Fatalf("armed before the wake source registered: %v", err)
		}
		if parkHooksArmed(c) || c.ParkRunHeld != nil || c.ParkLocalAuthorize != nil {
			t.Fatal("a refused arming set a hook")
		}
	}

	t.Run("the park sources alone", func(t *testing.T) {
		withMovesWakeSource(t, false)
		src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence)
		if err != nil {
			t.Fatal(err)
		}
		if _, ok := src.Evidence.(park.MoveSweepVerifier); !ok {
			t.Fatal("the registered evidence is not the sweep verifier")
		}
		shut(t, src)
		if err := registerMovesParks(src); err != nil {
			t.Fatal(err)
		}
		if p := claimMoveSources.Parks; p == nil || p.View == nil || p.Evidence == nil {
			t.Fatalf("registered %+v", p)
		}
		if err := registerMovesParks(src); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("a second registration: %v", err)
		}
		shut(t, src)
		if _, err := claimMoveSources.Parks.View(t.Context(), obs); err == nil {
			t.Fatal("the claim's registered park view answered before the wake source registered")
		}
	})

	t.Run("another source registered with the claim", func(t *testing.T) {
		withMovesWakeSource(t, false)
		src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence)
		if err != nil {
			t.Fatal(err)
		}
		if err := RegisterProductionClaimMoveSources(&parkRegistrationWake{}); err != nil {
			t.Fatal(err)
		}
		shut(t, src)
		if err := registerMovesWakeSource(&parkRegistrationWake{}); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("registered over another source: %v", err)
		}
		shut(t, src)
	})

	t.Run("this box's moves source", func(t *testing.T) {
		withMovesWakeSource(t, false)
		src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence)
		if err != nil {
			t.Fatal(err)
		}
		if err := registerMovesParks(src); err != nil {
			t.Fatal(err)
		}
		if err := registerMovesWakeSource(nil); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("a nil source: %v", err)
		}
		shut(t, src)
		if err := registerMovesWakeSource(&parkRegistrationWake{}); err != nil {
			t.Fatal(err)
		}
		if err := movesWakeSourceRegistered(); err != nil {
			t.Fatalf("the gate: %v", err)
		}
		if v, err := src.View(t.Context(), obs); err != nil || v.AuthoritativeRegion != "ie" {
			t.Fatalf("view %+v, %v", v, err)
		}
		if v, err := claimMoveSources.Parks.View(t.Context(), obs); err != nil || v.AuthoritativeRegion != "ie" {
			t.Fatalf("the claim's park view is not the moves view: %+v, %v", v, err)
		}
		c := moveCfg{OwnBucket: "data", Region: "ie"}
		if err := armMovesParks(t.Context(), &c, providers); err != nil || !parkHooksArmed(c) {
			t.Fatalf("not armed: %v", err)
		}
		if err := registerMovesWakeSource(&parkRegistrationWake{}); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("a second registration: %v", err)
		}
		if err := movesWakeSourceRegistered(); err != nil {
			t.Fatalf("a refused second registration shut the gate: %v", err)
		}
	})
}

// TestCommittedLaneViewNamesTheTargetWithItsMarker (the Task 3 carry): after
// the commit the target's lane park runs under the real view, which names
// the target's region authoritative and supplies the marker the target's
// probe published, read by its exact version; the lane is parked.
func TestCommittedLaneViewNamesTheTargetWithItsMarker(t *testing.T) {
	t.Parallel()
	h := newCloseHarness(t)
	ctx := context.Background()
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/") // the cut's holder, at the bare location
	body, err := json.Marshal(map[string]any{"op": p1ID, "incarnation": h.inc, "host": p1Host, "box": "uk-box", "installed_at": h.clock.Now(),
		"manifest_sha256": "x", "snapshot_txid": 1})
	if err != nil {
		t.Fatal(err)
	}
	markerVersion, err := h.vs.PutIfAbsent(ctx, "data-uk", location+"/generation.json", body)
	if err != nil {
		t.Fatal(err)
	}
	real := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-uk", "uk")
	var committed []park.View
	h.parkView = func(ctx context.Context, obs hostclaim.Observed) (park.View, error) {
		v, err := real(ctx, obs)
		if err == nil && obs.Claim.Op.Kind == hostclaim.KindLanding && obs.Claim.Op.Phase == hostclaim.PhaseCommitted {
			committed = append(committed, v)
		}
		return v, err
	}
	h.land(t)
	h.committedAndOpen(t)
	if len(committed) == 0 {
		t.Fatal("the committed lane park never asked the view")
	}
	for _, v := range committed {
		if v.Region != "uk" || v.AuthoritativeRegion != "uk" || !v.MarkerKnown || v.MarkerGeneration != p1ID ||
			v.Marker != (park.Ref{Key: location + "/generation.json", VersionID: markerVersion}) {
			t.Fatalf("the committed view %+v; want uk authoritative with the marker at %s", v, markerVersion)
		}
	}
	lane := moves.ScopeLocation(p1Host, p1ID) + "/"
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, lane) {
			t.Fatalf("the lane still holds %s after the committed park", k)
		}
	}
}

// TestRegisterProductionMovesParksNeedsBothSwitches (ruling H1): with either
// switch off the agent registers nothing. With both on it registers without
// CARLOS_REGION: the view's region is the claim's verified roster row's
// (fix round 1, region source), resolved when the view is first asked. What
// it registers stays shut (Q1) until the moves wake source registers right
// after it (registerProductionMovesWakeSource, Task 8): the parks alone
// leave the claim's park view refusing every host before it reads anything,
// so the claim's sweep is not widened.
func TestRegisterProductionMovesParksNeedsBothSwitches(t *testing.T) {
	withMovesWakeSource(t, false)
	env := func(vals map[string]string) func(string) string { return func(k string) string { return vals[k] } }
	hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1"}
	store := release.NewDirStore(t.TempDir())
	asked := 0
	roster := func(context.Context) (string, error) { asked++; return "ie", nil }
	for name, vals := range map[string]map[string]string{
		"claims off": {moves.SwitchEnv: "on"},
		"moves off":  {"CARLOS_HOST_CLAIM": "1", moves.SwitchEnv: "off"},
	} {
		if _, err := registerProductionMovesParks(t.Context(), env(vals), store, hib, roster, newMovesRefusals("ie-box", nil, t.Logf, time.Now)); err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		if claimMoveSources.Parks != nil {
			t.Fatalf("%s: registered", name)
		}
	}
	if _, err := registerProductionMovesParks(t.Context(), env(map[string]string{"CARLOS_HOST_CLAIM": "1", moves.SwitchEnv: "on"}), store, hib, roster, newMovesRefusals("ie-box", nil, t.Logf, time.Now)); err != nil {
		t.Fatalf("both switches on, no CARLOS_REGION: %v", err)
	}
	if claimMoveSources.Parks == nil || asked != 0 {
		t.Fatalf("registered %v, roster asked %d times at registration", claimMoveSources.Parks != nil, asked)
	}
	f := newVwFixture(t)
	obs := vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}})
	var dep *ClaimDependencyError
	if _, err := claimMoveSources.Parks.View(t.Context(), obs); !errors.As(err, &dep) || asked != 0 || claimMoveSources.Source != nil {
		t.Fatalf("the production park view answered (%v), roster asked %d times", err, asked)
	}
}

// TestHomeBoxArms (fix round 1, region source; home-region parks, part 1;
// claim Plan 13): a home box, whose placement code is "", is seen from
// park.HomeRegion, and its arms take the claim's engine only when the claim
// grants it the @home alias: the engine is named by the box's signed roster
// region and carries that region as its HomeRegion (the fleet's signed home
// designation). Without the grant, or with a home that is another region, a
// home box stays unarmed, even over an engine that names itself @home. An
// engine region of "" (unset) never arms, and neither does a box whose code
// is no placement, even when the engine's region spells the same string. A
// named box arms over its own named engine whatever the fleet's home is. The
// gate is open here; in production it stays shut until Task 8.
func TestHomeBoxArms(t *testing.T) {
	withMovesWakeSource(t, true)
	f := newVwFixture(t)
	f.rec.Region = ""
	view := movesParkViewAt(f.records(), f.bucket, "data", fixedRegion(park.HomeRegion))
	obs := vwObs(&hostclaim.Claim{Holder: f.holder(""), Op: hostclaim.Op{Kind: hostclaim.KindMoving, ID: vwMove, Phase: hostclaim.PhaseMoving, Exec: "box"}})
	f.rec.Move = &provision.MoveState{ID: vwMove, Incarnation: f.inc, From: "", To: "uk", State: provision.MoveCutting}
	v, err := view(t.Context(), obs)
	if err != nil || v.Region != park.HomeRegion || v.AuthoritativeRegion != park.HomeRegion || !v.MarkerKnown {
		t.Fatalf("the home view %+v, %v", v, err)
	}
	auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }
	arm := func(engineRegion, home, placement string) (moveCfg, error) {
		providers := ClaimProviders{moveParks: &claimMoveParks{
			local:  claimMoveLocalAuthorizer(func(context.Context) error { return nil }, auth),
			engine: park.Engine{Bucket: "data-home", Region: engineRegion, HomeRegion: home, Leases: (&renewTable{}).leases(), View: view, Authorize: auth},
			leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}}
		c := moveCfg{OwnBucket: "data-home", Region: placement}
		err := armMovesParks(t.Context(), &c, providers)
		return c, err
	}
	c, err := arm("ie", "ie", "")
	if err != nil {
		t.Fatalf("a home box the claim grants @home did not arm: %v", err)
	}
	if !parkHooksArmed(c) || c.ParkHomeRegion != "ie" {
		t.Fatalf("hooks unset or the home binding lost: %q", c.ParkHomeRegion)
	}
	for _, tc := range [][3]string{{"uk", "", "uk"}, {"uk", "ie", "uk"}} {
		if c, err := arm(tc[0], tc[1], tc[2]); err != nil || !parkHooksArmed(c) {
			t.Fatalf("a named box (engine %q, home %q) did not arm: %v", tc[0], tc[1], err)
		}
	}
	for _, tc := range [][3]string{
		{"ie", "", ""},               // no alias granted
		{"ie", "uk", ""},             // the fleet's home is another region
		{park.HomeRegion, "", ""},    // an engine naming itself @home, no grant
		{"", "", ""}, {"", "ie", ""}, // an unset engine region
		{"home", "", ""}, // the roster's own word, with no grant
		{park.HomeRegion, "", "home"}, {park.HomeRegion, "", park.HomeRegion}, {"IE", "", "IE"}, {"", "", "uk"},
	} {
		if c, err := arm(tc[0], tc[1], tc[2]); !errors.Is(err, hostbox.ErrChanged) || parkHooksArmed(c) {
			t.Fatalf("engine in %q, home %q, placement %q: armed (%v)", tc[0], tc[1], tc[2], err)
		}
	}
	if provision.ValidRegion(movesNoLiveRegion) || movesNoLiveRegion == park.HomeRegion {
		t.Fatalf("the no-live sentinel %q can be a box's region", movesNoLiveRegion)
	}
}

// TestParkViewRegionIsTheRosters (fix round 1, region source): the view
// asks the roster resolver once, caches a success, and a failure refuses
// that call and asks again on the next.
func TestParkViewRegionIsTheRosters(t *testing.T) {
	f := newVwFixture(t)
	calls, fail := 0, true
	resolve := cachedRegion(func(context.Context) (string, error) {
		calls++
		if fail {
			return "", hostbox.ErrChanged
		}
		return "uk", nil
	})
	view := movesParkViewAt(f.records(), f.bucket, "data", resolve)
	obs := vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}})
	if _, err := view(t.Context(), obs); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("a roster failure: %v", err)
	}
	fail = false
	for i := 0; i < 3; i++ {
		if v, err := view(t.Context(), obs); err != nil || v.Region != "uk" {
			t.Fatalf("view %+v, %v", v, err)
		}
	}
	if calls != 2 {
		t.Fatalf("the roster was asked %d times, want 2", calls)
	}
}

// TestLandingResidueParkUnderItsOwnLease (M5, M6): the landing takes its
// lease on the location at checking and holds it through published (spec
// §7.6). The leases are the claim's own shape over a table behind
// RenewingLeases' Open: the landing's own lease is a RenewingLeases one (as
// armPhase1's OwnLease is) and the park leases are AggregateLeases over the
// replica park lease set (as the claim's MoveParks engine's are). The
// residue park's scope resolves to the landing's own lock, so Engine.Run
// would wait on the landing itself; park.Engine.RunHeld borrows the held
// lease through AcquireHeld and LeaseScope, and the landing goes on to
// restored.
func TestLandingResidueParkUnderItsOwnLease(t *testing.T) {
	t.Parallel()
	location := "data-uk/" + cutRoot
	t.Run("under RunHeld", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		keys := h.residue(t)
		table := &renewTable{}
		lender := &countingLender{AggregateLeases: table.leases()}
		h.ownLease, h.parkLeases = table.ownLease, lender
		h.land(t)
		h.installed(t)
		if lender.held == 0 {
			t.Fatal("no residue park borrowed the landing's lease through the claim's leases")
		}
		for _, o := range table.opens {
			if o != location {
				t.Fatalf("a lease other than the landing's own was opened: %v", table.opens)
			}
		}
		for _, k := range keys {
			if _, ok := h.f.buckets["data-uk"][k]; ok && !strings.HasSuffix(k, "/generation.json") {
				t.Fatalf("%s is still at the location", k)
			}
		}
	})
	t.Run("Engine.Run waits on the landing's own lease", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		keys := h.residue(t)
		table := &renewTable{}
		h.ownLease, h.parkLeases = table.ownLease, table.leases()
		h.runHeld = func(ctx context.Context, e park.Engine, w park.Work, _ park.HeldLease) error { return e.Run(ctx, w) }
		for i := 0; i < 3; i++ {
			h.land(t)
		}
		if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
			t.Fatalf("the landing went past parking without RunHeld: %+v", m)
		}
		if got := table.holding(); len(got) != 1 || got[0] != location {
			t.Fatalf("the landing holds %v", got)
		}
		for _, k := range keys {
			if _, ok := h.f.buckets["data-uk"][k]; !ok {
				t.Fatalf("%s was parked under a lease nobody held", k)
			}
		}
	})
}

// countingLender is the claim's AggregateLeases, counting the held leases
// it lends (AcquireHeld).
type countingLender struct {
	hostbox.AggregateLeases
	held int
}

func (l *countingLender) AcquireHeld(ctx context.Context, bucket, scope string, own park.HeldLease) (park.HeldLease, error) {
	l.held++
	return l.AggregateLeases.AcquireHeld(ctx, bucket, scope, own)
}

// TestNoArmRunsWithoutTheThreeParkHooks (M17): every moves arm requires the
// claim's park leases, view and authorizer, the landing (and the recovery's
// install) RunHeld too, the finalization the claim's local authorizer;
// production's own wiring sets none of them until the wake-source gate
// opens.
func TestNoArmRunsWithoutTheThreeParkHooks(t *testing.T) {
	t.Parallel()
	three := map[string]func(*moveCfg){
		"no park leases":     func(c *moveCfg) { c.ParkLeases = nil },
		"no park view":       func(c *moveCfg) { c.ParkView = nil },
		"no park authorizer": func(c *moveCfg) { c.ParkAuthorize = nil },
	}
	runHeld := func(c *moveCfg) { c.ParkRunHeld = nil }
	local := func(c *moveCfg) { c.ParkLocalAuthorize = nil }
	arms := []struct {
		name  string
		cfg   func(t *testing.T) moveCfg
		armed func(moveCfg) bool
		extra map[string]func(*moveCfg)
	}{
		{"landing", func(t *testing.T) moveCfg { return newLandingHarness(t).lcfg }, landingArmed, map[string]func(*moveCfg){"no RunHeld": runHeld}},
		{"rollback target", func(t *testing.T) moveCfg { h := newRollbackHarness(t); return h.rollbackCfg(t) }, rollbackTargetArmed, nil},
		{"lane park (revert, abort, committed)", func(t *testing.T) moveCfg { return newRevertHarness(t).laneCfg() }, revertLaneArmed, nil},
		{"recovery", func(t *testing.T) moveCfg { h := newRecoverHarness(t, true); return h.sourceCfg(t) }, recoverySourceArmed, map[string]func(*moveCfg){"no RunHeld": runHeld}},
		{"source finalization", func(t *testing.T) moveCfg { return newFinalHarness(t).cfg }, sourceFinalArmed, map[string]func(*moveCfg){"no claim local authorizer": local}},
	}
	for _, a := range arms {
		t.Run(a.name, func(t *testing.T) {
			c := a.cfg(t)
			if !a.armed(c) {
				t.Fatal("fixture: the arm is not armed with every hook")
			}
			for name, unarm := range three {
				u := c
				unarm(&u)
				if a.armed(u) {
					t.Fatalf("%s: armed", name)
				}
			}
			for name, unarm := range a.extra {
				u := c
				unarm(&u)
				if a.armed(u) {
					t.Fatalf("%s: armed", name)
				}
			}
		})
	}
}

// TestNoArmRunsParkLocalWithoutTheClaimsLocalAuthorizer (M9, Q10): every arm
// that runs a park.Local (the landing's install, the target's half of a
// rollback, the source's recovery through the install, the source's
// finalization) requires the claim's MoveParks local authorizer, and the
// park.Local authorizer an install builds asks its own check and then the
// claim's: there is no fallback to its own rule alone.
func TestNoArmRunsParkLocalWithoutTheClaimsLocalAuthorizer(t *testing.T) {
	t.Parallel()
	arms := []struct {
		name  string
		cfg   func(t *testing.T) moveCfg
		armed func(moveCfg) bool
	}{
		{"landing", func(t *testing.T) moveCfg { return newLandingHarness(t).lcfg }, landingArmed},
		{"rollback target", func(t *testing.T) moveCfg { h := newRollbackHarness(t); return h.rollbackCfg(t) }, rollbackTargetArmed},
		{"recovery", func(t *testing.T) moveCfg { h := newRecoverHarness(t, true); return h.sourceCfg(t) }, recoverySourceArmed},
		{"source finalization", func(t *testing.T) moveCfg { return newFinalHarness(t).cfg }, sourceFinalArmed},
	}
	for _, a := range arms {
		t.Run(a.name, func(t *testing.T) {
			c := a.cfg(t)
			if !a.armed(c) {
				t.Fatal("fixture: the arm is not armed with every hook")
			}
			c.ParkLocalAuthorize = nil
			if a.armed(c) {
				t.Fatal("armed without the claim's local authorizer")
			}
		})
	}
	t.Run("the install's park.Local authorizer", func(t *testing.T) {
		h := newLandingHarness(t)
		ctx := t.Context()
		build := func(local func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error) *installRun {
			c := h.lcfg
			c.ParkLocalAuthorize = local
			rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, Incarnation: h.inc}}
			return newInstallRun(h.s, &movesWorker{cfg: c}, rec, installSource{OpID: p1ID, Host: p1Host, Role: roleTarget})
		}
		box := h.lcfg.Box
		ours := &hostclaim.Claim{Host: p1Host, Op: hostclaim.Op{ID: p1ID, Kind: hostclaim.KindLanding, Exec: box}}
		chore := hostclaim.Obligation{ID: "local", Op: p1ID, Kind: "park-local", Box: box}
		if err := build(nil).local.Authorize(ctx, ours, chore); err == nil {
			t.Fatal("the install's own rule stood alone without the claim's local authorizer")
		}
		asked := 0
		claim := func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error { asked++; return nil }
		if err := build(claim).local.Authorize(ctx, ours, chore); err != nil || asked != 1 {
			t.Fatalf("the claim's local authorizer was not asked: %v (%d)", err, asked)
		}
		other := &hostclaim.Claim{Host: p1Host, Op: hostclaim.Op{ID: "mv_other", Kind: hostclaim.KindLanding, Exec: box}}
		if err := build(claim).local.Authorize(ctx, other, chore); err == nil || asked != 1 {
			t.Fatalf("the install's own check did not run first: %v (%d)", err, asked)
		}
	})
}