package host

import (
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/carlosframework/platform/internal/park"
	"io"
	"log"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The Phase 1 source round (moves spec §7.3, §7.11, K5, K12), behind
// CARLOS_MOVES_PHASE1. It drives a move whose record carries an id: the
// staging lane under moves/<host>/<id>/ in the target bucket, written only
// through the guarded staging mirror while this box holds the staging-work
// lease; the per-operation receipts and ledger in this box's own bucket; the
// source's readiness report in the deployment bucket; and the promotion to
// ready (and, for an automatic cut, to cutting) by guarded CAS. The legacy
// worker (legacyMoveRound) is untouched and runs only with the switch off.
//
// The operator-visible sentences below are rows of the copy record
// (docs/superpowers/plans/2026-10-06-moves-phase1-copy.md), pending review;
// they reach `carlos move status` through the source report's S1.
const (
	s1NoDeployBucket = "this deployment has no deployment bucket for the staging lease"
	s1Nested         = "nested instances cannot move yet"
	s1NoSecretsKey   = "this box has no secrets key, so it cannot report the launch environment or seal the archive"
	s1CannotProbe    = "this box cannot probe the instance yet"
	s1NoReadyPath    = "the app's readiness path could not be read"
)

// archivePlainDigest and archiveBuild are the archive's digest and build,
// as variables so a test can make a file change under the walk.
var (
	archivePlainDigest = archive.PlainDigest
	archiveBuild       = archive.Build
)

func s1NotExec(host string) string {
	return host + " is not an exec-backed instance, so it cannot move"
}
func s1NoHibernate(host string) string { return host + " does not hibernate, so it cannot move" }
func s1Sidecar(host string) string {
	return host + " has a sidecar environment file, so it cannot move"
}
func s1Column(host, column string) string {
	return host + " has " + column + " set in the registry, so it cannot move"
}

// s1PrivateStorage is M-C17, the door's sentence repeated every round.
func s1PrivateStorage(host string) string {
	return host + " uses private storage. Instances with private storage cannot move yet."
}

// s1SpecialFile and s1OverCap are M-C14 and M-C15.
func s1SpecialFile(host, path string) string {
	return host + " keeps a file this move cannot copy: " + path + ". Only regular files and folders move."
}
func s1OverCap(host string) string {
	return host + " keeps more than 256 MiB of files beside its database. Keep large files in object storage."
}

// s1FilesPastCap is MV-N501 (#236): the files manifest a recovery of the
// host would restore holds an entry path past the readers' cap, so no
// recovery after the cut could restore it (sourceFilesRefusal).
func s1FilesPastCap(host string) string {
	return host + " has a files manifest with a path longer than 4095 bytes, so it cannot move. an operator must replace the manifest"
}

// s1FilePastCap is MV-N502 (#236 fix round 1): the files directory itself
// holds a path longer than the files tree's cap (archive.ErrPastTreeCap,
// one byte under the archive format's MV-N81 bound). The archive could
// carry it, but the target's prepared files manifest could not, so the
// landing would fail after the cut; the source refuses here instead.
func s1FilePastCap(host string) string {
	return host + " has a file whose path is longer than 4095 bytes, so it cannot move. shorten the path or remove the file"
}

// The rows say 4095, so the sentences do; this fails to compile if
// hostfiles.MaxPathBytes is ever changed without the rows.
const _ = uint(hostfiles.MaxPathBytes-4095) + uint(4095-hostfiles.MaxPathBytes)

// s1TooManyEntries and s1LongPath are MV-N79 and MV-N81, the archive's own
// rows, written here from data (the sentinel, the path the error carries)
// rather than taken from an error's text. s1ArchiveFailed is every archive
// error this round does not recognise; its text goes to the log only.
const (
	s1TooManyEntries = "archive: the files directory holds more than 131072 entries"
	s1ArchiveFailed  = "the files beside the database could not be archived"
)

func s1LongPath(path string) string {
	return "archive: refused path: " + path + " is longer than 4096 bytes"
}

// s1NoAnswer is M-C16.
func s1NoAnswer(host string) string {
	return host + " does not answer its readiness path. Declare moves.ready_path for this app."
}

// crossRefreshAfter is how old the source's cross-region rate may get before
// a round that copied nothing measures it again by recopying one staged
// object. Half of moves.MeasurementFreshness, so a quiet (hibernating)
// source keeps a fresh rate in its report while it waits for ready.
const crossRefreshAfter = moves.MeasurementFreshness / 2

// crossMinBytes is the least a round must copy for its rate to be recorded:
// a round of a few small files measures latency, not throughput.
const crossMinBytes = 1 << 20

// moveLedger is the per-operation ledger beside the copy receipts, at
// moves.LedgerKey in this box's own bucket (§6.3). It is what the rounds
// record about the lane: move.json's version (or "present" when a replay
// found it already there), the published archive by digest and by the
// version ids its PUTs returned, and the most recent copying round's
// cross-region rate.
type moveLedger struct {
	V               int               `json:"v"`
	MoveJSONVersion string            `json:"move_json_version,omitempty"`
	ArchivePlain    string            `json:"archive_plain,omitempty"`
	ArchiveSealed   string            `json:"archive_sealed,omitempty"`
	ArchiveVersions *moves.ArchiveRef `json:"archive_versions,omitempty"`
	LastRoundBytes  int64             `json:"last_round_bytes,omitempty"`
	// LastRoundSeconds and LastRoundAt time the most recent round that
	// copied at least crossMinBytes, or the recopy that refreshed it.
	LastRoundSeconds float64   `json:"last_round_seconds,omitempty"`
	LastRoundAt      time.Time `json:"last_round_at,omitempty"`
	// DeferredRounds counts the cut's final rounds that deferred an object
	// or moved more than the bar (§7.4 staged; three revert). LeaseUntil is
	// the seal's writers' bound (ruling G7), recorded beside it.
	DeferredRounds int       `json:"deferred_rounds,omitempty"`
	LeaseUntil     time.Time `json:"lease_until,omitempty"`
}

// moveJSONPresent is the ledger's note that move.json already existed when
// this source wrote it: its own lost-response PUT, or a foreign object. The
// source cannot tell which (M17 forbids the read), so the target's T9
// validates the object's fields.
const moveJSONPresent = "present"

func (l moveLedger) crossRate() (float64, time.Time) {
	if l.LastRoundBytes <= 0 || l.LastRoundSeconds <= 0 {
		return 0, time.Time{}
	}
	return float64(l.LastRoundBytes) / (1 << 20) / l.LastRoundSeconds, l.LastRoundAt
}

func readMoveLedger(ctx context.Context, m lease.MirrorStore, bucket, key string) (moveLedger, error) {
	body, err := m.Get(ctx, bucket, key)
	if errors.Is(err, lease.ErrMirrorNotFound) {
		return moveLedger{V: 1}, nil
	}
	if err != nil {
		return moveLedger{}, fmt.Errorf("read the move ledger %s/%s: %w", bucket, key, err)
	}
	var l moveLedger
	if err := json.Unmarshal(body, &l); err != nil || l.V != 1 {
		// An unreadable ledger is no evidence: start over. Every fact in it
		// is recomputed (move.json answers present, the archive is rebuilt).
		return moveLedger{V: 1}, nil
	}
	// Nor is a version id that pins nothing (final review finding 4): a
	// "null" recorded before the staging mirror refused it. The archive is
	// restaged and move.json asked for again (it answers present).
	if l.ArchiveVersions != nil && !l.ArchiveVersions.Pinned() {
		l.ArchivePlain, l.ArchiveSealed, l.ArchiveVersions = "", "", nil
	}
	if l.MoveJSONVersion != moveJSONPresent && !lease.UsableVersionID(l.MoveJSONVersion) {
		l.MoveJSONVersion = ""
	}
	return l, nil
}

func writeMoveLedger(ctx context.Context, m lease.MirrorStore, bucket, key string, l moveLedger) error {
	body, err := json.Marshal(l)
	if err != nil {
		return err
	}
	if err := m.Put(ctx, bucket, key, body); err != nil {
		return fmt.Errorf("write the move ledger %s/%s: %w", bucket, key, err)
	}
	return nil
}

// phase1MoveRound is the Phase 1 worker's step for one id'd record this box
// is the source of.
func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, fleet moveFleet) error {
	// The transitions another writer made (the console's doors, a declared
	// loss) are logged by the box that sees them (movesevents.go).
	w.observeMove(ctx, s, rec)
	switch rec.Move.State {
	case provision.MovePreparing, provision.MoveReady:
		return w.preparingRound(ctx, s, rec, fleet)
	case provision.MoveAborting:
		// The abort's source half (movesabort.go, Plan 5 Task 1): the lane
		// is the target region's to park.
		return w.abortSource(ctx, s, rec)
	case provision.MoveCutting:
		return w.runCut(ctx, s, rec, fleet)
	case provision.MoveReverting:
		// The record decided the revert; the claim follows it (§7.5), and
		// the source recovers (movesrevert.go, Plan 4 Task 6, ruling J7).
		// movesPass routes a reverting record by the participant rule
		// before it gets here; this arm is the same step.
		return w.revertSource(ctx, s, rec, &fleet)
	case provision.MoveLanding:
		// The target's (ruling G34): after the flip the record's region is
		// this box's and reconcile has minted the held row, so OwnsRoute
		// admits it here. checking to restored is Task 10's (movesland.go);
		// the probe and the commit are Plan 3's.
		return w.landingPass(ctx, s, rec)
	case provision.MoveCommitted:
		// The target's committed arm (Plan 3 Task 7, H9): the claim half
		// and the conditional open, every pass while committed.
		if movesParticipantRecord(w.cfg, rec) == roleTarget {
			return w.committedTarget(ctx, s, rec)
		}
	}
	return nil // later: Plans 4 and 5
}

