package host

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"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"
)

// The Phase 1 harness: the legacy moveHarness, its fake mirror extended
// with the staging writer's guarded At methods (F17) and a call log, a
// lease.VersionedStore over the same buckets, a staging-lease fake, a
// stepping clock, a MoveStart'ed record, and the host claim the request
// door would have written. fakeMoveMirror deliberately does NOT gain
// CopyVersioned: MirrorPrefix would then run the legacy tests in its
// versioned mode.

// stagingCall is one guarded staging request the fake received.
type stagingCall struct {
	Op, Bucket, Key string
	SignAt          time.Time
}

func (f *fakeMoveMirror) logAt(op string, signAt time.Time, bucket, key string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.at = append(f.at, stagingCall{op, bucket, key, signAt})
	if f.failAt != nil {
		return f.failAt(op, key)
	}
	return nil
}

func (f *fakeMoveMirror) nextVersion() string {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.versions++
	return fmt.Sprintf("ver-%04d", f.versions)
}

func (f *fakeMoveMirror) CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) {
	if err := f.logAt("copy", signAt, dstBucket, dstKey); err != nil {
		return "", "", err
	}
	etag, err := f.Copy(ctx, srcBucket, srcKey, srcETag, dstBucket, dstKey)
	if err != nil {
		return "", "", err
	}
	ver := f.nextVersion()
	f.setVersion(dstBucket, dstKey, ver)
	return etag, ver, nil
}

func (f *fakeMoveMirror) putAt(op string, signAt time.Time, bucket, key string, body []byte, createOnly bool) (string, error) {
	if err := f.logAt(op, signAt, bucket, key); err != nil {
		return "", err
	}
	f.bmu.Lock()
	_, exists := f.buckets[bucket][key]
	f.bmu.Unlock()
	if exists && createOnly {
		return "", lease.ErrExists
	}
	ver := f.nextVersion()
	f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
	f.setVersion(bucket, key, ver)
	return ver, nil
}

// setVersion records the current version of bucket/key. The legacy
// fakeMoveObject has no version field (its positional literals stay as
// they are), so versions live beside it.
func (f *fakeMoveMirror) setVersion(bucket, key, ver string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.vers == nil {
		f.vers = map[string]string{}
	}
	f.vers[bucket+"/"+key] = ver
	if f.keepHistory {
		if f.history == nil {
			f.history = map[string]fakeMoveObject{}
		}
		f.bmu.Lock()
		f.history[bucket+"/"+key+"@"+ver] = f.buckets[bucket][key]
		f.bmu.Unlock()
	}
}

// older is a non-current version of bucket/key the fake kept (keepHistory),
// as a versioned bucket keeps every version a later write superseded.
func (f *fakeMoveMirror) older(bucket, key, ver string) (fakeMoveObject, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	o, ok := f.history[bucket+"/"+key+"@"+ver]
	return o, ok && f.keepHistory
}

func (f *fakeMoveMirror) versionOf(bucket, key string) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.vers[bucket+"/"+key]
}

func (f *fakeMoveMirror) PutVersionedAt(_ context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) {
	return f.putAt("put", signAt, bucket, key, body, false)
}

func (f *fakeMoveMirror) PutReaderVersionedAt(_ context.Context, signAt time.Time, bucket, key string, r io.Reader, size int64, sha256Hex string) (string, error) {
	body, err := io.ReadAll(io.LimitReader(r, size+1))
	if err != nil {
		return "", err
	}
	if int64(len(body)) != size {
		return "", fmt.Errorf("fake: body is %d bytes, declared %d", len(body), size)
	}
	if sum := sha256.Sum256(body); hex.EncodeToString(sum[:]) != sha256Hex {
		return "", errors.New("fake: XAmzContentSHA256Mismatch")
	}
	return f.putAt("put", signAt, bucket, key, body, false)
}

func (f *fakeMoveMirror) PutIfAbsentVersionedAt(_ context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) {
	return f.putAt("put-if-absent", signAt, bucket, key, body, true)
}

func (f *fakeMoveMirror) DeleteAt(ctx context.Context, signAt time.Time, bucket, key string) error {
	if err := f.logAt("delete", signAt, bucket, key); err != nil {
		return err
	}
	return f.Delete(ctx, bucket, key)
}

