package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/superfly/ltx"

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

// The sealer of an offline operator move (docs/superpowers/plans/2026-10-20-
// offline-operator-move.md, Task 4): the one source box that turns "every
// source box holds, fences and reports" into a stop at the data layer, and
// records what the target must land.
//
// The tombstone is the linearization point of the whole move. Every wake on
// every box of the source region takes the host's single-writer lease
// (instances/<host>/lock.json in the source's own bucket) before it can
// write anything, and from the tombstone on that lease is held by
// moved:<id>. The order is the plan's (Review Focus 2):
//
//  1. this box's own report for the pass says held, fenced, quiesced, no
//     process and no layout journal, and no source box keeps a sidecar;
//  2. the local database is checkpointed unless it is behind the replica
//     (cfg.Checkpoint takes the lease, syncs and releases it);
//  3. lease.Tombstone, which never replaces a live holder;
//  4. only then the final position is read from the replica, and every
//     source box's local position must be at or below it;
//  5. CarryRequest, a final MirrorPrefix round with nothing deferred, and
//     the staged listing;
//  6. to 9. the files archive, the timers, the env and its pin, the release,
//     replication and next due;
//  10. seal.json, written once and never rewritten (putOfflineCreateOnly),
//     then the record's CAS to offline-sealed.
//
// Every step is an ensure. Until seal.json exists the whole seal is taken
// again on each pass (a tombstone that is already this move's is a no-op,
// the mirror resumes from its receipts, the archive is rebuilt and its
// objects overwritten), so a crash anywhere before it converges. Once
// seal.json exists the sealed position is fixed, and from then on the
// tombstone is never taken again: each pass reads lease.LockState first,
// and a tombstone that is absent, expired or anyone else's is a lost stop,
// as is a source report written since the seal at a position past the
// sealed one. The pass reports failed and pages: before open the operator
// aborts, from open on the runbook fixes it forward. Only a tombstone held
// by this move is re-asserted (lease.Reassert, If-Match on the ETag it
// reads, and lease.ErrTombstoneLost, never a new tombstone, when it went
// in between). After it is lifted (finish, abort) it is never re-asserted.

// The objects the sealer writes beside the sealed archive, under
// moves.OfflineArchivePrefix(host, id) in the target's data bucket. The
// landing (Task 5) reads them there.
const (
	// offlineArchiveSealed is the sealed files archive (archive.Seal of
	// archive.Build's output). Its sha256 is the seal's FilesSealed.
	offlineArchiveSealed = "files.sealed"
	// offlineArchiveManifest is its archive.Manifest (JSON), with the
	// sealed data key and every entry.
	offlineArchiveManifest = "files.json"
	// offlineTimersObject is the moves.TimersSnapshot of the row's active
	// timers. moves.TimersDigest of its rows is the seal's TimersDigest.
	offlineTimersObject = "timers.json"
	// offlineEnvReportObject is the source's moves.EnvReport (names and
	// HMACs, never a value), so the landing can name the variables that
	// differ (moves.CompareEnv) when the env pin does not match. Its
	// Digest is the seal's EnvDigest.
	offlineEnvReportObject = "env-report.json"
)

// offlineSealFresh is how old this box's own source report may be when the
// sealer reads it: the source step writes it earlier in the same pass, so
// anything older means that step failed this pass.
const offlineSealFresh = 2 * time.Minute

// offlinePageLostStop is the page rule of a lost stop: one page per move.
const offlinePageLostStop = "offline-lost-stop"

// offlineSealFaults is the sealer's test seam: a worker's entry is called
// after each step with the step's name, and an error returned is a crash
// there. Keyed by the worker, so tests that run side by side never share
// one. Empty in production.
var offlineSealFaults sync.Map // *movesWorker → func(point string) error

func (w *movesWorker) sealFault(point string) error {
	if f, ok := offlineSealFaults.Load(w); ok {
		return f.(func(string) error)(point)
	}
	return nil
}

// offlineSealerStep is the sealer's part of an offline move. offlineRound
// calls it on the box whose label is rec.Move.Sealer, after this box's own
// source step has written its report for the pass.
func (w *movesWorker) offlineSealerStep(ctx context.Context, s release.Store, rec provision.Record) error {
	m := rec.Move
	if !m.Offline() || w.cfg.Box == "" || m.Sealer != w.cfg.Box {
		return nil
	}
	switch m.State {
	case provision.OfflineSealing:
		return w.offlineSeal(ctx, s, rec)
	case provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded, provision.OfflineOpen:
		raw, err := w.offlineSealRecord(ctx, s, rec)
		if err != nil {
			return err
		}
		return w.offlineKeepStop(ctx, s, rec, raw)
	case provision.OfflineFinishing:
		return w.offlineFinishLift(ctx, s, rec)
	case provision.OfflineReverting:
		return w.offlineSealRevert(ctx, s, rec)
	}
	return nil
}

// tombstoneAt is where the host's single-writer lease lives in this box's
// own bucket: the activator's lease location, as ExecBackend.Acquire takes
// it for every wake.
func (c moveCfg) tombstoneAt(host string) (bucket, location, region string, err error) {
	if c.OwnBucket == "" {
		return "", "", "", errors.New("offline move: this box has no hibernate bucket")
	}
	return c.OwnBucket, activator.HibernatePrefix(c.OwnPrefix, host), c.OwnBucketRegion, nil
}

// tombState is the lease as lease.LockState read it: an absent lock has no
// owner and a zero expiry.
type tombState struct {
	owner   string
	expires time.Time
}

// mine reports a tombstone this move holds now. "Expired" is the lease's
// own reading (Acquire takes a lease the moment its expiry has passed), so
// it is judged against the wall clock the lease is written by.
func (t tombState) mine(id string) bool {
	return t.owner == moves.TombstoneOwner(id) && time.Now().Before(t.expires)
}

// free reports a lease nothing holds: absent, or past its expiry, whoever
// wrote it (a lifted tombstone included: lease.IsLiftedTombstone).
func (t tombState) free() bool {
	return t.owner == "" || !time.Now().Before(t.expires)
}

func (t tombState) String() string { return offlineLeaseText(t) }

