package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/instancekey"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// startIntoHomeT3 writes the record and claim for a move from the harness's
// named source (ie-box in ie) into the home placement, whose box is
// home-box: the record's move names the home's code, "", and the claim's
// target region is the placement-derived engine region, @home. The home box
// publishes its edges file with no placement, as a home box does.
func startIntoHomeT3(t *testing.T, h *phase1Harness) {
	t.Helper()
	ctx := context.Background()
	if h.src == "" {
		t.Fatal("startIntoHomeT3 needs a named source")
	}
	if _, err := h.s.Put(ctx, "control/edges/home-box.json",
		[]byte(`{"v":1,"box":"home-box","eip":"1.2.3.9","data_bucket":"data-home","data_bucket_region":"eu-west-1","hosts":{}}`), ""); err != nil {
		t.Fatal(err)
	}
	rec := provision.Record{Schema: 1, Host: p1Host, Slug: "s", App: "app", Account: "bab", Region: h.src, Channel: "stable", CreatedAt: p1Created}
	if err := provision.CreateRecord(ctx, h.s, rec); err != nil {
		t.Fatal(err)
	}
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveStartTransition(provision.MoveStart{
		ID: p1ID, Incarnation: h.inc, From: h.src, To: "", Cut: "manual", By: "op", Now: h.clock.Now()})); err != nil {
		t.Fatal(err)
	}
	obs, err := hostclaim.Update(ctx, h.s, hostclaim.Observed{Host: p1Host, Absent: true}, func(c *hostclaim.Claim) error {
		c.Holder = &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "app", p1Host), Account: "bab", App: "app",
			Incarnation: h.inc, CreatedAt: p1Created, Location: "instances/" + p1Host + "/" + h.inc, Generation: "cur", Stamp: 1}
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "ie-box"}
		c.Touched = hostclaim.Touched{Complete: true, Regions: []string{park.PlacementRegion(h.src)}, Boxes: []string{"ie-box"}}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.BeginMove(p1ID, "op", "ie-box", "home-box", park.PlacementRegion(""), h.clock.Now())); err != nil {
		t.Fatal(err)
	}
}

// A home box sets no CARLOS_REGION: agentOwnedVars adds it only when the
// box has a region, and removes a tenant-provided one otherwise. T3 runs
// here on both sides' production compositions: the source's launch report
// is its ExecBackend's LaunchEnvironment with region ie, and the home box's
// held environment is composeHeldEnv with the home box's own region, "", as
// armPhase1 wires it. Before home part 2's T3 rule the target lacked the
// source's CARLOS_REGION and failed region-missing; the harness's default
// fixture sets CARLOS_REGION=uk and would have hidden that.
func TestHomeTargetT3FromTheRealComposition(t *testing.T) {
	h := newPhase1Harness(t)
	src := &activator.ExecBackend{DataDir: h.cfg.DataDir, Region: h.src}
	h.cfg.LaunchEnv = func(string) (map[string]string, map[string]bool, error) {
		env, instance, _, err := src.LaunchLayers(h.route)
		return env, instance, err
	}
	startIntoHomeT3(t, h)
	h.pass(t)
	if _, ok := h.sourceReport(t).Env.Vars[moves.RegionVar]; !ok {
		t.Fatal("the named source's launch report carries no CARLOS_REGION; the witness needs it")
	}

	tc := h.targetCfg()
	tc.Region, tc.Box, tc.OwnBucket = "", "home-box", "data-home"
	tc.ComposeHeldEnv = composeHeldEnv(tc.SecretsPriv, tc.Region, false)
	h.targetPass(t, tc)
	r := h.targetReport(t)
	if r.Box != "home-box" || r.Region != "" {
		t.Fatalf("the report is not the home box's: box %q region %q", r.Box, r.Region)
	}
	if t3 := r.Checks[moves.CheckEnv]; !t3.OK {
		t.Fatalf("T3 on a home target that sets no region: %+v", t3)
	}
	if r.Inputs.EnvReport != h.sourceReport(t).Env.Digest {
		t.Fatalf("a passing T3 binds the source's env report: %q", r.Inputs.EnvReport)
	}
}

// A box that sets CARLOS_REGION is not the home box, even when a move into
// home names it: T3 fails it region-present (Review Focus 5), here with the
// same production composer given a region.
func TestHomeTargetT3RefusesABoxThatSetsARegion(t *testing.T) {
	h := newPhase1Harness(t)
	src := &activator.ExecBackend{DataDir: h.cfg.DataDir, Region: h.src}
	h.cfg.LaunchEnv = func(string) (map[string]string, map[string]bool, error) {
		env, instance, _, err := src.LaunchLayers(h.route)
		return env, instance, err
	}
	startIntoHomeT3(t, h)
	h.pass(t)

	tc := h.targetCfg()
	tc.Region, tc.Box, tc.OwnBucket = "", "home-box", "data-home"
	tc.ComposeHeldEnv = composeHeldEnv(tc.SecretsPriv, "uk", false)
	h.targetPass(t, tc)
	t3 := h.targetReport(t).Checks[moves.CheckEnv]
	if want := t3Differs([]string{moves.RegionVar + " (region-present)"}); t3.OK || t3.Reason != want {
		t.Fatalf("T3 = %+v, want %q", t3, want)
	}
}

// Home part 2, Task 3: the commit into home, through the production owners.
// Every witness below runs the move ca -> home on the intoHome fixture: the
// source ca-box in ca, the target home-box whose roster name is ie, whose
// placement code is "" and whose engine region for the move is @home.

// claimWatch is the harness's control store with a recorder on the host
// claim: every claim a transition writes is decoded and kept, so a chore
// the run created is seen even after an acknowledgement removed it, and so
// are every touched region and box and every quiet_after key the claim ever
// held. It embeds the DirStore so the store's optional interfaces stay.
type claimWatch struct {
	*release.DirStore
	mu     sync.Mutex
	writes []hostclaim.Claim
}

func (w *claimWatch) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := w.DirStore.Put(ctx, key, body, ifMatch)
	if err == nil && key == "control/hosts/"+p1Host+".json" {
		var c hostclaim.Claim
		if jerr := json.Unmarshal(body, &c); jerr == nil {
			w.mu.Lock()
			w.writes = append(w.writes, c)
			w.mu.Unlock()
		}
	}
	return etag, err
}

// since is every claim written after the n-th write.
func (w *claimWatch) since(n int) []hostclaim.Claim {
	w.mu.Lock()
	defer w.mu.Unlock()
	return append([]hostclaim.Claim(nil), w.writes[n:]...)
}

// watchClaims puts a claimWatch in front of the harness's store before
// anything captures it.
func watchClaims(out **claimWatch) p1Option {
	return func(h *phase1Harness) {
		w := &claimWatch{DirStore: h.s.(*release.DirStore)}
		h.s = w
		*out = w
	}
}

// stagingLeaseKeys records the key of every staging lease the harness's
// boxes take, as the production hook keys it (stagingLeaseKey over the
// record code the round passes).
type stagingLeaseKeys struct {
	mu   sync.Mutex
	keys []string
}

func (k *stagingLeaseKeys) wrap(c *moveCfg) {
	inner := c.StagingLease
	c.StagingLease = func(ctx context.Context, code, host string) (moves.SafeUntilGuard, func(context.Context) error, error) {
		key, err := stagingLeaseKey(code, host)
		if err != nil {
			return nil, nil, err
		}
		k.mu.Lock()
		k.keys = append(k.keys, key)
		k.mu.Unlock()
		return inner(ctx, code, host)
	}
}

func (k *stagingLeaseKeys) all() []string {
	k.mu.Lock()
	defer k.mu.Unlock()
	return append([]string(nil), k.keys...)
}

// homeLeaseKey is the home lane's staging lease for the harness's host.
var homeLeaseKey = "control/staging-leases/" + park.HomeRegion + "/" + p1Host

// onlyTheHomeToken is TestHomeTargetFreshStateShowsOnlyTheHomeToken's
// assertion over a witness that ran from clean claim state (the start CAS
// wrote intent.to @home and home_region ie, and nothing else names home):
// every chore created for the target side names @home and is owned by
// region:@home, touched only ever gained @home and the target box beside
// the source's, no quiet_after key is ie or "", and every staging lease
// taken for the home lane is keyed @home. A supplied ie would show here as
// a chore, a touched region, a quiet key or a lease key of its own.
func onlyTheHomeToken(t *testing.T, writes []hostclaim.Claim, leases []string, src, srcBox string) {
	t.Helper()
	if len(writes) == 0 {
		t.Fatal("fixture: the witness wrote no claim")
	}
	seen := map[string]bool{}
	for _, c := range writes {
		for _, o := range c.Obligations {
			if seen[o.ID] {
				continue
			}
			seen[o.ID] = true
			switch {
			case o.Kind == "park-local":
				if o.Owner != "box:"+o.Box || o.Region != "" {
					t.Fatalf("a park-local chore %+v", o)
				}
			case o.Region == src:
				if o.Owner != "region:"+src {
					t.Fatalf("a source-side chore %+v", o)
				}
			case o.Region != park.HomeRegion || o.Owner != "region:"+park.HomeRegion:
				t.Fatalf("a target-side chore not named %s: %+v", park.HomeRegion, o)
			}
		}
		for _, r := range c.Touched.Regions {
			if r != src && r != park.HomeRegion {
				t.Fatalf("touched gained %q: %+v", r, c.Touched)
			}
		}
		for _, b := range c.Touched.Boxes {
			if b != srcBox && b != "home-box" {
				t.Fatalf("touched gained box %q: %+v", b, c.Touched)
			}
		}
		for k := range c.QuietAfter {
			if k != src && k != park.HomeRegion {
				t.Fatalf("quiet_after gained %q: %v", k, c.QuietAfter)
			}
		}
		if c.Op.Intent != nil && c.Op.Intent["to"] != "" && c.Op.Intent["to"] != park.HomeRegion {
			t.Fatalf("the op's to is %q", c.Op.Intent["to"])
		}
	}
	for _, k := range leases {
		if k != homeLeaseKey && k != "control/staging-leases/"+src+"/"+p1Host {
			t.Fatalf("a staging lease keyed %q", k)
		}
	}
}

// laneKeysIn lists the keys under the move's staging lane in bucket.
func laneKeysIn(h *phase1Harness, bucket, id string) []string {
	prefix := moves.StagingPrefix(p1Host, id)
	var keys []string
	for k := range h.f.buckets[bucket] {
		if rest, ok := strings.CutPrefix(k, prefix); ok {
			keys = append(keys, rest)
		}
	}
	sort.Strings(keys)
	return keys
}

