//go:build linux

package host

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"reflect"
	"slices"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/console"
	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/instancekey"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/schedule"
	"github.com/carlosframework/platform/internal/status"
)

// The offline operator move end to end (Task 8 of docs/superpowers/plans/
// 2026-10-20-offline-operator-move.md): the console's real doors, in
// process, driven in the operator's order (start, seal, flip, open, finish),
// and two boxes in two regions, each ticked between the door calls in the
// agent's order (reconcileSerializedWith, the edges-file write, movesPass).
//
// Production wiring throughout: console.NewServer over the deployment store
// with an operator's CLI token; per box newMovesWorker armed by armOffline,
// a real registry and data directory, a real activator.Service, the
// production edges-file writer fed the identity edgesFileLoop builds from
// the box's own environment, the production build-state carry
// (moveCarry(newBuildStates)), a real litestream replica in data-ie written
// through lease.ReplicateWithLease, leasetest's S3 server for data-ie and
// data-uk, and the exec backend's real launch composition, restore and
// lease take.
//
// The fakes, and why:
//   - the activator's wake backend (fakeWakeBackend), which only the fence
//     needs, as the plan allows; the real ExecBackend does every lease take
//     and the restore;
//   - the operator page (sealWorld.arm's Page records instead of writing
//     the outbox), so the test can say no page fired;
//   - the box status snapshots, written once before the move as each box's
//     status loop would (status.WriteBox), because the console's fleet read
//     joins them and no status loop runs in a test;
//   - the reconcile Layout intent (offBox.reconcile names flat or nested)
//     instead of mintLayoutIntentWith, as T3 to T5's worlds do;
//   - the source's config.env is rendered once by RenderHeld before the
//     seal, standing in for config sync's render at rest;
//   - the agent process environment: both boxes live in one test process,
//     and the launch composer (childEnv) reads os.Environ(), so each box's
//     LaunchLayers runs with that box's own agent variables swapped into
//     the process environment for the call (withAgentEnv). That is what a
//     separate carlos-edge process per box gives production.
//   - for the nested source only, T5's raw-directory stand-in for the bind
//     mount (the fixture's mount is a symlink the nested reader refuses).

const e2eAddr = "ops@example.com"

// The agent environments of the two boxes: the variables measured on dev
// and flagship that differ from box to box (97f541f2). Each box's launch
// composer sees its own.
var (
	e2eIEAgentEnv = map[string]string{
		"SYSTEMD_EXEC_PID":      "1101",
		"MEMORY_PRESSURE_WATCH": "/sys/fs/cgroup/system.slice/carlos-edge.service/memory.pressure",
		"INVOCATION_ID":         "0f0e0d0c0b0a09080706050403020100",
		"AWS_REGION":            sealIERegion,
		"STATUS_BOX_REGION":     sealIERegion,
		"PLATFORM_EDGE_IPS":     "192.0.2.10",
	}
	e2eUKAgentEnv = map[string]string{
		"SYSTEMD_EXEC_PID":      "2202",
		"MEMORY_PRESSURE_WATCH": "/sys/fs/cgroup/system.slice/carlos-edge.service/memory.pressure.uk",
		"INVOCATION_ID":         "00010203040506070809",
		"AWS_REGION":            sealUKRegion,
		"STATUS_BOX_REGION":     sealUKRegion,
		"PLATFORM_EDGE_IPS":     "198.51.100.20,198.51.100.21",
	}
)

// envSwapMu serializes the agent-environment swap: the composer is called
// from one goroutine at a time in this test, and the lock keeps it so.
var envSwapMu sync.Mutex

// withAgentEnv runs compose with env as the process's agent environment,
// restored after the call.
func withAgentEnv(env map[string]string, compose func(registry.Route) (map[string]string, map[string]bool, []string, error)) func(registry.Route) (map[string]string, map[string]bool, []string, error) {
	return func(rt registry.Route) (map[string]string, map[string]bool, []string, error) {
		envSwapMu.Lock()
		defer envSwapMu.Unlock()
		type prev struct {
			v  string
			ok bool
		}
		old := map[string]prev{}
		for k, v := range env {
			pv, ok := os.LookupEnv(k)
			old[k] = prev{pv, ok}
			os.Setenv(k, v)
		}
		defer func() {
			for k, p := range old {
				if p.ok {
					os.Setenv(k, p.v)
				} else {
					os.Unsetenv(k)
				}
			}
		}()
		return compose(rt)
	}
}

// recordingStore is the deployment store every party uses (the console and
// both boxes), logging each write in order.
type recordingStore struct {
	*release.DirStore
	mu     sync.Mutex
	writes []string // "put <key>" or "delete <key>"
}

func (r *recordingStore) note(op, key string) {
	r.mu.Lock()
	r.writes = append(r.writes, op+" "+key)
	r.mu.Unlock()
}

func (r *recordingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	r.note("put", key)
	return r.DirStore.Put(ctx, key, body, ifMatch)
}

func (r *recordingStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	r.note("put", key)
	return r.DirStore.PutFrom(ctx, key, body, size, ifMatch)
}

func (r *recordingStore) Delete(ctx context.Context, key string) error {
	r.note("delete", key)
	return r.DirStore.Delete(ctx, key)
}

func (r *recordingStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
	r.note("delete", key)
	return r.DirStore.DeleteIfMatch(ctx, key, etag)
}

func (r *recordingStore) mark() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.writes)
}

func (r *recordingStore) since(m int) []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]string(nil), r.writes[m:]...)
}

// e2eView is the part of the console's status view the test drives by.
type e2eView struct {
	State   string                 `json:"state"`
	ID      string                 `json:"id"`
	Next    string                 `json:"next"`
	Waiting string                 `json:"waiting"`
	Failed  bool                   `json:"failed"`
	Last    *provision.MoveOutcome `json:"last"`
}

// e2eWorld is shop.example.com at home on ie-box (data-ie) with uk-box
// (data-uk) in uk, the console's doors over the same store.
type e2eWorld struct {
	*landWorld
	t     *testing.T
	rs    *recordingStore
	base  string // the doors' base URL
	token string
	ieEB  *activator.ExecBackend // ie-box's real lease take and restore

	id        string
	recBefore map[string]any
	storeMark int
	bucketsAt map[string]map[string]bool // bucket -> key@version seen before the move
	// The source's files tree and env/config bytes, before the move.
	srcTree               map[string]string
	srcEnvRaw, srcCfgRaw  []byte
	srcFiles, srcEnv, cfg string
	srcTimers             map[string]registry.Timer
	rows                  int
}