// srcRound is what one preparing round learned for the source report.
type srcRound struct {
	s1         *moves.Check // a refusal; nil means probe the instance
	ledger     moveLedger
	deltaBytes int64
	// replicaBytes is the replica's logical size this round
	// (replicaLogicalSize), carried as SourceReport.ReplicaBytes.
	replicaBytes int64
	// holder is the claim holder this round read under the lease, for
	// decideReady's layout rule (ruling T5-lease).
	holder hostclaim.Holder
}

// preparingRound is §7.3's source round (sourceRound), then Ready, the
// promotion (or the repin of a ready move whose pin a cut's admission
// cleared) and the automatic cut.
func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error {
	sr, published, err := w.sourceRound(ctx, s, rec, fleet)
	if err != nil || published == nil {
		return err
	}
	// The invariant: a round promotes or requests a cut only when its own
	// source report passed every check and was published. Any failing
	// check ends the round here; a retry or an error ended it above.
	if !sourcePassed(published) {
		w.notice(rec.Host+"#ready", "moves: %s: this round's source report does not pass (S1 %v, S2 %v); no promotion and no cut", rec.Host, published.S1.OK, published.S2.OK)
		return nil
	}
	return w.decideReady(ctx, s, rec, provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}, sr, published)
}

// sourceRound is §7.3's source round, in order: no deployment bucket
// refuses; the staging lease is taken and the claim revalidated under it;
// the box-step refusals and the one-box rule; the store binding; Touch;
// move.json; the replica lane; the archive; the source report. It returns
// the round and the report it published. A round that ended before any
// report (another worker holds the lease, a file changed under the
// archive) returns a nil report and no error: the next pass runs it again.
// The cut's `requested` runs it too (G13), and decides admission instead
// of readiness.
func (w *movesWorker) sourceRound(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) (*srcRound, *moves.SourceReport, error) {
	cfg := w.cfg
	id := rec.Move.ID
	report := func(sr *srcRound) (*srcRound, *moves.SourceReport, error) {
		published, err := w.writeSourceReport(ctx, s, rec, sr)
		return sr, published, err
	}
	// Round-5 finding 11: a lease in a per-participant bucket, or a
	// 15-minute one, cannot fence staging writers. Without the deployment
	// bucket there is no shared place for it, so the round refuses.
	if cfg.StagingLease == nil {
		return report(&srcRound{s1: &moves.Check{Reason: s1NoDeployBucket}})
	}
	tgt, ok := fleet.targets[rec.Move.To]
	if !ok {
		return nil, nil, fmt.Errorf("target region %s publishes no data bucket yet. waiting", rec.Move.To)
	}
	held, releaseLease, err := cfg.StagingLease(ctx, rec.Move.To, rec.Host)
	if err != nil {
		var he *lease.HeldError
		if errors.As(err, &he) {
			log.Printf("moves: %s: another worker holds the staging lease; this round waits", rec.Host)
			return nil, nil, nil
		}
		return nil, nil, fmt.Errorf("staging lease: %w", err)
	}
	defer func() {
		rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if err := releaseLease(rctx); err != nil {
			log.Printf("moves: %s: release the staging lease: %v", rec.Host, err)
		}
	}()
	// K5: revalidate the claim under the lease.
	obs, err := hostclaim.Read(ctx, s, rec.Host)
	if err != nil {
		return nil, nil, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent {
		return nil, nil, fmt.Errorf("move %s has no host claim", id)
	}
	if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil {
		return nil, nil, fmt.Errorf("move %s: %w", id, err)
	}
	if h := obs.Claim.Holder; h == nil || h.Incarnation != rec.Move.Incarnation {
		return nil, nil, fmt.Errorf("move %s: the host claim names another incarnation", id)
	}
	if cfg.Box != "" && obs.Claim.Op.Exec != cfg.Box {
		return nil, nil, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, obs.Claim.Op.Exec)
	}
	// The box-step refusals (§7.1: every box step re-checks), and the
	// one-box rule the door applied (D15, ruling G13): a region that has
	// gained a second box since is refused here, every round.
	if why, err := w.sourceRefusal(ctx, s, rec, obs.Claim.Holder); err != nil {
		return nil, nil, err
	} else if why != "" {
		return report(&srcRound{s1: &moves.Check{Reason: why}})
	}
	if why := oneBoxRefusal(fleet, rec); why != "" {
		return report(&srcRound{s1: &moves.Check{Reason: why}})
	}
	// K12 and T9: the target region and box are in touched before the
	// first staging write.
	targetBox := obs.Claim.Op.Intent["target"]
	if !slices.ContainsFunc(obs.Claim.Touched.Regions, func(region string) bool {
		return park.PlacementMatches(obs.Claim.Op.Intent["home_region"], rec.Move.To, region)
	}) || (targetBox != "" && !containsString(obs.Claim.Touched.Boxes, targetBox)) {
		var boxes []string
		if targetBox != "" {
			boxes = []string{targetBox}
		}
		if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Touch(id, []string{park.PlacementRegion(rec.Move.To)}, boxes)); err != nil {
			return nil, nil, fmt.Errorf("touch %s in the host claim: %w", rec.Move.To, err)
		}
	}
	stg, err := w.staging(ctx, tgt.DataBucketRegion)
	if err != nil {
		return nil, nil, err
	}
	own, err := w.mirror(ctx, tgt.DataBucketRegion)
	if err != nil {
		return nil, nil, err
	}
	gm := &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: cfg.Now}, dst: tgt.DataBucket, own: own}
	ledgerKey := moves.LedgerKey(rec.Host, id)
	ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey)
	if err != nil {
		return nil, nil, err
	}
	before := ledger
	saveLedger := func() error {
		if ledger == before {
			return nil
		}
		if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
			return err
		}
		before = ledger
		return nil
	}

	// move.json, create-only, before the first round. Once the ledger
	// knows it, it is not written again.
	if ledger.MoveJSONVersion == "" {
		body, err := json.Marshal(moves.MoveJSON{MoveID: id, Incarnation: rec.Move.Incarnation, Host: rec.Host, From: rec.Move.From, To: rec.Move.To})
		if err != nil {
			return nil, nil, err
		}
		ver, err := gm.PutIfAbsentVersioned(ctx, tgt.DataBucket, moves.MoveJSONKey(rec.Host, id), body)
		switch {
		case errors.Is(err, lease.ErrExists):
			ledger.MoveJSONVersion = moveJSONPresent
		case err != nil:
			return nil, nil, fmt.Errorf("write move.json: %w", err)
		default:
			ledger.MoveJSONVersion = ver
		}
		if err := saveLedger(); err != nil {
			return nil, nil, err
		}
	}

	// The replica lane: the claim's generation, flat under replica/.
	srcPrefix, keep := moves.ReplicaSource(*obs.Claim.Holder)
	dstPrefix := moves.ReplicaPrefix(rec.Host, id)
	receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: moves.ReceiptsKey(rec.Host, id)}
	var round lease.MirrorRound
	start := cfg.Now()
	for attempt := 0; ; attempt++ {
		round, err = lease.MirrorPrefixKeep(ctx, gm, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts, keep)
		// One retry, and only by calling again: a new check, a new
		// signature. A guard refusal is never retried.
		if err == nil || attempt == 1 || errors.Is(err, moves.ErrLeaseUnsafe) || errors.Is(err, moves.ErrInvalidSkew) {
			break
		}
		log.Printf("moves: %s: replica round failed, retrying once: %v", rec.Host, err)
	}
	if err != nil {
		return nil, nil, fmt.Errorf("replica lane: %w", err)
	}
	// The build state travels with the replica, in every round.
	if _, _, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, obs.Claim.Holder.Location, tgt.DataBucket, rec.Host, id); err != nil {
		return nil, nil, fmt.Errorf("build state lane: %w", err)
	}
	if round.BytesCopied >= crossMinBytes {
		ledger.LastRoundBytes, ledger.LastRoundSeconds, ledger.LastRoundAt = round.BytesCopied, cfg.Now().Sub(start).Seconds(), cfg.Now().UTC()
	} else if _, at := ledger.crossRate(); at.IsZero() || cfg.Now().Sub(at) > crossRefreshAfter {
		if err := w.refreshCrossRate(ctx, gm, srcPrefix, keep, tgt.DataBucket, dstPrefix, &ledger); err != nil {
			return nil, nil, err
		}
	}

	// The database's size as this round's replica states it, for Ready to
	// hold against the size the target's report priced (final review
	// finding 2). An unreadable header ends the round: no report, no
	// promotion.
	replicaBytes, err := w.replicaLogicalSize(ctx, gm, srcPrefix, keep)
	if err != nil {
		return nil, nil, err
	}
	sr := &srcRound{deltaBytes: round.DeltaBytes(), replicaBytes: replicaBytes, holder: *obs.Claim.Holder}
	// The files archive, rebuilt only when its plaintext digest changes. A
	// file that changed under the walk ends the round here, before any
	// report or readiness: the archive the target checked is no longer the
	// files, and the next round builds the one that is (review finding 2).
	why, retry, err := w.publishArchive(ctx, gm, rec, tgt.DataBucket, &ledger)
	if err != nil {
		return nil, nil, err
	}
	if err := saveLedger(); err != nil {
		return nil, nil, err
	}
	if retry {
		return nil, nil, nil
	}
	if why != "" {
		sr.s1 = &moves.Check{Reason: why}
	}
	sr.ledger = ledger
	return report(sr)
}