// TestMoveIntoHomeStagesTheHomeLane: the source round on ca-box stages the
// lane into the home box's bucket, data-home, under moves/<host>/<id>/,
// under the staging lease of the home lane and no other. Touch is a union:
// from a claim that names no home it adds @home and home-box; with ie and
// home-box already there it writes nothing; with ie and no home-box it adds
// @home and home-box and keeps ie. The only region it ever adds is @home.
func TestMoveIntoHomeStagesTheHomeLane(t *testing.T) {
	type tc struct {
		name    string
		opts    []p1Option
		seedBox bool
		writes  int
	}
	for _, c := range []tc{
		{name: "touched names no home", writes: 1},
		{name: "touched holds ie and home-box", opts: []p1Option{retainTouched("ie")}, seedBox: true, writes: 0},
		{name: "touched holds ie without home-box", opts: []p1Option{retainTouched("ie")}, writes: 1},
	} {
		t.Run(c.name, func(t *testing.T) {
			var watch *claimWatch
			h := newPhase1Harness(t, append([]p1Option{intoHome, watchClaims(&watch)}, c.opts...)...)
			leases := &stagingLeaseKeys{}
			leases.wrap(&h.cfg)
			h.start(t, "manual")
			if c.seedBox {
				rewriteP1Claim(t, h.s, func(cl *hostclaim.Claim) { cl.Touched.Boxes = append(cl.Touched.Boxes, "home-box") })
			}
			before := h.claim(t)
			beforeTouched, _ := json.Marshal(before.Touched)
			var atFirst *hostclaim.Claim
			h.f.failAt = func(op, key string) error {
				if atFirst == nil {
					atFirst = h.claim(t)
				}
				return nil
			}
			h.pass(t)
			if atFirst == nil {
				t.Fatal("no staging write happened")
			}
			if d := int(atFirst.Gen - before.Gen); d != c.writes {
				t.Fatalf("%d claim writes before the first staging write, want %d (touched %+v -> %+v)", d, c.writes, before.Touched, atFirst.Touched)
			}
			after := h.claim(t)
			for _, r := range before.Touched.Regions {
				if !containsString(after.Touched.Regions, r) {
					t.Fatalf("touched lost %q: %+v -> %+v", r, before.Touched, after.Touched)
				}
			}
			for _, b := range before.Touched.Boxes {
				if !containsString(after.Touched.Boxes, b) {
					t.Fatalf("touched lost box %q: %+v -> %+v", b, before.Touched, after.Touched)
				}
			}
			for _, r := range after.Touched.Regions {
				if !containsString(before.Touched.Regions, r) && r != park.HomeRegion {
					t.Fatalf("touched gained %q: %+v -> %+v", r, before.Touched, after.Touched)
				}
			}
			if !containsString(after.Touched.Boxes, "home-box") {
				t.Fatalf("touched has no home-box: %+v", after.Touched)
			}
			if c.writes == 0 {
				if afterTouched, _ := json.Marshal(after.Touched); string(afterTouched) != string(beforeTouched) {
					t.Fatalf("touched was rewritten: %s -> %s", beforeTouched, afterTouched)
				}
			} else if !containsString(after.Touched.Regions, park.HomeRegion) {
				t.Fatalf("the Touch did not add %s: %+v", park.HomeRegion, after.Touched)
			}
			// The lane is in data-home under moves/<host>/<id>/, and every
			// staging request went there.
			keys := laneKeysIn(h, "data-home", p1ID)
			for _, k := range []string{"move.json", "replica/0000/0000000000000001-0000000000000002.ltx", "replica/0001/0000000000000001-0000000000000002.ltx"} {
				if !containsString(keys, k) {
					t.Fatalf("the home lane lacks %s: %v", k, keys)
				}
			}
			for _, call := range h.f.stagingCalls() {
				if call.Bucket != "data-home" || !strings.HasPrefix(call.Key, moves.StagingPrefix(p1Host, p1ID)) {
					t.Fatalf("a staging request outside the home lane: %+v", call)
				}
			}
			// The lease of the home lane, and no other.
			got := leases.all()
			if len(got) == 0 {
				t.Fatal("the round took no staging lease")
			}
			for _, k := range got {
				if k != homeLeaseKey {
					t.Fatalf("a staging lease keyed %q, want %q", k, homeLeaseKey)
				}
			}
			if c.name == "touched names no home" {
				onlyTheHomeToken(t, watch.since(0), got, "ca", "ca-box")
			}
		})
	}
}

// TestMoveIntoHomeTargetChecksPass: on home-box, every readiness check is
// ok over the production composition (the home box sets no region). T8
// names the home by its name, ie, when it fails; T10 probes the source's
// bucket, data-ca; T12 refuses inside a quiet window recorded under ie, and
// again under @home, naming ie.
func TestMoveIntoHomeTargetChecksPass(t *testing.T) {
	t.Parallel()
	newHarness := func(t *testing.T) (*phase1Harness, moveCfg, *[]string) {
		h := newPhase1Harness(t, intoHome, realLaunchEnv)
		h.start(t, "manual")
		h.pass(t)
		tc := h.targetCfg()
		probed := &[]string{}
		tc.LaneProbe = func(_ context.Context, bucket, region, id string) error {
			*probed = append(*probed, bucket+"/"+region+"/"+id)
			return nil
		}
		return h, tc, probed
	}
	h, tc, probed := newHarness(t)
	h.targetPass(t, tc)
	r := h.targetReport(t)
	if r.Box != "home-box" || r.Region != "" {
		t.Fatalf("the report is not the home box's: box %q region %q", r.Box, r.Region)
	}
	for _, id := range []string{moves.CheckProtocol, moves.CheckSecrets, moves.CheckEnv, moves.CheckStorage, moves.CheckArchive, moves.CheckDisk, moves.CheckRelease,
		moves.CheckOneOwner, moves.CheckClaim, moves.CheckLane, moves.CheckBudget, moves.CheckQuiet, moves.CheckArmed} {
		if c, ok := r.Checks[id]; !ok || !c.OK {
			t.Fatalf("%s on the home box: %+v (present %v)", id, c, ok)
		}
	}
	if ok, why := r.Passing(); !ok {
		t.Fatalf("the home box's report does not pass: %s", why)
	}
	if len(*probed) == 0 {
		t.Fatal("T10 probed nothing")
	}
	for _, p := range *probed {
		if p != "data-ca/ca-central-1/"+p1ID {
			t.Fatalf("T10 probed %s, want the source's bucket data-ca", p)
		}
	}
	// T8 names the home by its name: a second placementless box makes the
	// home ambiguous, and the reason says ie, never @home or "".
	if _, err := h.s.Put(t.Context(), "control/edges/home-box-2.json",
		[]byte(`{"v":1,"box":"home-box-2","eip":"1.2.3.10","data_bucket":"data-home-2","data_bucket_region":"eu-west-1","hosts":{}}`), ""); err != nil {
		t.Fatal(err)
	}
	h.clock.Advance(time.Second)
	h.targetPass(t, tc)
	if t8 := h.targetReport(t).Checks[moves.CheckOneOwner]; t8.OK || t8.Reason != t8Count(2, "ie") {
		t.Fatalf("T8 with two home boxes: %+v, want %q", t8, t8Count(2, "ie"))
	}

	// T12 inside a quiet window, recorded under either spelling.
	for _, spelling := range []string{"ie", park.HomeRegion} {
		h, tc, _ := newHarness(t)
		until := h.clock.now.Add(time.Hour).UTC()
		rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.QuietAfter = map[string]time.Time{spelling: until} })
		h.targetPass(t, tc)
		if t12 := h.targetReport(t).Checks[moves.CheckQuiet]; t12.OK || t12.Reason != t12NotQuiet("ie", until) {
			t.Fatalf("T12 with quiet_after under %s: %+v, want %q", spelling, t12, t12NotQuiet("ie", until))
		}
	}
}

// intoHomeClose is the close harness for the move ca -> home with the
// production probe mode (Task 2): the real cut, flip and landing through
// restored, the probe through movesProbeOwners over the claim runtime.
// With residue, an earlier home stay of this incarnation is seeded in
// data-home first (a root generation and one named generation, the
// landing harness's residue), so the landing's parking step has work. It
// returns the residue's keys.
func intoHomeClose(t *testing.T, residue bool, opts ...p1Option) *intoHomeRun {
	t.Helper()
	return intoHomeCloseWith(t, residue, nil, opts...)
}

// intoHomeCloseWith is intoHomeClose with afterFlip run on the landing
// harness between the flip and the first landing pass.
func intoHomeCloseWith(t *testing.T, residue bool, afterFlip func(*landingHarness), opts ...p1Option) *intoHomeRun {
	t.Helper()
	r := &intoHomeRun{leases: &stagingLeaseKeys{}}
	lh := newLandingHarness(t, append([]p1Option{intoHome, realLaunchEnv, watchClaims(&r.watch)}, opts...)...)
	// Both boxes' configurations derive from the source's (targetCfg), so
	// one wrap records every staging lease the run takes.
	r.leases.wrap(&lh.cfg)
	if residue {
		r.residue = earlierHomeStay(t, lh)
	}
	lh.flip(t)
	if afterFlip != nil {
		afterFlip(lh)
	}
	h := newFinalHarnessFlipped(t, lh)
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.CutAttempts = 2
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	h.ownAuthorize = true
	r.h, r.p = h, h.productionProbe(t)
	// Before the probe the home box's env file gains bytes the prepared
	// manifest does not carry (a comment line, which changes no variable
	// and so not the env pin), and an adoption record sits beside it, as
	// TestLandingWakesEstablishThroughC8 stages them: the establishment then
	// has a revision to commit and a local effect to show.
	envPath := filepath.Join(h.tdir, p1Host+".env")
	before, err := os.ReadFile(envPath)
	if err != nil {
		t.Fatal(err)
	}
	r.env = append(append([]byte(nil), before...), []byte("# established on home-box\n")...)
	if err := os.WriteFile(envPath, r.env, 0o600); err != nil {
		t.Fatal(err)
	}
	r.adoption = filepath.Join(h.tdir, p1Host+instancekey.AdoptionSuffix)
	if err := os.WriteFile(r.adoption, []byte(`{"v":2}`), 0o600); err != nil {
		t.Fatal(err)
	}
	return r
}

// intoHomeRun is one into-home run: the close harness, its production
// probe, the residue planted, the env file and adoption record staged for
// C8, and the recorders of the claims written and the staging leases taken.
type intoHomeRun struct {
	h        *finalHarness
	p        *prodProbe
	residue  []string
	env      []byte
	adoption string
	watch    *claimWatch
	leases   *stagingLeaseKeys
}

// earlierHomeStay plants an earlier stay of this incarnation in data-home
// at the holder's location, as the landing harness's residue does: the
// root generation's level file, two level files of the named generation
// old, and the earlier stay's generation.json, here a marker that names
// old (the probe reads the location's current marker, and a home box that
// served the instance before left a well-formed one).
func earlierHomeStay(t *testing.T, h *landingHarness) []string {
	t.Helper()
	ctx := context.Background()
	marker, err := json.Marshal(genselect.Marker{Op: "old", Incarnation: h.inc, Host: p1Host, Box: "home-box", InstalledAt: h.clock.now.Add(-48 * time.Hour).UTC()})
	if err != nil {
		t.Fatal(err)
	}
	keys := []string{
		cutRoot + "/g-old/0000/0000000000000001-0000000000000001.ltx",
		cutRoot + "/g-old/0009/0000000000000001-0000000000000001.ltx",
		cutRoot + "/0000/0000000000000001-0000000000000001.ltx",
		cutRoot + "/generation.json",
	}
	for _, k := range keys {
		body := []byte("residue " + k)
		if strings.HasSuffix(k, "/generation.json") {
			body = marker
		}
		if _, err := h.vs.PutIfAbsent(ctx, "data-home", k, body); err != nil {
			t.Fatal(err)
		}
	}
	return keys
}