func newE2EWorld(t *testing.T, opts landOpts) *e2eWorld {
	t.Helper()
	ctx := context.Background()
	t.Setenv("CARLOS_HOST_CLAIM", "")
	// The test process's own values of the agent variables, so each box's
	// swap is visibly its own and the process value is restored after.
	for k := range e2eIEAgentEnv {
		t.Setenv(k, "test-process")
	}

	// The data buckets over the S3 wire protocol, on loopback.
	fake := leasetest.NewVersionedBucket()
	fake.AddBucket(sealIE)
	fake.AddBucket(sealUK)
	h, _ := leasetest.NewS3Server(fake)
	s3srv := httptest.NewServer(h)
	t.Cleanup(s3srv.Close)
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_SESSION_TOKEN", "")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
	t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "absent"))
	t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(t.TempDir(), "absent"))
	t.Setenv("CARLOS_S3_ENDPOINT", s3srv.URL)

	// One deployment store for everyone, recording every write.
	rs := &recordingStore{DirStore: release.NewDirStore(t.TempDir())}

	// The keys: the deployment's secrets key (the console's public half,
	// each box's private half) and the release key.
	keyB64, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := appconfig.ParsePrivateKey(keyB64, "test key")
	if err != nil {
		t.Fatal(err)
	}
	seed, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	relPriv, err := release.ParseSigningKey(seed, "test")
	if err != nil {
		t.Fatal(err)
	}
	relPub, err := release.ParsePublicKey(pubB64, "test")
	if err != nil {
		t.Fatal(err)
	}

	// The console: an operator, its CLI token, the ops account owning app.
	dir := t.TempDir()
	cst, err := console.Open(filepath.Join(dir, "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { cst.Close() })
	var srv *console.Server
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { srv.ServeHTTP(w, r) }))
	t.Cleanup(ts.Close)
	if srv, err = console.NewServer(console.Config{Origin: ts.URL, Keymail: "http://keymail.invalid",
		Registry: filepath.Join(dir, "registry.db"), Releases: rs, ReleasePub: relPub, Key: relPriv, AppScheme: "http",
		HomeRegion: "ie", SecretsPub: base64.StdEncoding.EncodeToString(priv.PublicKey().Bytes())}, cst); err != nil {
		t.Fatal(err)
	}
	now := time.Now()
	if err := cst.AddMember(console.OpsAccountID, e2eAddr, console.RoleOwner, now); err != nil {
		t.Fatal(err)
	}
	if err := cst.SetOperator(e2eAddr, true); err != nil {
		t.Fatal(err)
	}
	if _, err := cst.CreateAppIn(console.OpsAccountID, "app", e2eAddr, now); err != nil {
		t.Fatal(err)
	}
	userCode, deviceCode, err := cst.NewCLIAuth("ops@laptop", "127.0.0.1", now)
	if err != nil {
		t.Fatal(err)
	}
	if err := cst.ApproveCLIAuth(userCode, e2eAddr, now); err != nil {
		t.Fatal(err)
	}
	_, token, _, err := cst.PollCLIAuth(deviceCode, now)
	if err != nil || token == "" {
		t.Fatalf("no CLI token: %v", err)
	}

	// The deployment: the ops account, app's manifest and the instance's
	// record at home.
	wd := &offWorld{t: t, s: rs, acct: release.OpsAccount, now: now}
	if _, _, err := rs.Get(ctx, domains.AccountManifestKey(wd.acct)); errors.Is(err, os.ErrNotExist) {
		manifest, _ := json.Marshal(map[string]any{"id": console.OpsAccountID, "sqid": wd.acct, "name": "ops", "created_at": now})
		if _, err := rs.Put(ctx, domains.AccountManifestKey(wd.acct), manifest, ""); err != nil {
			t.Fatal(err)
		}
	}
	if err := provision.PutManifest(ctx, rs, wd.acct, "app", provision.Manifest{Schema: 1, App: "app", Domain: "example.com"}); err != nil {
		t.Fatal(err)
	}
	wd.create(offHost)

	// The boxes. ie-box is the home's one box; uk-box is uk's.
	sw := &sealWorld{offWorld: wd, fake: fake, keyB64: keyB64, priv: priv}
	sw.ie = sw.sourceBox("ie-box")
	sw.ie.env["CARLOS_HIBERNATE_REGION"] = sealIERegion
	sw.uk = sw.box("uk-box", "uk", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealUK, "CARLOS_SECRETS_KEY": keyB64,
		"CARLOS_HIBERNATE_REGION": sealUKRegion, "CARLOS_RELEASE_PUBKEY": pubB64})
	lw := &landWorld{t: t, sw: sw, wd: wd, s3: &landS3{f: fake}, ie: sw.ie, uk: sw.uk, opts: opts, relPriv: relPriv}
	shipVersion(t, rs, wd.acct, "app", sealVersion)
	shipVersion(t, rs, wd.acct, "app", "v2")
	lw.promote(sealVersion)
	lw.bundle(landStore)
	lw.ukEB = &activator.ExecBackend{DataDir: lw.uk.dd, OptDir: lw.uk.opt, Region: "uk"}
	lw.arm(lw.uk)
	w := &e2eWorld{landWorld: lw, t: t, rs: rs, base: ts.URL + "/api/cli/apps/" + wd.acct + "/app/instances/" + offHost + "/move/offline", token: token}

	// The source box's production build-state carry, and each box's launch
	// composer under its own agent environment.
	ie := lw.ie
	bs, bsErr := newBuildStates(&activator.Config{Bucket: sealIE, Region: sealIERegion, Prefix: sealPrefix}, ie.reg, ie.opt)
	ie.w.cfg.CarryRequest = moveCarry(bs, bsErr)
	ie.w.cfg.LaunchLayers = withAgentEnv(e2eIEAgentEnv, ie.w.cfg.LaunchLayers)
	lw.uk.w.cfg.LaunchLayers = withAgentEnv(e2eUKAgentEnv, lw.ukEB.LaunchLayers)
	w.ieEB = &activator.ExecBackend{Cfg: activator.Config{Bucket: sealIE, Region: sealIERegion, Prefix: sealPrefix}, DataDir: ie.dd, OptDir: ie.opt}

	// The source at rest, as a box that has served it: rows replicated in
	// two sessions, then a tail the stop never checkpointed.
	sw.instance(ie) // rows 0..19, replicated
	db := sw.dbPath(ie)
	sw.writeRows(ie, db, 20, 25, true)
	sw.writeRows(ie, db, 25, 30, false)
	w.rows = 30
	w.srcFiles, w.srcEnv, w.cfg = filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env"), filepath.Join(ie.dd, offHost+".config.env")
	if opts.nested {
		raw := sw.nest(ie)
		w.srcFiles, w.srcEnv, w.cfg = filepath.Join(raw, offHost), filepath.Join(raw, offHost+".env"), filepath.Join(raw, offHost+".config.env")
		for _, name := range []string{"launch.run.json", "launch.seed.json", "launch.gen.7.json"} {
			writeFile(t, filepath.Join(raw, name), "{}")
		}
		// T5's stand-in for the bind mount: the same two files, composed by
		// the same code, from the raw instance directory.
		ie.w.cfg.LaunchLayers = withAgentEnv(e2eIEAgentEnv, func(rt registry.Route) (map[string]string, map[string]bool, []string, error) {
			rt.DBPath, rt.Socket = provision.DefaultDBPath(raw, rt.Host), filepath.Join(provision.SocketDir, rt.Host+".sock")
			return (&activator.ExecBackend{DataDir: raw}).LaunchLayers(rt)
		})
	}
	if opts.noFiles {
		if err := os.RemoveAll(w.srcFiles); err != nil {
			t.Fatal(err)
		}
	}
	lw.srcFiles, lw.srcEnv, lw.srcConfig = w.srcFiles, w.srcEnv, w.cfg

	// Both boxes published as their loops publish them, and their status
	// snapshots.
	w.tickAll()
	for _, b := range []*offBox{ie, lw.uk} {
		if err := status.WriteBox(ctx, rs, status.Box{Label: b.name, GeneratedAt: time.Now(), Metrics: &status.BoxMetrics{}}); err != nil {
			t.Fatal(err)
		}
	}
	if !ie.lists(offHost) || lw.uk.lists(offHost) {
		t.Fatal("fixture: ie-box must serve the host and uk-box must not")
	}
	return w
}