var _ lease.StagingWriter = (*fakeMoveMirror)(nil)

// stagingCalls is the guarded call log, by op and key.
func (f *fakeMoveMirror) stagingCalls() []stagingCall {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]stagingCall(nil), f.at...)
}

// fakeVersioned is a lease.VersionedStore over the fake mirror's buckets,
// reading by exact version. deny refuses a call the way an IAM policy does.
type fakeVersioned struct {
	f     *fakeMoveMirror
	mu    sync.Mutex
	calls []string
	deny  func(op, bucket, key string) error
	// skipDelete, when it answers true for a key, makes DeleteVersion
	// return success without removing the version: a delete the store
	// accepted but whose listing still shows the version (Plan 2 Task 10,
	// the landing's Parked gate).
	skipDelete func(key string) bool
}

func (v *fakeVersioned) call(op, bucket, key string) error {
	v.mu.Lock()
	v.calls = append(v.calls, op+" "+bucket+"/"+key)
	v.mu.Unlock()
	if v.deny != nil {
		return v.deny(op, bucket, key)
	}
	return nil
}

func (v *fakeVersioned) ListVersions(_ context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
	if err := v.call("list-versions", bucket, prefix); err != nil {
		return nil, err
	}
	var out []lease.ObjectVersion
	for k, o := range v.f.buckets[bucket] {
		if strings.HasPrefix(k, prefix) {
			out = append(out, lease.ObjectVersion{Key: k, VersionID: v.versionID(bucket, k), ETag: o.ETag, Size: o.Size, IsLatest: true})
		}
	}
	return out, nil
}

// versionID is an object's version as S3 lists it: one written with no
// version (the source's replica, stored directly by a test) is "null",
// as S3 lists an object written while a bucket was not versioning.
func (v *fakeVersioned) versionID(bucket, key string) string {
	if ver := v.f.versionOf(bucket, key); ver != "" {
		return ver
	}
	return "null"
}

func (v *fakeVersioned) GetVersion(_ context.Context, bucket, key, versionID string) ([]byte, error) {
	if err := v.call("get-version", bucket, key); err != nil {
		return nil, err
	}
	o, ok := v.f.buckets[bucket][key]
	if !ok || versionID == "" || v.versionID(bucket, key) != versionID {
		if old, kept := v.f.older(bucket, key, versionID); kept && versionID != "" {
			return []byte(old.body), nil
		}
		return nil, lease.ErrMirrorNotFound
	}
	return []byte(o.body), nil
}

func (v *fakeVersioned) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
	b, err := v.GetVersion(ctx, bucket, key, versionID)
	if err != nil {
		return nil, err
	}
	return io.NopCloser(bytes.NewReader(b)), nil
}

// CopyVersion copies one exact version within the fake's buckets, as the
// park engine archives a landing's residue (Plan 2 Task 10). createOnly
// refuses an existing destination with lease.ErrDestinationExists.
func (v *fakeVersioned) CopyVersion(_ context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
	if err := v.call("copy-version", dstBucket, dstKey); err != nil {
		return "", "", err
	}
	o, ok := v.f.buckets[srcBucket][srcKey]
	if !ok || srcVersionID == "" || v.versionID(srcBucket, srcKey) != srcVersionID {
		old, kept := v.f.older(srcBucket, srcKey, srcVersionID)
		if !kept || srcVersionID == "" {
			return "", "", lease.ErrMirrorNotFound
		}
		o = old
	}
	if _, exists := v.f.buckets[dstBucket][dstKey]; exists && createOnly {
		return "", "", lease.ErrDestinationExists
	}
	ver := v.f.nextVersion()
	v.f.store(dstBucket, dstKey, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: o.Size}, o.body})
	v.f.setVersion(dstBucket, dstKey, ver)
	return ver, "etag-" + ver, nil
}

func (v *fakeVersioned) DeleteVersion(_ context.Context, bucket, key, versionID string) error {
	if err := v.call("delete-version", bucket, key); err != nil {
		return err
	}
	if v.skipDelete != nil && v.skipDelete(key) {
		return nil
	}
	// "null" is how S3 names an object written while the bucket was not
	// versioning (versionID), and deleting that version removes it.
	if _, ok := v.f.buckets[bucket][key]; ok && v.versionID(bucket, key) == versionID {
		delete(v.f.buckets[bucket], key)
	}
	return nil
}