// TestMoveIntoHomeCommitsAndCloses: the whole move ca -> home through the
// real cut, flip and landing, then the production probe owners and the
// claim runtime, as TestMoveFromHomeClosesThroughTheSourcePark does for
// part 1 with the fake probe.
func TestMoveIntoHomeCommitsAndCloses(t *testing.T) {
	r := intoHomeClose(t, true)
	// While the attempt is live, the owners' admission evidence (the one the
	// writer was bound to) accepts an admission on home-box in ie against
	// the op's @home, and refuses the same admission in another region.
	checked := false
	r.p.afterMaterialize = func(int) error {
		ctx := context.Background()
		o, err := hostclaim.Read(ctx, r.h.s, p1Host)
		if err != nil {
			return err
		}
		in, err := r.p.owners.Select(ctx, r.h.landingHarness.route, o, poProbe)
		if err != nil {
			return err
		}
		if in.Admission.Region != "ie" || o.Claim.Op.Intent["to"] != park.HomeRegion {
			t.Fatalf("fixture: an admission in %q against the op's %q", in.Admission.Region, o.Claim.Op.Intent["to"])
		}
		if err := r.p.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
			t.Fatalf("the evidence refused ie against %s: %v", park.HomeRegion, err)
		}
		bad := in.Admission
		bad.Region = "ca"
		if err := r.p.owners.AdmissionEvidence(ctx, o, bad, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("the evidence accepted an admission in ca: %v", err)
		}
		checked = true
		return nil
	}
	intoHomeCommitsAndCloses(t, r, 1)
	if !checked {
		t.Fatal("the evidence was never checked")
	}
	intoHomeMovesOutAgain(t, r)
}

// homeRegional is the home box's regional runner as the claim assembly
// builds it (claimassembly.go): hostbox.Regional over a park engine in the
// box's roster region ie with the fleet's home designation ie (so it owns
// @home too), on data-home, with the real moves park view seen from ie. The
// claim's authorizer and the regional owner check stand in as allowing.
func homeRegional(h *finalHarness) hostbox.Regional {
	return regionalRunner(h.phase1Harness, h.leases, "ie", "ie", "data-home")
}

// regionalRunner is a box's regional runner as the claim assembly builds
// it: hostbox.Regional over a park engine in the box's roster region with
// the box's verified home binding (home, "" for a named box), on its data
// bucket, with the real moves park view seen from its region.
func regionalRunner(h *phase1Harness, leases park.Leases, region, home, bucket string) hostbox.Regional {
	return hostbox.Regional{Engine: park.Engine{HomeRegion: home, Store: h.vs, Claims: h.s, Bucket: bucket, Region: region, Leases: claimParkLeases(leases),
		Now: h.clock.Now, NewID: obligationID, View: movesParkView(MovesRecordsFrom(h.s), h.vs, bucket, region), Authorize: allowParkAuthorize},
		Owner: func(context.Context, hostclaim.Observed) error { return nil }}
}

// residueChores is the op's park-generation chores, by generation.
func residueChores(c *hostclaim.Claim) map[string]hostclaim.Obligation {
	out := map[string]hostclaim.Obligation{}
	for _, o := range c.Obligations {
		if o.Op == p1ID && o.Kind == "park-generation" {
			out[o.Generation] = o
		}
	}
	return out
}

// intoHomeCommitsAndCloses drives an intoHomeClose run from restored (or
// probed) through the probe, the commit and the close, asserting each
// step. attempts is how many probe attempts the runtime admitted by the
// commit.
func intoHomeCommitsAndCloses(t *testing.T, r *intoHomeRun, attempts int) {
	t.Helper()
	ctx := context.Background()
	h, p, residue, env, adoption := r.h, r.p, r.residue, r.env, r.adoption
	// The landing's parking step queued one park-generation per residue
	// generation, in @home, owned by region:@home, and none for the
	// install's own.
	if residue != nil {
		gens := residueChores(h.claim(t))
		if len(gens) != 2 {
			t.Fatalf("residue chores: %+v", gens)
		}
		for _, g := range []string{hostclaim.RootGeneration, "old"} {
			o, ok := gens[g]
			if !ok || o.Region != park.HomeRegion || o.Owner != "region:"+park.HomeRegion || o.Park == nil || o.Park.Location != cutRoot {
				t.Fatalf("the residue chore for %s: %+v", g, o)
			}
		}
		if _, ok := gens[p1ID]; ok {
			t.Fatal("the landing queued a park of its own generation")
		}
	}

	// The probe, the commit and the open, through the production owners.
	h.land(t)
	h.committedAndOpen(t)
	if len(p.admissions) != attempts {
		t.Fatalf("admissions through the claim runtime: %+v", p.admissions)
	}
	a := p.admissions[len(p.admissions)-1]
	if a.Box != "home-box" || a.Region != "ie" || a.Mode != "cold" || (attempts == 1 && a.Generation != p1ID) {
		t.Fatalf("the probe's admission: %+v", a)
	}
	c := h.claim(t)
	if c.Op.Intent["to"] != park.HomeRegion || c.Op.Intent["home_region"] != "ie" || c.Holder.WriterBox != "home-box" {
		t.Fatalf("the claim after the probe: to %q home %q writer %q", c.Op.Intent["to"], c.Op.Intent["home_region"], c.Holder.WriterBox)
	}
	// The writer asked the owners' admission evidence for each admission,
	// and it passed: the admission's ie and the op's @home are one region
	// under home_region ie (SameRegion). The same evidence refuses the
	// admission in a region that is not the target's.
	if p.evidence != attempts {
		t.Fatalf("the writer's admission evidence ran %d times, want %d", p.evidence, attempts)
	}
	if !park.SameRegion(c.Op.Intent["home_region"], a.Region, c.Op.Intent["to"]) || a.Region == c.Op.Intent["to"] {
		t.Fatalf("fixture: the admission's %q and the op's %q are not two spellings of one region", a.Region, c.Op.Intent["to"])
	}

	// C8 established the prepared generation's files manifest in data-home
	// under <location>/files/g-<id>/.
	m, found, err := hostfiles.Load(ctx, p1FilesStore{vs: h.vs, bucket: "data-home"}, moves.GenerationFilesPrefix(cutRoot, p1ID))
	if err != nil || !found || m.Seq != 1 {
		t.Fatalf("the generation's files manifest in data-home: %+v %v %v", m, found, err)
	}
	if e, ok := m.Entries["upload.txt"]; !ok || !strings.HasPrefix(e.Object, moves.GenerationFilesPrefix(cutRoot, p1ID)) {
		t.Fatalf("the files manifest does not carry the instance's files from data-home: %+v", m.Entries)
	}
	if m.Platform[hostfiles.EnvItem].SHA256 != sha(env) {
		t.Fatalf("the established env item is not the env file on disk: %+v", m.Platform)
	}
	if _, has := m.Platform[instancekey.AdoptionItem]; has {
		t.Fatal("the adoption item was carried")
	}
	if _, err := os.Stat(adoption); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the adoption record remains: %v", err)
	}
	for k := range h.f.buckets["data-ca"] {
		if strings.Contains(k, "/files/g-"+p1ID+"/") {
			t.Fatalf("the generation's files were written to the source's bucket: %s", k)
		}
	}

	// The commit: the source's location park in ca, the lane park in
	// @home, the source's predecessor covered, quiet_after gains ca.
	src := h.chore(t, "ca", false)
	if src == nil || src.Owner != "region:ca" || src.Park.Location != cutRoot {
		t.Fatalf("the source's location park: %+v", src)
	}
	lane := h.chore(t, park.HomeRegion, true)
	if lane == nil || lane.Owner != "region:"+park.HomeRegion || lane.Park.Location != moves.ScopeLocation(p1Host, p1ID) {
		t.Fatalf("the lane park: %+v (%+v)", lane, c.Obligations)
	}
	if h.chore(t, "ie", true) != nil || h.chore(t, "ie", false) != nil {
		t.Fatalf("a chore named ie: %+v", c.Obligations)
	}
	covered := false
	for _, ad := range c.Holder.Admissions {
		if ad.Generation == p1ID && ad.Predecessor == "cur" {
			if ad.Region != "ie" || ad.PredecessorRegion != "ca" || !ad.Covered {
				t.Fatalf("the probe's admission after the commit: %+v", ad)
			}
			covered = true
		}
	}
	if !covered {
		t.Fatalf("no admission retires the source's generation: %+v", c.Holder.Admissions)
	}

	// After the horizon the home box's engine acknowledges the lane park
	// and the residue parks; the source parks its location; the record
	// closes in home.
	h.clock.Advance(24 * time.Hour)
	h.land(t)
	if h.chore(t, park.HomeRegion, true) != nil {
		t.Fatalf("the lane park was not acknowledged: %+v", h.claim(t).Obligations)
	}
	if residue != nil {
		// The residue parks are the home region's regional runner's: the
		// real hostbox.Regional on home-box, its engine in the roster's ie
		// with the @home alias, over the real park view seen from ie.
		r := homeRegional(h)
		for _, o := range residueChores(h.claim(t)) {
			if err := r.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); err != nil {
				t.Fatalf("the home box's regional runner on the residue park of %s: %v", o.Generation, err)
			}
		}
		if gens := residueChores(h.claim(t)); len(gens) != 0 {
			t.Fatalf("the residue parks were not acknowledged: %+v", gens)
		}
		for _, k := range residue {
			if _, ok := h.f.buckets["data-home"][k]; ok && !strings.HasSuffix(k, "/generation.json") {
				t.Fatalf("the earlier stay's %s is still at the location", k)
			}
		}
	}
	h.notClosed(t, "the source has not parked")
	h.sourcePass(t)
	h.done(t)
	rec := h.read(t, p1Host)
	if rec.Region != "" || rec.LastMove == nil || rec.LastMove.Outcome != "done" {
		t.Fatalf("the record after the close: region %q last %+v", rec.Region, rec.LastMove)
	}
	for k := range h.f.buckets["data-ca"] {
		if strings.HasPrefix(k, cutRoot+"/") {
			t.Fatalf("still at the source's location: %s", k)
		}
	}
	// The acknowledged parks set quiet_after: ca for the source's location,
	// and never the unset region.
	q := h.claim(t).QuietAfter
	if _, ok := q["ca"]; !ok {
		t.Fatalf("quiet_after %v names no ca", q)
	}
	if _, ok := q[""]; ok {
		t.Fatalf("quiet_after %v names the unset region", q)
	}
}

