//go:build linux

package host

import (
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"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/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The sealer (Task 4 of docs/superpowers/plans/2026-10-20-offline-operator-
// move.md), tested through offlineRound, the moves pass's dispatch, over
// the S3 wire protocol (leasetest.NewS3Server, with buckets data-ie and
// data-uk behind one endpoint), real litestream replicas written by
// lease.ReplicateWithLease, the real lease package, a real registry and
// activator per box, and the production edges-file writer for the target's
// coordinates. The deployment store is T3's DirStore world.

const (
	sealIE       = "data-ie"
	sealUK       = "data-uk"
	sealIERegion = "eu-west-1"
	sealUKRegion = "eu-west-2"
	sealPrefix   = "instances"
	sealVersion  = "v42"
)

type sealWorld struct {
	*offWorld
	fake   *leasetest.VersionedBucket
	keyB64 string
	priv   *ecdh.PrivateKey
	ie, uk *offBox
	pages  []string
	bodies []string // each page's body, beside pages
}

func newSealWorld(t *testing.T) *sealWorld {
	t.Helper()
	fake := leasetest.NewVersionedBucket()
	fake.AddBucket(sealIE)
	fake.AddBucket(sealUK)
	h, _ := leasetest.NewS3Server(fake)
	srv := httptest.NewServer(h) // 127.0.0.1 only
	t.Cleanup(srv.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", srv.URL)
	keyB64, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := appconfig.ParsePrivateKey(keyB64, "test key")
	if err != nil {
		t.Fatal(err)
	}
	sw := &sealWorld{offWorld: newOffWorld(t), fake: fake, keyB64: keyB64, priv: priv}
	sw.ie = sw.sourceBox("ie-a")
	sw.uk = sw.box("uk-box", "uk", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealUK, "CARLOS_SECRETS_KEY": keyB64})
	var last []byte
	if err := edgesFileWrite(context.Background(), sw.s, sw.uk.reg, edgesFile{Box: "uk-box", Placement: "uk", DataBucket: sealUK, DataBucketRegion: sealUKRegion}, "", &last, false); err != nil {
		t.Fatal(err)
	}
	return sw
}

// sourceBox is a home box armed as armMovesPass arms one: armOffline, then
// the hibernate coordinates, the S3 mirror, the production checkpoint and
// the box's exec backend for the launch environment.
func (sw *sealWorld) sourceBox(name string) *offBox {
	b := sw.box(name, "", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealIE, "CARLOS_SECRETS_KEY": sw.keyB64})
	sw.arm(b)
	return b
}

func (sw *sealWorld) arm(b *offBox) {
	c := &b.w.cfg
	c.OwnBucket, c.OwnBucketRegion, c.OwnPrefix = sealIE, sealIERegion, sealPrefix
	c.Mirror = func(ctx context.Context, dst string) (lease.MirrorStore, error) {
		return lease.NewS3Mirror(ctx, sealIERegion, dst)
	}
	// armMovesPass's Checkpoint, as it is wired there.
	c.Checkpoint = func(ctx context.Context, dbPath string) error {
		host := strings.TrimSuffix(filepath.Base(dbPath), ".db")
		rep, err := lease.ReplicateWithLease(ctx, dbPath, sealIE, activator.HibernatePrefix(sealPrefix, host), sealIERegion)
		if err != nil {
			return err
		}
		defer rep.Close(ctx)
		return rep.SyncAndWait(ctx)
	}
	c.CarryRequest = func(ctx context.Context, host, bucket, prefix string) error {
		sw.fake.Put(bucket, prefix+"/build-state.json", []byte(`{"request":"carried"}`))
		return nil
	}
	c.LaunchLayers = (&activator.ExecBackend{DataDir: b.dd}).LaunchLayers
	c.Page = func(_ context.Context, level, subject, body string) error {
		sw.pages = append(sw.pages, level+": "+subject)
		sw.bodies = append(sw.bodies, body)
		return nil
	}
}

// pass is a tick in the agent's order (reconcile, the edges file, the moves
// pass) whose moves pass is movesPass's offline part: the keep view's
// refresh from the record list, then offlineRound, returning the round's
// error rather than logging it.
func (sw *sealWorld) pass(b *offBox) error {
	sw.t.Helper()
	b.reconcile()
	b.edges()
	ctx := context.Background()
	refreshOfflineView(ctx, sw.s, b.w.cfg.Offline)
	return b.w.offlineRound(ctx, sw.s, sw.record())
}

func (sw *sealWorld) mustPass(b *offBox) {
	sw.t.Helper()
	if err := sw.pass(b); err != nil {
		sw.t.Fatal(err)
	}
}

func (sw *sealWorld) dbPath(b *offBox) string {
	sw.t.Helper()
	rt := b.row(offHost)
	if rt == nil {
		sw.t.Fatal("no row")
	}
	return rt.DBPath
}