// offlineLeaseText is the lease as the lost stop's page and the sealer's
// report name it (copy rows MV-O72 to MV-O75).
func offlineLeaseText(t tombState) string {
	switch {
	case t.owner == "":
		return "absent"
	case lease.IsLiftedTombstone(t.owner, t.expires) && t.expires.Unix() == 0:
		// LiftTombstone's expiry: the epoch.
		return fmt.Sprintf("lifted (%s)", t.owner)
	case t.free():
		return fmt.Sprintf("expired (%s at %s)", t.owner, t.expires.UTC().Format(time.RFC3339))
	}
	return fmt.Sprintf("held by %s until %s", t.owner, t.expires.UTC().Format(time.RFC3339))
}

// offlineLock reads the host's lease.
func (w *movesWorker) offlineLock(ctx context.Context, host string) (tombState, error) {
	bucket, loc, region, err := w.cfg.tombstoneAt(host)
	if err != nil {
		return tombState{}, err
	}
	_, owner, expires, err := lease.LockState(ctx, bucket, loc, region)
	if err != nil {
		return tombState{}, fmt.Errorf("read the lease of %s: %w", host, err)
	}
	return tombState{owner: owner, expires: expires}, nil
}

// offlineTarget is the target's coordinates as its edges file publishes
// them: the box at move.to (the lexically first, as every move reads the
// fleet), matched exactly and then by placement under the home binding.
func (w *movesWorker) offlineTarget(ctx context.Context, s release.Store, m *provision.MoveState) (edgesFile, error) {
	fleet, err := loadMoveFleet(ctx, s)
	if err != nil {
		return edgesFile{}, fmt.Errorf("read the fleet's edges files: %w", err)
	}
	if t, ok := fleet.targets[m.To]; ok {
		return t, nil
	}
	codes := make([]string, 0, len(fleet.targets))
	for code := range fleet.targets {
		codes = append(codes, code)
	}
	sort.Strings(codes)
	for _, code := range codes {
		if park.PlacementMatches(w.cfg.ParkHomeRegion, m.To, park.PlacementRegion(code)) {
			return fleet.targets[code], nil
		}
	}
	return edgesFile{}, fmt.Errorf("the target region %q publishes no data bucket yet", m.To)
}

// offlineMirror is the mirror pair between this box's bucket and region.
func (w *movesWorker) offlineMirror(ctx context.Context, region string) (lease.MirrorStore, error) {
	if w.cfg.Mirror == nil {
		return nil, errors.New("offline move: no S3 mirror on this box")
	}
	return w.mirror(ctx, region)
}

// replicaPosition is the newest TXID of the replica listed in objs under
// prefix (newestLTX), 0 when it holds no LTX file.
func replicaPosition(objs map[string]lease.MirrorObject, prefix string) uint64 {
	newest := newestLTX(objs, prefix, nil)
	if newest == "" {
		return 0
	}
	_, maxTX, err := ltx.ParseFilename(filepath.Base(newest))
	if err != nil {
		return 0
	}
	return uint64(maxTX)
}

// sealRefusal is a seal item that failed for a reason a later pass will
// meet again: the pass reports it (phase failed) and the operator aborts.
// Reason is the item's reason as `carlos move offline status` shows it.
// Each refusal is written as a literal where it is decided, so the copy
// gate reads every sentence a Reason can carry (copy rows MV-O76 to
// MV-O84); the two that carry an error's own text (the archive's refusal
// and the secrets key's) are box diagnostics (movesDiagnosticWrites).
type sealRefusal struct {
	item, Reason string
}

func (r *sealRefusal) Error() string { return r.item + ": " + r.Reason }

// offlineSeal is the sealer's step in offline-sealing.
func (w *movesWorker) offlineSeal(ctx context.Context, s release.Store, rec provision.Record) error {
	m, host := rec.Move, rec.Host
	sealKey := moves.OfflineSealKey(host, m.ID)
	if sealKey == "" {
		return fmt.Errorf("offline move %s: %s has no seal key", m.ID, host)
	}
	raw, _, err := s.Get(ctx, sealKey)
	switch {
	case err == nil:
		// Sealed on an earlier pass, which crashed before the CAS: the
		// position is fixed, so the stop is only kept, never taken again.
		return w.offlineKeepStop(ctx, s, rec, raw)
	case !errors.Is(err, os.ErrNotExist):
		return fmt.Errorf("read %s: %w", sealKey, err)
	}
	items := map[string]moves.Check{}
	seal, err := w.offlineTakeSeal(ctx, s, rec, items)
	var ref *sealRefusal
	if errors.As(err, &ref) {
		items[ref.item] = moves.Check{OK: false, Reason: ref.Reason}
		r := moves.OfflineReport{Op: m.ID, Host: host, Box: w.cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseFailed,
			At: w.cfg.now().UTC(), Items: items}
		return errors.Join(fmt.Errorf("offline move %s: %s cannot be sealed: %w", m.ID, host, err), writeOfflineReport(ctx, s, r))
	}
	if err != nil {
		return fmt.Errorf("offline move %s: sealing %s: %w", m.ID, host, err)
	}
	body, err := json.Marshal(seal)
	if err != nil {
		return err
	}
	stored, created, err := putOfflineCreateOnly(ctx, s, sealKey, body)
	if err != nil {
		return err
	}
	if created {
		log.Printf("moves: %s: offline move %s: sealed at TXID %d (%d staged objects, files %s)", host, m.ID, seal.FinalTXID, len(seal.Staged), seal.FilesPlain)
	}
	if err := w.sealFault("seal"); err != nil {
		return err
	}
	return w.offlineKeepStop(ctx, s, rec, stored)
}