func (v *fakeVersioned) PutIfAbsent(_ context.Context, bucket, key string, body []byte) (string, error) {
	if err := v.call("put-if-absent", bucket, key); err != nil {
		return "", err
	}
	if _, ok := v.f.buckets[bucket][key]; ok {
		return "", lease.ErrExists
	}
	ver := v.f.nextVersion()
	v.f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
	v.f.setVersion(bucket, key, ver)
	return ver, nil
}

func (v *fakeVersioned) callsOn(prefix string) []string {
	v.mu.Lock()
	defer v.mu.Unlock()
	var out []string
	for _, c := range v.calls {
		if strings.Contains(c, prefix) {
			out = append(out, c)
		}
	}
	return out
}

// fakeStagingLease is the staging-work lease: safe until `until`.
type fakeStagingLease struct {
	mu       sync.Mutex
	until    time.Time
	acquired int
	released int
	// unsafeAfter, when positive, makes every SafeUntil read after that
	// many answer the lease's start: too close to its end to sign.
	reads       int
	unsafeAfter int
}

func (l *fakeStagingLease) SafeUntil() (time.Time, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.reads++
	if l.unsafeAfter > 0 && l.reads > l.unsafeAfter {
		return l.until.Add(-moves.StagingLeaseTTL), nil
	}
	return l.until, nil
}

// stepClock advances a millisecond each read, so every timed span is
// positive and the order of events is the order of reads.
type stepClock struct {
	mu  sync.Mutex
	now time.Time
}

func (c *stepClock) Now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.now = c.now.Add(time.Millisecond)
	return c.now
}

func (c *stepClock) Advance(d time.Duration) {
	c.mu.Lock()
	c.now = c.now.Add(d)
	c.mu.Unlock()
}

const (
	p1Host = "a.go.example.com"
	p1ID   = "mv_abcdefghijklmnopqrstuvwxyz"
	p1ID2  = "mv_bcdefghijklmnopqrstuvwxyza"
)

var p1Created = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)

type phase1Harness struct {
	*moveHarness
	vs         *fakeVersioned
	lease      *fakeStagingLease
	clock      *stepClock
	inc        string
	route      registry.Route
	env        map[string]string // the source's launch environment
	optDir     string
	probe      moves.Check
	probePaths []string
	strict     moves.Check
	// worker1 is the source box's worker and worker2 the target's, each
	// kept across passes as one agent keeps its own.
	worker1, worker2 *movesWorker
	// src is the source box's placement code: "ie", or the home's "" for a
	// move out of home (fromHome).
	src string
	// The source box, its data bucket and the bucket's AWS region: ie-box,
	// data-ie and eu-west-1 by default.
	srcBox, srcBucket, srcAWS string
	// dst is the target box's placement code: "uk", or the home's "" for a
	// move into home (intoHome). dstBox, dstBucket and dstAWS name the
	// target box, its data bucket and the bucket's AWS region.
	dst, dstBox, dstBucket, dstAWS string
	// dstEngine is the target box's claim engine region, its roster name:
	// "uk", or the home box's roster name "ie" for a move into home (claim
	// Plan 13 keeps the roster name as a home box's engine region and grants
	// it the @home alias).
	dstEngine string
	// homeRegion is the fleet's signed home designation, "" when the
	// fixture names none. With it the claim's op carries home_region and the
	// target's ParkHomeRegion is set, as armMovesParks sets it.
	homeRegion string
	// admitted starts the move as the claim-mode door does
	// (hostclaim.BeginAdmittedMove with the roster's enrollments), not the
	// plain BeginMove.
	admitted bool
	// roster is the signed roster the fixture's boxes resolve through.
	roster p1Roster
	// retain is the claim state seeded at start as an earlier stay left it
	// (retainTouched, retainQuiet, retainChore).
	retain p1Retained
	// realEnv composes both sides' environments through the production
	// composer (realLaunchEnv): the source's launch report is its
	// ExecBackend's LaunchEnvironment, the target's held env composeHeldEnv
	// with the target's own region, each box's agent-owned variables its
	// own.
	realEnv bool
	// srcEnvFile is the source's own <host>.env, the instance layer the
	// archive carries: "APP_MODE=prod\n" by default (sourceEnvFile).
	srcEnvFile string
}