// instance gives b a live instance of offHost, as a box that has served it
// has: the row (minted ready by reconcile), its replication rung, next due
// and timers, a database written through lease.ReplicateWithLease into
// data-ie, the files directory, <host>.env with an instance key, the
// rendered configuration and a live build.
func (sw *sealWorld) instance(b *offBox) {
	t := sw.t
	t.Helper()
	b.tick()
	rt := b.row(offHost)
	if rt == nil || rt.HoldOp != "" {
		t.Fatalf("fixture: row %+v", rt)
	}
	if err := b.reg.SetReplication(offHost, registry.ReplicationBatched); err != nil {
		t.Fatal(err)
	}
	if err := b.reg.SetNextDue(offHost, 1_800_000_000); err != nil {
		t.Fatal(err)
	}
	for _, tm := range []registry.Timer{
		{Host: offHost, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", Path: "/jobs/nightly", NextAt: 1_800_000_100},
		{Host: offHost, Name: "reminder", Kind: "once", Path: "/jobs/remind", NextAt: 1_800_000_200},
		{Host: offHost, Name: "fired", Kind: "once", Path: "/jobs/fired", LastAt: 1_700_000_000, LastStatus: "200"},
	} {
		if err := b.reg.UpsertTimer(tm); err != nil {
			t.Fatal(err)
		}
	}
	writeFile(t, filepath.Join(b.dd, offHost, "instance.key"), "app key file")
	writeFile(t, filepath.Join(b.dd, offHost, "uploads", "a.png"), "png bytes")
	writeFile(t, filepath.Join(b.dd, offHost+".env"), "CARLOS_INSTANCE_KEY=the-instance-key\nCARLOS_ADMIN_TOKEN=token\n")
	writeFile(t, filepath.Join(b.dd, offHost+".config.env"), "APP_COLOUR=green\n")
	sw.writeRows(b, rt.DBPath, 0, 20, true)
	versions := filepath.Join(b.opt, "versions", sealVersion, sw.acct, "app")
	if err := os.MkdirAll(versions, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Dir(release.LiveLink(b.opt, offHost)), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(versions, release.LiveLink(b.opt, offHost)); err != nil {
		t.Fatal(err)
	}
}

// writeRows inserts rows [from, to) into the database at dbPath, then, with
// replicate, syncs it to data-ie under the lease as a wake's stop does.
func (sw *sealWorld) writeRows(b *offBox, dbPath string, from, to int, replicate bool) {
	t := sw.t
	t.Helper()
	db, err := sql.Open("sqlite", dbPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, q := range []string{`PRAGMA journal_mode=WAL`, `CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`} {
		if _, err := db.Exec(q); err != nil {
			t.Fatal(err)
		}
	}
	for i := from; i < to; i++ {
		if _, err := db.Exec(`INSERT INTO t (id, body) VALUES (?, ?)`, i, fmt.Sprintf("row-%d", i)); err != nil {
			t.Fatal(err)
		}
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	if !replicate {
		return
	}
	ctx := context.Background()
	rep, err := lease.ReplicateWithLease(ctx, dbPath, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
	if err != nil {
		t.Fatal(err)
	}
	if err := rep.SyncAndWait(ctx); err != nil {
		t.Fatal(err)
	}
	if err := rep.Close(ctx); err != nil {
		t.Fatal(err)
	}
}

func countRows(t *testing.T, dbPath string) int {
	t.Helper()
	db, err := sql.Open("sqlite", dbPath)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	var n int
	if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

// start moves offHost to the record's offline-stopping, ticks every source
// box until it holds and reports, and names the sealer and the boxes as the
// seal door does.
func (sw *sealWorld) startAndSeal(sealer string, boxes ...*offBox) {
	sw.t.Helper()
	sw.start()
	names := []string{}
	for _, b := range boxes {
		b.tick()
		names = append(names, b.name)
	}
	sw.move(provision.OfflineSeal(offID, sealer, names))
}

func (sw *sealWorld) mirror() lease.MirrorStore {
	sw.t.Helper()
	m, err := lease.NewS3Mirror(context.Background(), sealIERegion, sealUKRegion)
	if err != nil {
		sw.t.Fatal(err)
	}
	return m
}

func (sw *sealWorld) list(bucket, prefix string) map[string]lease.MirrorObject {
	sw.t.Helper()
	objs, err := sw.mirror().List(context.Background(), bucket, prefix)
	if err != nil {
		sw.t.Fatal(err)
	}
	return objs
}

func (sw *sealWorld) lock() (owner string, expires time.Time, generation int64) {
	sw.t.Helper()
	body, ok := sw.fake.Current(sealIE, sealPrefix+"/"+offHost+"/lock.json")
	if !ok {
		return "", time.Time{}, 0
	}
	var l litestream.Lease
	if err := json.Unmarshal(body, &l); err != nil {
		sw.t.Fatal(err)
	}
	return l.Owner, l.ExpiresAt, l.Generation
}

func (sw *sealWorld) sealRecord() (moves.OfflineReport, []byte, bool) {
	sw.t.Helper()
	raw, _, err := sw.s.Get(context.Background(), moves.OfflineSealKey(offHost, offID))
	if errors.Is(err, os.ErrNotExist) {
		return moves.OfflineReport{}, nil, false
	}
	if err != nil {
		sw.t.Fatal(err)
	}
	var r moves.OfflineReport
	if err := json.Unmarshal(raw, &r); err != nil {
		sw.t.Fatal(err)
	}
	return r, raw, true
}

// acquireAsAWake is what every wake does first (ExecBackend.Acquire).
func acquireAsAWake(ctx context.Context) (*lease.Held, error) {
	return lease.Acquire(ctx, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
}

func wantHeldByMove(t *testing.T, err error, what string) {
	t.Helper()
	var he *lease.HeldError
	if !errors.As(err, &he) || he.Owner != moves.TombstoneOwner(offID) {
		t.Fatalf("%s: %v, want a HeldError naming %s", what, err, moves.TombstoneOwner(offID))
	}
}

// openArchive fetches the sealed archive and its manifest from data-uk,
// verifies it and extracts it into a fresh flat files directory and env
// path, returning them.
func (sw *sealWorld) openArchive(seal moves.OfflineReport) (files, env string) {
	t := sw.t
	t.Helper()
	prefix := moves.OfflineArchivePrefix(offHost, offID)
	sealed, ok := sw.fake.Current(sealUK, prefix+offlineArchiveSealed)
	if !ok {
		t.Fatal("no sealed archive")
	}
	rawManifest, ok := sw.fake.Current(sealUK, prefix+offlineArchiveManifest)
	if !ok {
		t.Fatal("no archive manifest")
	}
	var m archive.Manifest
	if err := json.Unmarshal(rawManifest, &m); err != nil {
		t.Fatal(err)
	}
	bind := appconfig.Binding{Account: sw.acct, App: "app"}
	if err := archive.Verify(bytes.NewReader(sealed), m, sw.priv, bind); err != nil {
		t.Fatalf("Verify: %v", err)
	}
	sum := sha256.Sum256(sealed)
	if m.Plain != seal.FilesPlain || m.Sealed != seal.FilesSealed || "sha256:"+hex.EncodeToString(sum[:]) != seal.FilesSealed || m.Bytes != seal.FilesBytes || len(m.Entries) != seal.FilesEntries {
		t.Fatalf("manifest %+v against the seal %+v", m, seal)
	}
	var plain bytes.Buffer
	if err := archive.Open(bytes.NewReader(sealed), m.SealedDataKey, sw.priv, bind, &plain); err != nil {
		t.Fatalf("Open: %v", err)
	}
	parent := t.TempDir()
	files, env = filepath.Join(parent, offHost), filepath.Join(parent, offHost+".env")
	// A *bytes.Reader, not the buffer: zstd decodes a *bytes.Buffer under
	// 128 KiB from its Bytes() without draining it, and Extract's
	// end-of-stream check then finds the whole archive still unread.
	if err := archive.Extract(bytes.NewReader(plain.Bytes()), files, env); err != nil {
		t.Fatalf("Extract: %v", err)
	}
	if got, _, _, err := archive.PlainDigest(files, env); err != nil || got != seal.FilesPlain {
		t.Fatalf("the extraction digests to %s (%v), the seal says %s", got, err, seal.FilesPlain)
	}
	return files, env
}

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

// (a) The happy seal, end to end, and the passes after it.
func TestOfflineSealHappyPath(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	// A tail the stop never checkpointed: rows in the local database only.
	sw.writeRows(ie, sw.dbPath(ie), 20, 25, false)
	sw.startAndSeal("ie-a", ie)
	ctx := context.Background()

	sw.mustPass(ie)
	rec := sw.record()
	if rec.Move.State != provision.OfflineSealed {
		t.Fatalf("state %s, want sealed", rec.Move.State)
	}
	seal, raw, ok := sw.sealRecord()
	if !ok {
		t.Fatal("no seal record")
	}
	if err := moves.ValidateOfflineSeal(seal, offHost, offID); err != nil {
		t.Fatal(err)
	}
	src := rec.Move.Source
	if src == nil || src.Evidence != moves.OfflineSealEvidence(offHost, offID, raw) || src.FinalTXID != seal.FinalTXID ||
		src.Files == nil || src.Files.Plain != seal.FilesPlain || src.Files.Sealed != seal.FilesSealed || src.TimersSHA256 != seal.TimersDigest ||
		src.Release != sealVersion || src.EnvDigest != seal.EnvDigest || src.Replication != registry.ReplicationBatched || src.NextDue != 1_800_000_000 {
		t.Fatalf("record source %+v, seal %+v", src, seal)
	}

	// The tombstone, and a wake meets it.
	owner, expires, _ := sw.lock()
	if owner != moves.TombstoneOwner(offID) || time.Until(expires) < moves.TombstoneTTL-time.Minute {
		t.Fatalf("lock %s until %s", owner, expires)
	}
	if h, err := acquireAsAWake(ctx); err == nil {
		h.Release(ctx)
		t.Fatal("a wake took the lease after the seal")
	} else {
		wantHeldByMove(t, err, "a wake after the seal")
	}

	// The staged copy is the replica, object for object, and its listing is
	// the seal's.
	srcObjs := sw.list(sealIE, sealPrefix+"/"+offHost+"/")
	staged := sw.list(sealUK, activator.MovesPrefix(offHost))
	if len(staged) != len(seal.Staged) || len(staged) != len(srcObjs)-1 {
		t.Fatalf("staged %d objects, seal %d, source %d (lock.json included)", len(staged), len(seal.Staged), len(srcObjs))
	}
	for k, o := range staged {
		rel := strings.TrimPrefix(k, activator.MovesPrefix(offHost))
		if s := seal.Staged[rel]; s.Size != o.Size || s.ETag != o.ETag {
			t.Fatalf("%s staged %+v, sealed %+v", rel, o, s)
		}
		if so := srcObjs[sealPrefix+"/"+offHost+"/"+rel]; so.Size != o.Size {
			t.Fatalf("%s staged %+v, source %+v", rel, o, so)
		}
	}
	if _, ok := seal.Staged["build-state.json"]; !ok {
		t.Fatal("the carried build state is not staged")
	}
	if replicaPosition(staged, activator.MovesPrefix(offHost)) != seal.FinalTXID {
		t.Fatal("the staged newest TXID is not the seal's")
	}
	// The staged replica restores to every row, the tail included: the
	// checkpoint ran before the tombstone.
	restored := filepath.Join(t.TempDir(), "restored.db")
	if ok, err := lease.Restore(ctx, restored, sealUK, strings.TrimSuffix(activator.MovesPrefix(offHost), "/"), sealUKRegion); err != nil || !ok {
		t.Fatalf("restore the staged copy: %v %v", ok, err)
	}
	if n := countRows(t, restored); n != 25 {
		t.Fatalf("the staged copy holds %d rows, want 25", n)
	}

	// The archive opens, verifies and extracts to the same plaintext, with
	// the files and the env byte for byte.
	files, env := sw.openArchive(seal)
	for rel, want := range map[string]string{"instance.key": "app key file", "uploads/a.png": "png bytes"} {
		if got := readString(t, filepath.Join(files, rel)); got != want {
			t.Fatalf("%s = %q", rel, got)
		}
	}
	envBody := readString(t, filepath.Join(ie.dd, offHost+".env"))
	if readString(t, env) != envBody {
		t.Fatal("the env file did not travel byte for byte")
	}
	sum := sha256.Sum256([]byte(envBody))
	if seal.EnvSHA256 != hex.EncodeToString(sum[:]) || seal.KeySHA256 != instancekey.Digest("the-instance-key") {
		t.Fatalf("env %s key %s", seal.EnvSHA256, seal.KeySHA256)
	}

	// The timers snapshot (active rows only) and the env report.
	prefix := moves.OfflineArchivePrefix(offHost, offID)
	timersBody, _ := sw.fake.Current(sealUK, prefix+offlineTimersObject)
	snap, err := moves.DecodeTimers(timersBody)
	if err != nil || len(snap.Rows) != 2 || moves.TimersDigest(snap.Rows) != seal.TimersDigest {
		t.Fatalf("timers snapshot %+v %v", snap, err)
	}
	reportBody, _ := sw.fake.Current(sealUK, prefix+offlineEnvReportObject)
	var envReport moves.EnvReport
	if err := json.Unmarshal(reportBody, &envReport); err != nil || envReport.Digest != seal.EnvDigest || envReport.Vars["APP_COLOUR"] == "" || strings.Contains(string(reportBody), "green") {
		t.Fatalf("env report %s %v", reportBody, err)
	}
	if seal.Release != sealVersion || seal.Replication != registry.ReplicationBatched || seal.NextDue != 1_800_000_000 {
		t.Fatalf("seal %+v", seal)
	}

	// The living report: the tombstone and the pass's time.
	living := sw.report(moves.OfflineRoleSealer, "ie-a")
	if living.Phase != moves.OfflinePhaseSealed || living.TombstoneOwner != moves.TombstoneOwner(offID) || !living.TombstoneUntil.Equal(expires) || !living.At.Equal(sw.clock()) {
		t.Fatalf("living report %+v", living)
	}
	if ok, bad := living.SourceItemsOK(); !ok {
		t.Fatalf("living report items %v", bad)
	}

	// Later passes re-assert the expiry, keep the tombstone's generation,
	// never touch the seal record, and rewrite the living report.
	_, _, gen := sw.lock()
	for _, state := range []func(){func() {}, sw.flip, sw.landed} {
		state()
		time.Sleep(5 * time.Millisecond)
		sw.advance(time.Second)
		if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
			t.Fatal(err)
		}
		owner2, expires2, gen2 := sw.lock()
		if owner2 != owner || !expires2.After(expires) || gen2 != gen {
			t.Fatalf("after a later pass: %s until %s (generation %d), before %s until %s (generation %d)", owner2, expires2, gen2, owner, expires, gen)
		}
		expires = expires2
		living = sw.report(moves.OfflineRoleSealer, "ie-a")
		if !living.TombstoneUntil.Equal(expires2) || !living.At.Equal(sw.clock()) || living.Phase != moves.OfflinePhaseSealed {
			t.Fatalf("living report after a later pass: %+v", living)
		}
		if _, again, _ := sw.sealRecord(); !bytes.Equal(again, raw) {
			t.Fatal("the seal record was rewritten")
		}
	}
	if len(sw.pages) != 0 {
		t.Fatalf("paged: %v", sw.pages)
	}
}

// (b) A live writer's lease: with a local database the checkpoint cannot
// take it, without one the tombstone cannot; either way the state stays
// sealing and nothing is staged or sealed. A wake that races the tombstone
// meets it.
func TestOfflineSealWaitsForALiveWriter(t *testing.T) {
	ctx := context.Background()
	for _, local := range []bool{true, false} {
		t.Run(fmt.Sprintf("local database %v", local), func(t *testing.T) {
			sw := newSealWorld(t)
			ie := sw.ie
			sw.instance(ie)
			if !local {
				db := sw.dbPath(ie)
				for _, p := range []string{db, db + "-wal", db + "-shm", activatorMetaDir(db)} {
					if err := os.RemoveAll(p); err != nil {
						t.Fatal(err)
					}
				}
			}
			sw.startAndSeal("ie-a", ie)
			wake, err := acquireAsAWake(ctx)
			if err != nil {
				t.Fatal(err)
			}
			err = sw.pass(ie)
			var he *lease.HeldError
			if !errors.As(err, &he) || he.Owner == moves.TombstoneOwner(offID) {
				t.Fatalf("a live writer: %v", err)
			}
			if !local && !strings.Contains(err.Error(), "tombstone") {
				t.Fatalf("not refused at the tombstone: %v", err)
			}
			if sw.record().Move.State != provision.OfflineSealing {
				t.Fatal("the state moved")
			}
			if _, _, ok := sw.sealRecord(); ok {
				t.Fatal("sealed under a live writer")
			}
			if n := len(sw.list(sealUK, "moves/")); n != 0 {
				t.Fatalf("%d objects staged under a live writer", n)
			}
			if err := wake.Err(); err != nil {
				t.Fatalf("the writer lost its lease: %v", err)
			}
			if err := wake.Release(ctx); err != nil {
				t.Fatal(err)
			}

			// The race: a wake that comes for the lease the moment after
			// the tombstone is written is refused by it.
			raced := false
			offlineSealFaults.Store(ie.w, func(point string) error {
				if point == "tombstone" {
					raced = true
					h, err := acquireAsAWake(ctx)
					if err == nil {
						h.Release(ctx)
						return errors.New("a wake took the lease after the tombstone")
					}
					wantHeldByMove(t, err, "a wake racing the tombstone")
				}
				return nil
			})
			t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
			sw.mustPass(ie)
			if !raced || sw.record().Move.State != provision.OfflineSealed {
				t.Fatalf("raced %v, state %s", raced, sw.record().Move.State)
			}
		})
	}
}

// (c) A source box whose local position is past the replica's final
// position refuses the seal: the db-tail item fails, the report says
// failed, nothing is sealed and the state stays sealing.
func TestOfflineSealRefusesASourceAheadOfTheFinalPosition(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	ieB := sw.sourceBox("ie-b")
	ieB.tick()
	final := replicaPosition(sw.list(sealIE, sealPrefix+"/"+offHost+"/"), sealPrefix+"/"+offHost+"/")
	if final == 0 {
		t.Fatal("fixture: no replica")
	}
	// ie-b woke the host once and kept a position past the replica's.
	meta := filepath.Join(activatorMetaDir(provision.DefaultDBPath(ieB.dd, offHost)), "ltx", "0")
	writeFile(t, filepath.Join(meta, ltxName(final+1, final+5)), "ltx")
	sw.startAndSeal("ie-a", ie, ieB)
	if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.LocalTXID != final+5 {
		t.Fatalf("fixture: ie-b reports %d", r.LocalTXID)
	}
	err := sw.pass(ie)
	var ref *sealRefusal
	if !errors.As(err, &ref) || ref.item != "db-tail" || !strings.Contains(ref.Reason, "ie-b") {
		t.Fatalf("pass: %v", err)
	}
	r := sw.report(moves.OfflineRoleSealer, "ie-a")
	if r.Phase != moves.OfflinePhaseFailed || r.Items["db-tail"].OK || !r.Items["sidecar"].OK {
		t.Fatalf("report %+v", r)
	}
	if _, _, ok := sw.sealRecord(); ok || sw.record().Move.State != provision.OfflineSealing {
		t.Fatal("sealed past a source ahead of the replica")
	}
}

func ltxName(min, max uint64) string {
	return ltx.FormatFilename(ltx.TXID(min), ltx.TXID(max))
}

// (d) After the seal the stop is kept, never taken again: a tombstone
// stolen by another move, expired, lifted or deleted is a lost stop. The
// pass reports failed with the lease item down, pages once for the move,
// and leaves the lease exactly as it found it.
func TestOfflineSealLostStopIsNeverRetaken(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	sw.startAndSeal("ie-a", ie)
	sw.mustPass(ie)
	ctx := context.Background()
	key := sealPrefix + "/" + offHost + "/lock.json"
	ours, _ := sw.fake.Current(sealIE, key)
	lockBody := func(owner string, expires time.Time) []byte {
		b, err := json.Marshal(litestream.Lease{Generation: 1, Owner: owner, ExpiresAt: expires})
		if err != nil {
			t.Fatal(err)
		}
		return b
	}
	cases := []struct {
		name   string
		damage func()
	}{
		{"stolen by another move", func() {
			sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID2), time.Now().Add(time.Hour)))
		}},
		{"stolen by a wake", func() {
			sw.fake.Put(sealIE, key, lockBody("box-x", time.Now().Add(time.Hour)))
		}},
		{"expired", func() {
			sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID), time.Now().Add(-time.Second)))
		}},
		{"lifted", func() {
			if _, err := lease.LiftTombstone(ctx, sealIE, sealPrefix+"/"+offHost, sealIERegion, func(o string) bool { return o == moves.TombstoneOwner(offID) }); err != nil {
				t.Fatal(err)
			}
		}},
		{"deleted", func() { sw.fake.BareDelete(sealIE, key) }},
	}
	for _, c := range cases {
		// The stop holds: a pass is clean.
		sw.fake.Put(sealIE, key, ours)
		if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
			t.Fatalf("%s: before: %v", c.name, err)
		}
		c.damage()
		before := sw.fake.Snapshot()
		for i := range 2 {
			err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
			if err == nil || !strings.Contains(err.Error(), "lost stop") {
				t.Fatalf("%s, pass %d: %v", c.name, i, err)
			}
			if sw.fake.Snapshot() != before {
				t.Fatalf("%s, pass %d: the lease was written after the stop was lost", c.name, i)
			}
			r := sw.report(moves.OfflineRoleSealer, "ie-a")
			if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || !strings.Contains(r.Items["lease"].Reason, "lost stop") || r.TombstoneOwner != "" || !r.Items["files"].OK {
				t.Fatalf("%s: report %+v", c.name, r)
			}
		}
	}
	if len(sw.pages) != 1 || !strings.HasPrefix(sw.pages[0], "incident: ") {
		t.Fatalf("pages %v, want one incident", sw.pages)
	}
	if sw.record().Move.State != provision.OfflineSealed {
		t.Fatal("the record moved")
	}
}