// offlineTakeSeal runs steps 1 to 9 and returns the seal to write. items
// collects each item's check as it is decided; a *sealRefusal is the item
// that failed.
func (w *movesWorker) offlineTakeSeal(ctx context.Context, s release.Store, rec provision.Record, items map[string]moves.Check) (moves.OfflineReport, error) {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	var seal moves.OfflineReport

	// 1. The stop, as the source reports say it now: this box held, fenced,
	// quiesced, with no process and no layout journal; every box the seal
	// named held; no sidecar anywhere.
	if len(m.Boxes) == 0 {
		return seal, errors.New("the record names no source boxes")
	}
	reports, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleSource)
	if err != nil {
		return seal, fmt.Errorf("read the source reports: %w", err)
	}
	now := cfg.now().UTC()
	own, ok := reports[cfg.Box]
	if !ok || own.Phase != moves.OfflinePhaseStopped || !own.Held || !own.Fenced || !own.Quiesced || own.Procs != 0 || own.LayoutJournal || now.Sub(own.At) > offlineSealFresh {
		return seal, fmt.Errorf("this box has not stopped %s this pass: report found %v, held %v, fenced %v, quiesced %v, processes %d, layout journal %v, at %s",
			host, ok, own.Held, own.Fenced, own.Quiesced, own.Procs, own.LayoutJournal, own.At.Format(time.RFC3339))
	}
	boxes := append([]string(nil), m.Boxes...)
	sort.Strings(boxes)
	for _, b := range boxes {
		r, ok := reports[b]
		if !ok || !r.Held {
			return seal, fmt.Errorf("waiting for %s to report %s held", b, host)
		}
	}
	for _, b := range boxes {
		if reports[b].Sidecar {
			return seal, &sealRefusal{item: "sidecar", Reason: fmt.Sprintf("%s keeps a sidecar file for %s", b, host)}
		}
	}
	items["sidecar"] = moves.Check{OK: true, Reason: "no source box keeps a sidecar file"}

	rt, err := w.offlineRow(host)
	if err != nil {
		return seal, err
	}
	if rt == nil {
		return seal, fmt.Errorf("this box has no row for %s", host)
	}
	tgt, err := w.offlineTarget(ctx, s, m)
	if err != nil {
		return seal, err
	}
	// A staged move hands over in the destination its stage fixed (Astra
	// round 2, finding 2). A target that publishes anything else now is
	// not sealed into; the seal door refuses it first (MV-O192).
	if m.StagedAt != nil && (tgt.DataBucket != m.StageBucket || tgt.DataBucketRegion != m.StageRegion) {
		return seal, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion)
	}
	mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)
	if err != nil {
		return seal, err
	}
	bucket, loc, region, err := cfg.tombstoneAt(host)
	if err != nil {
		return seal, err
	}
	srcPrefix := loc + "/"
	dstPrefix := activator.MovesPrefix(host)
	dbPath := rt.DBPath
	if dbPath == "" {
		dbPath = cfg.flatDBPath(host, rt)
	}

	// 2. The checkpoint, while no tombstone is down: the local database is
	// synced under the lease, which the checkpoint releases again. That
	// carries a tail the stop failed to checkpoint, whether litestream had
	// already turned it into a local LTX file (the local position is past
	// the replica's) or it is still in the WAL (the positions are equal),
	// as the legacy cut re-checkpoints unconditionally. A local copy behind
	// the replica is a stale warm cache: the replica is the instance, and
	// syncing the stale copy into it is the one thing never done. A
	// tombstone already this move's means an earlier pass checkpointed and
	// took it, and nothing has woken since; the checkpoint could not take
	// the lease anyway.
	local, err := w.localState(host, rt)
	if err != nil {
		return seal, err
	}
	objs, err := mirror.List(ctx, bucket, srcPrefix)
	if err != nil {
		return seal, fmt.Errorf("list the replica: %w", err)
	}
	before, err := w.offlineLock(ctx, host)
	if err != nil {
		return seal, err
	}
	replica := replicaPosition(objs, srcPrefix)
	switch {
	case before.mine(m.ID):
	case local.LocalTXID < replica:
		log.Printf("moves: %s: offline move %s: the local copy (position %d) is behind the replica (%d); the replica is sealed as it is", host, m.ID, local.LocalTXID, replica)
	case lexists(dbPath):
		if cfg.Checkpoint == nil {
			return seal, errors.New("no checkpoint on this box")
		}
		if err := cfg.Checkpoint(ctx, dbPath); err != nil {
			return seal, fmt.Errorf("final checkpoint of %s: %w", dbPath, err)
		}
		log.Printf("moves: %s: offline move %s: checkpointed the local database (position %d, replica %d)", host, m.ID, local.LocalTXID, replica)
	case local.LocalTXID > replica:
		return seal, &sealRefusal{item: "db-tail", Reason: fmt.Sprintf("this box's local position %d is past the replica's %d and there is no database at %s to checkpoint", local.LocalTXID, replica, dbPath)}
	}
	if err := w.sealFault("checkpoint"); err != nil {
		return seal, err
	}

	// 3. The tombstone: from here no wake on any box can take the lease.
	owner := moves.TombstoneOwner(m.ID)
	if err := lease.Tombstone(ctx, bucket, loc, region, owner, moves.TombstoneTTL); err != nil {
		return seal, fmt.Errorf("tombstone %s/%s/lock.json: %w", bucket, loc, err)
	}
	if err := w.sealFault("tombstone"); err != nil {
		return seal, err
	}

	// 4. The final position, read only now, and no source box past it.
	objs, err = mirror.List(ctx, bucket, srcPrefix)
	if err != nil {
		return seal, fmt.Errorf("list the replica: %w", err)
	}
	final := replicaPosition(objs, srcPrefix)
	if final == 0 {
		return seal, &sealRefusal{item: "db", Reason: fmt.Sprintf("the replica at %s/%s holds no database", bucket, srcPrefix)}
	}
	mine, err := w.localState(host, rt) // after the checkpoint, not as reported before it
	if err != nil {
		return seal, err
	}
	for _, b := range boxes {
		pos := reports[b].LocalTXID
		if b == cfg.Box {
			pos = mine.LocalTXID
		}
		if pos > final {
			return seal, &sealRefusal{item: "db-tail", Reason: fmt.Sprintf("%s's local position %d is past the replica's final position %d", b, pos, final)}
		}
	}
	items["db-tail"] = moves.Check{OK: true, Reason: fmt.Sprintf("every source box is at or below TXID %d", final)}
	items["lease"] = moves.Check{OK: true, Reason: fmt.Sprintf("%s holds %s/lock.json", owner, loc)}
	if err := w.sealFault("final"); err != nil {
		return seal, err
	}

	// 5. The request record into the build state, then the byte-final
	// round into the target's staging lane and its listing.
	if cfg.CarryRequest != nil {
		if err := cfg.CarryRequest(ctx, host, bucket, loc); err != nil {
			return seal, fmt.Errorf("carry the request record into the build state: %w", err)
		}
		items["build-state"] = moves.Check{OK: true, Reason: "the request record is in the build state the replica prefix carries"}
	} else {
		items["build-state"] = moves.Check{OK: true, Reason: "this box keeps no request records; the build state travels as the replica prefix holds it"}
	}
	receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)}
	round, err := lease.MirrorPrefix(ctx, mirror, bucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts)
	if err != nil {
		return seal, err
	}
	if round.Deferred > 0 {
		return seal, fmt.Errorf("final mirror round: %d object(s) changed at the source during the copy", round.Deferred)
	}
	source, err := mirror.List(ctx, bucket, srcPrefix)
	if err != nil {
		return seal, fmt.Errorf("list the replica: %w", err)
	}
	listed, err := mirror.List(ctx, tgt.DataBucket, dstPrefix)
	if err != nil {
		return seal, fmt.Errorf("list the staged copy: %w", err)
	}
	staged := map[string]moves.StagedObject{}
	for k, o := range listed {
		staged[strings.TrimPrefix(k, dstPrefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag}
	}
	if err := sameListing(source, srcPrefix, staged); err != nil {
		return seal, fmt.Errorf("the staged copy is not the replica: %w", err)
	}
	if replicaPosition(listed, dstPrefix) != final {
		return seal, fmt.Errorf("the staged copy's newest TXID is %d, the replica's %d", replicaPosition(listed, dstPrefix), final)
	}
	items["db"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d objects staged at %s/%s, newest TXID %d", len(staged), tgt.DataBucket, dstPrefix, final)}
	if err := w.sealFault("mirror"); err != nil {
		return seal, err
	}

	// 6. The files and <host>.env, sealed, beside the staging lane.
	prefix := moves.OfflineArchivePrefix(host, m.ID)
	files, envPath, configPath, stray, err := w.offlineFilesOf(host, rt)
	if err != nil {
		return seal, fmt.Errorf("read what %s keeps: %w", host, err)
	}
	if len(stray) > 0 {
		sort.Strings(stray)
		// What the app keeps there is the app's data, in a place it reads
		// from: moving it into the files directory changes that place, and
		// removing it loses it. So the refusal offers neither (MV-O76).
		return seal, &sealRefusal{item: "files", Reason: fmt.Sprintf("%s keeps %s beside its files directory, and this move can carry only the files directory. see the offline move runbook", host, strings.Join(stray, ", "))}
	}
	a, err := w.offlineArchive(ctx, rec, mirror, tgt.DataBucket, prefix, files, envPath)
	if err != nil {
		return seal, err
	}
	items["files"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d files, %d bytes, sealed at %s%s", a.entries, a.bytes, prefix, offlineArchiveSealed)}
	if err := w.sealFault("archive"); err != nil {
		return seal, err
	}

	// 7. The timers.
	if cfg.Registry == nil {
		return seal, errors.New("no registry on this box")
	}
	rows, err := cfg.Registry.TimersFor(host)
	if err != nil {
		return seal, fmt.Errorf("read the timers: %w", err)
	}
	active := moves.ActiveTimers(rows)
	timersBody, err := moves.TimersSnapshot{V: 1, Host: host, Rows: active, TakenAt: now}.Encode()
	if err != nil {
		return seal, err
	}
	if err := mirror.Put(ctx, tgt.DataBucket, prefix+offlineTimersObject, timersBody); err != nil {
		return seal, fmt.Errorf("put the timers: %w", err)
	}
	timersDigest := moves.TimersDigest(rows)
	items["timers"] = moves.Check{OK: true, Reason: fmt.Sprintf("%d active timers at %s%s", len(active), prefix, offlineTimersObject)}
	if err := w.sealFault("timers"); err != nil {
		return seal, err
	}

	// 8. The env file, its effective instance key, and the env pin.
	envSHA, keySHA, err := offlineEnvDigests(envPath, configPath, a.envSHA)
	if err != nil {
		return seal, err
	}
	// The digest itself is the seal's EnvSHA256, not part of the reason.
	items["env"] = moves.Check{OK: true, Reason: "absent on the source, so none is carried"}
	if envSHA != "" {
		items["env"] = moves.Check{OK: true, Reason: "carried in the archive"}
	}
	if cfg.LaunchLayers == nil {
		return seal, &sealRefusal{item: "env-pin", Reason: "this box has no exec backend to compose the launch environment"}
	}
	if cfg.SecretsKey == "" {
		return seal, &sealRefusal{item: "env-pin", Reason: "this box holds no CARLOS_SECRETS_KEY to key the env report"}
	}
	env, instance, _, err := cfg.LaunchLayers(*rt)
	if err != nil {
		return seal, fmt.Errorf("compose the launch environment: %w", err)
	}
	mac, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
	if err != nil {
		return seal, &sealRefusal{item: "env-pin", Reason: err.Error()}
	}
	envReport := moves.ReportLaunch(mac, env, instance)
	envReportBody, err := json.Marshal(envReport)
	if err != nil {
		return seal, err
	}
	if err := mirror.Put(ctx, tgt.DataBucket, prefix+offlineEnvReportObject, envReportBody); err != nil {
		return seal, fmt.Errorf("put the env report: %w", err)
	}
	items["env-pin"] = moves.Check{OK: true, Reason: "launch environment digest " + envReport.Digest}
	if err := w.sealFault("env"); err != nil {
		return seal, err
	}

	// 9. The build, the replication rung and the next due time.
	if cfg.LiveVersion == nil {
		return seal, errors.New("no version store on this box")
	}
	version, err := cfg.LiveVersion(host)
	if err != nil {
		return seal, fmt.Errorf("read the live build: %w", err)
	}
	if version == "" {
		return seal, &sealRefusal{item: "release", Reason: fmt.Sprintf("no build is live for %s on this box", host)}
	}
	items["release"] = moves.Check{OK: true, Reason: "live build " + version}
	items["replication"] = moves.Check{OK: true, Reason: fmt.Sprintf("replication %q", rt.Replication)}
	items["next-due"] = moves.Check{OK: true, Reason: fmt.Sprintf("next due %d", rt.NextDue)}

	// The tombstone must still be this move's when the seal is written:
	// the position it records was read under it. One lost before the seal
	// is taken again on the next pass, which seals afresh.
	st, err := w.offlineLock(ctx, host)
	if err != nil {
		return seal, err
	}
	if !st.mine(m.ID) {
		return seal, fmt.Errorf("the lease is %s, not this move's tombstone; the next pass takes it again", st)
	}
	seal = moves.OfflineReport{
		V: moves.OfflineReportVersion, Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseSealed,
		At: now, Items: items, TombstoneOwner: owner,
		FinalTXID: final, Staged: staged,
		FilesPlain: a.plain, FilesSealed: a.sealed, FilesBytes: a.bytes, FilesEntries: a.entries,
		EnvSHA256: envSHA, KeySHA256: keySHA, EnvDigest: envReport.Digest,
		Release: version, TimersDigest: timersDigest, Replication: rt.Replication, NextDue: rt.NextDue,
	}
	if err := moves.ValidateOfflineSeal(seal, host, m.ID); err != nil {
		return seal, err
	}
	return seal, nil
}

// sameListing reports whether staged (keyed relative to the staging
// prefix) holds exactly the source's objects under srcPrefix, lock.json
// aside (the mirror never copies it), size for size. ETags may differ: a
// multipart source's copy gets a new one (lease.Receipts).
func sameListing(source map[string]lease.MirrorObject, srcPrefix string, staged map[string]moves.StagedObject) error {
	want := map[string]int64{}
	for k, o := range source {
		rel := strings.TrimPrefix(k, srcPrefix)
		if rel == "lock.json" {
			continue
		}
		want[rel] = o.Size
	}
	if len(want) != len(staged) {
		return fmt.Errorf("%d objects at the source, %d staged", len(want), len(staged))
	}
	for rel, size := range want {
		got, ok := staged[rel]
		if !ok || got.Size != size {
			return fmt.Errorf("%s: staged %v, source size %d", rel, got, size)
		}
	}
	return nil
}

// offlineFilesOf is what the archive carries for the row's layout, and
// what it cannot.
//
// The target lands the instance flat (the row is minted at flat paths), so
// the archive's files root is the app's files directory as the flat layout
// names it: <DataDir>/<host>, with <host>.env beside it. The layout pass
// maps the flat data directory onto the nested instance directory name for
// name (layout/extras.go: an app finds its files beside its database, and
// nested the database is <instance dir>/<host>.db), so a nested row's
// files directory is <instance dir>/<host>, read through the raw
// directory as every agent read of it is, and its env file is
// <instance dir>/<host>.env. Archiving the whole instance directory would
// land the app's files one level down on the target (<DataDir>/<host>/
// <host>/...), where the app would not find them.
//
// stray is every entry the app keeps beside its files directory, which
// the flat files directory cannot carry: on a nested row every entry of
// the instance directory that is neither the platform's
// (layout.IsNestedPlatformName: the seed marker and the launch descriptors
// included) nor <host>; on a flat row every entry of
// the data directory the layout's naming rules give this host
// (layout.FlatLeftovers; <host>-blobs is the live example) that is
// neither. The seal refuses on any, rather than leave it behind.
//
// A host that never kept a files directory (an app that writes nothing of
// its own) has none at the files path: the archive then carries the env
// file alone, FilesPlain is the digest of that, and the landing extracts
// an empty files directory.
//
// configPath is the rendered configuration beside the env file, which the
// effective instance key may come from.
func (w *movesWorker) offlineFilesOf(host string, rt *registry.Route) (files, envPath, configPath string, stray []string, err error) {
	dd := w.cfg.DataDir
	platform := func(name string) bool { return layout.IsNestedPlatformName(host, name) }
	if rt != nil && provision.NestedPaths(dd, *rt) {
		raw := layout.RawInstanceDir(dd, host)
		ents, err := os.ReadDir(raw)
		if err != nil && !errors.Is(err, os.ErrNotExist) {
			return "", "", "", nil, err
		}
		for _, e := range ents {
			if !platform(e.Name()) && e.Name() != host {
				stray = append(stray, filepath.Join(raw, e.Name()))
			}
		}
		return filepath.Join(raw, host), filepath.Join(raw, host+".env"), filepath.Join(raw, host+".config.env"), stray, nil
	}
	var hosts []string
	if w.cfg.Registry != nil {
		rows, err := w.cfg.Registry.List()
		if err != nil {
			return "", "", "", nil, err
		}
		for _, r := range rows {
			hosts = append(hosts, r.Host)
		}
	}
	for _, p := range layout.FlatLeftovers(dd, host, hosts) {
		if name := filepath.Base(p); !platform(name) && name != host {
			stray = append(stray, p)
		}
	}
	return filepath.Join(dd, host), filepath.Join(dd, host+".env"), filepath.Join(dd, host+".config.env"), stray, nil
}

// sealedArchive is what the seal records of the archive.
type sealedArchive struct {
	plain, sealed string
	bytes         int64
	entries       int
	// envSHA is the env entry's digest as the archive recorded it
	// ("sha256:<hex>"), "" when the archive carries no env.
	envSHA string
}

// offlineArchive builds the archive of files and envPath into a scratch
// directory on the data disk, seals it under the deployment's config key
// bound to the record's account and app, and puts it and its manifest
// under prefix in the target bucket. A file the archive refuses is a
// refusal of the files item, never a partial archive; a file that changed
// while it was read is retried. The files directory holds nothing of the
// platform's but the files marker, which every build leaves out, so no
// filter is needed (archive.Build is BuildFiltered with none).
func (w *movesWorker) offlineArchive(ctx context.Context, rec provision.Record, mirror lease.MirrorStore, bucket, prefix, files, envPath string) (sealedArchive, error) {
	cfg := w.cfg
	var out sealedArchive
	if cfg.ConfigPub == nil {
		return out, &sealRefusal{item: "files", Reason: "this box holds no CARLOS_SECRETS_KEY to seal the archive under"}
	}
	dir := moves.OfflineParkDir(cfg.DataDir, rec.Host, rec.Move.ID, moves.OfflineRoleSealer)
	if dir == "" {
		return out, fmt.Errorf("no scratch directory for %s under %q", rec.Host, cfg.DataDir)
	}
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return out, err
	}
	tmp, err := os.MkdirTemp(dir, "archive-")
	if err != nil {
		return out, err
	}
	defer os.RemoveAll(tmp)
	built, err := os.Create(filepath.Join(tmp, "archive.tar.zst"))
	if err != nil {
		return out, err
	}
	defer built.Close()
	plain, entries, size, err := archive.Build(files, envPath, built)
	switch {
	case errors.Is(err, archive.ErrChanged):
		return out, fmt.Errorf("archive the files: %w; built again next pass", err)
	case err != nil:
		return out, &sealRefusal{item: "files", Reason: err.Error()}
	}
	if _, err := built.Seek(0, io.SeekStart); err != nil {
		return out, err
	}
	sealedFile, err := os.Create(filepath.Join(tmp, "archive.sealed"))
	if err != nil {
		return out, err
	}
	defer sealedFile.Close()
	dataKey, sealedSHA, err := archive.Seal(built, cfg.ConfigPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealedFile)
	if err != nil {
		return out, fmt.Errorf("seal the archive: %w", err)
	}
	manifest, err := json.Marshal(archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Plain: plain, Sealed: sealedSHA, Bytes: size,
		Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize})
	if err != nil {
		return out, err
	}
	body, err := os.ReadFile(sealedFile.Name())
	if err != nil {
		return out, err
	}
	if err := mirror.Put(ctx, bucket, prefix+offlineArchiveSealed, body); err != nil {
		return out, fmt.Errorf("put the sealed archive: %w", err)
	}
	if err := mirror.Put(ctx, bucket, prefix+offlineArchiveManifest, manifest); err != nil {
		return out, fmt.Errorf("put the archive manifest: %w", err)
	}
	out = sealedArchive{plain: plain, sealed: sealedSHA, bytes: size, entries: len(entries)}
	for _, e := range entries {
		if e.Path == archive.EnvEntry {
			out.envSHA = e.SHA256
		}
	}
	return out, nil
}