// sourceEnvFile sets the source's own <host>.env.
func sourceEnvFile(body string) p1Option { return func(h *phase1Harness) { h.srcEnvFile = body } }

// realLaunchEnv composes both boxes' environments through the production
// composer instead of the h.env fixture (home part 2, Task 2b): the source
// pins what its ExecBackend would launch, and the target composes its own,
// so the box-owned variables differ between the two exactly as they do in
// production.
func realLaunchEnv(h *phase1Harness) { h.realEnv = true }

// heldEnv is the target's held environment for rec: the h.env fixture with
// the target's region rule, or under realLaunchEnv the production
// composition over the carried env with the target's region.
func (h *phase1Harness) heldEnv(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, error) {
	if h.realEnv {
		env, instance, _, _, err := composeHeldEnv(h.cfg.SecretsPriv, h.dst, false)(ctx, s, rec, carried)
		return env, instance, err
	}
	env := map[string]string{}
	for k, v := range h.env {
		env[k] = v
	}
	setHeldRegion(env, h.dst)
	return env, nil, nil
}

// p1Roster is the fixture's signed roster: each box's actor, as
// hostbox.Roster answers it.
type p1Roster map[string]hostbox.Actor

func (r p1Roster) Actor(_ context.Context, box string) (hostbox.Actor, error) {
	a, ok := r[box]
	if !ok {
		return hostbox.Actor{}, hostbox.ErrChanged
	}
	return a, nil
}

// p1Retained is claim state an earlier stay left, seeded at start in
// whichever spelling of home the option names: touched.regions, a
// quiet_after key, and an earlier residue chore owed in the home region.
type p1Retained struct {
	touched, quiet, chore string
}

// retainedChoreOp is the earlier op whose residue chore retainChore seeds,
// and retainedChoreGeneration the earlier stay's generation it parks.
const (
	retainedChoreOp         = "mv_earlierstayearlierstayearlie"
	retainedChoreGeneration = "old"
)

// The fixture's enrollments (the roster's 32-hex tokens).
const (
	p1SrcEnrollment = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1"
	p1DstEnrollment = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa2"
)

// p1Option adjusts a Phase 1 harness before it is built.
type p1Option func(*phase1Harness)

// fromHome builds the harness for a move out of the home placement: the
// source box, its record and its move name the home's code, "" (home-region
// parks, part 1). The source box keeps its name and bucket.
func fromHome(h *phase1Harness) { h.src = "" }

// intoHome builds the harness for a move into the home placement, the
// mirror of fromHome for the target (home part 2). The source is placed in a
// named code that is not the home's name (ca-box in ca, bucket data-ca), so
// no fixture name suggests home. The target is home-box: placement "",
// bucket data-home in eu-west-1, an edges file with a data bucket and no
// placement, and a held environment with no CARLOS_REGION. The roster names
// home-box in region ie and designates ie the fleet's home; the claim's op
// carries home_region ie (BeginAdmittedMove), and the target's
// ParkHomeRegion is ie, as armMovesParks sets it.
func intoHome(h *phase1Harness) {
	h.src, h.srcBox, h.srcBucket, h.srcAWS = "ca", "ca-box", "data-ca", "ca-central-1"
	h.dst, h.dstBox, h.dstBucket, h.dstAWS, h.dstEngine = "", "home-box", "data-home", "eu-west-1", "ie"
	h.homeRegion, h.admitted = "ie", true
}

// retainTouched seeds touched.regions with region, an earlier stay's
// spelling of home ("ie" or "@home").
func retainTouched(region string) p1Option {
	return func(h *phase1Harness) { h.retain.touched = region }
}

// retainQuiet seeds a past quiet_after key under region, either spelling.
func retainQuiet(region string) p1Option { return func(h *phase1Harness) { h.retain.quiet = region } }