// publish is one pass of edgesFileLoop on b: the identity it builds at arm
// from the box's own environment, the capability from the box's armed
// worker, through the production writer.
func (w *e2eWorld) publish(b *offBox) {
	w.t.Helper()
	sharedPool, inbound := edgesFileEligibility(b.getenv)
	aws := sealIERegion
	if b.region == "uk" {
		aws = sealUKRegion
	}
	ident := edgesFile{
		Box: b.name, BoxID: b.name, Region: aws, Placement: b.region,
		DataBucket:            strings.TrimSpace(b.getenv("CARLOS_HIBERNATE_BUCKET")),
		DataBucketRegion:      strings.TrimSpace(b.getenv("CARLOS_HIBERNATE_REGION")),
		HibernationConfigured: strings.TrimSpace(b.getenv("CARLOS_HIBERNATE_BUCKET")) != "",
		ServesSharedPool:      sharedPool,
		AcceptsInbound:        inbound,
		SecretsKeyFP:          edgesFileSecretsFP(b.getenv),
		Capabilities:          withOfflineCapability(edgesFileCapabilities(b.getenv), b.w.cfg.OfflineCapable && b.w.cfg.Offline != nil),
	}
	if err := edgesFileWrite(context.Background(), w.wd.s, b.reg, ident, "192.0.2.1", &b.last, false); err != nil {
		w.t.Fatal(err)
	}
}

// tick is the agent's order on one box: reconcile, the edges file, the
// moves pass. The boxes' clock is the wall clock, which the doors judge
// report freshness by.
func (w *e2eWorld) tick(b *offBox) {
	w.t.Helper()
	w.tickAt(b, time.Now())
}

// tickAt is tick with the boxes' clock reading at.
func (w *e2eWorld) tickAt(b *offBox, at time.Time) {
	w.t.Helper()
	w.setBoxClock(at)
	b.reconcile()
	w.publish(b)
	b.pass()
}

func (w *e2eWorld) setBoxClock(at time.Time) {
	w.wd.mu.Lock()
	w.wd.now = at
	w.wd.mu.Unlock()
}

func (w *e2eWorld) tickAll() {
	w.t.Helper()
	w.tick(w.ie)
	w.tick(w.uk)
}

// door posts to one of the doors (verb "" is start).
func (w *e2eWorld) door(verb, body string) (int, string) {
	w.t.Helper()
	u := w.base
	if verb != "" {
		u += "/" + verb
	}
	return w.do(http.MethodPost, u, body)
}

func (w *e2eWorld) do(method, u, body string) (int, string) {
	w.t.Helper()
	req, err := http.NewRequest(method, u, strings.NewReader(body))
	if err != nil {
		w.t.Fatal(err)
	}
	req.Header.Set("Authorization", "Bearer "+w.token)
	req.Header.Set("Content-Type", "application/json")
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		w.t.Fatal(err)
	}
	defer res.Body.Close()
	raw, _ := io.ReadAll(res.Body)
	return res.StatusCode, string(raw)
}

func (w *e2eWorld) decode(out string) e2eView {
	w.t.Helper()
	var v e2eView
	if err := json.Unmarshal([]byte(out), &v); err != nil {
		w.t.Fatalf("%v in %s", err, out)
	}
	return v
}

// accept posts a verb and wants the door's 202 with the state.
func (w *e2eWorld) accept(verb, body, wantState string) e2eView {
	w.t.Helper()
	code, out := w.door(verb, body)
	if code != http.StatusAccepted {
		w.t.Fatalf("%s = %d %s", verbName(verb), code, out)
	}
	v := w.decode(out)
	if v.State != wantState {
		w.t.Fatalf("%s: state %q, want %q (%s)", verbName(verb), v.State, wantState, out)
	}
	return v
}

func verbName(v string) string {
	if v == "" {
		return "start"
	}
	return v
}

func (w *e2eWorld) status() e2eView {
	w.t.Helper()
	code, out := w.do(http.MethodGet, w.base, "")
	if code != http.StatusOK {
		w.t.Fatalf("status = %d %s", code, out)
	}
	return w.decode(out)
}

// until ticks both boxes, then reads the status view, until ok says the
// step has reported.
func (w *e2eWorld) until(what string, ok func(e2eView) bool) e2eView {
	w.t.Helper()
	var v e2eView
	for range 6 {
		w.tickAll()
		if v = w.status(); ok(v) {
			return v
		}
	}
	w.t.Fatalf("%s: never reported; the view is %+v", what, v)
	return v
}

func nextIs(verb string) func(e2eView) bool {
	return func(v e2eView) bool { return v.Next == verb }
}

func (w *e2eWorld) recordRaw() []byte {
	w.t.Helper()
	raw, _, err := w.rs.Get(context.Background(), provision.RecordKey(w.wd.acct, "app", offHost))
	if err != nil {
		w.t.Fatal(err)
	}
	return raw
}

func asJSONMap(t *testing.T, raw []byte) map[string]any {
	t.Helper()
	out := map[string]any{}
	if err := json.Unmarshal(raw, &out); err != nil {
		t.Fatal(err)
	}
	return out
}

// requireRecordDiff: the record differs from the one read before start in
// region, move, last_move and record_seq only (move.updated_at lives in
// move).
func (w *e2eWorld) requireRecordDiff(when string) map[string]any {
	w.t.Helper()
	after := asJSONMap(w.t, w.recordRaw())
	allowed := map[string]bool{"region": true, "move": true, "last_move": true, "record_seq": true}
	keys := map[string]bool{}
	for k := range w.recBefore {
		keys[k] = true
	}
	for k := range after {
		keys[k] = true
	}
	for k := range keys {
		if !allowed[k] && !reflect.DeepEqual(w.recBefore[k], after[k]) {
			w.t.Errorf("%s: record field %q changed: %v -> %v", when, k, w.recBefore[k], after[k])
		}
	}
	return after
}

// bucketVersions is every key@version in bucket.
func (w *e2eWorld) bucketVersions(bucket string) map[string]bool {
	w.t.Helper()
	vers, err := w.sw.fake.ListVersions(context.Background(), bucket, "")
	if err != nil {
		w.t.Fatal(err)
	}
	out := map[string]bool{}
	for _, v := range vers {
		out[v.Key+"@"+v.VersionID] = true
	}
	return out
}