// readOwnFile reads a regular file without following a link, as the agent
// reads every file a nested tenant can write: exists false when there is
// none.
func readOwnFile(path string) (body []byte, exists bool, err error) {
	info, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, err
	}
	if !info.Mode().IsRegular() {
		return nil, true, fmt.Errorf("%s is not a regular file", path)
	}
	f, err := os.Open(path)
	if err != nil {
		return nil, true, err
	}
	defer f.Close()
	got, err := f.Stat()
	if err != nil {
		return nil, true, err
	}
	if !os.SameFile(info, got) {
		return nil, true, fmt.Errorf("%s changed while it was read", path)
	}
	body, err = io.ReadAll(io.LimitReader(f, instancekey.EnvReadLimit+1))
	if err != nil {
		return nil, true, err
	}
	if len(body) > instancekey.EnvReadLimit {
		return nil, true, fmt.Errorf("%s is larger than an env file", path)
	}
	return body, true, nil
}

// offlineEnvDigests is the env item's two values: the lowercase hex sha256
// of <host>.env's bytes ("" when the source has none), and the lowercase
// hex sha256 of the instance key a launch hands the app
// (instancekey.Effective over the rendered configuration and the own file,
// instancekey.Digest; "" when neither supplies one). archived is the env
// entry's digest as the archive recorded it ("sha256:<hex>", or "" for no
// env entry); the file read here must be the one archived.
func offlineEnvDigests(envPath, configPath, archived string) (envSHA, keySHA string, err error) {
	own, exists, err := readOwnFile(envPath)
	if err != nil {
		return "", "", fmt.Errorf("read the env file: %w", err)
	}
	if exists {
		sum := sha256.Sum256(own)
		envSHA = hex.EncodeToString(sum[:])
	}
	if want := strings.TrimPrefix(archived, "sha256:"); want != envSHA {
		return "", "", fmt.Errorf("the env file changed while it was archived (archived %q, now %q); sealed again next pass", want, envSHA)
	}
	config, _, err := readOwnFile(configPath)
	if err != nil {
		return "", "", fmt.Errorf("read the rendered configuration: %w", err)
	}
	if key, _ := instancekey.Effective(instancekey.ParseEnv(config), instancekey.ParseEnv(own)); key != "" {
		keySHA = instancekey.Digest(key)
	}
	return envSHA, keySHA, nil
}