// intoHomeMovesOutAgain: the closed move leaves the instance in home, and
// it can move out again by part 1's path. The roster places a move from
// home-box (roster ie, the fleet's home) to ca-box as "" -> ca; the start
// CAS begins it on the closed claim; and the home box's source round
// accepts it: it touches ca and ca-box, takes ca's staging lease and
// stages the new move's lane into data-ca.
func intoHomeMovesOutAgain(t *testing.T, r *intoHomeRun) {
	t.Helper()
	h := r.h
	ctx := context.Background()
	roster := hostactors.Roster{V: 1, DeploymentBucket: "deploy", Revision: 1, HomeRegion: "ie", Boxes: map[string]hostactors.Box{
		"home-box": {Region: "ie", State: "active", Enrollment: p1DstEnrollment},
		"ca-box":   {Region: "ca", State: "active", Enrollment: p1SrcEnrollment},
	}}
	if !hostops.ClaimMoveRosterPlaces(roster, "home-box", "ca-box", "", "ca", "ie") {
		t.Fatal("the roster does not place home-box -> ca-box as a move out of home")
	}
	if hostops.ClaimMoveRosterPlaces(roster, "home-box", "ca-box", "ie", "ca", "ie") {
		t.Fatal("the roster placed the home box's record as the named region ie")
	}
	rec := h.read(t, p1Host)
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveStartTransition(provision.MoveStart{
		ID: p1ID2, Incarnation: h.inc, From: rec.Region, To: "ca", Cut: "manual", By: "op", Now: h.clock.Now()})); err != nil {
		t.Fatal(err)
	}
	obs, err := hostclaim.Read(ctx, h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.BeginAdmittedMove(p1ID2, "op", "home-box", "ca-box", park.PlacementRegion("ca"), h.clock.Now(),
		hostclaim.MoveAdmission{HomeRegion: "ie", AppID: "app_" + p1Host, AdmissionID: p1ID2, SourceEnrollment: p1DstEnrollment, TargetEnrollment: p1SrcEnrollment})); err != nil {
		t.Fatalf("the start CAS of the move out of home: %v", err)
	}
	// The home box as the source: its own bucket, data directory, route
	// and launch environment.
	c := h.landingCfg(t)
	c.LaunchEnv = func(string) (map[string]string, map[string]bool, error) {
		env, instance, _, err := (&activator.ExecBackend{DataDir: h.tdir, Region: ""}).LaunchLayers(h.landingHarness.route)
		return env, instance, err
	}
	c.ReadyProbe, c.StrictNext = h.cfg.ReadyProbe, h.cfg.StrictNext
	before := len(r.leases.all())
	h.lease.mu.Lock()
	h.lease.until = h.clock.now.Add(moves.StagingLeaseTTL)
	h.lease.mu.Unlock()
	w := &movesWorker{cfg: c, mirrors: map[string]lease.MirrorStore{}}
	if err := movesPass(ctx, h.s, w); err != nil {
		t.Fatal(err)
	}
	m := h.read(t, p1Host).Move
	if m == nil || m.ID != p1ID2 || m.State != provision.MovePreparing {
		t.Fatalf("the move out of home after the source round: %+v", m)
	}
	keys := laneKeysIn(h.phase1Harness, "data-ca", p1ID2)
	replica := false
	for _, k := range keys {
		replica = replica || strings.HasPrefix(k, "replica/")
	}
	if !containsString(keys, "move.json") || !replica {
		t.Fatalf("the lane into ca: %v", keys)
	}
	if got := r.leases.all()[before:]; len(got) == 0 || got[0] != "control/staging-leases/ca/"+p1Host {
		t.Fatalf("the move out of home's staging leases: %v", got)
	}
	cl := h.claim(t)
	if !containsString(cl.Touched.Regions, "ca") || !containsString(cl.Touched.Boxes, "ca-box") || cl.Op.Intent["to"] != "ca" || cl.Op.Exec != "home-box" {
		t.Fatalf("the claim of the move out of home: %+v %+v", cl.Op, cl.Touched)
	}
}

// TestMoveIntoHomeRestartedProbe: the first probe attempt is admitted on
// home-box and the agent dies before readiness. The restarted attempt is
// admitted into a fresh generation whose predecessor is the first
// attempt's, in the region that admission recorded, ie; it is published
// fenced (TestMovesProbeOwnersPublishARestartedAttemptFenced's shape), on
// home-box into data-home; and the commit and the close proceed as for one
// attempt.
func TestMoveIntoHomeRestartedProbe(t *testing.T) {
	r := intoHomeClose(t, true)
	h, p := r.h, r.p
	published := false
	p.afterMaterialize = func(n int) error {
		switch n {
		case 1:
			return errors.New("test: the agent died before readiness")
		case 2:
			restartedAttemptPublishesFenced(t, h, p)
			published = true
		}
		return nil
	}
	h.holdProbed = true
	h.land(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || len(p.admissions) != 1 || p.admissions[0].Generation != p1ID {
		t.Fatalf("the first attempt: %+v admissions %+v", m, p.admissions)
	}
	first := p.admissions[0]
	if first.Box != "home-box" || first.Region != "ie" || first.PredecessorRegion != "ca" || first.Predecessor != "cur" {
		t.Fatalf("the first attempt's admission: %+v", first)
	}

	// The agent restarts: a fresh worker, and the claim runtime's second
	// attempt, published fenced while it is live.
	h.freshTarget()
	h.land(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbed || !published {
		t.Fatalf("the restarted attempt: %+v published %v", m, published)
	}
	if len(p.admissions) != 2 {
		t.Fatalf("admissions: %+v", p.admissions)
	}

	// The commit and the close, as for one attempt.
	h.holdProbed = false
	intoHomeCommitsAndCloses(t, r, 2)
}

// restartedAttemptPublishesFenced runs inside the restarted attempt, after
// Materialize: the admission is a fresh generation whose predecessor is the
// first attempt's generation in the first admission's spelling, ie, and
// the fresh generation is published fenced, never unfenced and never
// before its own first root, then over the location's marker in data-home.
func restartedAttemptPublishesFenced(t *testing.T, h *finalHarness, p *prodProbe) {
	t.Helper()
	ctx := context.Background()
	first, again := p.admissions[0], p.admissions[1]
	if again.Box != "home-box" || again.Region != "ie" || again.Mode != "cold" || again.Generation == p1ID || again.Predecessor != p1ID || again.PredecessorRegion != first.Region {
		t.Fatalf("the restarted attempt's admission: %+v (first %+v)", again, first)
	}
	admitted, in := p.materialized, p.materializedIn
	if g := admitted.Claim.Holder.Generation; g != again.Generation || in.Admission.NewGeneration != g {
		t.Fatalf("fixture: the materialized holder is at %s, its inputs admit %s", g, in.Admission.NewGeneration)
	}
	if err := p.owners.Publish(ctx, admitted, in); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("an unfenced marker for a fresh generation: %v", err)
	}
	synced := 0
	pub := ClaimPublication{Sync: func(context.Context) error { synced++; return nil }, Check: func(context.Context) error { return nil }}
	if _, err := p.owners.PublishFenced(ctx, admitted, in, pub); !errors.Is(err, hostbox.ErrChanged) || synced != 1 {
		t.Fatalf("a marker before the root: %v synced=%d", err, synced)
	}
	root := cutRoot + "/g-" + again.Generation + "/0009/" + ltx.FormatFilename(1, ltx.TXID(h.n))
	hdr, err := (&ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 4, MinTXID: 1, MaxTXID: ltx.TXID(h.n)}).MarshalBinary()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := h.vs.PutIfAbsent(ctx, "data-home", root, hdr); err != nil {
		t.Fatal(err)
	}
	receipt, err := p.owners.PublishFenced(ctx, admitted, in, pub)
	if err != nil || receipt.Root.Key != root || receipt.MarkerVersion == "" {
		t.Fatalf("publish: %+v %v", receipt, err)
	}
	if err := p.owners.PublicationEvidence(ctx, admitted); err != nil {
		t.Fatal(err)
	}
	raw, err := h.vs.GetVersion(ctx, "data-home", cutRoot+"/generation.json", receipt.MarkerVersion)
	if err != nil {
		t.Fatal(err)
	}
	var marker genselect.Marker
	if err := json.Unmarshal(raw, &marker); err != nil || marker.Op != again.Generation || marker.Box != "home-box" || marker.Incarnation != h.inc {
		t.Fatalf("the published marker: %+v %v", marker, err)
	}
	for k := range h.f.buckets["data-ca"] {
		if strings.HasPrefix(k, cutRoot+"/g-"+again.Generation+"/") {
			t.Fatalf("the restarted attempt wrote into the source's bucket: %s", k)
		}
	}
}

// cleanHomeStart asserts the claim a witness starts from: the start CAS
// wrote intent.to @home and home_region ie, and nothing else in the claim
// names home in either spelling (no touched entry, no quiet key, no chore).
func cleanHomeStart(t *testing.T, c *hostclaim.Claim) {
	t.Helper()
	if c.Op.Intent["to"] != park.HomeRegion || c.Op.Intent["home_region"] != "ie" {
		t.Fatalf("the start CAS: %+v", c.Op.Intent)
	}
	for _, r := range c.Touched.Regions {
		if r == "ie" || r == park.HomeRegion {
			t.Fatalf("touched names home at the start: %+v", c.Touched)
		}
	}
	if containsString(c.Touched.Boxes, "home-box") || len(c.QuietAfter) != 0 || len(c.Obligations) != 0 {
		t.Fatalf("the start is not clean: touched %+v quiet %v obligations %+v", c.Touched, c.QuietAfter, c.Obligations)
	}
}

// homeTokenSeen counts what a witness created for the target side, so the
// fresh-state proof is never vacuous: chores in @home, and whether touched
// ever held @home and home-box.
func homeTokenSeen(writes []hostclaim.Claim) (chores int, touched bool) {
	seen := map[string]bool{}
	for _, c := range writes {
		for _, o := range c.Obligations {
			if o.Region == park.HomeRegion && !seen[o.ID] {
				seen[o.ID] = true
				chores++
			}
		}
		if containsString(c.Touched.Regions, park.HomeRegion) && containsString(c.Touched.Boxes, "home-box") {
			touched = true
		}
	}
	return chores, touched
}

// TestHomeTargetFreshStateShowsOnlyTheHomeToken (Review Focus 1, supplied
// regions are @home; the runtime proof): each box-side into-home witness of
// Tasks 3 and 4 runs once from clean claim state, where a wrong spelling
// cannot hide (the hostops witnesses, TestMoveIntoHomeDeclareLost and
// TestMoveIntoHomeFencedTargetRelease, assert the same from their own clean
// state with onlyHomeToken). A
// chore queued with ie would be a new obligation named ie; a Touch with ie
// would add ie to touched; a staging lease keyed with ie would show in the
// recorded keys; an exact comparison with intent.to (Abort, Revert,
// TargetReleased and the fenced and lost releases) would refuse and stop
// the witness. hostclaim.Commit compares nothing with intent.to: given ie
// it queues the lane park under ie, and only this check's chore assertion
// catches it, as it does QueueStagingPark's. After each, every
// target-side chore the run ever created is @home and owned by
// region:@home, touched gained only @home and home-box, quiet_after never
// named ie or "", and every home lane lease is keyed @home (a dirty
// return's lane is the source's, under ca's lease).
func TestHomeTargetFreshStateShowsOnlyTheHomeToken(t *testing.T) {
	type run struct {
		writes []hostclaim.Claim
		leases []string
		chores int // the least number of @home chores the witness creates
		// noHomeLease: the witness's recorder starts after the cut, so it
		// sees no lease of the home lane; returnLane: the witness writes a
		// dirty return's lane into the source's region, under ca's lease.
		noHomeLease, returnLane bool
	}
	for name, witness := range map[string]func(t *testing.T) run{
		"the source round stages the home lane": func(t *testing.T) run {
			var watch *claimWatch
			h := newPhase1Harness(t, intoHome, watchClaims(&watch))
			leases := &stagingLeaseKeys{}
			leases.wrap(&h.cfg)
			h.start(t, "manual")
			cleanHomeStart(t, h.claim(t))
			h.pass(t)
			h.targetPass(t, h.targetCfg())
			h.pass(t)
			if m := h.read(t, p1Host).Move; m.State != provision.MoveReady {
				t.Fatalf("the move did not reach ready: %+v", m)
			}
			return run{writes: watch.since(0), leases: leases.all()}
		},
		"the landing, the commit and the close": func(t *testing.T) run {
			r := intoHomeClose(t, true)
			intoHomeCommitsAndCloses(t, r, 1)
			// Two residue parks and the lane park.
			return run{writes: r.watch.since(0), leases: r.leases.all(), chores: 3}
		},
		"the restarted probe": func(t *testing.T) run {
			r := intoHomeClose(t, false)
			r.p.afterMaterialize = func(n int) error {
				if n == 1 {
					return errors.New("test: the agent died before readiness")
				}
				return nil
			}
			r.h.land(t)
			r.h.freshTarget()
			intoHomeCommitsAndCloses(t, r, 2)
			return run{writes: r.watch.since(0), leases: r.leases.all(), chores: 1}
		},
		// Task 4.
		"the abort": func(t *testing.T) run {
			writes, leases := intoHomeAbort(t)
			return run{writes: writes, leases: leases, chores: 1}
		},
		"the pre-flip revert": func(t *testing.T) run {
			writes, leases := intoHomePreFlipRevert(t)
			return run{writes: writes, leases: leases, chores: 1}
		},
		"the clean rollback": func(t *testing.T) run {
			writes, leases := intoHomeCleanRollback(t)
			// The home location's cleaning park and the lane park.
			return run{writes: writes, leases: leases, chores: 2, noHomeLease: true}
		},
		"the dirty return": func(t *testing.T) run {
			writes, leases := intoHomeDirtyReturn(t)
			return run{writes: writes, leases: leases, chores: 1, noHomeLease: true, returnLane: true}
		},
	} {
		t.Run(name, func(t *testing.T) {
			got := witness(t)
			// The claim the start CAS left is the first written with the
			// move's op: clean, as cleanHomeStart requires.
			for i, c := range got.writes {
				if c.Op.Kind == hostclaim.KindMoving {
					cleanHomeStart(t, &got.writes[i])
					break
				}
			}
			onlyTheHomeToken(t, got.writes, got.leases, "ca", "ca-box")
			chores, touched := homeTokenSeen(got.writes)
			if chores < got.chores || !touched {
				t.Fatalf("the witness created %d chores in %s (want at least %d), touched held it %v", chores, park.HomeRegion, got.chores, touched)
			}
			home := 0
			for _, k := range got.leases {
				switch {
				case k == homeLeaseKey:
					home++
				case got.returnLane && k == "control/staging-leases/ca/"+p1Host:
				default:
					t.Fatalf("a staging lease keyed %q", k)
				}
			}
			if home == 0 && !got.noHomeLease {
				t.Fatal("the witness took no staging lease of the home lane")
			}
		})
	}
}