// retainChore seeds an earlier op's residue chore owed under region (the
// roster name "ie" for home), after the start, as an earlier stay left it:
// a park-generation of the generation retainedChoreGeneration, scoped to
// the holder's location, the shape hostclaim.QueueResidueParks queues.
func retainChore(region string) p1Option { return func(h *phase1Harness) { h.retain.chore = region } }

func newPhase1Harness(t *testing.T, opts ...p1Option) *phase1Harness {
	t.Helper()
	h := &phase1Harness{moveHarness: newMoveHarness(t), lease: &fakeStagingLease{}, clock: &stepClock{now: time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)}, src: "ie",
		srcEnvFile: "APP_MODE=prod\n", srcBox: "ie-box", srcBucket: "data-ie", srcAWS: "eu-west-1", dst: "uk", dstBox: "uk-box", dstBucket: "data-uk", dstAWS: "eu-west-2", dstEngine: "uk"}
	for _, o := range opts {
		o(h)
	}
	h.cfg.Region = h.src
	h.cfg.OwnBucket, h.cfg.OwnBucketRegion = h.srcBucket, h.srcAWS
	h.roster = p1Roster{
		h.srcBox: {Box: h.srcBox, Enrollment: p1SrcEnrollment, Region: park.PlacementRegion(h.src), FleetHomeRegion: h.homeRegion, Bucket: h.srcBucket, Eligible: true},
		h.dstBox: {Box: h.dstBox, Enrollment: p1DstEnrollment, Region: h.dstEngine, FleetHomeRegion: h.homeRegion, Bucket: h.dstBucket, Eligible: true},
	}
	if h.dst == "" {
		// The home box's roster entry equals the fleet's home designation:
		// the claim grants it the @home alias (claim Plan 13).
		a := h.roster[h.dstBox]
		a.HomeRegion = h.homeRegion
		h.roster[h.dstBox] = a
	}
	h.vs = &fakeVersioned{f: h.f}
	h.lease.until = h.clock.now.Add(moves.StagingLeaseTTL)
	inc, err := hostclaim.Incarnation("bab", "app", p1Host, p1Created)
	if err != nil {
		t.Fatal(err)
	}
	h.inc = inc
	privB64, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := appconfig.ParsePrivateKey(privB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	h.optDir = t.TempDir()
	h.route = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "ready",
		Socket: provision.DefaultSocket(p1Host), DBPath: provision.DefaultDBPath(h.cfg.DataDir, p1Host), Provisioned: true}
	h.env = map[string]string{"CARLOS_REGION": h.src, "CARLOS_CONTROL_SOCKET": "/run/carlos/" + p1Host + ".sock.control", "APP_MODE": "prod"}
	h.probe, h.strict = moves.Check{OK: true}, moves.Check{OK: true}
	ctx := context.Background()

	h.cfg.Enabled = true
	h.cfg.Box = h.srcBox
	h.cfg.Now = h.clock.Now
	h.cfg.OptDir = h.optDir
	h.cfg.DeployBucket = "deploy"
	h.cfg.StagingLease = func(context.Context, string, string) (moves.SafeUntilGuard, func(context.Context) error, error) {
		h.lease.mu.Lock()
		h.lease.acquired++
		h.lease.mu.Unlock()
		return h.lease, func(context.Context) error {
			h.lease.mu.Lock()
			h.lease.released++
			h.lease.mu.Unlock()
			return nil
		}, nil
	}
	h.cfg.Staging = func(context.Context, string) (lease.StagingWriter, error) { return h.f, nil }
	h.cfg.Versioned = func(context.Context, string, string) (lease.VersionedStore, error) { return h.vs, nil }
	h.cfg.SecretsKey, h.cfg.SecretsPriv, h.cfg.SecretsPub = privB64, priv, priv.PublicKey()
	h.cfg.LaunchEnv = func(string) (map[string]string, map[string]bool, error) { return h.env, nil, nil }
	if h.realEnv {
		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
		}
	}
	h.cfg.FilesDir = func(host string) (string, string) {
		return filepath.Join(h.cfg.DataDir, host), filepath.Join(h.cfg.DataDir, host+".env")
	}
	h.cfg.ReadyProbe = func(_ context.Context, _, path string) moves.Check {
		h.probePaths = append(h.probePaths, path)
		return h.probe
	}
	h.cfg.StrictNext = func(string) moves.Check { return h.strict }
	h.cfg.Route = func(host string) (*registry.Route, error) {
		if host != p1Host {
			return nil, nil
		}
		rt := h.route
		return &rt, nil
	}

	// The source's files and carried .env, its live release v7, and its
	// replica: two generations under the scoped location (g-old and the
	// live g-cur), plus a generation.json and a lock.json beside them.
	files := filepath.Join(h.cfg.DataDir, p1Host)
	if err := os.MkdirAll(files, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(files, "upload.txt"), []byte("hello"), 0o644); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host+".env"), []byte(h.srcEnvFile), 0o600); err != nil {
		t.Fatal(err)
	}
	pointLiveForTest(t, h.optDir, p1Host, "v7")
	loc := "instances/" + p1Host + "/" + inc + "/"
	h.f.store(h.srcBucket, loc+"g-cur/0000/0000000000000001-0000000000000002.ltx",
		fakeMoveObject{lease.MirrorObject{ETag: "cur-a", Size: 2 << 20}, string(ltxForTest(t, 4096, 4, 1, 2))})
	h.f.store(h.srcBucket, loc+"g-cur/0001/0000000000000001-0000000000000002.ltx",
		fakeMoveObject{lease.MirrorObject{ETag: "cur-b", Size: 4096}, string(ltxForTest(t, 4096, 4, 1, 2))})
	h.f.putSized(h.srcBucket, loc+"g-old/0000/0000000000000001-0000000000000001.ltx", "old-a", 4096)
	h.f.put(h.srcBucket, loc+"generation.json", "gen")
	h.f.put(h.srcBucket, loc+"g-cur/lock.json", "lock")

	// The source's edges file: ie publishes its data bucket too (T10).
	if _, err := h.s.Put(ctx, "control/edges/"+h.srcBox+".json",
		[]byte(`{"v":1,"box":"`+h.srcBox+`","eip":"1.2.3.5","placement":"`+h.src+`","data_bucket":"`+h.srcBucket+`","data_bucket_region":"`+h.srcAWS+`","hosts":{}}`), ""); err != nil {
		t.Fatal(err)
	}
	// The target's edges file, when it is not newMoveHarness's uk-box: a
	// home box publishes its data bucket and no placement.
	if h.dstBox != "uk-box" {
		if err := h.s.Delete(ctx, "control/edges/uk-box.json"); err != nil {
			t.Fatal(err)
		}
		placement := ""
		if h.dst != "" {
			placement = `"placement":"` + h.dst + `",`
		}
		if _, err := h.s.Put(ctx, "control/edges/"+h.dstBox+".json",
			[]byte(`{"v":1,"box":"`+h.dstBox+`","eip":"1.2.3.4",`+placement+`"data_bucket":"`+h.dstBucket+`","data_bucket_region":"`+h.dstAWS+`","hosts":{}}`), ""); err != nil {
			t.Fatal(err)
		}
	}
	// The app's instances manifest: it declares no readiness path.
	if err := provision.PutManifest(ctx, h.s, "bab", "app", provision.Manifest{Schema: 1}); err != nil {
		t.Fatal(err)
	}
	// The console's record (T2).
	if _, err := h.s.Put(ctx, moves.ConsoleKey, []byte(`{"v":1,"protocol":1,"secrets_key_fp":"`+appconfig.KeyFingerprint(priv.PublicKey())+`"}`), ""); err != nil {
		t.Fatal(err)
	}
	return h
}