// newBucketKeys is the keys written or deleted in bucket since snap.
func (w *e2eWorld) newBucketKeys(bucket string, snap map[string]bool) []string {
	w.t.Helper()
	var out []string
	seen := map[string]bool{}
	for kv := range w.bucketVersions(bucket) {
		if snap[kv] {
			continue
		}
		k := kv[:strings.LastIndex(kv, "@")]
		if !seen[k] {
			seen[k] = true
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out
}

func (w *e2eWorld) snapBuckets() {
	w.bucketsAt = map[string]map[string]bool{sealIE: w.bucketVersions(sealIE), sealUK: w.bucketVersions(sealUK)}
}

// requireClosedWriteSet: since start, the deployment store saw writes only
// in the plan's closed write set, and the data buckets only under the
// move's own data keys. control/hosts/, control/host-census/,
// control/host-actors.json, the app's admission and the domains claim are
// named, so a write there reads as what it is.
func (w *e2eWorld) requireClosedWriteSet(when string) {
	t := w.t
	t.Helper()
	recKey := provision.RecordKey(w.wd.acct, "app", offHost)
	admission, err := hostops.AdmissionKey(w.wd.acct, "app")
	if err != nil {
		t.Fatal(err)
	}
	forbidden := []string{"control/hosts/", "control/host-census/", "control/host-actors.json", admission, domains.ClaimsPrefix, domains.ClaimKey(offHost)}
	allowed := func(key string) bool {
		switch {
		case key == recKey, key == release.SerialKey,
			strings.HasPrefix(key, "control/offline-moves/"),
			strings.HasPrefix(key, "control/config-ack/"),
			strings.HasPrefix(key, edgesFilePrefix):
			return true
		}
		return false
	}
	ops := w.rs.since(w.storeMark)
	if !slices.ContainsFunc(ops, func(op string) bool { return strings.HasPrefix(op, "put control/offline-moves/") }) {
		t.Fatalf("%s: the recording saw no report written: %v", when, ops)
	}
	for _, op := range ops {
		key := op[strings.IndexByte(op, ' ')+1:]
		for _, f := range forbidden {
			if strings.HasPrefix(key, f) {
				t.Errorf("%s: the move wrote the claim's key %s (%s)", when, key, op)
			}
		}
		if !allowed(key) {
			t.Errorf("%s: a write outside the closed write set: %s", when, op)
		}
	}
	loc := activator.HibernatePrefix(sealPrefix, offHost) + "/"
	for _, k := range w.newBucketKeys(sealIE, w.bucketsAt[sealIE]) {
		if !strings.HasPrefix(k, loc) && k != "move-receipts/"+offHost+".json" {
			t.Errorf("%s: data-ie written outside the move's keys: %s", when, k)
		}
	}
	for _, k := range w.newBucketKeys(sealUK, w.bucketsAt[sealUK]) {
		if !strings.HasPrefix(k, activator.MovesPrefix(offHost)) && !strings.HasPrefix(k, moves.OfflineArchivePrefix(offHost, w.id)) {
			t.Errorf("%s: data-uk written outside the move's keys: %s", when, k)
		}
	}
}

// begin captures the state the move is judged against, then starts it
// through the door and ticks until the seal door would accept.
func (w *e2eWorld) begin() {
	t := w.t
	t.Helper()
	w.recBefore = asJSONMap(t, w.recordRaw())
	timers, err := w.ie.reg.TimersFor(offHost)
	if err != nil {
		t.Fatal(err)
	}
	w.srcTimers = map[string]registry.Timer{}
	for _, tm := range timers {
		w.srcTimers[tm.Name] = tm
	}
	w.storeMark = w.rs.mark()
	w.snapBuckets()

	v := w.accept("", `{"region":"uk"}`, provision.OfflineStopping)
	w.id = v.ID
	if !strings.HasPrefix(w.id, "mv_") {
		t.Fatalf("move id %q", w.id)
	}
	w.until("the source's stop", nextIs("seal"))
	w.requireRecordDiff("after start")
	if w.ie.lists(offHost) {
		t.Fatal("ie-box still lists the host after its stop")
	}

	w.renderSourceConfig()
}

// renderSourceConfig stands in for config sync's render at rest: the
// source's config.env from the bundle (store binding and secret), and the
// source's bytes the move is judged against.
func (w *e2eWorld) renderSourceConfig() {
	t := w.t
	t.Helper()
	hr, err := w.ie.w.cfg.RenderHeld(context.Background(), w.wd.s, *w.ie.row(offHost), w.id, nil, func(string) error { return nil })
	if err != nil {
		t.Fatal(err)
	}
	if hr.Path != w.cfg {
		// The nested fixture's sockets are not production's, so the render
		// took the flat path (T5's note): production renders into the
		// instance directory.
		if err := os.Rename(hr.Path, w.cfg); err != nil {
			t.Fatal(err)
		}
	}
	if isDir(w.srcFiles) {
		w.srcTree = tree(t, w.srcFiles)
	}
	w.srcEnvRaw, w.srcCfgRaw = mustRead(t, w.srcEnv), mustRead(t, w.cfg)
	if !bytes.Contains(w.srcCfgRaw, []byte("STORE_BUCKET")) {
		t.Fatalf("fixture: the source's config carries no store binding: %q", w.srcCfgRaw)
	}
}

func mustRead(t *testing.T, p string) []byte {
	t.Helper()
	b, err := os.ReadFile(p)
	if err != nil {
		t.Fatal(err)
	}
	return b
}

// sealAndFlip drives seal, the sealer's passes and flip, and asserts the
// stop between them: the tombstone holds data-ie against every wake.
func (w *e2eWorld) seal() {
	t := w.t
	t.Helper()
	w.accept("seal", "", provision.OfflineSealing)
	w.until("the seal", nextIs("flip"))
	if st := w.wd.record().Move.State; st != provision.OfflineSealed {
		t.Fatalf("state %s after the seal", st)
	}
	w.requireStopped("after the seal")
}

// requireStopped: a wake's lease take on data-ie meets this move's
// tombstone, through lease.Acquire and through ie-box's ExecBackend (the one
// lease take for every wake source), and the row stays held. Until open,
// the activator also refuses a request for the held row at its fence (the
// source lifts the fence once it has parked, at open).
func (w *e2eWorld) requireStopped(when string) {
	t := w.t
	t.Helper()
	rt := w.requireTombstoned(when)
	if !w.ie.svc.Fenced(offHost) {
		t.Fatalf("%s: ie-box's activator is not fenced", when)
	}
	proxied := false
	rec := httptest.NewRecorder()
	w.ie.svc.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+offHost+"/", nil), rt,
		http.HandlerFunc(func(http.ResponseWriter, *http.Request) { proxied = true }))
	if proxied || rec.Code < 500 {
		t.Fatalf("%s: a request on ie-box was answered %d (proxied %v)", when, rec.Code, proxied)
	}
}

// requireTombstoned is requireStopped's lease half, and the held row.
func (w *e2eWorld) requireTombstoned(when string) registry.Route {
	t := w.t
	t.Helper()
	ctx := context.Background()
	_, err := acquireAsAWake(ctx)
	var he *lease.HeldError
	if !errors.As(err, &he) || he.Owner != moves.TombstoneOwner(w.id) {
		t.Fatalf("%s: lease.Acquire on data-ie = %v, want a HeldError naming %s", when, err, moves.TombstoneOwner(w.id))
	}
	rt := w.ie.row(offHost)
	if rt == nil || rt.HoldOp != w.id || rt.Status == "ready" {
		t.Fatalf("%s: ie-box's row %+v is not held by the move", when, rt)
	}
	if l, err := w.ieEB.Acquire(ctx, *rt); err == nil || !strings.Contains(err.Error(), moves.TombstoneOwner(w.id)) {
		if l != nil {
			l.Release(ctx)
		}
		t.Fatalf("%s: ie-box's exec backend took the lease: %v", when, err)
	}
	return *rt
}

func (w *e2eWorld) flipAndLand() {
	w.t.Helper()
	w.accept("flip", "", provision.OfflineLanding)
	w.requireRecordDiff("after flip")
	w.until("the landing", nextIs("open"))
}

// open drives open and ticks until the console would take finish: the
// target opened and every source box parked.
func (w *e2eWorld) open() {
	w.t.Helper()
	w.accept("open", "", provision.OfflineOpen)
	w.until("the open", nextIs("finish"))
	w.requireRecordDiff("after open")
}

// finish drives both finish calls and the passes between.
func (w *e2eWorld) finish(wantOutcome string) {
	t := w.t
	t.Helper()
	code, out := w.door("finish", "")
	if code != http.StatusAccepted {
		t.Fatalf("finish = %d %s", code, out)
	}
	if v := w.decode(out); v.State != "none" {
		if v.State != provision.OfflineFinishing {
			t.Fatalf("finish: state %q", v.State)
		}
		w.until("the finishing", nextIs("finish"))
		w.accept("finish", "", "none")
	}
	v := w.status()
	if v.State != "none" || v.Last == nil || v.Last.ID != w.id || v.Last.Outcome != wantOutcome {
		t.Fatalf("after finish: %+v (last %+v)", v, v.Last)
	}
}