// obligationsJSON is the claim's obligation list as stored, for byte
// comparison.
func obligationsJSON(t *testing.T, obs []hostclaim.Obligation) string {
	t.Helper()
	b, err := json.Marshal(obs)
	if err != nil {
		t.Fatal(err)
	}
	return string(b)
}

// otherThan is the obligations not in region.
func otherThan(obs []hostclaim.Obligation, region string) []hostclaim.Obligation {
	var out []hostclaim.Obligation
	for _, o := range obs {
		if o.Region != region {
			out = append(out, o)
		}
	}
	return out
}

// retainedID is a fixed obligation id for a seeded chore.
func retainedID(n int) string { return fmt.Sprintf("%032x", 0xa000+n) }

// TestHomeTargetRetainedChoresKeepTheirSpelling (Review Focus 1, retained
// state keeps its spelling): the residue chores and the lane chore are
// already owed when the landing and the commit queue theirs, under ie and
// then under @home. The landing's QueueResidueParks(@home) leaves the
// obligation list byte-identical; the commit's QueueStagingPark(@home)
// leaves every chore but the source's new location park byte-identical. No
// second chore, no rewrite of ie to @home. The home box's engines then
// acknowledge each seeded chore once, and the move closes in home.
func TestHomeTargetRetainedChoresKeepTheirSpelling(t *testing.T) {
	for _, spelling := range []string{"ie", park.HomeRegion} {
		t.Run(spelling, func(t *testing.T) {
			ctx := context.Background()
			r, ids := retainedChoresRun(t, spelling)
			h := r.h

			// Each retained chore is acknowledged once, in its spelling: the
			// lane by the home box's moves engine, the residue by its
			// regional runner.
			h.clock.Advance(24 * time.Hour)
			h.land(t)
			reg := homeRegional(h)
			for _, g := range []string{hostclaim.RootGeneration, "old"} {
				if err := reg.Run(ctx, park.Work{Host: p1Host, ID: ids[g].ID, Op: p1ID}); err != nil {
					t.Fatalf("the regional runner on the residue park of %s owed under %s: %v", g, spelling, err)
				}
			}
			ackedOnce(t, h, ids, spelling)
			// A second run of an acknowledged chore writes nothing.
			gen := h.claim(t).Gen
			if err := reg.Run(ctx, park.Work{Host: p1Host, ID: ids["old"].ID, Op: p1ID}); err != nil || h.claim(t).Gen != gen {
				t.Fatalf("a second acknowledgement: %v (claim gen %d to %d)", err, gen, h.claim(t).Gen)
			}
			h.sourcePass(t)
			h.done(t)
		})
	}
}

// ackedOnce asserts each chore in ids is no longer owed and its retirement
// holds exactly one park manifest in data-home, and that no home chore is
// still owed.
func ackedOnce(t *testing.T, h *finalHarness, ids map[string]hostclaim.Obligation, spelling string) {
	t.Helper()
	c := h.claim(t)
	for name, o := range ids {
		for _, owed := range c.Obligations {
			if owed.ID == o.ID {
				t.Fatalf("the %s chore owed under %s was not acknowledged: %+v", name, spelling, owed)
			}
		}
		n := 0
		for k := range h.f.buckets["data-home"] {
			if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/"+o.Retirement+"/") && strings.HasSuffix(k, "/manifest.json") {
				n++
			}
		}
		if n != 1 {
			t.Fatalf("the %s chore's retirement %s holds %d manifests, want 1", name, o.Retirement, n)
		}
	}
	for _, o := range c.Obligations {
		if o.Region == park.HomeRegion || o.Region == "ie" {
			t.Fatalf("a home chore is still owed: %+v", o)
		}
	}
}

// retainedChoresRun is an into-home run whose residue chores and lane
// chore are already owed under spelling when the landing and the commit
// queue theirs; it asserts both queues leave them as they are, and returns
// at landing/committed with the seeded chores by name (@root, old, lane).
func retainedChoresRun(t *testing.T, spelling string) (*intoHomeRun, map[string]hostclaim.Obligation) {
	t.Helper()
	ids := map[string]hostclaim.Obligation{}
	var seeded, atQueue string
	r := intoHomeCloseWith(t, true, func(lh *landingHarness) {
		// Owed before the landing's parking step: one park-generation
		// per residue generation, with the landing's own scope.
		rec := lh.read(t, p1Host)
		rewriteP1Claim(t, lh.s, func(c *hostclaim.Claim) {
			scope := moves.ResidueScope(*c.Holder, *rec.Move.FlippedAt)
			for i, g := range []string{hostclaim.RootGeneration, "old"} {
				s := scope
				o := hostclaim.Obligation{ID: retainedID(2 * i), Op: p1ID, Kind: "park-generation", Region: spelling, Generation: g, Retirement: retainedID(2*i + 1), Owner: "region:" + spelling, Park: &s}
				ids[g] = o
				c.Obligations = append(c.Obligations, o)
			}
		})
		seeded = obligationsJSON(t, lh.claim(t).Obligations)
		lh.landFault = func(point string) error {
			if point == "after-residue-queued" && atQueue == "" {
				atQueue = obligationsJSON(t, lh.claim(t).Obligations)
			}
			return nil
		}
	})
	h, p := r.h, r.p
	if atQueue == "" {
		t.Fatal("the landing never queued its residue parks")
	}
	if atQueue != seeded {
		t.Fatalf("QueueResidueParks(%s) over chores owed under %s changed the obligations:\n%s\nwas\n%s", park.HomeRegion, spelling, atQueue, seeded)
	}

	// Owed before the commit: the lane's park with the commit's own
	// scope.
	var beforeCommit, afterCommit []hostclaim.Obligation
	h.landFault = func(point string) error {
		switch point {
		case "before-commit":
			if beforeCommit != nil {
				return nil
			}
			rec := h.read(t, p1Host)
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
				_, staging, err := moves.CommitScopes(c, rec.Move.FlippedAt)
				if err != nil {
					t.Fatal(err)
				}
				o := hostclaim.Obligation{ID: retainedID(10), Op: p1ID, Kind: "park-replica", Region: spelling, Retirement: retainedID(11), Owner: "region:" + spelling, Park: &staging}
				ids["lane"] = o
				c.Obligations = append(c.Obligations, o)
			})
			beforeCommit = h.claim(t).Obligations
		case "after-claim-committed":
			afterCommit = h.claim(t).Obligations
		}
		return nil
	}
	h.land(t)
	h.committedAndOpen(t)
	if len(p.admissions) != 1 || afterCommit == nil {
		t.Fatalf("the commit: admissions %+v, claim captured %v", p.admissions, afterCommit != nil)
	}
	if got, want := obligationsJSON(t, otherThan(afterCommit, "ca")), obligationsJSON(t, beforeCommit); got != want {
		t.Fatalf("the commit over a lane park owed under %s changed the home chores:\n%s\nwas\n%s", spelling, got, want)
	}
	if src := afterCommit[len(afterCommit)-1]; len(afterCommit) != len(beforeCommit)+1 || src.Region != "ca" || src.Kind != "park-replica" || src.Park.Location != cutRoot {
		t.Fatalf("the commit queued %+v beside the retained chores", afterCommit[len(beforeCommit):])
	}
	return r, ids
}

// TestHomeRegionalAcknowledgesBothSpellings (Ask A2's witness): the real
// hostbox.Regional on home-box (roster region ie, home ie) acknowledges a
// lane park-replica and the residue park-generations owed under @home, and
// the same owed under ie, each exactly once; ca-box's runner (region ca, no
// home) acknowledges none of them.
func TestHomeRegionalAcknowledgesBothSpellings(t *testing.T) {
	for _, spelling := range []string{park.HomeRegion, "ie"} {
		t.Run(spelling, func(t *testing.T) {
			ctx := context.Background()
			r, ids := retainedChoresRun(t, spelling)
			h := r.h
			h.clock.Advance(24 * time.Hour)
			ca := regionalRunner(h.phase1Harness, h.leases, "ca", "", "data-ca")
			home := homeRegional(h)
			for name, o := range ids {
				gen := h.claim(t).Gen
				if err := ca.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); !errors.Is(err, hostbox.ErrUnsupported) || h.claim(t).Gen != gen {
					t.Fatalf("ca's runner on the %s chore owed under %s: %v (claim gen %d to %d)", name, spelling, err, gen, h.claim(t).Gen)
				}
			}
			for name, o := range ids {
				if err := home.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); err != nil {
					t.Fatalf("home-box's runner on the %s chore owed under %s: %v", name, spelling, err)
				}
				gen := h.claim(t).Gen
				if err := home.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); err != nil || h.claim(t).Gen != gen {
					t.Fatalf("a second acknowledgement of the %s chore: %v (claim gen %d to %d)", name, err, gen, h.claim(t).Gen)
				}
			}
			ackedOnce(t, h, ids, spelling)
		})
	}
}

// Home part 2, Task 4: revert, rollback, abort, loss and conversion with a
// home target. Each witness also runs from clean claim state in
// TestHomeTargetFreshStateShowsOnlyTheHomeToken.

// homeLaneCfg is home-box's configuration with its moves park engine armed
// (region @home, ParkHomeRegion ie) over its own data directory and lease
// table.
func homeLaneCfg(h *phase1Harness, c moveCfg, dir string, leases *fakeParkLeases) moveCfg {
	c.DataDir = dir
	armParkFakes(&c, h.s, leases, park.PlacementRegion(c.Region), roleTarget)
	return c
}