// buildStateName is the build-state object's name beside a lease
// (release.BuildStateKey) and in a lane's state/ prefix.
const buildStateName = "build-state.json"

// mirrorBuildState is the build-state lane (settings spec rev 11 §4.3a):
// the instance's build-state.json, beside the lease at location in
// srcBucket, mirrored to opID's lane state/ prefix in dstBucket, and
// nothing else. The destination must know which build last ran against
// this data. A source with no build state leaves none in the lane (the
// mirror deletes a stale copy) and answers a nil object. The preparing
// rounds, the cut's final round and a rollback's return lane all call it.
// Receipts live in srcBucket beside the replica lane's, under their own
// key. The object it answers is the lane copy by the version this round
// copied or vouched for (round.Versions, a versioned mirror only), for the
// seal: a later write by a stale round is never what a seal names (ruling
// R-SEAL-STATE). Its VersionID is "" on a mirror that records none.
func mirrorBuildState(ctx context.Context, m lease.MirrorStore, srcBucket, location, dstBucket, host, opID string) (lease.MirrorRound, *moves.SealedObject, error) {
	src := strings.TrimSuffix(location, "/") + "/"
	dst := moves.BuildStateKey(host, opID)
	keep := func(rel string) bool { return rel == buildStateName }
	round, err := lease.MirrorPrefixKeep(ctx, m, srcBucket, src, dstBucket, strings.TrimSuffix(dst, buildStateName),
		lease.Receipts{Bucket: srcBucket, Key: moves.BuildStateReceiptsKey(host, opID)}, keep)
	if err != nil {
		return round, nil, err
	}
	listed, err := m.List(ctx, srcBucket, src+buildStateName)
	if err != nil {
		return round, nil, fmt.Errorf("list the build state: %w", err)
	}
	o, ok := listed[src+buildStateName]
	if !ok {
		return round, nil, nil
	}
	return round, &moves.SealedObject{Key: dst, VersionID: round.Versions[buildStateName], Size: o.Size}, nil
}