// requireTombstoneLifted: data-ie's lock is absent or a lifted tombstone.
func (w *e2eWorld) requireTombstoneLifted() {
	w.t.Helper()
	_, owner, expires, err := lease.LockState(context.Background(), sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
	if err != nil || !(owner == "" || lease.IsLiftedTombstone(owner, expires)) {
		w.t.Fatalf("data-ie's lock is %q until %v (%v), want it lifted", owner, expires, err)
	}
}

// requireTarget: what uk-box holds after open, against the source as it
// was before the move.
func (w *e2eWorld) requireTarget() {
	t := w.t
	t.Helper()
	uk := w.uk
	if w.opts.noFiles {
		if got := tree(t, filepath.Join(uk.dd, offHost)); len(got) != 1 {
			t.Fatalf("a source with no files directory landed %v", got)
		}
	} else if got := tree(t, filepath.Join(uk.dd, offHost)); !reflect.DeepEqual(got, w.srcTree) {
		t.Fatalf("files differ:\nsource %v\ntarget %v", w.srcTree, got)
	}
	if got := mustRead(t, filepath.Join(uk.dd, offHost+".env")); !bytes.Equal(got, w.srcEnvRaw) {
		t.Fatalf("<host>.env differs:\nsource %q\ntarget %q", w.srcEnvRaw, got)
	}
	ukCfg := mustRead(t, filepath.Join(uk.dd, offHost+".config.env"))
	if !bytes.Equal(ukCfg, w.srcCfgRaw) {
		t.Fatalf("<host>.config.env differs:\nsource %q\ntarget %q", w.srcCfgRaw, ukCfg)
	}
	// The instance key the app is handed: byte for byte the source's.
	srcKey, _ := instancekey.Effective(instancekey.ParseEnv(w.srcCfgRaw), instancekey.ParseEnv(w.srcEnvRaw))
	dstKey, _ := instancekey.Effective(instancekey.ParseEnv(ukCfg), instancekey.ParseEnv(mustRead(t, filepath.Join(uk.dd, offHost+".env"))))
	if srcKey == "" || dstKey != srcKey {
		t.Fatalf("instance key: source %q, target %q", srcKey, dstKey)
	}
	rt := uk.row(offHost)
	if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
		t.Fatalf("uk-box's row after open: %+v", rt)
	}
	// The launch the target composes, under its own agent environment: the
	// store binding, its secret, the key; and the agent's own variables
	// are uk-box's, not ie-box's (the pin matched across them).
	env, _, _, err := uk.w.cfg.LaunchLayers(*rt)
	if err != nil {
		t.Fatal(err)
	}
	for k, want := range map[string]string{"STORE_BUCKET": landStore, "STORE_SECRET_KEY": landSecret, "CARLOS_INSTANCE_KEY": srcKey,
		"SYSTEMD_EXEC_PID": e2eUKAgentEnv["SYSTEMD_EXEC_PID"], "PLATFORM_EDGE_IPS": e2eUKAgentEnv["PLATFORM_EDGE_IPS"]} {
		if env[k] != want {
			t.Fatalf("the target launches with %s=%q, want %q", k, env[k], want)
		}
	}
	// Timers, NextDue and replication arrived.
	rows, err := uk.reg.TimersFor(offHost)
	if err != nil || len(rows) != 2 || moves.TimersDigest(rows) != w.sealDigest() {
		t.Fatalf("timers on the target %+v %v", rows, err)
	}
	for _, tm := range rows {
		want, ok := w.srcTimers[tm.Name]
		if !ok || tm.Kind != want.Kind || tm.Spec != want.Spec || tm.Path != want.Path || tm.NextAt != want.NextAt {
			t.Fatalf("timer %+v on the target, the source had %+v", tm, want)
		}
	}
	if rt.NextDue != 1_800_000_000 || rt.Replication != registry.ReplicationBatched {
		t.Fatalf("row columns %+v", rt)
	}
	// The edges loop's pass on the registry change the release made.
	w.publish(uk)
	w.publish(w.ie)
	if !uk.lists(offHost) {
		t.Fatal("uk-box's edges file does not list the opened host")
	}
	if w.ie.lists(offHost) {
		t.Fatal("ie-box's edges file lists the moved host")
	}
}

func (w *e2eWorld) sealDigest() string {
	w.t.Helper()
	raw, _, err := w.wd.s.Get(context.Background(), moves.OfflineSealKey(offHost, w.id))
	if err != nil {
		w.t.Fatal(err)
	}
	var r moves.OfflineReport
	if err := json.Unmarshal(raw, &r); err != nil {
		w.t.Fatal(err)
	}
	return r.TimersDigest
}

// firstWake is the target's first wake: the real exec backend's restore
// promotes moves/<host>/ into data-uk's own prefix, and the database passes
// the integrity check holding every row written before the stop.
func (w *e2eWorld) firstWake() {
	t := w.t
	t.Helper()
	ctx := context.Background()
	m, err := lease.NewS3Mirror(ctx, sealUKRegion, sealUKRegion)
	if err != nil {
		t.Fatal(err)
	}
	rt := w.uk.row(offHost)
	eb := &activator.ExecBackend{Cfg: activator.Config{Bucket: sealUK, Region: sealUKRegion, Prefix: sealPrefix}, DataDir: w.uk.dd, OptDir: w.uk.opt, Mirror: m}
	l, err := eb.Acquire(ctx, *rt)
	if err != nil {
		t.Fatalf("the first wake's lease on data-uk: %v", err)
	}
	defer l.Release(ctx)
	fresh, err := eb.Restore(ctx, *rt)
	if err != nil || fresh {
		t.Fatalf("restore: fresh %v, %v", fresh, err)
	}
	if err := lease.CheckDB(rt.DBPath); err != nil {
		t.Fatal(err)
	}
	if n := countRows(t, rt.DBPath); n != w.rows {
		t.Fatalf("the target restored %d rows, want %d", n, w.rows)
	}
	if got := w.s3.current(t, sealUK, activator.MovesPrefix(offHost)); len(got) != 0 {
		t.Fatalf("staging left after the promotion: %v", keysOf(got))
	}
}

// requireRetired: ie-box's parked report lists every source-retire path
// as parked, absent or left with a reason; what it parked is in the park
// directory and not in place; nothing it had is gone.
func (w *e2eWorld) requireRetired() moves.OfflineReport {
	t := w.t
	t.Helper()
	ie := w.ie
	r := w.wd.reportFor(w.id, moves.OfflineRoleSource, "ie-box")
	if r.Phase != moves.OfflinePhaseParked {
		t.Fatalf("ie-box reports %s, want parked", r.Phase)
	}
	for _, rp := range r.Retired {
		switch rp.Action {
		case moves.RetiredParked:
			if lexists(rp.Path) {
				t.Errorf("%s is reported parked but is still in place", rp.Path)
			}
		case moves.RetiredAbsent:
			if lexists(rp.Path) && !w.opts.nested {
				t.Errorf("%s is reported absent but exists", rp.Path)
			}
		case moves.RetiredLeft:
			if rp.Reason == "" {
				t.Errorf("%s left with no reason", rp.Path)
			}
		default:
			t.Errorf("%s: action %q", rp.Path, rp.Action)
		}
	}
	if !w.opts.nested {
		// Every path of the manifest's source-retire row, each with the
		// action this source must get ("" is parked or absent, as the
		// database's companions are whatever litestream left).
		db := provision.DefaultDBPath(ie.dd, offHost)
		files := moves.RetiredParked
		if w.opts.noFiles {
			files = moves.RetiredAbsent
		}
		want := map[string]string{
			db: moves.RetiredParked, db + "-wal": "", db + "-shm": "",
			db + "-litestream": moves.RetiredAbsent, activatorMetaDir(db): moves.RetiredParked,
			filepath.Join(ie.dd, offHost):                     files,
			filepath.Join(ie.dd, offHost+".env"):              moves.RetiredParked,
			filepath.Join(ie.dd, offHost+".config.env"):       moves.RetiredParked,
			filepath.Join(ie.dd, "programs", offHost+".json"): moves.RetiredAbsent,
			filepath.Join(ie.opt, "live", offHost):            moves.RetiredLeft,
		}
		got := map[string]string{}
		for _, rp := range r.Retired {
			got[rp.Path] = rp.Action
		}
		for p, action := range want {
			a, ok := got[p]
			if !ok || (action != "" && a != action) || (action == "" && a != moves.RetiredParked && a != moves.RetiredAbsent) {
				t.Errorf("source-retire %s: %q (listed %v), want %q", p, a, ok, action)
			}
		}
		if len(got) != len(want) {
			t.Errorf("the inventory lists %d paths, want %d: %+v", len(got), len(want), r.Retired)
		}
	} else {
		// Nested: the raw directory's contents, each listed, the platform's
		// launch descriptors included; the directory and its view are left.
		raw := layout.RawInstanceDir(ie.dd, offHost)
		listed := map[string]string{}
		for _, rp := range r.Retired {
			listed[rp.Path] = rp.Action
		}
		for _, name := range []string{offHost, offHost + ".env", offHost + ".config.env", offHost + ".db", "launch.run.json", "launch.gen.7.json"} {
			if listed[filepath.Join(raw, name)] != moves.RetiredParked {
				t.Errorf("the nested source's %s: %q, want parked", name, listed[filepath.Join(raw, name)])
			}
		}
		if listed[raw] != moves.RetiredLeft {
			t.Errorf("the raw instance directory: %q, want left", listed[raw])
		}
	}
	w.requireParkedBytes()
	return r
}