// pointLiveForTest makes optDir/live/<host> point at version v, the shape
// release.LiveVersion reads.
func pointLiveForTest(t *testing.T, optDir, host, v string) {
	t.Helper()
	target := filepath.Join(optDir, "versions", v, "bab", "app")
	if err := os.MkdirAll(target, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(optDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	link := filepath.Join(optDir, "live", host)
	os.Remove(link)
	if err := os.Symlink(target, link); err != nil {
		t.Fatal(err)
	}
}

// start writes the record (MoveStart'ed, id p1ID, cut mode cut) and the
// claim the door's CAS leaves: serving under exec ie-box, then moving with
// target uk-box. The claim has touched ie and ie-box only. Under intoHome
// the source is ca-box, the target home-box and the move's region the
// placement-derived @home, begun with the admission as the claim-mode door
// begins it (BeginAdmittedMove, home_region ie). The retained options add
// the claim state an earlier stay left.
func (h *phase1Harness) start(t *testing.T, cut string) provision.Record {
	t.Helper()
	ctx := context.Background()
	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: h.dst, Cut: cut, 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: h.srcBox}
		c.Touched = hostclaim.Touched{Complete: true, Regions: []string{park.PlacementRegion(h.src)}, Boxes: []string{h.srcBox}}
		if h.retain.touched != "" {
			c.Touched.Regions = append(c.Touched.Regions, h.retain.touched)
		}
		if h.retain.quiet != "" {
			c.QuietAfter = map[string]time.Time{h.retain.quiet: h.clock.Now().Add(-time.Hour).UTC()}
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	to := park.PlacementRegion(h.dst)
	begin := hostclaim.BeginMove(p1ID, "op", h.srcBox, h.dstBox, to, h.clock.Now())
	if h.admitted {
		begin = hostclaim.BeginAdmittedMove(p1ID, "op", h.srcBox, h.dstBox, to, h.clock.Now(), hostclaim.MoveAdmission{HomeRegion: h.homeRegion,
			AppID: "app_" + p1Host, AdmissionID: p1ID, SourceEnrollment: h.roster[h.srcBox].Enrollment, TargetEnrollment: h.roster[h.dstBox].Enrollment})
	}
	if _, err := hostclaim.Update(ctx, h.s, obs, begin); err != nil {
		t.Fatal(err)
	}
	if h.retain.chore != "" {
		// An earlier stay's residue chore in the home region, owed under
		// the spelling the earlier op wrote: one generation of the earlier
		// stay, parked under the location's scope, as QueueResidueParks
		// queues a landing's residue. Seeded after the start: the start
		// guard refuses a move into a region with a chore owed.
		rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
			c.Obligations = append(c.Obligations, hostclaim.Obligation{ID: "chore-earlier", Op: retainedChoreOp, Kind: "park-generation", Region: h.retain.chore,
				Generation: retainedChoreGeneration, Owner: "region:" + h.retain.chore, Retirement: "retirement-earlier",
				Park: &hostclaim.ParkScope{Location: c.Holder.Location, Incarnation: h.inc, EvidenceAt: h.clock.Now().Add(-time.Hour).UTC(), RequestLimit: moves.ParkRequestLimit}})
		})
	}
	return h.read(t, p1Host)
}