// oneBoxRefusal is the one-box rule (D15), re-checked from the edges
// snapshot every round (ruling G13): either region publishing more than one
// box refuses with M-C21's sentence. It counts every box that publishes the
// placement, as T8 does for the target.
func oneBoxRefusal(fleet moveFleet, rec provision.Record) string {
	for _, region := range []string{rec.Move.To, rec.Move.From} {
		if len(fleet.boxes[region]) > 1 {
			return s1ManyBoxes(region)
		}
	}
	return ""
}

// s1ManyBoxes is M-C21 as the source says it; the home region has no code
// of its own on a box.
func s1ManyBoxes(region string) string {
	if region == "" {
		region = "the home region"
	}
	return region + " has more than one box. Moves between regions with several boxes are not supported yet."
}

// refreshCrossRate measures the cross-region path when no recent round did:
// it recopies the largest object of the lane (guarded, as every staging
// write) and times it. The recopy lands a new version of an object the lane
// already holds with the same bytes; the receipts keep naming the earlier
// version, which holds the same bytes too.
func (w *movesWorker) refreshCrossRate(ctx context.Context, gm *guardedMirror, srcPrefix string, keep func(string) bool, dstBucket, dstPrefix string, ledger *moveLedger) error {
	src, err := gm.List(ctx, w.cfg.OwnBucket, srcPrefix)
	if err != nil {
		return fmt.Errorf("list the replica source: %w", err)
	}
	var key string
	var obj lease.MirrorObject
	for k, o := range src {
		rel := strings.TrimPrefix(k, srcPrefix)
		if keep(rel) && (o.Size > obj.Size || (o.Size == obj.Size && k < key) || key == "") {
			key, obj = k, o
		}
	}
	if key == "" || obj.Size <= 0 {
		return nil // nothing to measure with yet
	}
	start := w.cfg.Now()
	if _, _, err := gm.CopyVersioned(ctx, w.cfg.OwnBucket, key, obj.ETag, dstBucket, dstPrefix+strings.TrimPrefix(key, srcPrefix)); err != nil {
		if errors.Is(err, lease.ErrCopySourceChanged) {
			return nil // the next round copies it anyway, and times that
		}
		return fmt.Errorf("measure the cross-region copy: %w", err)
	}
	if d := w.cfg.Now().Sub(start); d > 0 {
		ledger.LastRoundBytes, ledger.LastRoundSeconds, ledger.LastRoundAt = obj.Size, d.Seconds(), w.cfg.Now().UTC()
	}
	return nil
}