// offlineSealRecord reads seal.json, nil when it is not there.
func (w *movesWorker) offlineSealRecord(ctx context.Context, s release.Store, rec provision.Record) ([]byte, error) {
	key := moves.OfflineSealKey(rec.Host, rec.Move.ID)
	raw, _, err := s.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("read %s: %w", key, err)
	}
	return raw, nil
}

// decodeSeal parses and checks the seal record; nil raw is no seal.
func decodeSeal(raw []byte, host, id string) (*moves.OfflineReport, error) {
	if raw == nil {
		return nil, nil
	}
	var seal moves.OfflineReport
	if err := json.Unmarshal(raw, &seal); err != nil {
		return nil, fmt.Errorf("the seal record does not parse: %w", err)
	}
	if err := moves.ValidateOfflineSeal(seal, host, id); err != nil {
		return nil, err
	}
	return &seal, nil
}

// offlineKeepStop is every pass after the seal, until the tombstone is
// lifted: read the lease first; a tombstone that is not this move's any
// more is a lost stop, never taken again. Then the source reports: one
// written since the seal at a position past the sealed one is a lost stop
// too (a write the stop did not hold off, which the staged copy lacks).
// Only then is this move's tombstone re-asserted. In
// offline-sealing (a seal written by a pass that crashed before the CAS)
// it also makes the CAS. Then it rewrites the living sealer report: the
// seal's values and items, the tombstone's owner and expiry, and At.
func (w *movesWorker) offlineKeepStop(ctx context.Context, s release.Store, rec provision.Record, raw []byte) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	seal, err := decodeSeal(raw, host, m.ID)
	if err != nil {
		return fmt.Errorf("offline move %s: %s: %w; it is never rewritten, so abort the move", m.ID, host, err)
	}
	if seal == nil && m.State == provision.OfflineSealing {
		return fmt.Errorf("offline move %s: %s has no seal record", m.ID, host)
	}
	st, err := w.offlineLock(ctx, host)
	if err != nil {
		return err
	}
	if !st.mine(m.ID) {
		return w.offlineLeaseLost(ctx, s, rec, seal, st)
	}
	if seal != nil {
		if err := w.offlineCheckPositions(ctx, s, rec, seal); err != nil {
			return err
		}
	}
	bucket, loc, region, err := cfg.tombstoneAt(host)
	if err != nil {
		return err
	}
	if err := w.sealFault("reassert"); err != nil {
		return err
	}
	if err := lease.Reassert(ctx, bucket, loc, region, moves.TombstoneOwner(m.ID), moves.TombstoneTTL); err != nil {
		var held *lease.HeldError
		if errors.As(err, &held) {
			return w.offlineLeaseLost(ctx, s, rec, seal, tombState{owner: held.Owner, expires: held.ExpiresAt})
		}
		if errors.Is(err, lease.ErrTombstoneLost) {
			// Gone between the read above and Reassert's own: the same
			// lost stop, and Reassert wrote nothing. Read it again for
			// what it is now (absent, when even that read fails).
			now, rerr := w.offlineLock(ctx, host)
			if rerr != nil {
				now = tombState{}
			}
			return w.offlineLeaseLost(ctx, s, rec, seal, now)
		}
		return fmt.Errorf("offline move %s: re-assert the tombstone of %s: %w", m.ID, host, err)
	}
	if st, err = w.offlineLock(ctx, host); err != nil {
		return err
	}
	if !st.mine(m.ID) {
		return w.offlineLeaseLost(ctx, s, rec, seal, st)
	}
	if m.State == provision.OfflineSealing {
		src := provision.MoveSource{
			FinalTXID: seal.FinalTXID, Replication: seal.Replication, NextDue: seal.NextDue,
			Files:        &provision.MoveFilesDigest{Plain: seal.FilesPlain, Sealed: seal.FilesSealed, Bytes: seal.FilesBytes, Entries: seal.FilesEntries},
			TimersSHA256: seal.TimersDigest, Release: seal.Release, EnvDigest: seal.EnvDigest,
			Evidence: moves.OfflineSealEvidence(host, m.ID, raw),
		}
		_, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, provision.OfflineSealedBy(m.ID, cfg.Box, src))
		switch {
		case errors.Is(err, provision.ErrMoveConflict):
			log.Printf("moves: %s: offline move %s: the record moved on before the seal's CAS: %v", host, m.ID, err)
		case err != nil:
			return fmt.Errorf("offline move %s: record the seal of %s: %w", m.ID, host, err)
		default:
			nudgeSerial(ctx, s)
			log.Printf("moves: %s: offline move %s: sealed", host, m.ID)
		}
		if err := w.sealFault("cas"); err != nil {
			return err
		}
	}
	living := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer}
	if seal != nil {
		living = *seal
	}
	living.Phase = moves.OfflinePhaseSealed
	living.At = cfg.now().UTC()
	living.TombstoneOwner = st.owner
	living.TombstoneUntil = st.expires.UTC()
	return writeOfflineReport(ctx, s, living)
}