func (h *phase1Harness) claim(t *testing.T) *hostclaim.Claim {
	t.Helper()
	obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
	if err != nil || obs.Absent {
		t.Fatalf("claim: %v absent=%v", err, obs.Absent)
	}
	return obs.Claim
}

// pass runs one source pass.
func (h *phase1Harness) pass(t *testing.T) {
	t.Helper()
	if err := movesPass(context.Background(), h.s, h.w()); err != nil {
		t.Fatal(err)
	}
}

// w is one worker kept across the harness's passes, as one agent keeps it.
func (h *phase1Harness) w() *movesWorker {
	if h.worker1 == nil {
		h.worker1 = &movesWorker{cfg: h.cfg, mirrors: map[string]lease.MirrorStore{}}
	}
	h.worker1.cfg = h.cfg
	return h.worker1
}

// targetCfg is the uk box's configuration over the same store and buckets,
// its hooks stubbed to pass (each test turns one off). Under intoHome it is
// home-box's: placement "", ParkHomeRegion ie, and a held environment with
// no CARLOS_REGION.
func (h *phase1Harness) targetCfg() moveCfg {
	c := h.cfg
	c.Region, c.Box, c.OwnBucket, c.OwnBucketRegion = h.dst, h.dstBox, h.dstBucket, h.dstAWS
	if h.dst == "" {
		c.ParkHomeRegion = h.homeRegion
	}
	c.OwnsRoute = func(string) bool { return false }
	c.Route = func(string) (*registry.Route, error) { return nil, nil }
	c.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) {
		env, instance, err := h.heldEnv(ctx, s, rec, carried)
		return env, instance, nil, provision.MoveInputs{BundleETags: map[string]string{"default": "e1", "environment": ""}}, err
	}
	c.ProbeStore = func(context.Context, map[string]string, string) (string, error) { return "", nil }
	c.DiskFree = func(string) (uint64, uint64, error) { return 100 << 30, 90 << 30, nil }
	c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check {
		return moves.Check{OK: true}
	}
	c.LaneProbe = func(context.Context, string, string, string) error { return nil }
	c.MeasureS3 = func(context.Context, string) (moves.MeasuredRate, error) {
		return moves.MeasuredRate{MiBPerS: 250, At: h.clock.Now(), Bytes: 32 << 20}, nil
	}
	c.MeasureDisk = func(context.Context) (moves.MeasuredRate, moves.FilesRate, error) {
		now := h.clock.Now()
		return moves.MeasuredRate{MiBPerS: 500, At: now, Bytes: 128 << 20}, moves.FilesRate{PerS: 1600, At: now, Files: 512}, nil
	}
	// The cut guard's T13: the target states it can land and roll back,
	// as every box will once the wake-source gate opens (armStandIns).
	armStandIns(&c, h.s, roleTarget)
	return c
}