// requireParkedBytes: nothing the source had is gone. The app's files, env
// and config are in the park directory byte for byte, and the database with
// every row.
func (w *e2eWorld) requireParkedBytes() {
	t := w.t
	t.Helper()
	park := moves.OfflineParkDir(w.ie.dd, offHost, w.id, moves.OfflineRoleSource)
	if w.opts.nested {
		park = filepath.Join(park, "instance")
		for _, name := range []string{"launch.run.json", "launch.gen.7.json"} {
			if !lexists(filepath.Join(park, name)) {
				t.Errorf("the nested source's %s was not parked", name)
			}
		}
	}
	if !w.opts.noFiles {
		if got := tree(t, filepath.Join(park, offHost)); !reflect.DeepEqual(got, w.srcTree) {
			t.Fatalf("the parked files differ:\nsource %v\nparked %v", w.srcTree, got)
		}
	}
	if got := mustRead(t, filepath.Join(park, offHost+".env")); !bytes.Equal(got, w.srcEnvRaw) {
		t.Fatal("the parked env differs")
	}
	if got := mustRead(t, filepath.Join(park, offHost+".config.env")); !bytes.Equal(got, w.srcCfgRaw) {
		t.Fatal("the parked config differs")
	}
	if n := countRows(t, filepath.Join(park, offHost+".db")); n != w.rows {
		t.Fatalf("the parked database holds %d rows, want %d", n, w.rows)
	}
}

// reportFor reads one box's report for op.
func (wd *offWorld) reportFor(op, role, box string) moves.OfflineReport {
	wd.t.Helper()
	body, _, err := wd.s.Get(context.Background(), moves.OfflineReportKey(offHost, op, role, box))
	if err != nil {
		wd.t.Fatalf("no %s report from %s: %v", role, box, err)
	}
	var r moves.OfflineReport
	if err := json.Unmarshal(body, &r); err != nil {
		wd.t.Fatal(err)
	}
	if err := moves.ValidateOfflineReport(r); err != nil {
		wd.t.Fatal(err)
	}
	return r
}

// TestOfflineMoveE2E is the two-box witness: the whole offline move through
// the console's doors and both boxes' production passes.
func TestOfflineMoveE2E(t *testing.T) {
	t.Run("done", func(t *testing.T) { offlineMoveDone(t, landOpts{}) })
	t.Run("nested source with launch files", func(t *testing.T) { offlineMoveDone(t, landOpts{nested: true}) })
	t.Run("no files directory", func(t *testing.T) { offlineMoveDone(t, landOpts{noFiles: true}) })
	t.Run("rollback after a tampered archive", offlineMoveRollback)
	// Harness Run 9, finding 7: the doors driven with nothing between them
	// but what production runs (each door's serial bump, the boxes' own
	// nudges and the offline re-assert loop), first back to back, then with
	// the operator three minutes behind at every step, past the doors' two
	// minutes of belief in a report.
	t.Run("doors back to back", func(t *testing.T) { offlineMoveDoorsApart(t, 0) })
	t.Run("doors three minutes apart", func(t *testing.T) { offlineMoveDoorsApart(t, 3*time.Minute) })
}

// e2eLoops are what acts on the two boxes between door calls in production,
// and nothing else: each box's fast-converge tick (fastTick, the real serial
// rule: a moved serial runs the pass set once per clamp, and a nudge made
// during the pass runs it again), whose pass set is the agent's order
// (reconcile, the edges file, the moves pass); and each box's offline
// re-assert tick (offlineReassert). Nothing else ticks a box.
type e2eLoops struct {
	w      *e2eWorld
	fast   map[*offBox]*fastDeps
	tickAt time.Time // fastTick's own clock, stepped past the clamp each round
	boxAt  time.Time // what the boxes' clock reads during a serial pass
	passes int       // serial passes run
	again  int       // re-assert ticks run
}

func (w *e2eWorld) loops() *e2eLoops {
	l := &e2eLoops{w: w, fast: map[*offBox]*fastDeps{}, tickAt: time.Now()}
	for _, b := range []*offBox{w.ie, w.uk} {
		// The sweep's registry is empty: the sweep stages releases and is
		// not what this test is about. The serial leg is.
		l.fast[b] = &fastDeps{reg: restartTestRegistry(w.t), store: w.rs, fullPass: func(context.Context) {
			l.passes++
			w.tickAt(b, l.boxAt)
		}}
	}
	l.round() // the boot read seeds each box's serial
	return l
}

// round is one fastTick on each box; it reports whether either ran its pass.
func (l *e2eLoops) round() bool {
	before := l.passes
	l.tickAt = l.tickAt.Add(fastFullClamp + time.Second)
	for _, b := range []*offBox{l.w.ie, l.w.uk} {
		fastTick(context.Background(), l.fast[b], l.tickAt)
	}
	return l.passes > before
}

// after is what the boxes do once a door has bumped the serial, when the
// operator runs the next door gap later: the serial passes at the time of
// the bump, then the re-assert loop's ticks through the gap, up to now.
func (l *e2eLoops) after(gap time.Duration) {
	t := l.w.t
	t.Helper()
	l.boxAt = time.Now().Add(-gap)
	rounds := 0
	for l.round() {
		if rounds++; rounds > 4 {
			t.Fatal("the serial kept moving after four rounds of passes")
		}
	}
	if gap > 0 {
		// Without the re-assert, a door would read the reports as the bump's
		// passes left them: older than it believes. The status view says
		// so, and never as a process (MV-O31).
		if v := l.w.status(); v.Next != "" || !strings.Contains(v.Waiting, "is not current") {
			t.Fatalf("%s behind the bump, the view read %+v; want a report that is not current", gap, v)
		}
	}
	for at := l.boxAt.Add(offlineReassertEvery); !at.After(time.Now()); at = at.Add(offlineReassertEvery) {
		l.reassert(at)
	}
}