// (e) A crash after each step, then the next pass: one tombstone (one
// generation, created once), staged objects copied once and identical, the
// archive the seal names, and a sealed record.
func TestOfflineSealReplaysAfterACrashAtEachStep(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	sw.writeRows(ie, sw.dbPath(ie), 20, 23, false)
	sw.startAndSeal("ie-a", ie)
	ctx := context.Background()
	points := []string{"checkpoint", "tombstone", "final", "mirror", "archive", "timers", "env", "seal", "cas"}
	t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
	var gen int64
	var stagedAfter map[string]lease.MirrorObject
	for i, crash := range points {
		hit := false
		offlineSealFaults.Store(ie.w, func(point string) error {
			if point == crash {
				hit = true
				return errors.New("crash at " + point)
			}
			return nil
		})
		err := sw.pass(ie)
		if !hit || err == nil || !strings.Contains(err.Error(), "crash at "+crash) {
			t.Fatalf("crash at %s: hit %v, %v", crash, hit, err)
		}
		if i >= 1 { // the tombstone is down from the crash after it on
			owner, _, g := sw.lock()
			if owner != moves.TombstoneOwner(offID) || (gen != 0 && g != gen) {
				t.Fatalf("after the crash at %s: lock %s generation %d (was %d)", crash, owner, g, gen)
			}
			gen = g
		}
		if crash == "mirror" {
			stagedAfter = sw.list(sealUK, activator.MovesPrefix(offHost))
		}
		wantState := provision.OfflineSealing
		if crash == "cas" {
			wantState = provision.OfflineSealed
		}
		if s := sw.record().Move.State; s != wantState {
			t.Fatalf("after the crash at %s: state %s", crash, s)
		}
	}
	offlineSealFaults.Delete(ie.w)
	sw.mustPass(ie)
	if sw.record().Move.State != provision.OfflineSealed {
		t.Fatal("not sealed after the replays")
	}
	if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || r.TombstoneOwner != moves.TombstoneOwner(offID) {
		t.Fatalf("living report %+v", r)
	}
	// One tombstone: created once (one version with the move's owner and
	// no expiry change before the seal), one generation throughout.
	if _, _, g := sw.lock(); g != gen {
		t.Fatalf("generation %d, was %d", g, gen)
	}
	versions, err := sw.fake.ListVersions(ctx, sealIE, sealPrefix+"/"+offHost+"/lock.json")
	if err != nil {
		t.Fatal(err)
	}
	created := 0
	for _, v := range versions {
		if v.DeleteMarker {
			continue
		}
		body, err := sw.fake.GetVersion(ctx, sealIE, v.Key, v.VersionID)
		if err != nil {
			t.Fatal(err)
		}
		var l litestream.Lease
		if err := json.Unmarshal(body, &l); err == nil && l.Owner == moves.TombstoneOwner(offID) && l.Generation != gen {
			t.Fatalf("a tombstone of another generation %d", l.Generation)
		}
		if l.Owner == moves.TombstoneOwner(offID) {
			created++
		}
	}
	// The tombstone is written once before the seal (the replays after it
	// find it this move's and write nothing); after the seal the re-asserts
	// of the last two passes rewrite it, same generation.
	if created != 3 {
		t.Fatalf("%d versions of the tombstone", created)
	}
	// The staged objects: identical to those after the first mirror round,
	// and each copied once.
	final := sw.list(sealUK, activator.MovesPrefix(offHost))
	if len(final) != len(stagedAfter) {
		t.Fatalf("staged %d objects, %d after the first round", len(final), len(stagedAfter))
	}
	for k, o := range final {
		if stagedAfter[k] != o {
			t.Fatalf("%s changed: %+v, was %+v", k, o, stagedAfter[k])
		}
		vs, err := sw.fake.ListVersions(ctx, sealUK, k)
		if err != nil || len(vs) != 1 {
			t.Fatalf("%s has %d versions (%v)", k, len(vs), err)
		}
	}
	seal, _, ok := sw.sealRecord()
	if !ok {
		t.Fatal("no seal record")
	}
	sw.openArchive(seal)
	if n := len(seal.Staged); n != len(final) {
		t.Fatalf("seal names %d staged objects, %d staged", n, len(final))
	}
}