// homeLaneChore is the op's lane park owed in @home.
func homeLaneChore(c *hostclaim.Claim) *hostclaim.Obligation {
	for _, o := range c.Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) && o.Region != "ca" {
			return &o
		}
	}
	return nil
}

// TestMoveIntoHomeAbortParksTheHomeLane (Ask A2's witness): an abort before
// the cut. Abort queues the lane park in @home, owned by region:@home;
// home-box's moves engine archives the lane in data-home; past the horizon
// a named box's regional runner (ca) refuses the chore and home-box's
// hostbox.Regional (roster region ie, home ie) acknowledges it; the record
// clears only after that.
func TestMoveIntoHomeAbortParksTheHomeLane(t *testing.T) {
	intoHomeAbort(t)
}

func intoHomeAbort(t *testing.T) ([]hostclaim.Claim, []string) {
	t.Helper()
	ctx := context.Background()
	var watch *claimWatch
	h := newPhase1Harness(t, intoHome, watchClaims(&watch))
	leases := &stagingLeaseKeys{}
	leases.wrap(&h.cfg)
	h.start(t, "manual")
	cleanHomeStart(t, h.claim(t))
	h.pass(t)
	staged := laneKeysIn(h, "data-home", p1ID)
	if len(staged) == 0 {
		t.Fatal("fixture: the round staged nothing in data-home")
	}
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil {
		t.Fatal(err)
	}
	h.pass(t)
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseAborting || c.Op.Exec != "ca-box" || c.Op.Intent["to"] != park.HomeRegion {
		t.Fatalf("the claim after the abort: %+v", c.Op)
	}
	o := homeLaneChore(c)
	if o == nil || o.Region != park.HomeRegion || o.Owner != "region:"+park.HomeRegion || o.Park.Incarnation != h.inc {
		t.Fatalf("the abort's lane park: %+v (%+v)", o, c.Obligations)
	}
	waiting := func(why string) {
		t.Helper()
		if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveAborting {
			t.Fatalf("%s: the record cleared: %+v", why, m)
		}
	}
	// home-box's moves engine archives the lane; the horizon has not passed.
	tleases := &fakeParkLeases{}
	tc := homeLaneCfg(h, h.targetCfg(), t.TempDir(), tleases)
	h.targetPass(t, tc)
	if left := laneKeysIn(h, "data-home", p1ID); len(left) != 0 {
		t.Fatalf("the home lane still holds %v", left)
	}
	if homeLaneChore(h.claim(t)) == nil {
		t.Fatal("the lane park was acknowledged before its horizon")
	}
	h.pass(t)
	waiting("before the horizon")
	// Past the horizon: a named box's runner refuses the chore, the home
	// box's acknowledges it.
	until := o.Park.EvidenceAt.Add(park.SignatureAge + o.Park.RequestLimit)
	h.clock.Advance(until.Sub(h.clock.now) + time.Second)
	work := park.Work{Host: p1Host, ID: o.ID, Op: p1ID}
	if err := regionalRunner(h, tleases, "ca", "", "data-ca").Run(ctx, work); !errors.Is(err, hostbox.ErrUnsupported) {
		t.Fatalf("ca's regional runner on the home lane park: %v", err)
	}
	if homeLaneChore(h.claim(t)) == nil {
		t.Fatal("ca's runner acknowledged the home lane park")
	}
	h.pass(t)
	waiting("ca's runner refused")
	if err := regionalRunner(h, tleases, "ie", "ie", "data-home").Run(ctx, work); err != nil {
		t.Fatalf("home-box's regional runner on the lane park: %v", err)
	}
	if homeLaneChore(h.claim(t)) != nil {
		t.Fatal("the lane park was not acknowledged")
	}
	if q := h.claim(t).QuietAfter; q["ie"] != (time.Time{}) || q[""] != (time.Time{}) {
		t.Fatalf("quiet_after %v", q)
	}
	// Only now does the record clear.
	h.pass(t)
	rec := h.read(t, p1Host)
	if rec.Move != nil || rec.Region != "ca" || rec.LastMove == nil || rec.LastMove.Outcome != "aborted" {
		t.Fatalf("the record after the abort: region %q move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ca-box" || len(c.Obligations) != 0 {
		t.Fatalf("the claim after the abort: %+v %+v", c.Op, c.Obligations)
	}
	for k := range h.f.buckets["data-ca"] {
		if strings.HasPrefix(k, "parked/") {
			t.Fatalf("the abort parked in the source's bucket: %s", k)
		}
	}
	return watch.since(0), leases.all()
}

// TestMoveIntoHomeCleanRollback: the home target's probe, through the
// production probe owners, is admitted on home-box and fails before any
// write, until its attempts are spent. The rollback releases clean:
// TargetReleased names @home, which matches op.intent.to exactly, and the
// cleaning park of the home location is queued in @home. The source then
// reopens on ca-box through the production recovery owners, in two stages:
// before the admission the clean ReturnToSource has put the holder back on
// src_gen and the owners' baseline restores from src_gen; after it the
// holder is on the fresh generation the admission named, the admission is
// clean-return with no writes lost, in ca, and its predecessor is the failed
// home target's last admission, in the spelling that admission recorded:
// ie. The target's cleaning then queues the lane park in @home, home-box's
// regional runner acknowledges both after the horizon, and the record ends
// rolled_back in ca with nothing parked in data-ca.
func TestMoveIntoHomeCleanRollback(t *testing.T) {
	intoHomeCleanRollback(t)
}

func intoHomeCleanRollback(t *testing.T) ([]hostclaim.Claim, []string) {
	t.Helper()
	ctx := context.Background()
	var watch *claimWatch
	h := newRecoverHarness(t, false, intoHome, realLaunchEnv, watchClaims(&watch))
	leases := &stagingLeaseKeys{}
	leases.wrap(&h.cfg)
	cleanHomeStart(t, firstMoving(t, watch.since(0)))
	// Two attempts, each admitted through the production owners and the
	// claim runtime and failing before any write. The second is a fresh
	// generation with its own stamp: its restore retires the first's
	// installation and re-tags its inventory (RetireSuperseded).
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.CutAttempts = 2
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	p := h.productionProbe(t)
	p.afterMaterialize = func(int) error { return errors.New("test: the probe failed") }
	for i := 0; i < 8 && h.read(t, p1Host).Move.State != provision.MoveRollingBack; i++ {
		h.rollbackPass(t)
	}
	if len(p.admissions) != 2 || p.stopped != 1 {
		t.Fatalf("the probe's attempts: admissions %+v, predecessors stopped %d", p.admissions, p.stopped)
	}
	failed := p.admissions[len(p.admissions)-1]
	if failed.Box != "home-box" || failed.Region != "ie" || failed.Generation == p1ID || failed.Predecessor != p1ID {
		t.Fatalf("the home target's last admission: %+v", failed)
	}
	for i := 0; i < 4 && h.rb(t).TargetPhase != provision.RbTargetReleased; i++ {
		h.rollbackPass(t)
	}
	rb := h.rb(t)
	if rb.TargetPhase != provision.RbTargetReleased || rb.Dirty {
		t.Fatalf("the target did not release clean: %+v", rb)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseReleased || c.Op.Intent["to"] != park.HomeRegion || c.Op.Intent["rollback"] != rb.ID {
		t.Fatalf("the claim after TargetReleased: %+v", c.Op)
	}
	location := false
	for _, o := range c.Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Park != nil && o.Park.Location == cutRoot {
			if o.Region != park.HomeRegion || o.Owner != "region:"+park.HomeRegion {
				t.Fatalf("the cleaning park of the home location: %+v", o)
			}
			location = true
		}
	}
	if !location {
		t.Fatalf("TargetReleased queued no park of the home location: %+v", c.Obligations)
	}

	// The source's recovery through the production recovery owners. Its
	// bucket is versioned, as production's is: the harness stored the
	// source's replica directly, so each object gets its version here.
	srcGen := c.Op.Journal[hostclaim.JournalSrcGen]
	src := sourceRecovery(t, h)
	src.beforeAdmit = func(o hostclaim.Observed, in ClaimLaunchInputs) {
		// Stage one: the clean ReturnToSource put the holder back on
		// src_gen, and the owners restore from src_gen.
		if o.Claim.Holder.Generation != srcGen || srcGen != "cur" || in.Baseline == nil || in.Baseline.Generation != srcGen || in.Prepared {
			t.Fatalf("before the admission: holder at %q (src_gen %q), baseline %+v", o.Claim.Holder.Generation, srcGen, in.Baseline)
		}
		if a := in.Admission; a.Coverage != "clean-return" || a.Box != "ca-box" || a.Region != "ca" || a.Mode != "cold" || a.PredecessorRegion != failed.Region {
			t.Fatalf("the selected admission: %+v", a)
		}
	}
	h.srcPass(t)
	if src.wakes != 1 || src.materialized.Claim == nil {
		t.Fatalf("the recovery's starts: %d, materialized %v", src.wakes, src.materialized.Claim != nil)
	}
	// Stage two: the holder is on the fresh generation the runtime admitted
	// and restored; the admission is clean-return, in ca, after the home
	// target's last admission in ie.
	ah := src.admitted.Claim.Holder
	last := ah.Admissions[len(ah.Admissions)-1]
	if ah.Generation != src.materializedIn.Admission.NewGeneration || ah.Generation == srcGen || ah.WriterBox != "ca-box" {
		t.Fatalf("the holder after the admission: %+v", ah)
	}
	if gen, err := hostbox.ReadGenerationMarker(h.srcRoute.DBPath + ".gen"); err != nil || gen != ah.Generation {
		t.Fatalf("the restore's generation marker %q %v, want %s", gen, err, ah.Generation)
	}
	if last.Coverage != "clean-return" || last.WritesLost != nil || len(last.AdditionalWritesLost) != 0 || last.Region != "ca" || last.Box != "ca-box" {
		t.Fatalf("the clean return's admission: %+v", last)
	}
	if last.Predecessor != failed.Generation || last.PredecessorRegion != "ie" {
		t.Fatalf("the clean return's predecessor %q in %q, want the home target's %q in ie", last.Predecessor, last.PredecessorRegion, failed.Generation)
	}
	h.opened(t)

	// The target's cleaning: the lane park in @home, then both parks
	// acknowledged by home-box's regional runner after their horizon.
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned {
		t.Fatalf("the target did not clean: %+v", rb)
	}
	owed := h.claim(t).Obligations
	var lane *hostclaim.Obligation
	for i, o := range owed {
		if o.Op == p1ID && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) {
			lane = &owed[i]
		}
		if o.Op == p1ID && (o.Region != park.HomeRegion || o.Owner != "region:"+park.HomeRegion) {
			t.Fatalf("a chore of the rollback outside @home: %+v", o)
		}
	}
	if lane == nil {
		t.Fatalf("no lane park: %+v", owed)
	}
	h.clock.Advance(24 * time.Hour)
	home := homeRegionalOn(h.phase1Harness, h.leases)
	for _, o := range owed {
		if o.Op != p1ID {
			continue
		}
		if err := home.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); err != nil {
			t.Fatalf("home-box's regional runner on %+v: %v", o, err)
		}
	}
	h.srcPass(t)
	final := h.read(t, p1Host)
	if final.Move != nil || final.Region != "ca" || final.LastMove == nil || final.LastMove.Outcome != "rolled_back" {
		t.Fatalf("the record after the rollback: region %q move %+v last %+v", final.Region, final.Move, final.LastMove)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ca-box" || len(c.Obligations) != 0 {
		t.Fatalf("the claim after the rollback: %+v %+v", c.Op, c.Obligations)
	}
	// Nothing the source serves was parked: data-ca holds no archive, and
	// src_gen's objects are where they were.
	cur := 0
	for k := range h.f.buckets["data-ca"] {
		if strings.HasPrefix(k, "parked/") {
			t.Fatalf("the rollback parked in the source's bucket: %s", k)
		}
		if strings.HasPrefix(k, cutLineage) {
			cur++
		}
	}
	if cur == 0 {
		t.Fatal("src_gen's objects are gone from data-ca")
	}
	return watch.since(0), leases.all()
}