func (l *e2eLoops) reassert(at time.Time) {
	l.w.setBoxClock(at)
	for _, b := range []*offBox{l.w.ie, l.w.uk} {
		b.w.offlineReassert(context.Background(), l.w.rs)
	}
	l.again++
}

// offlineMoveDoorsApart drives start to open with the operator gap behind
// each door's bump, and wants every door accepted on its first or second try,
// the second after one more re-assert tick. It logs the time each door took
// to its answer and from start to open.
func offlineMoveDoorsApart(t *testing.T, gap time.Duration) {
	w := newE2EWorld(t, landOpts{})
	l := w.loops()
	tries := map[string]int{}
	door := func(verb, body, want string) e2eView {
		t.Helper()
		for try := 1; ; try++ {
			code, out := w.door(verb, body)
			if code == http.StatusAccepted {
				tries[verbName(verb)] = try
				v := w.decode(out)
				if v.State != want {
					t.Fatalf("%s: state %q, want %q", verbName(verb), v.State, want)
				}
				return v
			}
			if try == 2 {
				t.Fatalf("%s refused twice, %s behind the previous step: %d %s", verbName(verb), gap, code, out)
			}
			l.reassert(time.Now())
		}
	}
	next := func(step, verb string) {
		t.Helper()
		if v := w.status(); v.Next != verb {
			t.Fatalf("after %s, %s behind: next %q, want %q (waiting %q)", step, gap, v.Next, verb, v.Waiting)
		}
	}
	timers, err := w.ie.reg.TimersFor(offHost)
	if err != nil {
		t.Fatal(err)
	}
	w.srcTimers = map[string]registry.Timer{}
	for _, tm := range timers {
		w.srcTimers[tm.Name] = tm
	}
	began := time.Now()
	var took []string
	timed := func(name string, f func()) {
		at := time.Now()
		f()
		took = append(took, fmt.Sprintf("%s %s", name, time.Since(at).Round(time.Millisecond)))
	}

	timed("start", func() {
		w.id = door("", `{"region":"uk"}`, provision.OfflineStopping).ID
		l.after(gap)
	})
	next("start", "seal")
	w.renderSourceConfig()
	timed("seal", func() {
		door("seal", "", provision.OfflineSealing)
		l.after(gap)
	})
	if st := w.wd.record().Move.State; st != provision.OfflineSealed {
		t.Fatalf("state %s after the seal's passes", st)
	}
	next("seal", "flip")
	timed("flip", func() {
		door("flip", "", provision.OfflineLanding)
		l.after(gap)
	})
	if st := w.wd.record().Move.State; st != provision.OfflineLanded {
		t.Fatalf("state %s after the flip's passes", st)
	}
	next("flip", "open")
	timed("open", func() {
		door("open", "", provision.OfflineOpen)
		l.after(0)
	})
	next("open", "finish")
	total := time.Since(began).Round(time.Millisecond)
	t.Logf("operator gap %s: start to open in %s of test time (%s); %d serial passes, %d re-assert ticks; tries %v",
		gap, total, strings.Join(took, ", "), l.passes, l.again, tries)
	// The move is the real one: the target holds what the source had.
	w.requireTarget()
}

func offlineMoveDone(t *testing.T, opts landOpts) {
	w := newE2EWorld(t, opts)
	ctx := context.Background()

	// The env pin's premise: the two boxes compose under different agent
	// environments, each its own.
	for _, c := range []struct {
		b   *offBox
		env map[string]string
	}{{w.ie, e2eIEAgentEnv}, {w.uk, e2eUKAgentEnv}} {
		rt := registry.Route{Host: offHost, Kind: "instance", DBPath: provision.DefaultDBPath(c.b.dd, offHost), Socket: filepath.Join(provision.SocketDir, offHost+".sock")}
		env, _, _, err := c.b.w.cfg.LaunchLayers(rt)
		if err != nil {
			t.Fatal(err)
		}
		for k, v := range c.env {
			if env[k] != v {
				t.Fatalf("%s composes %s=%q, want its own %q", c.b.name, k, env[k], v)
			}
		}
	}
	if os.Getenv("SYSTEMD_EXEC_PID") != "test-process" {
		t.Fatal("the agent environment swap leaked into the process")
	}

	w.begin()
	w.seal()
	w.flipAndLand()
	if r := w.wd.reportFor(w.id, moves.OfflineRoleTarget, "uk-box"); !r.Items["env-pin"].OK {
		t.Fatalf("env-pin across the two agent environments: %+v", r.Items["env-pin"])
	}
	w.requireStopped("after the landing")
	w.open()

	w.requireTarget()
	retired := w.requireRetired()
	if rt := w.ie.row(offHost); rt == nil {
		t.Fatal("ie-box's row went before finishing")
	}
	w.requireTombstoned("after open")
	w.requireSourceSchedulerQuiet()
	// The writes so far, before the first wake (which is not the tool's).
	w.requireClosedWriteSet("through open")
	w.firstWake()
	w.storeMark = w.rs.mark()
	w.snapBuckets()

	w.finish("done")
	w.requireClosedWriteSet("from open to the end")
	w.requireTombstoneLifted()
	if rt := w.ie.row(offHost); rt != nil {
		t.Fatalf("ie-box still has a row after finish: %+v", rt)
	}
	w.tickAll()
	if w.ie.lists(offHost) || !w.uk.lists(offHost) {
		t.Fatalf("after finish: ie lists %v, uk lists %v", w.ie.lists(offHost), w.uk.lists(offHost))
	}
	w.requireRetiredAfterFinish(retired)

	after := w.requireRecordDiff("after finish")
	if after["region"] != "uk" {
		t.Fatalf("region %v", after["region"])
	}
	if _, ok := after["move"]; ok {
		t.Fatalf("a move is left on the record: %v", after["move"])
	}
	rec := w.wd.record()
	if rec.LastMove == nil || rec.LastMove.Outcome != "done" || rec.LastMove.ID != w.id {
		t.Fatalf("last_move %+v", rec.LastMove)
	}
	if len(w.sw.pages) != 0 {
		t.Fatalf("pages fired: %v", w.sw.pages)
	}
	// The source's lease is free once the move has finished.
	held, err := acquireAsAWake(ctx)
	if err != nil {
		t.Fatalf("data-ie after finish: %v", err)
	}
	held.Release(ctx)
}