// (f) The revert: the sealer lifts this move's tombstone, deletes exactly
// the staging lane and the archive prefix in the target's bucket, clears
// the receipts and lifts the fence; a wake can take the lease again. A
// lease that is not this move's is left alone. Every part is an ensure.
func TestOfflineSealRevertLiftsAndUnfences(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	sw.startAndSeal("ie-a", ie)
	sw.mustPass(ie)
	ctx := context.Background()
	// Neighbours the revert must not touch.
	sw.fake.Put(sealUK, "moves/other.example.com/0000/x.ltx", []byte("x"))
	sw.fake.Put(sealUK, "moves/@offline/other.example.com/"+offID2+"/files.sealed", []byte("x"))
	sw.fake.Put(sealUK, sealPrefix+"/"+offHost+"/0000/y.ltx", []byte("y"))
	// Copy receipts, as a round over a multipart source leaves them.
	sw.fake.Put(sealIE, activator.MoveReceiptsKey(offHost), []byte(`{"v":1}`))
	sw.flip()
	sw.abort()
	if sw.record().Move.State != provision.OfflineReverting {
		t.Fatal("fixture: not reverting")
	}
	if !ie.svc.Fenced(offHost) {
		t.Fatal("fixture: not fenced")
	}
	// The sealer's own step lifts the fence too (the source step does it
	// as well, earlier in the same round).
	if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
		t.Fatal(err)
	}
	if ie.svc.Fenced(offHost) {
		t.Fatal("still fenced after the revert")
	}
	owner, expires, _ := sw.lock()
	if !lease.IsLiftedTombstone(owner, expires) || owner != moves.TombstoneOwner(offID) {
		t.Fatalf("lock %s %s, want this move's lifted tombstone", owner, expires)
	}
	for _, prefix := range []string{activator.MovesPrefix(offHost), moves.OfflineArchivePrefix(offHost, offID)} {
		if n := len(sw.list(sealUK, prefix)); n != 0 {
			t.Fatalf("%s holds %d objects", prefix, n)
		}
	}
	for _, k := range []string{"moves/other.example.com/0000/x.ltx", "moves/@offline/other.example.com/" + offID2 + "/files.sealed", sealPrefix + "/" + offHost + "/0000/y.ltx"} {
		if _, ok := sw.fake.Current(sealUK, k); !ok {
			t.Fatalf("the revert deleted %s", k)
		}
	}
	if _, ok := sw.fake.Current(sealIE, activator.MoveReceiptsKey(offHost)); ok {
		t.Fatal("the receipts are still there")
	}
	if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseReverted {
		t.Fatalf("report %+v", r)
	}
	// Again: nothing changes, and the lifted tombstone is not taken again.
	before := sw.fake.Snapshot()
	sw.mustPass(ie)
	if sw.fake.Snapshot() != before {
		t.Fatal("a second revert pass wrote to the buckets")
	}
	wake, err := acquireAsAWake(ctx)
	if err != nil {
		t.Fatalf("a wake after the revert: %v", err)
	}
	// A lease that is not this move's (a live wake now) is left alone.
	if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
		t.Fatal(err)
	}
	if err := wake.Err(); err != nil {
		t.Fatalf("the revert disturbed a wake's lease: %v", err)
	}
	wake.Release(ctx)
	// Another move's tombstone is never lifted by this one.
	other, err := json.Marshal(litestream.Lease{Generation: 9, Owner: moves.TombstoneOwner(offID2), ExpiresAt: time.Now().Add(time.Hour)})
	if err != nil {
		t.Fatal(err)
	}
	sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/lock.json", other)
	if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
		t.Fatal(err)
	}
	if owner, expires, _ := sw.lock(); owner != moves.TombstoneOwner(offID2) || !time.Now().Before(expires) {
		t.Fatalf("another move's tombstone was lifted: %s %s", owner, expires)
	}
}