// offlineSealerRefresh is the sealer's half of the report refresh
// (offlineReassert), from offline-sealed to offline-landed: offlineKeepStop's
// checks without its write to the lease. It reads the seal and the lease, a
// tombstone that is not this move's live one is the lost stop it always is,
// it checks the source positions against the seal, and it rewrites the
// living report with the tombstone's owner and expiry exactly as read. It
// never re-asserts the tombstone (fix round 1, Astra's finding 2: a refresh
// every 15 s through offlineKeepStop pushed the expiry out seven days each
// time); the sealer's ordinary pass keeps doing that, once a pass.
func (w *movesWorker) offlineSealerRefresh(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	if !m.Offline() || cfg.Box == "" || m.Sealer != cfg.Box {
		return nil
	}
	switch m.State {
	case provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded:
	default:
		return nil
	}
	raw, err := w.offlineSealRecord(ctx, s, rec)
	if err != nil {
		return err
	}
	seal, err := decodeSeal(raw, host, m.ID)
	if err != nil {
		return fmt.Errorf("offline move %s: %s: %w; it is never rewritten, so abort the move", m.ID, host, err)
	}
	st, err := w.offlineLock(ctx, host)
	if err != nil {
		return err
	}
	if !st.mine(m.ID) {
		return w.offlineLeaseLost(ctx, s, rec, seal, st)
	}
	if seal != nil {
		if err := w.offlineCheckPositions(ctx, s, rec, seal); err != nil {
			return err
		}
	}
	living := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer}
	if seal != nil {
		living = *seal
	}
	living.Phase = moves.OfflinePhaseSealed
	living.At = cfg.now().UTC()
	living.TombstoneOwner = st.owner
	living.TombstoneUntil = st.expires.UTC()
	return writeOfflineReport(ctx, s, living)
}