// sourceRefusal is the box-step re-check (§6.2, §7.1, F14): the reason this
// source cannot move, or "". A registry or bucket read that fails is an
// error and ends the round. holder is the claim's, as the round read it
// under the staging lease.
func (w *movesWorker) sourceRefusal(ctx context.Context, s release.Store, rec provision.Record, holder *hostclaim.Holder) (string, error) {
	cfg := w.cfg
	if cfg.Route == nil {
		return "", errors.New("no registry to re-check the source against")
	}
	rt, err := cfg.Route(rec.Host)
	if err != nil {
		return "", fmt.Errorf("read the registry row: %w", err)
	}
	if rt == nil {
		return "", fmt.Errorf("%s has no registry row on this box", rec.Host)
	}
	if why := routeRefusal(*rt, cfg.DataDir); why != "" {
		return why, nil
	}
	if fileExists(filepath.Join(cfg.DataDir, rec.Host+"-sidecar.env")) {
		return s1Sidecar(rec.Host), nil
	}
	// The door's private-storage check (Task 4), repeated before the first
	// staging write of every round.
	if env := provision.InstanceStoreEnv(&rec); env == "" || appconfig.ValidEnvironmentName(env) {
		store, _, err := objstore.GetIn(ctx, s, rec.Account, rec.App, env)
		if err != nil {
			return "", fmt.Errorf("read the instance's storage binding: %w", err)
		}
		if store != nil {
			return s1PrivateStorage(rec.Host), nil
		}
	}
	if cfg.SecretsPub == nil || cfg.SecretsKey == "" {
		return s1NoSecretsKey, nil
	}
	return w.sourceFilesRefusal(ctx, rec, holder)
}

// movesFilesPreflightTimeout bounds sourceFilesRefusal's reads together:
// creating the versioned reader, listing the manifest key's versions for
// each generation of the chain, and downloading the selected manifest. A
// files manifest is one JSON object, small next to what a round copies, so
// a healthy read takes well under a second; 30 seconds absorbs a slow
// bucket, is a quarter of the claim's own request limit
// (hostbox.RequestLimit), and is far inside the 45-minute staging lease the
// round holds (moves.StagingLeaseTTL). Past it the round ends with an
// error and no report, and the next pass tries again.
const movesFilesPreflightTimeout = 30 * time.Second

// movesFilesPreflightBound is the bound in force: movesFilesPreflightTimeout,
// a variable only so a test can watch a stalled read end without waiting
// the full 30 seconds.
var movesFilesPreflightBound = movesFilesPreflightTimeout

// sourceFilesRefusal is the cut's files preflight (#236; Paul's ruling of
// 2026-10-15): the files manifest a recovery of this host would restore
// must pass the readers' path cap, or the move never reaches its cut. A
// manifest written before the cap (a8cebad) can hold a path past it, and
// every reader refuses that manifest terminally: a recovery that selected
// it would fail its establishment and leave the source offline after the
// cut. It runs in every preparing round and in the cut's requested round
// (G13), before any staging write, stop or cut.
//
// The manifest is the one the establishment selects (movesFilesSelect):
// the newest manifest of the holder's admitted chain in this region. That
// is the one a revert restores (its holder never leaves the source), and
// the one a clean return's src_gen bridge restores, since the flip
// journals the holder's generation as src_gen and every later write of a
// manifest is held to the cap by Commit. A dirty return restores from a
// prepared manifest its install builds through the capped writers, so
// nothing published before the cap reaches it. The rule is the
// establishment's own (movesFilesRestorable). A host with no files
// manifest has nothing to check here. A read that fails ends the round.
//
// It is deny-only (census ruling moves-overcap-cut-236, platform-1e,
// 2026-10-15; binding): it can only refuse a move, never allow anything.
// Passing it grants nothing, and it never skips or short-circuits the
// establishment's own check of the same manifest, which reads through the
// claim's identity-bound store (hostbox.RequestStore) and applies
// movesFilesRestorable and ValidateManifest again. Its reads go through
// cfg.Versioned, the moves worker's own reader, and are bounded by
// movesFilesPreflightTimeout. A change that lets it authorize anything
// needs a new census ruling.
func (w *movesWorker) sourceFilesRefusal(ctx context.Context, rec provision.Record, holder *hostclaim.Holder) (string, error) {
	cfg := w.cfg
	if holder == nil || holder.Location == "" {
		return "", nil // no location, no files manifest (a legacy holder)
	}
	if cfg.Versioned == nil {
		return "", errors.New("no versioned reader to check the files manifest with")
	}
	// The round runs under the staging lease and the pass mutex every moves
	// pass shares, and the instances loop calls it with no deadline: a
	// stalled listing or download must end the round, not hold both.
	ctx, cancel := context.WithTimeout(ctx, movesFilesPreflightBound)
	defer cancel()
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return "", fmt.Errorf("the data bucket, to check the files manifest: %w", err)
	}
	m, key, version, found, err := movesFilesSelect(ctx, vs, cfg.OwnBucket, holder.Location, movesFilesChain(holder, cfg.ParkHomeRegion, cfg.parkRegion()))
	if err != nil {
		return "", fmt.Errorf("read the files manifest: %w", err)
	}
	if !found {
		return "", nil
	}
	if err := movesFilesRestorable(m); err != nil {
		if !errors.Is(err, hostfiles.ErrTooDeep) {
			return "", fmt.Errorf("the files manifest %s@%s: %w", key, version, err)
		}
		log.Printf("moves: %s: the files manifest %s@%s cannot be restored (%v); an operator must replace it", rec.Host, key, version, err)
		return s1FilesPastCap(rec.Host), nil
	}
	return "", nil
}

// routeRefusal is the registry half of the box-step re-check. Backing is
// refused whenever it is set: §6.2 names only Backing (unit), and the plan
// refuses any non-empty value, which is the stricter reading (F14).
func routeRefusal(rt registry.Route, dataDir string) string {
	switch {
	case rt.Kind != "instance" || rt.Platform:
		return s1NotExec(rt.Host)
	case !rt.Hibernate:
		return s1NoHibernate(rt.Host)
	case rt.DBPath != provision.DefaultDBPath(dataDir, rt.Host):
		return s1Nested
	}
	for _, c := range []struct {
		name string
		set  bool
	}{
		{"caps", rt.Caps != ""}, {"unit", rt.Unit != ""}, {"backing", rt.Backing != ""},
		{"tunnel", rt.Tunnel != ""}, {"fleet_box", rt.FleetBox != ""}, {"pool", rt.Pool != ""},
		{"canary", rt.Canary}, {"addr", rt.Addr != ""}, {"addrs", rt.Addrs != ""}, {"health", rt.Health != ""},
		{"replica", rt.Replica}, {"replicas", rt.Replicas != ""},
	} {
		if c.set {
			return s1Column(rt.Host, c.name)
		}
	}
	return ""
}