// The revert before anything was sealed (abort while sealing failed on a
// refused item) lifts the tombstone the failed seal took, and a revert
// with no tombstone at all is a clean pass.
func TestOfflineSealRevertBeforeTheSeal(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	writeFile(t, filepath.Join(ie.dd, offHost, "big"), "")
	if err := os.Truncate(filepath.Join(ie.dd, offHost, "big"), archive.Cap+1); err != nil {
		t.Fatal(err)
	}
	sw.startAndSeal("ie-a", ie)
	if err := sw.pass(ie); err == nil {
		t.Fatal("sealed over the cap")
	}
	if owner, _, _ := sw.lock(); owner != moves.TombstoneOwner(offID) {
		t.Fatal("fixture: no tombstone")
	}
	sw.abort()
	sw.mustPass(ie)
	if owner, expires, _ := sw.lock(); !lease.IsLiftedTombstone(owner, expires) {
		t.Fatalf("lock %s %s after the revert", owner, expires)
	}
	if n := len(sw.list(sealUK, "moves/")); n != 0 {
		t.Fatalf("%d objects left in the staging prefixes", n)
	}
	sw.mustPass(ie)
}

// The files item: a file over the cap, a FIFO, and an entry of the host's
// beside its files directory are failed items (the report says which),
// never a partial archive and never a seal; once the file is gone the next
// pass seals.
func TestOfflineSealFailsAFileOverTheCapAndAFIFO(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	sw.startAndSeal("ie-a", ie)
	files := filepath.Join(ie.dd, offHost)
	big := filepath.Join(files, "big.bin")
	writeFile(t, big, "")
	if err := os.Truncate(big, archive.Cap+1); err != nil {
		t.Fatal(err)
	}
	fifo := filepath.Join(files, "pipe")
	blobs := filepath.Join(ie.dd, offHost+"-blobs")
	cases := []struct {
		name  string
		setup func()
		want  string
	}{
		{"over the cap", func() {}, strings.TrimPrefix(archive.ErrOverCap.Error(), "archive: ")},
		{"a FIFO", func() {
			if err := os.Remove(big); err != nil {
				t.Fatal(err)
			}
			if err := syscall.Mkfifo(fifo, 0o644); err != nil {
				t.Fatal(err)
			}
		}, strings.TrimPrefix(archive.ErrSpecialFile.Error(), "archive: ") + ": pipe"},
		// The flat layout's rules give <host>-blobs to the host, and the
		// archive carries the files directory alone.
		{"an entry beside the files directory", func() {
			if err := os.Remove(fifo); err != nil {
				t.Fatal(err)
			}
			writeFile(t, filepath.Join(blobs, "b"), "blob")
		}, blobs},
	}
	for _, c := range cases {
		c.setup()
		err := sw.pass(ie)
		var ref *sealRefusal
		if !errors.As(err, &ref) || ref.item != "files" || !strings.Contains(ref.Reason, c.want) {
			t.Fatalf("%s: %v", c.name, err)
		}
		r := sw.report(moves.OfflineRoleSealer, "ie-a")
		if r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK || !r.Items["db"].OK {
			t.Fatalf("%s: report %+v", c.name, r)
		}
		if _, _, ok := sw.sealRecord(); ok || sw.record().Move.State != provision.OfflineSealing {
			t.Fatalf("%s: sealed", c.name)
		}
		if n := len(sw.list(sealUK, moves.OfflineArchivePrefix(offHost, offID))); n != 0 {
			t.Fatalf("%s: %d archive objects written", c.name, n)
		}
	}
	if err := os.RemoveAll(blobs); err != nil {
		t.Fatal(err)
	}
	sw.mustPass(ie)
	if sw.record().Move.State != provision.OfflineSealed {
		t.Fatal("not sealed once the files were fixed")
	}
	if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || !r.Items["files"].OK {
		t.Fatalf("report %+v", r)
	}
}