// TestMoveIntoHomeDirtyReturn: the home target's probe, through the
// production probe owners, is admitted on home-box, writes, and fails. The
// rollback releases dirty: home-box seals its installation and writes the
// return lane into data-ca under the staging lease
// control/staging-leases/ca/<host>. The source installs the return through
// the shared installer, and the recovery owners admit it into the return
// generation, sealed, establishing its files from the return's prepared
// manifest through the real C8 provider. probed queues the return lane's
// park in ca. The pre-move generations park after the return (D6): no
// park-generation is owed before the target has returned, and the source's
// install queues them in ca.
func TestMoveIntoHomeDirtyReturn(t *testing.T) {
	intoHomeDirtyReturn(t)
}

func intoHomeDirtyReturn(t *testing.T) ([]hostclaim.Claim, []string) {
	t.Helper()
	ctx := context.Background()
	var watch *claimWatch
	h := newRecoverHarness(t, false, intoHome, realLaunchEnv, watchClaims(&watch))
	leases := &stagingLeaseKeys{}
	leases.wrap(&h.cfg)
	cleanHomeStart(t, firstMoving(t, watch.since(0)))
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.CutAttempts = 2
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	// Each attempt is admitted through the production owners and the claim
	// runtime, the app writes, and the attempt fails. The second's restore
	// retires the first's installation (RetireSuperseded).
	p := h.productionProbe(t)
	p.afterMaterialize = func(int) error {
		h.targetChanged(t)
		return errors.New("test: the probe failed after a write")
	}
	residueBefore := func(why string) {
		t.Helper()
		for _, o := range h.claim(t).Obligations {
			if o.Kind == "park-generation" {
				t.Fatalf("%s: a pre-move generation's park is owed already: %+v", why, o)
			}
		}
	}
	for i := 0; i < 6 && h.read(t, p1Host).Move.State != provision.MoveRollingBack; i++ {
		h.rollbackPass(t)
	}
	if len(p.admissions) != 2 || p.stopped != 1 || p.admissions[1].Box != "home-box" || p.admissions[1].Region != "ie" || p.admissions[1].Predecessor != p1ID {
		t.Fatalf("the home target's admissions: %+v (predecessors stopped %d)", p.admissions, p.stopped)
	}
	before := len(leases.all())
	for i := 0; i < 6 && h.rb(t).TargetPhase != provision.RbTargetReturned; i++ {
		h.rollbackPass(t)
		residueBefore("the target's release")
	}
	rb := h.rb(t)
	if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("the target did not return dirty: %+v", rb)
	}
	// The return lane in data-ca, under ca's staging lease.
	lane := moves.ScopeLocation(p1Host, rb.ID) + "/"
	inCA := 0
	for k := range h.f.buckets["data-ca"] {
		if strings.HasPrefix(k, lane) {
			inCA++
		}
	}
	for k := range h.f.buckets["data-home"] {
		if strings.HasPrefix(k, lane) {
			t.Fatalf("the return lane was written into data-home: %s", k)
		}
	}
	if inCA == 0 {
		t.Fatal("the return lane holds nothing in data-ca")
	}
	if got := leases.all()[before:]; len(got) == 0 || slicesAll(got, "control/staging-leases/ca/"+p1Host) != len(got) {
		t.Fatalf("the return lane's staging leases: %v", got)
	}

	// The source's recovery through the claim runtime on ca-box: the
	// production recovery owners, the real C8 provider establishing from the
	// return's prepared manifest.
	rec := sourceRecovery(t, h)
	// Under CARLOS_HOST_CLAIM (the probe mode sets it) the source's
	// configured step publishes the live release through the held release;
	// the stand-in answers as that publication would.
	h.releaseLive = func(context.Context, release.Store, provision.Record, string) moves.Check {
		return moves.Check{OK: true}
	}
	rec.beforeAdmit = func(o hostclaim.Observed, in ClaimLaunchInputs) {
		residueQueued := false
		for _, ob := range o.Claim.Obligations {
			if ob.Kind == "park-generation" && ob.Region == "ca" && ob.Owner == "region:ca" {
				residueQueued = true
			}
		}
		if !residueQueued {
			t.Fatalf("the source's install queued no park of the pre-move generations: %+v", o.Claim.Obligations)
		}
		if a := in.Admission; !in.Prepared || a.NewGeneration != rb.ID || a.Coverage != "sealed" || a.Box != "ca-box" || a.Region != "ca" || a.PredecessorRegion != "ie" {
			t.Fatalf("the dirty return's selected admission: %+v prepared %v", a, in.Prepared)
		}
	}
	h.srcPass(t)
	if rec.wakes != 1 || rec.materialized.Claim == nil || rec.materializedIn.Admission.NewGeneration != rb.ID {
		t.Fatalf("the recovery's starts: %d, materialized %+v", rec.wakes, rec.materializedIn.Admission)
	}
	got := h.opened(t)
	if got.ID != rb.ID {
		t.Fatalf("opened %+v", got)
	}
	// C8: the return generation's prepared manifest in data-ca is the one
	// the establishment opened.
	m, found, err := hostfiles.Load(ctx, p1FilesStore{vs: h.vs, bucket: "data-ca"}, moves.GenerationFilesPrefix(cutRoot, rb.ID))
	if err != nil || !found {
		t.Fatalf("the return generation's files manifest in data-ca: %v %v", found, err)
	}
	if _, ok := m.Entries["upload.txt"]; !ok || m.Platform[hostfiles.EnvItem].SHA256 == "" {
		t.Fatalf("the return's files manifest: %+v", m)
	}
	// probed queued the return lane's park in ca.
	returnLane := 0
	for _, o := range h.claim(t).Obligations {
		if o.Kind == "park-replica" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, rb.ID) {
			if o.Region != "ca" || o.Owner != "region:ca" {
				t.Fatalf("the return lane's park: %+v", o)
			}
			returnLane++
		}
	}
	if returnLane != 1 {
		t.Fatalf("return lane parks: %d", returnLane)
	}
	return watch.since(0), leases.all()
}

// slicesAll counts the entries of xs equal to x.
func slicesAll(xs []string, x string) int {
	n := 0
	for _, s := range xs {
		if s == x {
			n++
		}
	}
	return n
}

// TestMoveIntoHomePreFlipRevert: the cut stops the source on ca-box and
// fails before the flip (TestCutDeadlinePassedReverts' shape, stopped with
// execution given up). Revert(id, ca-box, @home, &laneScope) matches
// op.intent.to exactly and queues the lane park in @home, owned by
// region:@home. The source's revert probe re-admits on ca-box through the
// production recovery owners, uncovered (TestRevertProbeAdmitsUncovered's
// shape), in ca, after the source's own generation in ca. home-box's moves
// engine archives the lane in data-home; past the horizon home-box's
// hostbox.Regional acknowledges it, and the revert closes. Nothing in data-ca
// is parked.
func TestMoveIntoHomePreFlipRevert(t *testing.T) {
	intoHomePreFlipRevert(t)
}

func intoHomePreFlipRevert(t *testing.T) ([]hostclaim.Claim, []string) {
	t.Helper()
	ctx := context.Background()
	var watch *claimWatch
	// The cut harness under a landing harness: its source replica is a real
	// one, which the recovery's cold restore restores.
	h := newRevertHarnessOn(t, newLandingHarness(t, intoHome, realLaunchEnv, watchClaims(&watch)).cutHarness)
	leases := &stagingLeaseKeys{}
	leases.wrap(&h.cfg)
	h.fromStopped(t)
	cleanHomeStart(t, firstMoving(t, watch.since(0)))
	if m := h.move(t); m.FlippedAt != nil || m.To != "" {
		t.Fatalf("the fixture flipped: %+v", m)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ca-box" || c.Op.Intent["to"] != park.HomeRegion {
		t.Fatalf("the revert's claim: %+v", c.Op)
	}
	lane := homeLaneChore(c)
	if lane == nil || lane.Region != park.HomeRegion || lane.Owner != "region:"+park.HomeRegion {
		t.Fatalf("Revert's lane park: %+v (%+v)", lane, c.Obligations)
	}
	staged := laneKeysIn(h.phase1Harness, "data-home", p1ID)
	if len(staged) == 0 {
		t.Fatal("fixture: the home lane is empty")
	}

	// The source's revert probe through the production recovery owners.
	h.arm(t)
	srcGen := c.Holder.Generation
	rec := revertRecovery(t, h)
	rec.beforeAdmit = func(o hostclaim.Observed, in ClaimLaunchInputs) {
		if a := in.Admission; a.Box != "ca-box" || a.Region != "ca" || a.PredecessorRegion != "ca" || a.Coverage != "" || a.Mode != "cold" || a.ObservedGeneration != srcGen ||
			in.Baseline == nil || in.Baseline.Generation != srcGen {
			t.Fatalf("the revert's selected admission: %+v baseline %+v", a, in.Baseline)
		}
	}
	h.pass(t)
	h.reopened(t)
	if rec.wakes != 1 || rec.materialized.Claim == nil {
		t.Fatalf("the revert probe's starts: %d, materialized %v", rec.wakes, rec.materialized.Claim != nil)
	}
	ah := rec.admitted.Claim.Holder
	last := ah.Admissions[len(ah.Admissions)-1]
	if ah.Generation == srcGen || ah.WriterBox != "ca-box" || last.Region != "ca" || last.PredecessorRegion != "ca" || last.Predecessor != srcGen || last.Coverage != "" {
		t.Fatalf("the revert's admission: holder %+v last %+v", ah, last)
	}
	// Uncovered: the source's own writes are recorded as possibly lost when
	// a writer session held them (the harness's serving holder never had
	// one, so there is nothing to name).
	if prior := rec.observed[0].Claim.Holder; prior.Session != 0 && (last.WritesLost == nil || last.WritesLost.Box != prior.WriterBox || last.WritesLost.Session != prior.Session) {
		t.Fatalf("the revert's writes lost: %+v (prior holder %+v)", last.WritesLost, prior)
	}

	// home-box's moves engine archives the lane; the horizon has not passed.
	h.tPass(t)
	if left := laneKeysIn(h.phase1Harness, "data-home", p1ID); len(left) != 0 {
		t.Fatalf("the home lane still holds %v", left)
	}
	if homeLaneChore(h.claim(t)) == nil {
		t.Fatal("the lane park was acknowledged before its horizon")
	}
	h.pass(t)
	h.reopened(t)
	// Past the horizon home-box's regional runner acknowledges it, and the
	// revert closes.
	h.pastHorizon(t, lane)
	if err := homeRegionalOn(h.phase1Harness, h.tleases).Run(ctx, park.Work{Host: p1Host, ID: lane.ID, Op: p1ID}); err != nil {
		t.Fatalf("home-box's regional runner on the lane park: %v", err)
	}
	if homeLaneChore(h.claim(t)) != nil {
		t.Fatal("the lane park was not acknowledged")
	}
	h.pass(t)
	h.closed(t)
	if rec := h.read(t, p1Host); rec.Region != "ca" {
		t.Fatalf("the record after the revert: %+v", rec)
	}
	for k := range h.f.buckets["data-ca"] {
		if strings.HasPrefix(k, "parked/") {
			t.Fatalf("the revert parked in the source's bucket: %s", k)
		}
	}
	return watch.since(0), leases.all()
}