func fileExists(path string) bool {
	_, err := os.Lstat(path)
	return err == nil
}

func containsString(xs []string, x string) bool {
	for _, v := range xs {
		if v == x {
			return true
		}
	}
	return false
}

// publishArchive builds, seals and stages the files archive when its
// plaintext digest differs from the one the ledger published, and records
// the new one. It returns an S1 reason for an archive this move can never
// carry, or retry for a file that changed while it was read: the round
// then ends, and the next one tries again.
func (w *movesWorker) publishArchive(ctx context.Context, gm *guardedMirror, rec provision.Record, dstBucket string, ledger *moveLedger) (why string, retry bool, err error) {
	cfg := w.cfg
	dir, envPath := cfg.FilesDir(rec.Host)
	plain, _, _, err := archivePlainDigest(dir, envPath)
	if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry {
		return why, retry, ferr
	}
	if plain == ledger.ArchivePlain && ledger.ArchiveVersions != nil {
		return "", false, nil
	}
	a, why, retry, err := w.buildArchive(rec)
	if err != nil || why != "" || retry {
		return why, retry, err
	}
	ref, err := a.stage(ctx, gm, dstBucket, rec.Host, rec.Move.ID)
	if err != nil {
		return "", false, err
	}
	ledger.ArchivePlain, ledger.ArchiveSealed, ledger.ArchiveVersions = a.Plain, a.Sealed, ref
	log.Printf("moves: %s: staged the files archive %s (%d files, %d bytes)", rec.Host, a.Plain, a.Entries, a.Bytes)
	return "", false, nil
}

// builtArchive is one built and sealed archive on the data disk, in the
// move's archive directory (archiveDir): <Name>.sealed and <Name>.json.
// The preparing round keeps the newest one there, so the cut can write the
// very object the target verified into move-snapshots/ (ruling G14).
type builtArchive struct {
	Name       string `json:"name"`   // files-<sealed hex>
	Plain      string `json:"plain"`  // the plaintext digest
	Sealed     string `json:"sealed"` // the sealed object's sha256
	Bytes      int64  `json:"bytes"`  // plaintext bytes
	Entries    int    `json:"entries"`
	SealedSize int64  `json:"sealed_size"`
	dir        string
}

func (a builtArchive) sealedPath() string   { return filepath.Join(a.dir, a.Name+".sealed") }
func (a builtArchive) manifestPath() string { return filepath.Join(a.dir, a.Name+".json") }

// archiveDir is where a move's built archive lives on the data disk, under
// its journal directory: never a shared tmpfs.
func archiveDir(dataDir, host, op string) string {
	return filepath.Join(moves.JournalDir(dataDir, host, op), "archive")
}

// archiveName is the object name for a sealed digest.
func archiveName(sealed string) string { return "files-" + digestHex(sealed) }

// buildArchive builds and seals the files archive into the move's archive
// directory, streamed to disk so a 256 MiB archive is never held in
// memory, and removes any older one there. Sealing is randomized (§8.2):
// every build is a new object with a new name. It returns an S1 reason for
// an archive this move can never carry, or retry for a file that changed
// while it was read.
func (w *movesWorker) buildArchive(rec provision.Record) (a builtArchive, why string, retry bool, err error) {
	cfg := w.cfg
	dir, envPath := cfg.FilesDir(rec.Host)
	adir := archiveDir(cfg.DataDir, rec.Host, rec.Move.ID)
	if err := os.MkdirAll(adir, 0o700); err != nil {
		return a, "", false, err
	}
	tmp, err := os.MkdirTemp(adir, "build-")
	if err != nil {
		return a, "", false, err
	}
	defer os.RemoveAll(tmp)
	built, err := os.Create(filepath.Join(tmp, "archive.tar.zst"))
	if err != nil {
		return a, "", false, err
	}
	defer built.Close()
	plain, entries, bytes, err := archiveBuild(dir, envPath, built)
	if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry {
		return a, why, retry, ferr
	}
	if _, err := built.Seek(0, io.SeekStart); err != nil {
		return a, "", false, err
	}
	sealedTmp := filepath.Join(tmp, "archive.sealed")
	sealed, err := os.Create(sealedTmp)
	if err != nil {
		return a, "", false, err
	}
	defer sealed.Close()
	// Seal hashes what it writes: sealedSHA is the sealed file's sha256,
	// the payload hash its upload is signed with.
	dataKey, sealedSHA, err := archive.Seal(built, cfg.SecretsPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealed)
	if err != nil {
		return a, "", false, fmt.Errorf("seal the archive: %w", err)
	}
	if err := sealed.Sync(); err != nil {
		return a, "", false, err
	}
	st, err := sealed.Stat()
	if err != nil {
		return a, "", false, err
	}
	m := archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Incarnation: rec.Move.Incarnation, Plain: plain,
		Sealed: sealedSHA, Bytes: bytes, Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize}
	manifest, err := json.Marshal(m)
	if err != nil {
		return a, "", false, err
	}
	a = builtArchive{Name: archiveName(sealedSHA), Plain: plain, Sealed: sealedSHA, Bytes: bytes, Entries: len(entries), SealedSize: st.Size(), dir: adir}
	if err := writeFileSync(filepath.Join(tmp, "manifest.json"), manifest); err != nil {
		return a, "", false, err
	}
	if err := os.Rename(sealedTmp, a.sealedPath()); err != nil {
		return a, "", false, err
	}
	if err := os.Rename(filepath.Join(tmp, "manifest.json"), a.manifestPath()); err != nil {
		return a, "", false, err
	}
	if err := syncDir(adir); err != nil {
		return a, "", false, err
	}
	pruneArchives(adir, a.Name)
	return a, "", false, nil
}