// A nested instance, made nested by the real layout pass, which takes the
// flat files directory <data>/<host> into the instance directory as
// <instance dir>/<host>, where the app finds it beside its database. The
// archive is read through the raw instance directory and carries that
// files directory and the env file, never the database, its litestream
// state, the rendered configuration or any other platform entry; it
// extracts flat with the app's files at the flat files directory's root,
// where the flat app looks. An entry beside the files directory that the
// flat layout cannot carry refuses the files item until it is gone.
func TestOfflineSealArchivesANestedInstanceFlat(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	raw := sw.nest(ie)
	// More platform entries, which stay behind: the generation and files
	// markers, the adoption record, the seed marker of a seeded database,
	// and the launch descriptors (fixed and per generation) as a real
	// nested host keeps them.
	for _, name := range []string{offHost + ".db.gen", ".files-gen", offHost + instancekey.AdoptionSuffix, offHost + ".db" + activator.SeedMarkerSuffix,
		"launch.run.json", "launch.seed.json", "launch.gen.7.json"} {
		writeFile(t, filepath.Join(raw, name), "platform")
	}
	// An entry of the app's beside its files directory, which no flat
	// files directory can hold.
	stray := filepath.Join(raw, offHost+"-blobs", "b")
	writeFile(t, stray, "blob")
	sw.startAndSeal("ie-a", ie)
	err := sw.pass(ie)
	var ref *sealRefusal
	if !errors.As(err, &ref) || ref.item != "files" || !strings.Contains(ref.Reason, offHost+"-blobs") {
		t.Fatalf("a stray entry: %v", err)
	}
	// The refusal says the move cannot carry it, and offers no fix that
	// moves or removes the app's data (review I3).
	if !strings.Contains(ref.Reason, "this move can carry only the files directory. see the offline move runbook") || strings.Contains(ref.Reason, "move it") || strings.Contains(ref.Reason, "remove") {
		t.Fatalf("the refusal: %q", ref.Reason)
	}
	for _, name := range []string{"launch.run.json", "launch.gen.7.json", ".db.seed"} {
		if strings.Contains(ref.Reason, name) {
			t.Fatalf("the refusal names the platform's %s: %q", name, ref.Reason)
		}
	}
	if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK {
		t.Fatalf("report %+v", r)
	}
	if err := os.RemoveAll(filepath.Dir(stray)); err != nil {
		t.Fatal(err)
	}
	sw.mustPass(ie)
	if sw.record().Move.State != provision.OfflineSealed {
		t.Fatal("not sealed")
	}
	seal, _, _ := sw.sealRecord()
	files, env := sw.openArchive(seal)
	got := map[string]bool{}
	if err := filepath.Walk(files, func(p string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !info.IsDir() {
			rel, _ := filepath.Rel(files, p)
			got[rel] = true
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if len(got) != 2 || !got["instance.key"] || !got[filepath.Join("uploads", "a.png")] {
		t.Fatalf("the archive carried %v, want the app's two files", got)
	}
	if readString(t, env) != readString(t, filepath.Join(raw, offHost+".env")) {
		t.Fatal("the env file did not travel byte for byte")
	}
	if seal.KeySHA256 != instancekey.Digest("the-instance-key") {
		t.Fatal("the key digest")
	}
}

// nest makes b's instance of offHost nested with the real layout pass,
// which takes the flat files directory <data>/<host> into the instance
// directory as <instance dir>/<host>, and returns the raw instance
// directory.
func (sw *sealWorld) nest(ie *offBox) string {
	t := sw.t
	t.Helper()
	ctx := context.Background()
	// The bind mount at <dd>/instances, as a link in the fixture.
	if err := os.MkdirAll(filepath.Join(ie.dd, ".instances"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(".instances", filepath.Join(ie.dd, "instances")); err != nil {
		t.Fatal(err)
	}
	// The layout pass works on the row's own paths; the sockets live in a
	// short directory of the test's, as the pass's tests keep them.
	sd, err := os.MkdirTemp("", "ly")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(sd) })
	rt := ie.row(offHost)
	if err := ie.reg.SetInstancePaths(offHost, rt.DBPath, rt.Socket, rt.DBPath, filepath.Join(sd, offHost+".sock")); err != nil {
		t.Fatal(err)
	}
	pass := &layout.Pass{Registry: ie.reg, DataDir: ie.dd, SocketDir: sd, SocketsGID: -1, Now: time.Now, SweptAtStart: true, Logf: t.Logf}
	res := pass.Converge(ctx, layout.ModeNested)
	migrated := false
	for _, r := range res {
		if r.Host == offHost && r.Outcome == "migrated" {
			migrated = true
		}
	}
	if !migrated {
		t.Fatalf("the layout pass: %+v", res)
	}
	ie.nested = true
	raw := layout.RawInstanceDir(ie.dd, offHost)
	if rt := ie.row(offHost); !provision.NestedPaths(ie.dd, *rt) {
		t.Fatalf("fixture: row %+v is not nested", rt)
	}
	if lexists(filepath.Join(ie.dd, offHost)) {
		t.Fatal("fixture: the layout pass left the flat files directory")
	}
	for _, name := range []string{offHost + ".db", offHost + ".env", offHost + ".config.env", ".activator-" + offHost + ".db-litestream", filepath.Join(offHost, "instance.key")} {
		if !lexists(filepath.Join(raw, name)) {
			t.Fatalf("fixture: the layout pass did not move %s into the instance directory", name)
		}
	}
	return raw
}

// Finishing lifts the tombstone only once every box the seal named reports
// gone: until then each pass keeps the stop. After the lift a wake takes
// the lease, the report says lifted, and a later pass changes nothing.
func TestOfflineSealFinishingLiftsOnlyAfterEverySourceIsGone(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	ieB := sw.sourceBox("ie-b")
	sw.startAndSeal("ie-a", ie, ieB)
	sw.mustPass(ie)
	ieB.tick()
	if sw.record().Move.State != provision.OfflineSealed {
		t.Fatal("not sealed")
	}
	sw.flip()
	sw.landed()
	sw.open()
	sw.mustPass(ie)
	sw.mustPass(ieB)
	for _, b := range []string{"ie-a", "ie-b"} {
		if r := sw.report(moves.OfflineRoleSource, b); r.Phase != moves.OfflinePhaseParked {
			t.Fatalf("%s: %+v", b, r)
		}
	}
	sw.finishing()
	ctx := context.Background()
	// ie-a goes; ie-b has not run a pass since.
	for range 3 {
		sw.mustPass(ie)
	}
	if r := sw.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseGone {
		t.Fatalf("ie-a: %+v", r)
	}
	owner, _, _ := sw.lock()
	if owner != moves.TombstoneOwner(offID) {
		t.Fatal("fixture: no tombstone")
	}
	if _, err := acquireAsAWake(ctx); err == nil {
		t.Fatal("the tombstone was lifted before ie-b was gone")
	}
	if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || r.TombstoneOwner != moves.TombstoneOwner(offID) {
		t.Fatalf("sealer report while waiting: %+v", r)
	}
	for range 3 {
		sw.mustPass(ieB)
	}
	if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.Phase != moves.OfflinePhaseGone {
		t.Fatalf("ie-b: %+v", r)
	}
	sw.mustPass(ie)
	if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseLifted {
		t.Fatalf("sealer report: %+v", r)
	}
	owner, expires, _ := sw.lock()
	if !lease.IsLiftedTombstone(owner, expires) || owner != moves.TombstoneOwner(offID) {
		t.Fatalf("lock %s %s", owner, expires)
	}
	before := sw.fake.Snapshot()
	sw.mustPass(ie)
	if sw.fake.Snapshot() != before {
		t.Fatal("a pass after the lift wrote to the buckets")
	}
	wake, err := acquireAsAWake(ctx)
	if err != nil {
		t.Fatalf("a wake after the lift: %v", err)
	}
	wake.Release(ctx)
	if len(sw.pages) != 0 {
		t.Fatalf("paged: %v", sw.pages)
	}
}

// A host that never kept a files directory (seen on real nested hosts: the
// instance directory holds the database, the env files and the launch
// descriptors, and no <host>) seals with an empty files item: the archive
// carries the env file alone, FilesPlain is archive.PlainDigest of no files
// directory and that env file (FilesEntries 1), and it extracts to an
// empty files directory beside the env file. Nested and flat alike.
func TestOfflineSealAHostWithNoFilesDirectory(t *testing.T) {
	for _, nested := range []bool{true, false} {
		t.Run(map[bool]string{true: "nested", false: "flat"}[nested], func(t *testing.T) {
			sw := newSealWorld(t)
			ie := sw.ie
			sw.instance(ie)
			files := filepath.Join(ie.dd, offHost)
			envPath := filepath.Join(ie.dd, offHost+".env")
			if nested {
				raw := sw.nest(ie)
				files, envPath = filepath.Join(raw, offHost), filepath.Join(raw, offHost+".env")
				writeFile(t, filepath.Join(raw, "launch.run.json"), "{}")
			}
			if err := os.RemoveAll(files); err != nil {
				t.Fatal(err)
			}
			sw.startAndSeal("ie-a", ie)
			sw.mustPass(ie)
			if sw.record().Move.State != provision.OfflineSealed {
				t.Fatal("not sealed")
			}
			seal, _, _ := sw.sealRecord()
			want, wantBytes, wantEntries, err := archive.PlainDigest(files, envPath)
			if err != nil {
				t.Fatal(err)
			}
			if seal.FilesPlain != want || seal.FilesBytes != wantBytes || seal.FilesEntries != wantEntries || wantEntries != 1 {
				t.Fatalf("seal files %s %d bytes %d entries, want %s %d bytes 1 entry (env only)", seal.FilesPlain, seal.FilesBytes, seal.FilesEntries, want, wantBytes)
			}
			if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || !r.Items["files"].OK {
				t.Fatalf("report %+v", r)
			}
			got, env := sw.openArchive(seal)
			ents, err := os.ReadDir(got)
			if err != nil || len(ents) != 0 {
				t.Fatalf("the landing's files directory: %v entries, %v", len(ents), err)
			}
			if readString(t, env) != readString(t, envPath) {
				t.Fatal("the env file did not travel byte for byte")
			}
			if lexists(files) {
				t.Fatal("the seal made a files directory on the source")
			}
		})
	}
}

// The tombstone goes between the sealer's LockState and Reassert's own
// read: Reassert reports it lost (lease.ErrTombstoneLost) and writes
// nothing, so the sealer pages a lost stop once and never takes it again.
func TestOfflineSealLostBetweenTheReadAndTheReassert(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	sw.startAndSeal("ie-a", ie)
	sw.mustPass(ie)
	ctx := context.Background()
	key := sealPrefix + "/" + offHost + "/lock.json"
	var after string
	t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
	offlineSealFaults.Store(ie.w, func(point string) error {
		if point == "reassert" {
			sw.fake.BareDelete(sealIE, key)
			after = sw.fake.Snapshot()
		}
		return nil
	})
	err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
	if err == nil || !strings.Contains(err.Error(), "lost stop") {
		t.Fatalf("pass: %v", err)
	}
	if after == "" || sw.fake.Snapshot() != after {
		t.Fatal("the bucket was written after the tombstone went")
	}
	if _, ok := sw.fake.Current(sealIE, key); ok {
		t.Fatal("the tombstone was taken again")
	}
	r := sw.report(moves.OfflineRoleSealer, "ie-a")
	if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || r.Items["lease"].Reason != "lost stop: "+offlineLeaseCause(tombState{}) {
		t.Fatalf("report %+v", r)
	}
	// The next pass finds it gone at LockState: still nothing written,
	// still one page.
	offlineSealFaults.Delete(ie.w)
	if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err == nil || !strings.Contains(err.Error(), "lost stop") {
		t.Fatalf("second pass: %v", err)
	}
	if sw.fake.Snapshot() != after {
		t.Fatal("the second pass wrote to the bucket")
	}
	if len(sw.pages) != 1 {
		t.Fatalf("pages %v, want one", sw.pages)
	}
}

// The lost stop's page says abort only while abort can still put the
// instance back. From open on, abort is refused, so it points to the
// status and the runbook instead. Each state in its own world: the page
// goes out once per move.
func TestOfflineSealLostStopPageByState(t *testing.T) {
	for _, state := range []string{provision.OfflineSealed, provision.OfflineOpen, provision.OfflineFinishing} {
		t.Run(state, func(t *testing.T) {
			sw := newSealWorld(t)
			ie := sw.ie
			sw.instance(ie)
			sw.startAndSeal("ie-a", ie)
			sw.mustPass(ie)
			ctx := context.Background()
			if state != provision.OfflineSealed {
				sw.flip()
				sw.landed()
				sw.open()
			}
			if state == provision.OfflineFinishing {
				sw.finishing()
			}
			if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
				t.Fatalf("the stop held: %v", err)
			}
			if got := sw.record().Move.State; got != state {
				t.Fatalf("fixture: state %s", got)
			}
			sw.fake.BareDelete(sealIE, sealPrefix+"/"+offHost+"/lock.json")
			if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err == nil || !strings.Contains(err.Error(), "lost stop") {
				t.Fatalf("lost: %v", err)
			}
			if len(sw.pages) != 1 || sw.pages[0] != "incident: offline move of "+offHost+" lost its stop" {
				t.Fatalf("pages %v", sw.pages)
			}
			body := sw.bodies[0]
			if !strings.HasPrefix(body, "The offline move "+offID+" of "+offHost+" is "+state+", and its stop no longer holds: the source's lease is absent, not this move's tombstone. ") {
				t.Fatalf("body: %s", body)
			}
			abort := strings.HasSuffix(body, "The move cannot go on. Read carlos move offline status --host "+offHost+" and abort the move.")
			runbook := strings.HasSuffix(body, "It cannot be aborted once it is open. Read carlos move offline status --host "+offHost+" and follow the offline move runbook.")
			if wantAbort := state == provision.OfflineSealed; abort != wantAbort || runbook == wantAbort {
				t.Fatalf("body in %s: %s", state, body)
			}
		})
	}
}

// The backstop behind the holds and the fence (review M2): a source report
// written since the seal at a position past the sealed one is a lost stop
// in the db-tail item, even with the tombstone held. Nothing is written to
// the lease, and the page names the box.
func TestOfflineSealLostStopOnASourcePastTheSealedPosition(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	ieB := sw.sourceBox("ie-b")
	sw.startAndSeal("ie-a", ie, ieB)
	sw.mustPass(ie)
	if sw.record().Move.State != provision.OfflineSealed {
		t.Fatal("not sealed")
	}
	seal, _, _ := sw.sealRecord()
	ctx := context.Background()
	// A pass with every report at or below the seal keeps the stop.
	ieB.tick()
	if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
		t.Fatalf("at the seal: %v", err)
	}
	// ie-b writes past the sealed position, and reports it.
	meta := filepath.Join(activatorMetaDir(provision.DefaultDBPath(ieB.dd, offHost)), "ltx", "0")
	writeFile(t, filepath.Join(meta, ltxName(seal.FinalTXID+1, seal.FinalTXID+3)), "ltx")
	ieB.tick()
	if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.LocalTXID != seal.FinalTXID+3 || r.At.Before(seal.At) {
		t.Fatalf("fixture: ie-b reports %d at %s", r.LocalTXID, r.At)
	}
	before := sw.fake.Snapshot()
	err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
	if err == nil || !strings.Contains(err.Error(), "lost stop") {
		t.Fatalf("past the seal: %v", err)
	}
	if sw.fake.Snapshot() != before {
		t.Fatal("the lease was written after the stop was lost")
	}
	cause := offlinePastSealCause("ie-b", seal.FinalTXID+3, seal.FinalTXID)
	r := sw.report(moves.OfflineRoleSealer, "ie-a")
	if r.Phase != moves.OfflinePhaseFailed || r.Items["db-tail"].OK || r.Items["db-tail"].Reason != "lost stop: "+cause || r.TombstoneOwner != "" || !r.Items["lease"].OK {
		t.Fatalf("report %+v", r)
	}
	if len(sw.bodies) != 1 || !strings.Contains(sw.bodies[0], "its stop no longer holds: "+cause+".") {
		t.Fatalf("pages %v %v", sw.pages, sw.bodies)
	}
	if owner, _, _ := sw.lock(); owner != moves.TombstoneOwner(offID) {
		t.Fatalf("the tombstone is %s", owner)
	}
}

// Fix round 1, Astra's finding 2: the report refresh (offlineReassert) is
// report-only for the sealer. Repeated refreshes rewrite the sealer's report
// with the tombstone's expiry as read and leave the lock object byte for
// byte; an expired, missing or replaced tombstone makes the refresh report
// the lost stop, never a fresh report, and the lock is not written.
func TestOfflineSealerRefreshNeverRenewsTheTombstone(t *testing.T) {
	sw := newSealWorld(t)
	ie := sw.ie
	sw.instance(ie)
	sw.startAndSeal("ie-a", ie)
	sw.mustPass(ie) // seals, takes the tombstone, records sealed
	if st := sw.record().Move.State; st != provision.OfflineSealed {
		t.Fatalf("fixture: state %s", st)
	}
	sw.mustPass(ie) // the view learns sealed, as every pass refreshes it
	ctx := context.Background()
	key := sealPrefix + "/" + offHost + "/lock.json"
	ours, ok := sw.fake.Current(sealIE, key)
	if !ok {
		t.Fatal("fixture: no tombstone")
	}
	_, expires, _ := sw.lock()
	for i := range 3 {
		time.Sleep(20 * time.Millisecond) // a renewal would write a later expiry
		sw.advance(15 * time.Second)
		if n := ie.w.offlineReassert(ctx, sw.s); n != 1 {
			t.Fatalf("refresh %d acted on %d records", i, n)
		}
		if now, _ := sw.fake.Current(sealIE, key); !bytes.Equal(now, ours) {
			t.Fatalf("refresh %d rewrote the tombstone:\nbefore %s\nafter  %s", i, ours, now)
		}
		r := sw.report(moves.OfflineRoleSealer, "ie-a")
		if r.Phase != moves.OfflinePhaseSealed || !r.At.Equal(sw.clock()) || !r.TombstoneUntil.Equal(expires.UTC()) || r.TombstoneOwner != moves.TombstoneOwner(offID) {
			t.Fatalf("refresh %d: report %+v, want sealed at %v with the expiry %v as read", i, r, sw.clock(), expires)
		}
	}

	lockBody := func(owner string, exp time.Time) []byte {
		b, err := json.Marshal(litestream.Lease{Generation: 1, Owner: owner, ExpiresAt: exp})
		if err != nil {
			t.Fatal(err)
		}
		return b
	}
	for _, c := range []struct {
		name   string
		damage func()
	}{
		{"expired", func() { sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID), time.Now().Add(-time.Second))) }},
		{"missing", func() { sw.fake.BareDelete(sealIE, key) }},
		{"replaced", func() { sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID2), time.Now().Add(time.Hour))) }},
	} {
		sw.fake.Put(sealIE, key, ours)
		sw.advance(15 * time.Second)
		ie.w.offlineReassert(ctx, sw.s)
		if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed {
			t.Fatalf("%s: before the damage the report is %s", c.name, r.Phase)
		}
		c.damage()
		before := sw.fake.Snapshot()
		sw.advance(15 * time.Second)
		ie.w.offlineReassert(ctx, sw.s)
		if sw.fake.Snapshot() != before {
			t.Fatalf("%s: the refresh wrote the lease", c.name)
		}
		r := sw.report(moves.OfflineRoleSealer, "ie-a")
		if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || !strings.Contains(r.Items["lease"].Reason, "lost stop") || r.TombstoneOwner != "" {
			t.Fatalf("%s: report %+v, want the lost stop", c.name, r)
		}
	}
}