// offlineCheckPositions is the backstop behind the holds and the fence:
// every source report written since the seal (at or after its At) must be
// at or below the sealed final position. One past it is a box that wrote
// what the staged copy does not hold, so the sealed position is no longer
// the instance's: a lost stop, in the db-tail item.
func (w *movesWorker) offlineCheckPositions(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport) error {
	reports, err := readOfflineReports(ctx, s, rec.Host, rec.Move.ID, moves.OfflineRoleSource)
	if err != nil {
		return fmt.Errorf("read the source reports: %w", err)
	}
	boxes := make([]string, 0, len(reports))
	for b := range reports {
		boxes = append(boxes, b)
	}
	sort.Strings(boxes)
	for _, b := range boxes {
		r := reports[b]
		if r.At.Before(seal.At) || r.LocalTXID <= seal.FinalTXID {
			continue
		}
		return w.offlineLostStop(ctx, s, rec, seal, "db-tail", offlinePastSealCause(b, r.LocalTXID, seal.FinalTXID))
	}
	return nil
}

// offlineLeaseLost is the lost stop of a tombstone that is absent,
// expired, lifted or someone else's, in the lease item.
func (w *movesWorker) offlineLeaseLost(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport, st tombState) error {
	return w.offlineLostStop(ctx, s, rec, seal, "lease", offlineLeaseCause(st))
}

// offlineLeaseCause and offlinePastSealCause are the two causes of a lost
// stop as its page and the failed item say them (MV-O69, MV-O70).
func offlineLeaseCause(st tombState) string {
	return fmt.Sprintf("the source's lease is %s, not this move's tombstone", offlineLeaseText(st))
}

func offlinePastSealCause(box string, pos, final uint64) string {
	return fmt.Sprintf("%s reports TXID %d, past the sealed position %d", box, pos, final)
}