// stage puts the archive's two objects into the staging lane's files/
// through the guard (the sealed object streamed from disk) and returns
// them by the versions the PUTs answered.
func (a builtArchive) stage(ctx context.Context, gm *guardedMirror, dstBucket, host, op string) (*moves.ArchiveRef, error) {
	manifest, err := os.ReadFile(a.manifestPath())
	if err != nil {
		return nil, err
	}
	prefix := moves.FilesPrefix(host, op)
	ref := &moves.ArchiveRef{ManifestKey: prefix + a.Name + ".json", SealedObjectKey: prefix + a.Name + ".sealed", Plain: a.Plain}
	if ref.SealedVersion, err = gm.PutFileVersioned(ctx, dstBucket, ref.SealedObjectKey, a.sealedPath(), a.SealedSize, digestHex(a.Sealed)); err != nil {
		return nil, fmt.Errorf("stage the sealed archive: %w", err)
	}
	if ref.ManifestVersion, err = gm.PutVersioned(ctx, dstBucket, ref.ManifestKey, manifest); err != nil {
		return nil, fmt.Errorf("stage the archive manifest: %w", err)
	}
	return ref, nil
}

// pruneArchives removes every archive in dir but keep. Best effort: a file
// left behind costs disk until the move's directory is removed.
func pruneArchives(dir, keep string) {
	ents, err := os.ReadDir(dir)
	if err != nil {
		return
	}
	for _, e := range ents {
		name := e.Name()
		if strings.HasPrefix(name, "files-") && strings.TrimSuffix(strings.TrimSuffix(name, ".sealed"), ".json") != keep {
			os.Remove(filepath.Join(dir, name))
		}
	}
}

// writeFileSync writes body to path and syncs it.
func writeFileSync(path string, body []byte) error {
	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
	if err != nil {
		return err
	}
	if _, err := f.Write(body); err != nil {
		f.Close()
		return err
	}
	if err := f.Sync(); err != nil {
		f.Close()
		return err
	}
	return f.Close()
}

// archiveFailure maps an archive error to an S1 reason (a standing refusal
// the operator must act on), a retry (a file changed under the walk), or
// nothing. The reason is always a recorded sentence chosen by errors.Is,
// with a path only from the *archive.PathError the error carries; the
// error's own text, which can name anything the walk met, is logged and
// never published (ruling T15-sinks). An error this does not recognise
// gets s1ArchiveFailed.
func archiveFailure(host string, err error) (why string, retry bool, ferr error) {
	if err == nil {
		return "", false, nil
	}
	if errors.Is(err, archive.ErrChanged) {
		log.Printf("moves: %s: %v; the archive is rebuilt next round", host, err)
		return "", true, nil
	}
	log.Printf("moves: %s: archive the files: %v", host, err)
	var pe *archive.PathError
	hasPath := errors.As(err, &pe) && pe.Path != ""
	switch {
	case errors.Is(err, archive.ErrSpecialFile) && hasPath:
		return s1SpecialFile(host, pe.Path), false, nil // M-C14, hard links too
	case errors.Is(err, archive.ErrOverCap):
		return s1OverCap(host), false, nil // M-C15
	case errors.Is(err, archive.ErrPastTreeCap):
		return s1FilePastCap(host), false, nil // MV-N502
	case errors.Is(err, archive.ErrTooManyEntries):
		return s1TooManyEntries, false, nil // MV-N79
	case errors.Is(err, archive.ErrBadPath) && hasPath:
		return s1LongPath(pe.Path), false, nil // MV-N81
	}
	return s1ArchiveFailed, false, nil
}

// digestHex is "sha256:<hex>" without its algorithm, for an object name.
func digestHex(d string) string {
	_, h, ok := strings.Cut(d, ":")
	if !ok {
		return d
	}
	if _, err := hex.DecodeString(h); err != nil {
		return d
	}
	return h
}

// sourcePassed is every check a source report carries: S1 and S2.
func sourcePassed(r *moves.SourceReport) bool { return r != nil && r.S1.OK && r.S2.OK }

// writeSourceReport publishes the source's readiness report (§7.3): the
// launch environment as HMACs, S1 (a refusal, or the instance's own answer
// to its readiness path), S2, the published archive, the live release and
// the cross-region rate. It is replaced every round.
func (w *movesWorker) writeSourceReport(ctx context.Context, s release.Store, rec provision.Record, r *srcRound) (*moves.SourceReport, error) {
	cfg := w.cfg
	now := cfg.Now().UTC()
	sr := moves.SourceReport{V: 1, Box: cfg.Box, MoveID: rec.Move.ID, At: now,
		FilesPlain: r.ledger.ArchivePlain, Archive: r.ledger.ArchiveVersions}
	sr.CrossRate, sr.CrossRateAt = r.ledger.crossRate()
	if r.s1 != nil {
		sr.S1 = *r.s1
	} else {
		sr.S1 = w.probeSource(ctx, s, rec)
	}
	sr.S2 = moves.Check{Reason: "the next spawn of this host is not strict"}
	if cfg.StrictNext != nil {
		sr.S2 = cfg.StrictNext(rec.Host)
		// NextSpawnStrict prefixes its reasons with "S2: ", which Ready adds.
		sr.S2.Reason = strings.TrimPrefix(sr.S2.Reason, "S2: ")
	}
	if st, err := os.Stat(provision.DefaultDBPath(cfg.DataDir, rec.Host)); err == nil && st.Mode().IsRegular() {
		sr.DBBytes = st.Size()
	}
	sr.ReplicaBytes = r.replicaBytes
	if cfg.OptDir != "" {
		if v, err := release.LiveVersion(cfg.OptDir, rec.Host); err == nil {
			sr.Release = v
		} else {
			log.Printf("moves: %s: live version: %v", rec.Host, err)
		}
	}
	if r.s1 == nil && cfg.LaunchEnv != nil {
		key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
		if err != nil {
			sr.S1 = moves.Check{Reason: s1NoSecretsKey}
		} else if env, instance, err := cfg.LaunchEnv(rec.Host); err != nil {
			return nil, fmt.Errorf("compose the launch environment: %w", err)
		} else {
			sr.Env = moves.ReportLaunch(key, env, instance)
		}
	}
	body, err := json.Marshal(sr)
	if err != nil {
		return nil, err
	}
	if err := moves.PutReplace(ctx, s, moves.SourceReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), body); err != nil {
		return nil, err
	}
	return &sr, nil
}