// requireSourceSchedulerQuiet is harness Run 9's defect 8. After open the
// source keeps its row, held, until finishing, its timers with it, and open
// has lifted the fence. The source's own schedule runner, built as the agent
// builds it (newScheduleDeps over ie-box's registry, the deployment store,
// its activator and data directory), ticks with every timer due: it writes
// nothing for the host. No env file at either layout's path (the tick used
// to mint one, a fresh admin token, in the parked instance directory), no
// timer row moves, and no store write.
func (w *e2eWorld) requireSourceSchedulerQuiet() {
	t := w.t
	t.Helper()
	ie := w.ie
	if ie.svc.Fenced(offHost) {
		t.Fatal("fixture: ie-box is still fenced after open")
	}
	// Run 9's app declared a one-minute schedule; the row is the source's
	// as the converger writes it, due now. (The fixture's own timers are
	// the seal's, kept for the digest, and are not all ones the runner can
	// fire.)
	if err := ie.reg.UpsertTimer(registry.Timer{Host: offHost, Name: "every-minute", Kind: schedule.KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: 1}); err != nil {
		t.Fatal(err)
	}
	before, err := ie.reg.TimersFor(offHost)
	if err != nil || len(before) == 0 {
		t.Fatalf("fixture: ie-box keeps no timers for the host after open (%v)", err)
	}
	mark := w.rs.mark()
	sd := newScheduleDeps(ie.reg, w.rs, ie.svc, nil, ie.dd)
	sd.runner.Tick(context.Background())
	// A delivery the tick started holds the host's claim until it ends.
	for deadline := time.Now().Add(10 * time.Second); sd.runner.Busy(offHost); {
		if time.Now().After(deadline) {
			t.Fatal("the source's tick did not finish")
		}
		time.Sleep(5 * time.Millisecond)
	}
	for _, p := range []string{
		filepath.Join(ie.dd, offHost+".env"),
		filepath.Join(provision.InstanceDir(ie.dd, offHost), offHost+".env"),
		filepath.Join(layout.RawInstanceDir(ie.dd, offHost), offHost+".env"),
	} {
		if lexists(p) {
			t.Fatalf("the source's scheduler wrote %s after open", p)
		}
	}
	after, err := ie.reg.TimersFor(offHost)
	// The runner's daily sweep of finished once-rows is box-wide and is not
	// a tick of the host (the timers digest leaves that history out, joint
	// N14): only the live rows are compared.
	live := func(ts []registry.Timer) []registry.Timer {
		return slices.DeleteFunc(slices.Clone(ts), func(tm registry.Timer) bool { return tm.Kind == "once" && tm.NextAt == 0 })
	}
	before, after = live(before), live(after)
	if err != nil || !reflect.DeepEqual(after, before) {
		t.Fatalf("the source's tick moved the host's timers (%v):\nbefore %+v\nafter  %+v", err, before, after)
	}
	if ops := w.rs.since(mark); len(ops) != 0 {
		t.Fatalf("the source's tick wrote to the store: %v", ops)
	}
}

// requireRetiredAfterFinish: finishing removed the row and nothing parked
// came back or went.
func (w *e2eWorld) requireRetiredAfterFinish(at moves.OfflineReport) {
	t := w.t
	t.Helper()
	park := moves.OfflineParkDir(w.ie.dd, offHost, w.id, moves.OfflineRoleSource)
	if !isDir(park) {
		t.Fatal("the park directory is gone after finish")
	}
	for _, rp := range at.Retired {
		if rp.Action == moves.RetiredParked && lexists(rp.Path) {
			t.Fatalf("%s came back after finish", rp.Path)
		}
	}
	r := w.wd.reportFor(w.id, moves.OfflineRoleSource, "ie-box")
	if r.Phase != moves.OfflinePhaseGone {
		t.Fatalf("ie-box reports %s after finish, want gone", r.Phase)
	}
	w.requireParkedBytes()
}

// A tampered archive fails the landing; open refuses; abort lifts the stop
// and unfences; the source serves again with every row.
func offlineMoveRollback(t *testing.T) {
	w := newE2EWorld(t, landOpts{})
	ctx := context.Background()
	w.begin()
	w.seal()

	// Tamper with the sealed archive in data-uk.
	key := moves.OfflineArchivePrefix(offHost, w.id) + offlineArchiveSealed
	body, ok := w.sw.fake.Current(sealUK, key)
	if !ok || len(body) < 64 {
		t.Fatal("fixture: no sealed archive")
	}
	tampered := bytes.Clone(body)
	tampered[len(tampered)/2] ^= 0xff
	w.sw.fake.Put(sealUK, key, tampered)

	w.accept("flip", "", provision.OfflineLanding)
	var v e2eView
	for range 4 {
		w.tickAll()
		if v = w.status(); v.Failed {
			break
		}
	}
	r := w.wd.reportFor(w.id, moves.OfflineRoleTarget, "uk-box")
	if !v.Failed || r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK {
		t.Fatalf("the landing of a tampered archive: view %+v, report %s %+v", v, r.Phase, r.Items["files"])
	}
	if code, out := w.door("open", ""); code != http.StatusConflict {
		t.Fatalf("open on a failed landing = %d %s", code, out)
	}
	if rt := w.uk.row(offHost); rt == nil || rt.HoldOp != w.id {
		t.Fatalf("uk-box's row %+v, want held by the move", rt)
	}

	w.accept("abort", "", provision.OfflineReverting)
	w.until("the revert", nextIs("finish"))
	w.finish("aborted")
	w.requireClosedWriteSet("through the abort")
	after := w.requireRecordDiff("after the abort")
	if r, ok := after["region"]; ok && r != "" {
		t.Fatalf("region %v after the abort, want home", r)
	}
	if _, ok := after["move"]; ok {
		t.Fatalf("a move is left on the record: %v", after["move"])
	}

	// The source: the row ready and listed, unfenced, the tombstone gone.
	w.tickAll()
	rt := w.ie.row(offHost)
	if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
		t.Fatalf("ie-box's row after the abort: %+v", rt)
	}
	if !w.ie.lists(offHost) {
		t.Fatal("ie-box does not list the host after the abort")
	}
	if w.ie.svc.Fenced(offHost) {
		t.Fatal("ie-box is still fenced after the abort")
	}
	w.requireTombstoneLifted()

	// The target: no row; what it extracted is parked, not deleted.
	if rt := w.uk.row(offHost); rt != nil {
		t.Fatalf("uk-box has a row after the abort: %+v", rt)
	}
	for _, name := range []string{offHost, offHost + ".env", offHost + ".config.env"} {
		if lexists(filepath.Join(w.uk.dd, name)) {
			t.Errorf("uk-box keeps %s in place after the abort", name)
		}
	}
	if parked := w.parkedTarget(); !parked[offHost+".config.env"] {
		t.Errorf("uk-box parked %v, want what it rendered", parked)
	}
	if w.uk.lists(offHost) {
		t.Fatal("uk-box lists the host after the abort")
	}
	// Both staging prefixes are empty, and data-uk's own prefix untouched.
	for _, p := range []string{activator.MovesPrefix(offHost), moves.OfflineArchivePrefix(offHost, w.id), activator.HibernatePrefix(sealPrefix, offHost) + "/"} {
		if got := w.s3.current(t, sealUK, p); len(got) != 0 {
			t.Fatalf("data-uk %s holds %v after the abort", p, keysOf(got))
		}
	}

	// The source serves again with no write lost: the wake takes the
	// lease, the warm cache is the database, and it holds every row; so
	// does the replica.
	l, err := w.ieEB.Acquire(ctx, *rt)
	if err != nil {
		t.Fatalf("a wake on ie-box after the abort: %v", err)
	}
	defer l.Release(ctx)
	if fresh, err := w.ieEB.Restore(ctx, *rt); err != nil || fresh {
		t.Fatalf("restore on ie-box: fresh %v, %v", fresh, err)
	}
	if err := lease.CheckDB(rt.DBPath); err != nil {
		t.Fatal(err)
	}
	if n := countRows(t, rt.DBPath); n != w.rows {
		t.Fatalf("ie-box's database holds %d rows, want %d", n, w.rows)
	}
	scratch := filepath.Join(t.TempDir(), "replica.db")
	if ok, err := lease.Restore(ctx, scratch, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion); err != nil || !ok {
		t.Fatalf("restore data-ie's replica: %v %v", ok, err)
	}
	if n := countRows(t, scratch); n != w.rows {
		t.Fatalf("data-ie's replica holds %d rows, want %d", n, w.rows)
	}
	if len(w.sw.pages) != 0 {
		t.Fatalf("pages fired: %v", w.sw.pages)
	}
}

// parkedTarget is the base names uk-box parked for the move.
func (w *e2eWorld) parkedTarget() map[string]bool {
	out := map[string]bool{}
	for _, role := range []string{moves.OfflineRoleTarget, moves.OfflineParkTargetBefore} {
		ents, _ := os.ReadDir(moves.OfflineParkDir(w.uk.dd, offHost, w.id, role))
		for _, e := range ents {
			out[e.Name()] = true
		}
	}
	return out
}