// sourceRecovery binds the recover harness's source box (ca-box) to the
// production mode: the recovery adapter's wake runs the claim runtime on
// ca-box (newProdWake's start: Admit, Prepare, Restore, Reserve and
// Materialize through the production recovery owners and the real C8).
// The source is first made what a served source leaves (sourceServed).
func sourceRecovery(t *testing.T, h *recoverHarness) *prodProbe {
	t.Helper()
	sourceServed(t, h.phase1Harness, h.sdir)
	p := newProdWake(t, h.phase1Harness, prodSide{box: h.srcBox, code: h.src, dataDir: h.sdir, bucket: h.srcBucket, aws: h.srcAWS,
		route:  func() registry.Route { return h.srcRoute },
		inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.Rollback.RecoveryInputs }})
	h.recoveryWake = p.recoveryWake
	return p
}

// revertRecovery binds a revert harness's source box (ca-box) to the
// production mode, as sourceRecovery does for a rollback: the revert's
// recovery adapter wakes the claim runtime on ca-box.
func revertRecovery(t *testing.T, h *revertHarness) *prodProbe {
	t.Helper()
	sourceServed(t, h.phase1Harness, h.cfg.DataDir)
	p := newProdWake(t, h.phase1Harness, prodSide{box: h.srcBox, code: h.src, dataDir: h.cfg.DataDir, bucket: h.srcBucket, aws: h.srcAWS,
		route:  func() registry.Route { return h.route },
		inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.RecoveryInputs }})
	h.cfg.recoveryProbe = func(ctx context.Context, s release.Store, r provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
		h.probes = append(h.probes, req)
		w := h.w()
		a := recoveryProbeAdapter{Box: w.cfg.Box, Claims: true, Route: w.cfg.Route, Current: w.recoveryInputs, Wake: p.recoveryWake}
		return a.probe(ctx, s, r, req)
	}
	return p
}

// sourceServed makes the source box and its bucket what serving the
// instance under claims leaves (the harness built them by hand): the
// bucket versioned with a generation.json naming src_gen (sourceMarker);
// the source's installation in the claim's inventory at the flip's source
// stamp; src_gen's files manifest in its bucket, built from the source's
// files and env; and src_gen's origin recorded, as an admission through the
// real CAS (or the #250 backfill) records it.
func sourceServed(t *testing.T, h *phase1Harness, dataDir string) {
	t.Helper()
	ctx := context.Background()
	sourceMarker(t, h)
	c := h.claim(t)
	var err error
	// The source's installation: the flip's journalled stamp and src_gen,
	// or, for a revert whose holder never left the source, the holder's.
	stamp, gen := c.Holder.Stamp, c.Holder.Generation
	if raw, ok := c.Op.Journal[hostclaim.JournalSourceStamp]; ok {
		if stamp, err = strconv.ParseUint(raw, 10, 64); err != nil || stamp == 0 {
			t.Fatalf("fixture: the claim's source stamp %q: %v", raw, err)
		}
		gen = c.Op.Journal[hostclaim.JournalSrcGen]
	}
	// The box's guard directories and its installation record at that
	// stamp, unless a step of the recovery already wrote it.
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(dataDir, d), 0o770); err != nil {
			t.Fatal(err)
		}
	}
	record := filepath.Join(dataDir, "hostguard", p1Host+".json")
	if _, err := os.Stat(record); errors.Is(err, os.ErrNotExist) {
		body := fmt.Sprintf(`{"v":1,"host":%q,"incarnation":%q,"active_stamp":%d,"retired_stamps":[]}`, p1Host, h.inc, stamp)
		if err := os.WriteFile(record, []byte(body), 0o660); err != nil {
			t.Fatal(err)
		}
	}
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		if c.Inventory == nil {
			c.Inventory = map[string]hostclaim.InventoryEntry{}
		}
		c.Inventory[h.srcBox] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": stamp, p1Host + ".db.gen": stamp, p1Host: stamp, p1Host + ".env": stamp}}
	})
	if _, err = buildPreparedManifest(ctx, h.vs, h.srcBucket, cutRoot, gen, p1ID, stamp, filepath.Join(dataDir, p1Host), filepath.Join(dataDir, p1Host+".env"), true,
		hostfiles.WakeState{V: 1, Host: p1Host, Timers: []hostfiles.WakeTimer{}}); err != nil {
		t.Fatalf("src_gen's files manifest: %v", err)
	}
	recordOrigins(t, h.s, p1Host, hostclaim.OriginTuple{Incarnation: h.inc, Location: cutRoot, Region: park.PlacementRegion(h.src), Generation: gen})
}

// firstMoving is the first claim written with the move's op: the start
// CAS's.
func firstMoving(t *testing.T, writes []hostclaim.Claim) *hostclaim.Claim {
	t.Helper()
	for i := range writes {
		if writes[i].Op.Kind == hostclaim.KindMoving {
			return &writes[i]
		}
	}
	t.Fatal("no claim was written with the move's op")
	return nil
}

// sourceMarker makes the source's bucket what a source that served the
// instance leaves: every object versioned (the harness stored the replica
// directly), and the location's generation.json a marker naming src_gen,
// cur, on ca-box (the harness's placeholder is not one).
func sourceMarker(t *testing.T, h *phase1Harness) {
	t.Helper()
	body, err := json.Marshal(genselect.Marker{Op: "cur", Incarnation: h.inc, Host: p1Host, Box: h.srcBox, InstalledAt: p1Created})
	if err != nil {
		t.Fatal(err)
	}
	h.f.store(h.srcBucket, cutRoot+"/generation.json", fakeMoveObject{lease.MirrorObject{ETag: "marker", Size: int64(len(body))}, string(body)})
	h.f.setVersion(h.srcBucket, cutRoot+"/generation.json", h.f.nextVersion())
	for k := range h.f.buckets[h.srcBucket] {
		if h.f.versionOf(h.srcBucket, k) == "" {
			h.f.setVersion(h.srcBucket, k, h.f.nextVersion())
		}
	}
}

// homeRegionalOn is home-box's regional runner over a harness's store and
// lease table.
func homeRegionalOn(h *phase1Harness, leases park.Leases) hostbox.Regional {
	return regionalRunner(h, leases, "ie", "ie", "data-home")
}

// TestCommittedLandingIntoHomeConvertsOnTargetFence (Ask H1's witness):
// after the box's commit of a move into home (uk -> home, the home box dst
// in roster region ie, the fleet's home), dst is fenced as terminated, and
// region HA's committed-landing conversion turns the claim into a failover
// relocation: intent.region is the roster name ie, move_to_region the
// move's @home, the home binding kept. The commit's chores (the lane park,
// owed under @home or, retained, under ie; the source's location park in
// uk) come through byte-identical, each in the spelling it was owed in.
func TestCommittedLandingIntoHomeConvertsOnTargetFence(t *testing.T) {
	for _, lane := range []string{park.HomeRegion, "ie"} {
		t.Run("lane owed under "+lane, func(t *testing.T) {
			ctx := t.Context()
			w := newLandingWorldHome(t, true, nil, "ie")
			rec, err := provision.GetRecord(ctx, w.store, "bab", "jam", w.obs.Host)
			if err != nil {
				t.Fatal(err)
			}
			o, err := hostclaim.Read(ctx, w.store, w.obs.Host)
			if err != nil {
				t.Fatal(err)
			}
			source, staging, err := moves.CommitScopes(o.Claim, rec.Move.FlippedAt)
			if err != nil {
				t.Fatal(err)
			}
			id := o.Claim.Op.ID
			if lane != park.HomeRegion {
				// The lane park already owed in its retained spelling.
				if o, err = hostclaim.Update(ctx, w.store, o, func(c *hostclaim.Claim) error {
					s := staging
					c.Obligations = append(c.Obligations, hostclaim.Obligation{ID: retainedID(20), Op: id, Kind: "park-replica", Region: lane, Retirement: retainedID(21), Owner: "region:" + lane, Park: &s})
					return nil
				}); err != nil {
					t.Fatal(err)
				}
			}
			// The box's commit, as commitClaim writes it: the record's codes
			// through park.PlacementRegion, the target @home.
			if o, err = hostclaim.Update(ctx, w.store, o, hostclaim.KeepWriterSession(id, hostclaim.Commit(id, park.PlacementRegion(rec.Move.From), "src", park.PlacementRegion(rec.Move.To), source, staging, newClaimID))); err != nil {
				t.Fatal(err)
			}
			if o.Claim.Op.Phase != hostclaim.PhaseCommitted || o.Claim.Op.Intent["to"] != park.HomeRegion {
				t.Fatalf("the commit: %+v", o.Claim.Op)
			}
			committed := o.Claim.Obligations
			lanes := 0
			for _, ob := range committed {
				if ob.Park != nil && ob.Park.Location == moves.ScopeLocation(w.obs.Host, id) {
					lanes++
					if ob.Region != lane || ob.Owner != "region:"+lane {
						t.Fatalf("the lane park after the commit: %+v", ob)
					}
				}
			}
			if lanes != 1 {
				t.Fatalf("the commit owes %d lane parks: %+v", lanes, committed)
			}

			if err := convertFencedLanding(w.homeWorker(w.enrollment["wrk"], "ie", "ie"), w.store)(ctx, w.store, o); err != nil {
				t.Fatalf("convert: %v", err)
			}
			after, err := hostclaim.Read(ctx, w.store, w.obs.Host)
			if err != nil {
				t.Fatal(err)
			}
			c := after.Claim
			if c.Op.Kind != hostclaim.KindRelocating || c.Op.Journal["conversion"] != "committed-landing" || c.Op.Journal["conversion_landing_phase"] != hostclaim.PhaseCommitted {
				t.Fatalf("not converted from committed: %+v", c.Op)
			}
			if c.Op.Intent["region"] != "ie" || c.Op.Intent["move_to_region"] != park.HomeRegion || c.Op.Intent["home_region"] != "ie" || c.Op.Journal["move_from_region"] != "uk" {
				t.Fatalf("intent %v journal from %q", c.Op.Intent, c.Op.Journal["move_from_region"])
			}
			for _, before := range committed {
				kept := false
				for _, now := range c.Obligations {
					if now.ID == before.ID {
						if obligationsJSON(t, []hostclaim.Obligation{now}) != obligationsJSON(t, []hostclaim.Obligation{before}) {
							t.Fatalf("the conversion rewrote %+v as %+v", before, now)
						}
						kept = true
					}
				}
				if !kept {
					t.Fatalf("the conversion dropped %+v", before)
				}
			}
			for _, now := range c.Obligations {
				if now.Region == "" {
					t.Fatalf("a chore in the unset region: %+v", now)
				}
				// From clean state (the lane not retained), every chore is
				// the source's (uk) or @home: nothing supplied the roster
				// name.
				if lane == park.HomeRegion && now.Region != "uk" && (now.Region != park.HomeRegion || now.Owner != "region:"+park.HomeRegion) {
					t.Fatalf("a chore %+v after a commit from clean state", now)
				}
			}
		})
	}
}