// probeSource is S1: the instance's own answer to its readiness path,
// woken if it sleeps, never from the response cache.
func (w *movesWorker) probeSource(ctx context.Context, s release.Store, rec provision.Record) moves.Check {
	if w.cfg.ReadyProbe == nil {
		return moves.Check{Reason: s1CannotProbe}
	}
	// "/" only when the manifest was read and declares no path: an app
	// whose declared path fails must not pass on a healthy homepage while
	// its manifest cannot be read (review finding 5).
	m, err := provision.GetManifest(ctx, s, rec.Account, rec.App)
	if err != nil {
		log.Printf("moves: %s: S1: instances manifest: %v", rec.Host, err)
		return moves.Check{Reason: s1NoReadyPath}
	}
	return w.cfg.ReadyProbe(ctx, rec.Host, m.ReadyPathOrDefault())
}

// decideReady reads the target's report and applies §7.3's rule to it and
// to source, the report this round just published (and that passed); on
// ready from preparing it pins the inputs and promotes, and an automatic
// cut then marks the cut requested (§7.4). The caller reaches it only for a
// round whose own report passed, so a failing round never requests a cut.
// Every CAS is guarded by the op identity the round read: a replacement
// move is ErrMoveConflict and the round ends.
func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec provision.Record, g provision.MoveGuard, r *srcRound, source *moves.SourceReport) error {
	cfg := w.cfg
	state := rec.Move.State
	// A ready move with no pin is one a cut's admission sent back (G1): the
	// same rule pins it again, from this passing round, before any cut.
	repin := state == provision.MoveReady && rec.Move.Inputs == nil
	// Ruling T5-lease (fix round 2): a holder whose location is not the
	// activator's lease prefix cannot be cut, so a ready move on one is
	// neither pinned again nor cut automatically. Nothing is written: the
	// door refuses such a move at request time and the cut's admission
	// refuses a replay, so a move that got here waits, with a notice,
	// until an operator aborts it.
	// (Its promotion to ready still happens, so the operator sees it there.)
	_, layoutErr := writerLease(cfg.OwnPrefix, rec.Host, r.holder)
	if layoutErr != nil && (repin || rec.Move.Cut == "auto") {
		w.notice(rec.Host+"#layout", "moves: %s: move %s cannot be cut (%v); no repin and no automatic cut", rec.Host, g.ID, layoutErr)
		if repin {
			return nil
		}
	}
	// The cut guard (Paul's ruling, 2026-10-11): a cut this box refused
	// because it cannot recover the move (cutRun.refuseUnarmed) is not
	// pinned again while that holds, so an automatic cut does not loop
	// through the refusal. The first cut is still requested, and its
	// refusal records the reason the operator reads. A target that cannot
	// land fails T13, so Ready below keeps it from pinning.
	if repin && !recoverySourceArmed(cfg) {
		w.notice(rec.Host+"#unarmed", "moves: %s: move %s is not pinned again: %s", rec.Host, g.ID, provision.RefusedSourceUnarmed)
		return nil
	}
	if state == provision.MovePreparing || repin {
		target, err := readTargetReport(ctx, s, rec)
		if err != nil {
			return err
		}
		ok, why := moves.Ready(r.deltaBytes, cfg.ReadyDelta, target, source, cfg.Now())
		if !ok {
			w.notice(rec.Host+"#ready", "moves: %s: not ready: %s", rec.Host, why)
			return nil
		}
		// moves.Ready has required target.Inputs.Release to equal the
		// source's, so what is pinned is what T7 checked (review finding 3).
		in := target.Inputs
		mut := provision.MoveGuardedPromoteReady(g, cfg.Now(), in)
		if repin {
			mut = provision.MoveGuardedRepin(g, in)
		}
		var pre provision.Record
		after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, mut))
		if err != nil {
			if errors.Is(err, provision.ErrMoveConflict) {
				log.Printf("moves: %s: the record no longer holds move %s as %s without a pin (%v); this round ends", rec.Host, g.ID, state, err)
				return nil
			}
			return err
		}
		w.logTransition(ctx, s, pre, *after)
		nudgeSerial(ctx, s)
		if repin {
			log.Printf("moves: %s: move %s pinned its inputs again", rec.Host, g.ID)
		} else {
			log.Printf("moves: %s: move %s is ready (last round moved %d bytes; bar %d)", rec.Host, g.ID, r.deltaBytes, cfg.ReadyDelta)
		}
		state = provision.MoveReady
	}
	if state == provision.MoveReady && rec.Move.Cut == "auto" && layoutErr == nil {
		var pre provision.Record
		after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveGuardedCut(g)))
		if err != nil {
			if errors.Is(err, provision.ErrMoveConflict) {
				log.Printf("moves: %s: the automatic cut of %s was refused (%v)", rec.Host, g.ID, err)
				return nil
			}
			return err
		}
		w.logTransition(ctx, s, pre, *after)
		nudgeSerial(ctx, s)
		log.Printf("moves: %s: move %s cut requested automatically", rec.Host, g.ID)
	}
	return nil
}

func readSourceReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.SourceReport, error) {
	var r moves.SourceReport
	ok, err := readReport(ctx, s, moves.SourceReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), &r)
	if !ok {
		return nil, err
	}
	return &r, nil
}

func readTargetReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.TargetReport, error) {
	var r moves.TargetReport
	ok, err := readReport(ctx, s, moves.TargetReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID)), &r)
	if !ok {
		return nil, err
	}
	return &r, nil
}

// readReport reads a report: (false, nil) when absent or unreadable as a
// report (Ready then says no report), an error when the read failed.
func readReport(ctx context.Context, s release.Store, key string, into any) (bool, error) {
	body, _, err := s.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("read %s: %w", key, err)
	}
	if err := json.Unmarshal(body, into); err != nil {
		log.Printf("moves: %s does not parse: %v", key, err)
		return false, nil
	}
	return true, nil
}