// setHeldRegion is the agent's rule for the target's held environment
// (agentOwnedVars): CARLOS_REGION is the box's placement when it has one,
// and absent on a home box.
func setHeldRegion(env map[string]string, placement string) {
	if placement == "" {
		delete(env, "CARLOS_REGION")
		return
	}
	env["CARLOS_REGION"] = placement
}

// targetPass runs one pass as the target box.
func (h *phase1Harness) targetPass(t *testing.T, c moveCfg) {
	t.Helper()
	if h.worker2 == nil {
		h.worker2 = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	}
	h.worker2.cfg = c
	if err := movesPass(context.Background(), h.s, h.worker2); err != nil {
		t.Fatal(err)
	}
}

func (h *phase1Harness) targetReport(t *testing.T) *moves.TargetReport {
	t.Helper()
	r, err := readTargetReport(context.Background(), h.s, h.read(t, p1Host))
	if err != nil {
		t.Fatal(err)
	}
	return r
}

func (h *phase1Harness) sourceReport(t *testing.T) *moves.SourceReport {
	t.Helper()
	r, err := readSourceReport(context.Background(), h.s, h.read(t, p1Host))
	if err != nil {
		t.Fatal(err)
	}
	return r
}

func (h *phase1Harness) ledger(t *testing.T) moveLedger {
	t.Helper()
	l, err := readMoveLedger(context.Background(), h.f, h.srcBucket, moves.LedgerKey(p1Host, p1ID))
	if err != nil {
		t.Fatal(err)
	}
	return l
}

// ltxForTest is a real LTX file (lz4-compressed pages) for a database of
// commit pages of pageSize bytes, covering minTX to maxTX. A snapshot
// (minTX 1) carries every page; an increment carries page 1 alone. Pages
// are zeros, so the file is far smaller than the database it describes.
func ltxForTest(t *testing.T, pageSize, commit uint32, minTX, maxTX uint64) []byte {
	t.Helper()
	var buf bytes.Buffer
	enc, err := ltx.NewEncoder(&buf)
	if err != nil {
		t.Fatal(err)
	}
	hdr := ltx.Header{Version: ltx.Version, PageSize: pageSize, Commit: commit, MinTXID: ltx.TXID(minTX), MaxTXID: ltx.TXID(maxTX), Timestamp: 1000}
	if minTX > 1 {
		hdr.PreApplyChecksum = ltx.ChecksumFlag | 1
	}
	if err := enc.EncodeHeader(hdr); err != nil {
		t.Fatal(err)
	}
	pages := uint32(1)
	if minTX == 1 {
		pages = commit
	}
	page := make([]byte, pageSize)
	for pgno := uint32(1); pgno <= pages; pgno++ {
		if pgno == ltx.LockPgno(pageSize) {
			continue
		}
		if err := enc.EncodePage(ltx.PageHeader{Pgno: pgno}, page); err != nil {
			t.Fatal(err)
		}
	}
	enc.SetPostApplyChecksum(ltx.ChecksumFlag | 2)
	if err := enc.Close(); err != nil {
		t.Fatal(err)
	}
	return buf.Bytes()
}