// offlineLostStop reports a stop that no longer holds after the seal:
// cause says why, and item is the seal item it fails. The tombstone is
// never taken again (a wake may have written the replica in between, so
// the sealed position is no longer the source's), the living report says
// failed with that item not OK and no tombstone owner (so open refuses),
// and an operator is paged once per move.
func (w *movesWorker) offlineLostStop(ctx context.Context, s release.Store, rec provision.Record, seal *moves.OfflineReport, item, cause string) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer}
	if seal != nil {
		r = *seal
	}
	r.Phase = moves.OfflinePhaseFailed
	r.At = cfg.now().UTC()
	r.TombstoneOwner, r.TombstoneUntil = "", time.Time{}
	items := map[string]moves.Check{}
	for k, v := range r.Items {
		items[k] = v
	}
	items[item] = moves.Check{OK: false, Reason: "lost stop: " + cause}
	r.Items = items
	werr := writeOfflineReport(ctx, s, r)
	subject, body := offlineLostStopPage(host, m.ID, m.State, cause)
	w.firePage(ctx, host, m.ID, offlinePageLostStop, alerting.LevelIncident, subject, body)
	log.Printf("moves: %s: offline move %s: LOST STOP: %s", host, m.ID, cause)
	return errors.Join(fmt.Errorf("offline move %s: %s: lost stop: %s", m.ID, host, cause), werr)
}

// offlineLostStopPage is the lost stop's page (copy rows MV-O66 to
// MV-O68; the item's reason is MV-O71): its subject, and a body for the move's state. Before open,
// abort puts the instance back with nothing lost, so the body says to
// abort. From open on, abort is refused (MV-O40) and the move is fixed
// forward by hand, so the body points to the status and the runbook.
func offlineLostStopPage(host, id, state, cause string) (subject, body string) {
	subject = fmt.Sprintf("offline move of %s lost its stop", host)
	if state == provision.OfflineOpen || state == provision.OfflineFinishing {
		body = fmt.Sprintf("The offline move %s of %s is %s, and its stop no longer holds: %s. It cannot be aborted once it is open. Read carlos move offline status --host %s and follow the offline move runbook.", id, host, state, cause, host)
		return subject, body
	}
	body = fmt.Sprintf("The offline move %s of %s is %s, and its stop no longer holds: %s. The move cannot go on. Read carlos move offline status --host %s and abort the move.", id, host, state, cause, host)
	return subject, body
}

// offlineFinishLift is the sealer's step in offline-finishing. Until every
// box the seal named reports gone (it parked at open and reconcile has
// removed its row since), the stop is kept as before. Then the tombstone
// is lifted (only this move's, with the ETag read), the copy receipts are
// cleared and the report says lifted. A lease that is already free (lifted
// by an earlier pass) is left as it is; one someone else holds live is a
// lost stop.
func (w *movesWorker) offlineFinishLift(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	reports, err := readOfflineReports(ctx, s, host, m.ID, moves.OfflineRoleSource)
	if err != nil {
		return fmt.Errorf("read the source reports: %w", err)
	}
	gone := len(m.Boxes) > 0
	for _, b := range m.Boxes {
		if r, ok := reports[b]; !ok || r.Phase != moves.OfflinePhaseGone {
			gone = false
		}
	}
	raw, err := w.offlineSealRecord(ctx, s, rec)
	if err != nil {
		return err
	}
	if !gone {
		return w.offlineKeepStop(ctx, s, rec, raw)
	}
	st, err := w.offlineLock(ctx, host)
	if err != nil {
		return err
	}
	switch {
	case st.mine(m.ID):
		bucket, loc, region, err := cfg.tombstoneAt(host)
		if err != nil {
			return err
		}
		if _, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) }); err != nil {
			return fmt.Errorf("offline move %s: lift the tombstone of %s: %w", m.ID, host, err)
		}
		log.Printf("moves: %s: offline move %s: tombstone lifted", host, m.ID)
	case st.free():
	default:
		seal, _ := decodeSeal(raw, host, m.ID)
		return w.offlineLeaseLost(ctx, s, rec, seal, st)
	}
	if err := w.sealFault("lift"); err != nil {
		return err
	}
	own, err := w.offlineMirror(ctx, cfg.OwnBucketRegion)
	if err != nil {
		return err
	}
	clearOfflineReceipts(ctx, own, cfg.OwnBucket, host)
	r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseLifted, At: cfg.now().UTC()}
	return writeOfflineReport(ctx, s, r)
}

// clearOfflineReceipts is clearMoveReceipts as an ensure: receipts that are
// already gone are not deleted again, so a pass that repeats finish or
// revert writes nothing (on a versioned bucket every DELETE of an absent
// key leaves one more delete marker).
func clearOfflineReceipts(ctx context.Context, m lease.MirrorStore, bucket, host string) {
	rc := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)}
	if _, err := m.Get(ctx, rc.Bucket, rc.Key); errors.Is(err, lease.ErrMirrorNotFound) {
		return
	}
	clearMoveReceipts(ctx, m, rc, host)
}

// offlineSealRevert is the sealer's step in offline-reverting: lift the
// tombstone if it is this move's (a lease anyone else holds, or none, is
// left alone), delete the two prefixes this move wrote in the target's
// bucket (the staged replica and the archive), clear the copy receipts,
// lift this move's fence (the source step lifts it too) and report
// reverted. Every part is an ensure.
func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	bucket, loc, region, err := cfg.tombstoneAt(host)
	if err != nil {
		return err
	}
	lifted, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) })
	var held *lease.HeldError
	switch {
	case err == nil:
		if lifted {
			log.Printf("moves: %s: offline move %s: tombstone lifted on abort", host, m.ID)
		}
	case errors.As(err, &held):
		w.notice(host+"#offline-revert-lease", "moves: %s: offline move %s: the lease is %s's, not this move's; left as it is", host, m.ID, held.Owner)
	default:
		return fmt.Errorf("offline move %s: lift the tombstone of %s: %w", m.ID, host, err)
	}
	if err := w.sealFault("revert-lift"); err != nil {
		return err
	}
	tgt, err := w.offlineTarget(ctx, s, m)
	if err != nil {
		return err
	}
	mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)
	if err != nil {
		return err
	}
	for _, prefix := range []string{activator.MovesPrefix(host), moves.OfflineArchivePrefix(host, m.ID)} {
		if prefix == "" || prefix == "moves/" {
			return fmt.Errorf("offline move %s: no staging prefix for %s", m.ID, host)
		}
		if err := lease.DeletePrefix(ctx, mirror, tgt.DataBucket, prefix); err != nil {
			return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, tgt.DataBucket, prefix, err)
		}
	}
	clearOfflineReceipts(ctx, mirror, bucket, host)
	w.offlineLiftFence(host, m)
	r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()}
	return writeOfflineReport(ctx, s, r)
}