package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// The cut on the source (moves spec §5.3, §7.4, §7.5), behind
// CARLOS_MOVES_PHASE1. One pass drives one cutting record as far as it can:
// each phase does its work and then advances cut_phase by one guarded CAS,
// and the next phase runs in the same pass on the re-read record. Nothing
// is carried in memory between passes: every phase resumes from the
// record, the claim, the deployment bucket, this box's own bucket and the
// op journal (moves.Journal, <DataDir>/.moves/<host>/<op>/).
//
// Phase names (ruling G8). A phase is written when the previous phase's
// work is complete, before its own: `stopping` before the hold and the
// stop, `stopped` once the tree is proven stopped, `staged` once the final
// round is sealed, `archived` once the seal names the archive. So a record
// at X means "X's work may have begun, and everything before X is done",
// and a replay of X redoes X's work from durable state. Each phase's
// comment says what a replay finds.
//
// The deadline (§7.4): before any work, a phase past cut_deadline stops
// the cut. Before anything stopped (admitted) that is back to ready; from
// stopping on it is a revert, whose claim half follows the record once the
// source has given up execution (revertClaim).

// cutLandingWindow is the landing deadline the flip writes (now + 15 min).
const cutLandingWindow = 15 * time.Minute

// cutRequestLimit is the staging park's request limit, the longest a
// staging request runs (lease's stagingRequestTimeout).
const cutRequestLimit = moves.ParkRequestLimit

// cutMaxDeferred is how many final rounds may defer before the cut reverts
// (§7.4 staged: three deferred rounds).
const cutMaxDeferred = 3

// cutStepsPerPass bounds the phases one pass runs.
const cutStepsPerPass = 8

// Op journal steps the cut writes (moves.Journal).
const (
	cutStepEvidence    = "evidence"     // §5.3 steps 1 to 4, the StoppedReport JSON
	cutStepCoverage    = "coverage"     // cutCoverage: the stopped database's position against N
	cutStepArchive     = "archive"      // the cut archive's identity, before any PUT
	cutStepRevertBound = "revert_bound" // the revert's staging park evidence_at
)

// cutCoverage is the journalled coverage decision (ruling 3): whether the
// stopped database's own position equalled N after the final checkpoint.
type cutCoverage struct {
	N       uint64 `json:"n"`
	Covered bool   `json:"covered"`
}

// cutRun is one pass's state for one cutting record.
type cutRun struct {
	w     *movesWorker
	s     release.Store
	fleet moveFleet
	rec   provision.Record
	g     provision.MoveGuard
	// The staging lease and the guarded mirror, taken once per pass by the
	// first phase that writes staging (ruling G6) and released at its end.
	gm      *guardedMirror
	release func(context.Context) error
}

// runCut is the MoveCutting arm of phase1MoveRound.
func (w *movesWorker) runCut(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error {
	c := &cutRun{w: w, s: s, fleet: fleet, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}}
	defer c.releaseLease()
	for i := 0; i < cutStepsPerPass; i++ {
		advanced, err := c.step(ctx)
		if err != nil || !advanced {
			return err
		}
		fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
		if err != nil {
			return fmt.Errorf("re-read the record: %w", err)
		}
		if fresh.Move == nil || fresh.Move.ID != c.g.ID || fresh.Move.Incarnation != c.g.Incarnation || fresh.Move.State != provision.MoveCutting {
			return nil
		}
		c.rec = *fresh
	}
	return nil
}

func (c *cutRun) cfg() moveCfg { return c.w.cfg }

// fault is the test seam after a side effect.
func (c *cutRun) fault(point string) error {
	if f := c.w.cfg.cutFault; f != nil {
		return f(point)
	}
	return nil
}

// step runs the record's current phase once. advanced is true when it
// wrote the record (a phase, back to ready, a revert or the flip).
func (c *cutRun) step(ctx context.Context) (bool, error) {
	m := c.rec.Move
	if m.State != provision.MoveCutting {
		return false, nil
	}
	if m.CutDeadline != nil && c.cfg().Now().After(*m.CutDeadline) {
		switch m.CutPhase {
		case provision.CutAdmitted:
			return c.backToReady(ctx, "the cut's deadline passed before anything stopped")
		case provision.CutStopping:
			return c.revert(ctx, provision.RevertStopEvidenceBudget)
		case provision.CutStopped, provision.CutStaged, provision.CutArchived:
			return c.revert(ctx, provision.RevertCutDeadline)
		}
	}
	switch m.CutPhase {
	case provision.CutRequested:
		return c.requested(ctx)
	case provision.CutAdmitted:
		return c.admitted(ctx)
	case provision.CutStopping:
		return c.stopping(ctx)
	case provision.CutStopped:
		return c.stopped(ctx)
	case provision.CutStaged:
		return c.staged(ctx)
	case provision.CutArchived:
		return c.archived(ctx)
	}
	return false, nil
}

// cutTransition writes one guarded record transition. A conflict is
// someone else's decision (an abort, a console write): logged, and the
// pass ends without error.
func (c *cutRun) cutTransition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, c.s, c.rec.Account, c.rec.App, c.rec.Host, observedMove(&pre, mut))
	if err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: %s of %s refused (%v); the record moved on", c.rec.Host, what, c.g.ID, err)
			return false, nil
		}
		return false, fmt.Errorf("%s: %w", what, err)
	}
	c.w.logTransition(ctx, c.s, pre, *after)
	return true, nil
}

func (c *cutRun) advance(ctx context.Context, from, to string, stamp func(*provision.MoveState)) (bool, error) {
	ok, err := c.cutTransition(ctx, "the cut "+from+" to "+to, provision.MoveCutPhase(c.g, from, to, stamp))
	if ok {
		log.Printf("moves: %s: cut %s is %s", c.rec.Host, c.g.ID, to)
	}
	return ok, err
}

// backToReady undoes an admission whose checks are not fresh or whose
// inputs moved (§7.4 admitted): nothing has stopped. The pin is cleared and
// the next passing readiness round pins again (G1).
func (c *cutRun) backToReady(ctx context.Context, why string) (bool, error) {
	ok, err := c.cutTransition(ctx, "back to ready", provision.MoveBackToReady(c.g))
	if ok {
		log.Printf("moves: %s: cut %s goes back to ready: %s", c.rec.Host, c.g.ID, why)
		nudgeSerial(ctx, c.s)
	}
	return ok, err
}

// revert starts the pre-flip recovery (§7.4 reverting): the record first,
// then the claim half once execution is none (revertClaim).
func (c *cutRun) revert(ctx context.Context, reason string) (bool, error) {
	ok, err := c.cutTransition(ctx, "revert", provision.MoveRevert(c.g, reason))
	if !ok {
		return ok, err
	}
	log.Printf("moves: %s: cut %s reverts: %s", c.rec.Host, c.g.ID, reason)
	nudgeSerial(ctx, c.s)
	if err := c.w.revertClaim(ctx, c.s, c.rec); err != nil {
		// The record decided; the next pass's reverting arm writes the
		// claim half (phase1MoveRound).
		log.Printf("moves: %s: the claim half of the revert waits: %v", c.rec.Host, err)
	}
	return true, nil
}

// requested (§7.4): one source round under the staging lease, with a fresh
// source report (G13), then Ready on that round and the target's report,
// then the pinned inputs against the current ones; the admission stamps
// cut_at, the budget and the deadline. Any check that fails goes back to
// ready, so readiness reruns. A replay finds requested and runs the round
// again: nothing here is written but the reports and the lane, which every
// round rewrites.
func (c *cutRun) requested(ctx context.Context) (bool, error) {
	cfg := c.cfg()
	if refused, wrote, err := c.refuseUnarmed(ctx); refused || err != nil {
		return wrote, err
	}
	sr, published, err := c.w.sourceRound(ctx, c.s, c.rec, c.fleet)
	if err != nil || published == nil {
		return false, err
	}
	if !sourcePassed(published) {
		return c.backToReady(ctx, "this round's source report does not pass")
	}
	target, err := readTargetReport(ctx, c.s, c.rec)
	if err != nil {
		return false, err
	}
	if ok, why := moves.Ready(sr.deltaBytes, cfg.ReadyDelta, target, published, cfg.Now()); !ok {
		return c.backToReady(ctx, why)
	}
	if same, err := c.inputsHold(ctx); err != nil || !same {
		if err != nil {
			return false, err
		}
		return c.backToReady(ctx, "the inputs pinned at readiness have changed")
	}
	if refused, wrote, err := c.refuseHolder(ctx); refused || err != nil {
		return wrote, err
	}
	now := cfg.Now()
	budget := target.CutBudgetS
	return c.cutTransition(ctx, "the admission", provision.MoveAdmit(c.g, now, budget, moves.CutDeadline(now, float64(budget)), target.AttemptsAdmitted))
}

// admitted (§7.4): the pinned inputs still match the current bundle and
// overlay versions, the release and the launch report, and the claim
// holder is one this box can stop under the writer's lease (ruling
// T5-lease), or back to ready. A replay finds admitted and compares again;
// nothing has stopped.
func (c *cutRun) admitted(ctx context.Context) (bool, error) {
	same, err := c.inputsHold(ctx)
	if err != nil {
		return false, err
	}
	if !same {
		return c.backToReady(ctx, "the inputs pinned at readiness have changed")
	}
	if refused, wrote, err := c.refuseHolder(ctx); refused || err != nil {
		return wrote, err
	}
	if refused, wrote, err := c.refuseUnarmed(ctx); refused || err != nil {
		return wrote, err
	}
	return c.advance(ctx, provision.CutAdmitted, provision.CutStopping, nil)
}

// refuseUnarmed is the cut guard (Paul's ruling, 2026-10-11): a cut starts
// only when the target's latest report states that its box can land the
// move and roll it back (T13, moves.TargetReport.Armed) and this box can
// recover the move if the cut fails (recoverySourceArmed). Otherwise the
// cut goes back to ready with the reason recorded, before this box takes
// the staging lease, holds or stops anything, or seals the claim: it runs
// first at requested and last at admitted, the step before stopping. It
// fails closed: no report, or an older box's report with no T13, is
// unarmed. The refusal stands while the cause does: decideReady pins no
// move again while this box cannot recover, and the target's failing T13
// keeps Ready from pinning it (§7.3), so nothing loops and the operator
// reads why in `carlos move status`.
//
// refused and wrote are apart (fix round 1): refused is the guard's
// verdict, wrote whether the refusal's record write landed. A refused cut
// ends the step either way. When another writer moved the record first
// (ErrMoveConflict, which cutTransition logs and answers false) nothing
// was written, the pass ends, and the next pass reads the record again;
// the cut never carries on to the source's round or to stopping.
func (c *cutRun) refuseUnarmed(ctx context.Context) (refused, wrote bool, err error) {
	reason := ""
	if !recoverySourceArmed(c.cfg()) {
		reason = provision.RefusedSourceUnarmed
	} else {
		target, err := readTargetReport(ctx, c.s, c.rec)
		if err != nil {
			return false, false, err
		}
		if !target.Armed() {
			reason = provision.RefusedTargetUnarmed
		}
	}
	if reason == "" {
		return false, false, nil
	}
	log.Printf("moves: %s: cut %s refused: %s", c.rec.Host, c.g.ID, reason)
	if err := c.fault("refusal"); err != nil {
		return true, false, err
	}
	wrote, err = c.cutTransition(ctx, "the refusal", provision.MoveBackToReadyRefused(c.g, reason))
	if wrote {
		nudgeSerial(ctx, c.s)
	}
	return true, wrote, err
}

// refuseHolder reads the claim fresh and, when its holder is not at the
// activator's lease location (holderSupported), sends the cut back to
// ready with RefusedReplicaLayout recorded (ruling T5-lease). The
// refusal stands as long as the claim does: a passing round pins again,
// the next cut is refused the same way, and the operator reads why.
//
// refused and wrote are apart, as in refuseUnarmed (fix round 1): a
// refused cut ends the step whether or not the refusal's record write
// landed. A write another writer beat (ErrMoveConflict) writes nothing,
// the pass ends, and the next pass reads the record again.
func (c *cutRun) refuseHolder(ctx context.Context) (refused, wrote bool, err error) {
	obs, err := c.claim(ctx)
	if err != nil {
		return false, false, err
	}
	if err := c.holderSupported(*obs.Claim.Holder); err != nil {
		log.Printf("moves: %s: cut %s refused: %v", c.rec.Host, c.g.ID, err)
		if err := c.fault("holder-refusal"); err != nil {
			return true, false, err
		}
		wrote, err := c.cutTransition(ctx, "the refusal", provision.MoveBackToReadyRefused(c.g, provision.RefusedReplicaLayout))
		if wrote {
			nudgeSerial(ctx, c.s)
		}
		return true, wrote, err
	}
	return false, false, nil
}

// inputsHold compares the record's pin with the current inputs.
func (c *cutRun) inputsHold(ctx context.Context) (bool, error) {
	cur, err := c.w.currentInputs(ctx, c.s, c.rec)
	if err != nil {
		return false, err
	}
	return sameInputs(c.rec.Move.Inputs, cur), nil
}

// currentInputs are the inputs the admission compares (§7.4 admitted): the
// bundle and overlay versions, the live release and the launch report's
// digest, as this box would publish them now. The digest is the
// move-portable one (moves.PortableDigest), since the pin it is compared
// with crosses to the target.
func (w *movesWorker) currentInputs(ctx context.Context, s release.Store, rec provision.Record) (provision.MoveInputs, error) {
	return w.inputsDigested(ctx, s, rec, moves.PortableDigest)
}

// recoveryInputs are the inputs a recovery probe records and then checks on
// this same box (movesrecoveryprobe.go): currentInputs' fields, with the
// digest of the whole composed environment (moves.LocalDigest), box-owned
// variables included. The record and the check are made on one box just
// before the wake, so the box's own variables are part of what runs and
// must not change between them either (fix round 2).
func (w *movesWorker) recoveryInputs(ctx context.Context, s release.Store, rec provision.Record) (provision.MoveInputs, error) {
	return w.inputsDigested(ctx, s, rec, func(key []byte, env map[string]string, _ map[string]bool) string {
		return moves.LocalDigest(key, env)
	})
}

func (w *movesWorker) inputsDigested(ctx context.Context, s release.Store, rec provision.Record, digest func(key []byte, env map[string]string, instance map[string]bool) string) (provision.MoveInputs, error) {
	cfg := w.cfg
	if cfg.InputVersions == nil || cfg.LaunchEnv == nil {
		return provision.MoveInputs{}, errors.New("no way to read the current inputs on this box")
	}
	bundle, overlay, err := cfg.InputVersions(ctx, s, rec)
	if err != nil {
		return provision.MoveInputs{}, fmt.Errorf("read the configuration versions: %w", err)
	}
	live, err := release.LiveVersion(cfg.OptDir, rec.Host)
	if err != nil {
		return provision.MoveInputs{}, fmt.Errorf("read the live release: %w", err)
	}
	key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
	if err != nil {
		return provision.MoveInputs{}, err
	}
	env, instance, err := cfg.LaunchEnv(rec.Host)
	if err != nil {
		return provision.MoveInputs{}, fmt.Errorf("compose the launch environment: %w", err)
	}
	// The probe-only variables are dropped as the owners' checks drop them
	// (launchVars): the probe sets its own, whatever the files say.
	vars := make(map[string]string, len(env))
	for k, v := range env {
		if !probeOnlyVars[k] {
			vars[k] = v
		}
	}
	return provision.MoveInputs{BundleETags: bundle, OverlayETag: overlay, Release: live, EnvReport: digest(key, vars, instance)}, nil
}

// sameInputs is the admission's comparison: every bundle etag, the overlay
// etag, the release and the launch report. The files digest is not one of
// them: a file written before the stop is carried by the cut's own archive.
func sameInputs(pin *provision.MoveInputs, cur provision.MoveInputs) bool {
	if pin == nil || pin.OverlayETag != cur.OverlayETag || pin.Release != cur.Release || pin.EnvReport != cur.EnvReport {
		return false
	}
	if len(pin.BundleETags) != len(cur.BundleETags) {
		return false
	}
	for k, v := range pin.BundleETags {
		if cv, ok := cur.BundleETags[k]; !ok || cv != v {
			return false
		}
	}
	return true
}

// claim reads the host claim and checks it is this move's moving op for
// the record's incarnation.
func (c *cutRun) claim(ctx context.Context) (hostclaim.Observed, error) {
	obs, err := hostclaim.Read(ctx, c.s, c.rec.Host)
	if err != nil {
		return obs, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent {
		return obs, fmt.Errorf("move %s has no host claim", c.g.ID)
	}
	if err := hostclaim.RequireOp(c.g.ID)(obs.Claim); err != nil {
		return obs, fmt.Errorf("move %s: %w", c.g.ID, err)
	}
	if h := obs.Claim.Holder; h == nil || h.Incarnation != c.rec.Move.Incarnation {
		return obs, fmt.Errorf("move %s: the host claim names another incarnation", c.g.ID)
	}
	return obs, nil
}

// dbPath is the row's database path, or the flat default.
func (c *cutRun) dbPath() string {
	cfg := c.cfg()
	if cfg.Route != nil {
		if rt, err := cfg.Route(c.rec.Host); err == nil && rt != nil && rt.DBPath != "" {
			return rt.DBPath
		}
	}
	return provision.DefaultDBPath(cfg.DataDir, c.rec.Host)
}

func (c *cutRun) journal() (*moves.Journal, error) {
	return moves.ReadJournal(c.cfg().DataDir, c.rec.Host, c.g.ID)
}

// stopping (§5.3, §7.4): the row held moving under the move id, the wake
// fence, the timer drain, then the evidence of steps 1 to 4. A tree not
// proven stopped waits for the next pass, until the deadline reverts with
// "stop evidence budget". Proven, the evidence goes into the op journal
// (local: the report's replica step, N and digests come at stopped) and
// the phase advances.
//
// A replay finds stopping and runs it all again from the box's state:
// every step is idempotent (the hold under the same op, the fence at the
// same generation, the drain, and Evidence, which reads only durable
// state). No in-memory state is needed (Review Focus 1).
func (c *cutRun) stopping(ctx context.Context) (bool, error) {
	cfg := c.cfg()
	host := c.rec.Host
	obs, err := c.claim(ctx)
	if err != nil {
		return false, err
	}
	if err := hostclaim.RequireMoveOp(c.g.ID, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil {
		return false, fmt.Errorf("move %s: %w", c.g.ID, err)
	}
	if cfg.Box == "" || obs.Claim.Op.Exec != cfg.Box {
		return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", c.g.ID, obs.Claim.Op.Exec)
	}
	// Admission refused any other holder; a claim that differs now is an
	// error, never a stop (fail closed).
	if err := c.holderSupported(*obs.Claim.Holder); err != nil {
		return false, fmt.Errorf("move %s: %w", c.g.ID, err)
	}
	if cfg.HoldMoving == nil || cfg.Fence == nil || cfg.TimersDrained == nil || cfg.Evidence == nil {
		return false, errors.New("this box cannot stop an instance for a move: a hook is not wired")
	}
	if err := cfg.HoldMoving(host, c.g.ID); err != nil {
		return false, fmt.Errorf("hold the row: %w", err)
	}
	gen := moveFenceGeneration(c.rec.Move)
	if err := cfg.Fence(ctx, host, gen); err != nil {
		return false, fmt.Errorf("fence against wakes: %w", err)
	}
	c.w.fenced.Store(host, gen)
	if err := cfg.TimersDrained(ctx, host); err != nil {
		return false, fmt.Errorf("drain the timers: %w", err)
	}
	rep, err := cfg.Evidence(ctx, host, obs.Claim.Holder.Location, []string{c.dbPath()})
	if err != nil {
		return false, fmt.Errorf("stop evidence: %w", err)
	}
	if !rep.TreeStopped() {
		c.w.notice(host+"#evidence", "moves: %s: the instance is not proven stopped yet (%s); retrying until the cut's deadline", host, failedSteps(rep))
		if strayRefused(rep) {
			c.w.pageStray(ctx, c.rec)
		}
		return false, nil
	}
	j, err := c.journal()
	if err != nil {
		return false, err
	}
	body, err := json.Marshal(rep)
	if err != nil {
		return false, err
	}
	j.Phase = provision.CutStopping
	j.Mark(cutStepEvidence, string(body))
	if err := j.Commit(cfg.DataDir); err != nil {
		return false, err
	}
	if err := c.fault("after-evidence"); err != nil {
		return false, err
	}
	return c.advance(ctx, provision.CutStopping, provision.CutStopped, nil)
}

// failedSteps names the evidence steps that did not pass, for the log.
func failedSteps(r moves.StoppedReport) string {
	var out []string
	for _, id := range []string{moves.StepTree, moves.StepStray, moves.StepFDs, moves.StepLease} {
		if c, ok := r.Steps[id]; !ok || !c.OK {
			out = append(out, id)
		}
	}
	return strings.Join(out, ", ")
}

// stopped (§5.3 step 5, §7.4): the stopped source's report, its execution
// given up, then the final round and the seal.
//
// While the claim's executor is still this box: the final checkpoint under
// the claim's lineage (G3), N, the database and timers digests, the files'
// plain digest, the complete stopped-source.json (create-only, K13), the
// record's source baselines, and then exec to none with the cut's session
// in one claim write (K3). Once exec is none, a replay reads the published
// report instead: nothing on the source's replica is written again.
//
// Then, under the staging lease: the timers snapshot into timers/ and the
// final mirror round into replica/. A round that deferred or moved more
// than the bar is counted in the ledger (three revert); a round that
// cannot name a usable version for every current key runs again. The seal
// goes into the claim with the op's phase sealed in one CAS, and only then
// does the record say staged (G8).
//
// A replay finds stopped and repeats from the claim's state: the
// checkpoint is idempotent on a stopped database, the report and the
// snapshots are create-only and read back, every claim write is
// conditional, and the final round rewrites staging, which the seal pins by
// version.
func (c *cutRun) stopped(ctx context.Context) (bool, error) {
	cfg := c.cfg()
	obs, err := c.claim(ctx)
	if err != nil {
		return false, err
	}
	op := obs.Claim.Op
	if op.Kind != hostclaim.KindMoving || (op.Phase != hostclaim.PhaseMoving && op.Phase != hostclaim.PhaseSealed) {
		return false, fmt.Errorf("move %s: the claim is %s/%q", c.g.ID, op.Kind, op.Phase)
	}
	holder := *obs.Claim.Holder
	if err := c.holderSupported(holder); err != nil {
		return false, fmt.Errorf("move %s: %w", c.g.ID, err)
	}
	stoppedKey := moves.StoppedKey(moves.ReportPrefix(c.rec.Account, c.rec.App, c.rec.Host, c.g.ID), "source")
	var report moves.StoppedReport
	switch {
	case cfg.Box != "" && op.Exec == cfg.Box:
		if report, err = c.giveUpExecution(ctx, obs, stoppedKey); err != nil {
			return false, err
		}
		if report.V == 0 {
			return false, nil // reverted, or waiting
		}
	case op.Exec == "" && op.Journal[hostclaim.JournalEvidenceSource] == stoppedKey:
		if report, err = readStoppedReport(ctx, c.s, stoppedKey); err != nil {
			return false, err
		}
	default:
		return false, fmt.Errorf("move %s: the claim's executor is %q and cites %q", c.g.ID, op.Exec, op.Journal[hostclaim.JournalEvidenceSource])
	}
	return c.finalRound(ctx, holder, report, stoppedKey)
}

// giveUpExecution is stopped's first half, run while this box executes.
// It returns a zero report when the phase ended without one (a revert, or
// a file that changed).
//
// A revert's entry runs it too (revertSource, Plan 4 Task 6, ruling J7), on a
// record already reverting: it then records no baselines (MoveRecordSource is
// a cut's, at stopped; a revert from stopping has none, J8), and a files
// directory that cannot be archived is no reason to revert again: the report
// carries no files digest, and nothing compares one (§7.4 evidence).
func (c *cutRun) giveUpExecution(ctx context.Context, obs hostclaim.Observed, stoppedKey string) (moves.StoppedReport, error) {
	cfg := c.cfg()
	host := c.rec.Host
	holder := *obs.Claim.Holder
	reverting := c.rec.Move.State == provision.MoveReverting
	j, err := c.journal()
	if err != nil {
		return moves.StoppedReport{}, err
	}
	var ev moves.StoppedReport
	if raw, ok := j.Steps[cutStepEvidence]; !ok || json.Unmarshal([]byte(raw), &ev) != nil || !ev.TreeStopped() {
		// The journal lost the evidence (a rebuilt data directory): the row
		// is still held and fenced, so ask the box again.
		if cfg.Evidence == nil {
			return moves.StoppedReport{}, errors.New("no stop evidence on this box")
		}
		if ev, err = cfg.Evidence(ctx, host, holder.Location, []string{c.dbPath()}); err != nil {
			return moves.StoppedReport{}, fmt.Errorf("stop evidence: %w", err)
		}
		if !ev.TreeStopped() {
			return moves.StoppedReport{}, fmt.Errorf("the instance is no longer proven stopped (%s)", failedSteps(ev))
		}
	}
	own, vs, err := c.ownStores(ctx)
	if err != nil {
		return moves.StoppedReport{}, err
	}
	prefix, keep := moves.ReplicaSource(holder)
	dbPath := c.dbPath()
	fromDB, err := regularFile(dbPath)
	if err != nil {
		return moves.StoppedReport{}, err
	}
	if fromDB {
		if cfg.FinalCheckpoint == nil {
			return moves.StoppedReport{}, errors.New("no final checkpoint on this box")
		}
		check := func(ctx context.Context) error { return c.stillExecuting(ctx) }
		if err := cfg.FinalCheckpoint(ctx, host, dbPath, holder, check); err != nil {
			return moves.StoppedReport{}, fmt.Errorf("final checkpoint: %w", err)
		}
	}
	if err := c.fault("after-checkpoint"); err != nil {
		return moves.StoppedReport{}, err
	}
	n, err := c.finalTXID(ctx, own, prefix, keep)
	if err != nil {
		return moves.StoppedReport{}, err
	}
	var dbSHA string
	if fromDB {
		if dbSHA, err = fileSHA256(dbPath); err != nil {
			return moves.StoppedReport{}, fmt.Errorf("digest the stopped database: %w", err)
		}
		// Coverage (ruling 3): the stopped database's own position against
		// N, decided here, with the database present after its checkpoint,
		// and journalled so the seal can still state it if the disk is lost
		// before the final round. Without a database nothing is journalled
		// and the seal says uncovered.
		if cfg.LocalPosition == nil {
			return moves.StoppedReport{}, errors.New("no local position on this box")
		}
		pos, err := cfg.LocalPosition(dbPath)
		if err != nil {
			return moves.StoppedReport{}, fmt.Errorf("read the stopped database's position: %w", err)
		}
		body, err := json.Marshal(cutCoverage{N: n, Covered: pos == n})
		if err != nil {
			return moves.StoppedReport{}, err
		}
		j.Mark(cutStepCoverage, string(body))
		if err := j.Commit(cfg.DataDir); err != nil {
			return moves.StoppedReport{}, err
		}
	}
	timersSHA, err := c.snapshotTimers(ctx, own, vs)
	if err != nil {
		return moves.StoppedReport{}, err
	}
	dir, envPath := cfg.FilesDir(host)
	plain, _, _, err := archivePlainDigest(dir, envPath)
	if why, retry, ferr := archiveFailure(host, err); ferr != nil || why != "" || retry {
		switch {
		case why != "" && reverting:
			log.Printf("moves: %s: the revert's stopped report carries no files digest: %s", host, why)
			plain = ""
		case why != "":
			_, rerr := c.revert(ctx, provision.RevertArchiveRefused)
			return moves.StoppedReport{}, rerr
		default:
			return moves.StoppedReport{}, ferr
		}
	}
	var rt registry.Route
	if cfg.Route != nil {
		if r, err := cfg.Route(host); err != nil {
			return moves.StoppedReport{}, fmt.Errorf("read the registry row: %w", err)
		} else if r != nil {
			rt = *r
		}
	}
	report := ev
	report.V, report.Role, report.Box, report.Incarnation, report.ClaimGen = 1, "source", cfg.Box, holder.Incarnation, obs.Claim.Gen
	report.Host, report.OpID = host, c.g.ID
	report.Steps = map[string]moves.Check{}
	for k, v := range ev.Steps {
		report.Steps[k] = v
	}
	report.Steps[moves.StepReplica] = moves.Check{OK: true}
	report.FinalTXID, report.DBSHA256, report.FilesPlain, report.TimersSHA256 = n, dbSHA, plain, timersSHA
	if !report.Complete() {
		return moves.StoppedReport{}, fmt.Errorf("the stopped report is not complete (%s)", failedSteps(report))
	}
	// K13: create-only. A report already there is this cut's own replay
	// when it names the same host, op, role and N; it is then the report.
	if report, err = publishStoppedReport(ctx, c.s, stoppedKey, report); err != nil {
		return moves.StoppedReport{}, err
	}
	if err := c.fault("after-report"); err != nil {
		return moves.StoppedReport{}, err
	}
	if !reverting {
		src := provision.MoveSource{Replication: rt.Replication, NextDue: rt.NextDue, FinalTXID: report.FinalTXID, DBSHA256: report.DBSHA256,
			Files: &provision.MoveFilesDigest{Plain: report.FilesPlain}, TimersSHA256: report.TimersSHA256, Evidence: stoppedKey}
		if ok, err := c.cutTransition(ctx, "the source baselines", provision.MoveRecordSource(c.g, src)); !ok {
			if err == nil {
				err = errors.New("the record moved on")
			}
			return moves.StoppedReport{}, err
		}
		if err := c.fault("after-record-source"); err != nil {
			return moves.StoppedReport{}, err
		}
	}
	session := strconv.FormatUint(holder.Session, 10)
	if _, err := hostclaim.Update(ctx, c.s, obs, func(cl *hostclaim.Claim) error {
		if err := hostclaim.ExecToNone(c.g.ID, cfg.Box, stoppedKey, hostclaim.JournalEvidenceSource)(cl); err != nil {
			return err
		}
		return hostclaim.SetJournal(c.g.ID, map[string]string{hostclaim.JournalCutSession: session})(cl)
	}); err != nil {
		return moves.StoppedReport{}, fmt.Errorf("give up execution: %w", err)
	}
	log.Printf("moves: %s: stopped at TXID %d; execution given up, citing %s", host, report.FinalTXID, stoppedKey)
	if err := c.fault("after-exec-none"); err != nil {
		return moves.StoppedReport{}, err
	}
	return report, nil
}

// stillExecuting is the final checkpoint's authority: every replica
// request it makes first re-reads the claim, which must still name this
// move with this box executing.
func (c *cutRun) stillExecuting(ctx context.Context) error {
	obs, err := c.claim(ctx)
	if err != nil {
		return err
	}
	if obs.Claim.Op.Kind != hostclaim.KindMoving || obs.Claim.Op.Exec == "" || obs.Claim.Op.Exec != c.cfg().Box {
		return fmt.Errorf("move %s: this box no longer executes the host", c.g.ID)
	}
	return nil
}

// publishStoppedReport writes the report create-only. On a conflict the
// existing report is read: the same host, op, role and N is a replay and
// the existing report is returned; anything else is an error.
func publishStoppedReport(ctx context.Context, s release.Store, key string, r moves.StoppedReport) (moves.StoppedReport, error) {
	body, err := json.Marshal(r)
	if err != nil {
		return r, err
	}
	if _, err := s.Put(ctx, key, body, ""); err == nil {
		return r, nil
	} else if !errors.Is(err, release.ErrCASConflict) {
		return r, fmt.Errorf("write %s: %w", key, err)
	}
	prev, err := readStoppedReport(ctx, s, key)
	if err != nil {
		return r, err
	}
	if prev.Host != r.Host || prev.OpID != r.OpID || prev.Role != r.Role || prev.FinalTXID != r.FinalTXID {
		return r, fmt.Errorf("%s already names %s %s %s at TXID %d, not this stop at %d", key, prev.Host, prev.OpID, prev.Role, prev.FinalTXID, r.FinalTXID)
	}
	return prev, nil
}

func readStoppedReport(ctx context.Context, s release.Store, key string) (moves.StoppedReport, error) {
	body, _, err := s.Get(ctx, key)
	if err != nil {
		return moves.StoppedReport{}, fmt.Errorf("read %s: %w", key, err)
	}
	var r moves.StoppedReport
	if err := json.Unmarshal(body, &r); err != nil || r.V != 1 || !r.Complete() {
		return moves.StoppedReport{}, fmt.Errorf("%s is not a complete stopped report", key)
	}
	return r, nil
}

// ownStores are this box's own bucket as a mirror (reads and plain writes)
// and by version (create-only writes).
func (c *cutRun) ownStores(ctx context.Context) (lease.MirrorStore, lease.VersionedStore, error) {
	cfg := c.cfg()
	tgt, ok := c.fleet.targets[c.rec.Move.To]
	if !ok {
		return nil, nil, fmt.Errorf("target region %s publishes no data bucket yet. waiting", c.rec.Move.To)
	}
	own, err := c.w.mirror(ctx, tgt.DataBucketRegion)
	if err != nil {
		return nil, nil, err
	}
	if cfg.Versioned == nil {
		return nil, nil, errors.New("no versioned store on this box")
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return nil, nil, err
	}
	return own, vs, nil
}

// finalTXID is N (§5.1 step 1): the highest max TXID among the claim
// generation's level files in this box's own replica, the set the lane
// mirrors (G3). After the final checkpoint it is the newest level-0
// file's; a compaction only ever rewrites below it.
func (c *cutRun) finalTXID(ctx context.Context, own lease.MirrorStore, prefix string, keep func(string) bool) (uint64, error) {
	objs, err := own.List(ctx, c.cfg().OwnBucket, prefix)
	if err != nil {
		return 0, fmt.Errorf("list the replica: %w", err)
	}
	newest := newestLTX(objs, prefix, keep)
	if newest == "" {
		return 0, errors.New("the replica holds no level file")
	}
	_, maxTX, err := ltx.ParseFilename(filepath.Base(newest))
	if err != nil {
		return 0, err
	}
	return uint64(maxTX), nil
}

// snapshotTimers writes the host's timer rows create-only to
// move-snapshots/<host>/<op>/timers.json in this box's bucket and returns
// their digest. A snapshot already there is this cut's own (a replay): it
// is read back and its rows are the ones digested.
func (c *cutRun) snapshotTimers(ctx context.Context, own lease.MirrorStore, vs lease.VersionedStore) (string, error) {
	cfg := c.cfg()
	if cfg.Timers == nil {
		return "", errors.New("no timer rows on this box")
	}
	rows, err := cfg.Timers(c.rec.Host)
	if err != nil {
		return "", fmt.Errorf("read the timers: %w", err)
	}
	body, err := moves.TimersSnapshot{V: 1, Host: c.rec.Host, Rows: rows, TakenAt: cfg.Now().UTC()}.Encode()
	if err != nil {
		return "", err
	}
	key := moves.SnapshotsPrefix(c.rec.Host, c.g.ID) + "timers.json"
	if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, key, body); errors.Is(err, lease.ErrExists) {
		prev, err := own.Get(ctx, cfg.OwnBucket, key)
		if err != nil {
			return "", fmt.Errorf("read %s: %w", key, err)
		}
		snap, err := moves.DecodeTimers(prev)
		if err != nil {
			return "", err
		}
		rows = snap.Rows
	} else if err != nil {
		return "", fmt.Errorf("snapshot the timers: %w", err)
	}
	return moves.TimersDigest(rows), nil
}

// lease takes the staging lease once per pass (G6) and revalidates the
// claim under it (K5). waiting is true when another worker holds it.
func (c *cutRun) lease(ctx context.Context) (waiting bool, err error) {
	if c.gm != nil {
		return false, nil
	}
	cfg := c.cfg()
	if cfg.StagingLease == nil {
		return false, errors.New("no deployment bucket for the staging lease")
	}
	tgt, ok := c.fleet.targets[c.rec.Move.To]
	if !ok {
		return false, fmt.Errorf("target region %s publishes no data bucket yet. waiting", c.rec.Move.To)
	}
	held, rel, err := cfg.StagingLease(ctx, c.rec.Move.To, c.rec.Host)
	if err != nil {
		var he *lease.HeldError
		if errors.As(err, &he) {
			log.Printf("moves: %s: another worker holds the staging lease; the cut waits", c.rec.Host)
			return true, nil
		}
		return false, fmt.Errorf("staging lease: %w", err)
	}
	stg, err := c.w.staging(ctx, tgt.DataBucketRegion)
	if err != nil {
		rel(context.Background())
		return false, err
	}
	own, err := c.w.mirror(ctx, tgt.DataBucketRegion)
	if err != nil {
		rel(context.Background())
		return false, err
	}
	c.gm = &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: cfg.Now}, dst: tgt.DataBucket, own: own}
	c.release = rel
	return false, nil
}

func (c *cutRun) releaseLease() {
	if c.release == nil {
		return
	}
	rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := c.release(rctx); err != nil {
		log.Printf("moves: %s: release the staging lease: %v", c.rec.Host, err)
	}
	c.release, c.gm = nil, nil
}

// sealedClaim revalidates the claim under the lease for the phases after
// execution was given up: this move's moving op with exec none and this
// stop's evidence, at one of the phases given.
func (c *cutRun) sealedClaim(ctx context.Context, stoppedKey string, phases ...string) (hostclaim.Observed, error) {
	obs, err := c.claim(ctx)
	if err != nil {
		return obs, err
	}
	op := obs.Claim.Op
	if op.Kind != hostclaim.KindMoving || !containsString(phases, op.Phase) || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceSource] != stoppedKey {
		return obs, fmt.Errorf("move %s: the claim is %s/%q exec %q, not given up for this cut", c.g.ID, op.Kind, op.Phase, op.Exec)
	}
	return obs, nil
}

// bound is the writers' horizon for the seal (G7): the latest moment a
// request this pass signed can still land, and never earlier than the
// bound already recorded.
func (c *cutRun) bound(prev time.Time) time.Time {
	b := prev
	if last := c.gm.lastSigned(); !last.IsZero() {
		if t := last.Add(moves.SignatureLife + moves.DefaultSkew); t.After(b) {
			b = t
		}
	}
	return b.UTC()
}

// finalRound is stopped's second half: the timers and the replica into
// staging under the lease, then the seal.
func (c *cutRun) finalRound(ctx context.Context, holder hostclaim.Holder, report moves.StoppedReport, stoppedKey string) (bool, error) {
	cfg := c.cfg()
	host, id := c.rec.Host, c.g.ID
	if waiting, err := c.lease(ctx); waiting || err != nil {
		return false, err
	}
	obs, err := c.sealedClaim(ctx, stoppedKey, hostclaim.PhaseMoving, hostclaim.PhaseSealed)
	if err != nil {
		return false, err
	}
	own, _, err := c.ownStores(ctx)
	if err != nil {
		return false, err
	}
	dst := c.gm.dst
	timersKey := moves.SnapshotsPrefix(host, id) + "timers.json"
	timersBody, err := own.Get(ctx, cfg.OwnBucket, timersKey)
	if err != nil {
		return false, fmt.Errorf("read %s: %w", timersKey, err)
	}
	stagedTimers := moves.TimersPrefix(host, id) + "timers.json"
	tver, err := c.gm.PutVersioned(ctx, dst, stagedTimers, timersBody)
	if err != nil {
		return false, fmt.Errorf("stage the timers: %w", err)
	}
	if !lease.UsableVersionID(tver) {
		return false, fmt.Errorf("the staged timers have no usable version id %q", tver)
	}
	prefix, keep := moves.ReplicaSource(holder)
	dstPrefix := moves.ReplicaPrefix(host, id)
	receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: moves.ReceiptsKey(host, id)}
	var round lease.MirrorRound
	for attempt := 0; ; attempt++ {
		round, err = lease.MirrorPrefixKeep(ctx, c.gm, cfg.OwnBucket, prefix, dst, dstPrefix, receipts, keep)
		if err == nil || attempt == 1 || errors.Is(err, moves.ErrLeaseUnsafe) || errors.Is(err, moves.ErrInvalidSkew) {
			break
		}
		log.Printf("moves: %s: final round failed, retrying once: %v", host, err)
	}
	if err != nil {
		return false, fmt.Errorf("final round: %w", err)
	}
	// The build state of the stopped instance, so the lane carries the
	// final state and not a round-old one (settings spec rev 11 §4.3a). A
	// state that changed under the copy is not final: the pass retries.
	// The seal names the version this round copied (ruling R-SEAL-STATE).
	// This box's request record goes into it first (ruling R-SPEC-43A-2):
	// execution is given up, so no wake writes it now.
	if cfg.CarryRequest != nil {
		if err := cfg.CarryRequest(ctx, host, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/")); err != nil {
			return false, fmt.Errorf("final round: carry the request record into the build state: %w", err)
		}
	}
	stRound, stateObj, err := mirrorBuildState(ctx, c.gm, cfg.OwnBucket, holder.Location, dst, host, id)
	if err != nil {
		return false, fmt.Errorf("final round: build state lane: %w", err)
	}
	if stRound.Deferred > 0 {
		return false, fmt.Errorf("final round: the build state changed during the copy; retrying the cut")
	}
	if err := c.fault("after-final-round"); err != nil {
		return false, err
	}
	ledgerKey := moves.LedgerKey(host, id)
	ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey)
	if err != nil {
		return false, err
	}
	// §7.4 staged: a round that deferred is not byte-final; nor is one that
	// moved more than the bar the cut was priced for (G27).
	if round.Deferred > 0 || round.DeltaBytes() > cfg.ReadyDelta {
		ledger.DeferredRounds++
		if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
			return false, err
		}
		log.Printf("moves: %s: final round %d deferred %d object(s) and moved %d bytes", host, ledger.DeferredRounds, round.Deferred, round.DeltaBytes())
		if ledger.DeferredRounds >= cutMaxDeferred {
			return c.revert(ctx, provision.RevertDeferredRounds)
		}
		nudgeSerial(ctx, c.s)
		return false, nil
	}
	// G27: every kept current source key by a usable version, else another
	// round.
	src, err := own.List(ctx, cfg.OwnBucket, prefix)
	if err != nil {
		return false, fmt.Errorf("list the replica: %w", err)
	}
	var replica []moves.SealedObject
	for k, o := range src {
		rel := strings.TrimPrefix(k, prefix)
		if !keep(rel) {
			continue
		}
		ver := round.Versions[rel]
		if !lease.UsableVersionID(ver) {
			c.w.notice(host+"#seal", "moves: %s: the final round has no usable version for %s; another round", host, rel)
			return false, nil
		}
		replica = append(replica, moves.SealedObject{Key: dstPrefix + rel, VersionID: ver, Size: o.Size})
	}
	var state []moves.SealedObject
	if stateObj != nil {
		if !lease.UsableVersionID(stateObj.VersionID) {
			c.w.notice(host+"#seal", "moves: %s: the final round has no usable version for the build state; another round", host)
			return false, nil
		}
		state = []moves.SealedObject{*stateObj}
	}
	// Coverage starts false (ruling 3): a database still here is asked for
	// its position against N; one that is gone is covered only if the
	// checkpoint journalled that it was, at this N (giveUpExecution).
	fromDB, err := regularFile(c.dbPath())
	if err != nil {
		return false, err
	}
	covered := false
	if fromDB {
		if cfg.LocalPosition == nil {
			return false, errors.New("no local position on this box")
		}
		pos, err := cfg.LocalPosition(c.dbPath())
		if err != nil {
			return false, fmt.Errorf("read the stopped database's position: %w", err)
		}
		covered = pos == report.FinalTXID
	} else if j, err := c.journal(); err != nil {
		return false, err
	} else if raw, ok := j.Steps[cutStepCoverage]; ok {
		var cov cutCoverage
		if json.Unmarshal([]byte(raw), &cov) == nil && cov.N == report.FinalTXID {
			covered = cov.Covered
		}
	}
	now := cfg.Now().UTC()
	seal := moves.Seal{V: moves.SealVersion, N: report.FinalTXID, Replica: replica,
		Timers:     []moves.SealedObject{{Key: stagedTimers, VersionID: tver, Size: int64(len(timersBody))}},
		State:      state,
		SrcCovered: covered, FromDB: fromDB, SealedAt: now}
	seal.LeaseUntil = c.bound(now)
	if err := seal.Validate(); err != nil {
		return false, err
	}
	if !seal.ReachesN() {
		return false, fmt.Errorf("the sealed replica does not end at TXID %d", seal.N)
	}
	if _, err := hostclaim.Update(ctx, c.s, obs, sealStopped(id, stoppedKey, seal)); err != nil {
		return false, fmt.Errorf("seal the cut: %w", err)
	}
	if err := c.fault("after-seal"); err != nil {
		return false, err
	}
	ledger.LeaseUntil = seal.LeaseUntil
	if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
		return false, err
	}
	log.Printf("moves: %s: sealed %d replica objects at TXID %d", host, len(replica), seal.N)
	return c.advance(ctx, provision.CutStopped, provision.CutStaged, nil)
}

// sealStopped is the stopped seal's claim write: the op's phase "" to
// sealed (or already sealed, a replay) and the seal, in one CAS, under
// exec none and this stop's evidence.
func sealStopped(id, stoppedKey string, seal moves.Seal) func(*hostclaim.Claim) error {
	return func(c *hostclaim.Claim) error {
		if c.Op.Exec != "" || c.Op.Journal[hostclaim.JournalEvidenceSource] != stoppedKey {
			return fmt.Errorf("%w: the seal needs exec none and this stop's evidence", hostclaim.ErrWrongState)
		}
		if c.Op.Phase == hostclaim.PhaseMoving {
			if err := hostclaim.SetMovePhase(id, hostclaim.KindMoving, hostclaim.PhaseMoving, hostclaim.PhaseSealed)(c); err != nil {
				return err
			}
		} else if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseSealed)(c); err != nil {
			return err
		}
		return hostclaim.SetJournal(id, map[string]string{hostclaim.JournalSeal: seal.Encode()})(c)
	}
}

// journalledSeal reads the seal the claim carries.
func journalledSeal(c *hostclaim.Claim) (moves.Seal, error) {
	raw, ok := c.Op.Journal[hostclaim.JournalSeal]
	if !ok {
		return moves.Seal{}, fmt.Errorf("%w: the claim carries no seal", moves.ErrBadSeal)
	}
	return moves.DecodeSeal(raw)
}

// staged (§7.4, §8.3, codex 19, G14): the final archive of the stopped
// files directory, its identity fixed before any PUT, written create-only
// to move-snapshots/ and staged into files/, then the seal with the
// archive's two versions, and only then the record's archived with the
// archive's digests. A replay finds staged and redoes the archive from its
// journalled identity (cutArchive), then the seal and the phase.
func (c *cutRun) staged(ctx context.Context) (bool, error) {
	a, ok, err := c.sealArchive(ctx)
	if err != nil || !ok {
		return false, err
	}
	return c.advance(ctx, provision.CutStaged, provision.CutArchived, func(m *provision.MoveState) {
		if m.Source == nil {
			m.Source = &provision.MoveSource{}
		}
		m.Source.Files = &provision.MoveFilesDigest{Plain: a.Plain, Sealed: a.Sealed, Bytes: a.Bytes, Entries: a.Entries}
	})
}

// sealArchive is staged's work, shared with archived's replay: the archive
// published and the claim's seal naming it. ok is false when the phase
// ended without it (a revert, the lease held elsewhere).
func (c *cutRun) sealArchive(ctx context.Context) (builtArchive, bool, error) {
	cfg := c.cfg()
	src := c.rec.Move.Source
	if src == nil || src.Files == nil || src.Evidence == "" {
		return builtArchive{}, false, fmt.Errorf("move %s: the record has no source baselines", c.g.ID)
	}
	if waiting, err := c.lease(ctx); waiting || err != nil {
		return builtArchive{}, false, err
	}
	obs, err := c.sealedClaim(ctx, src.Evidence, hostclaim.PhaseSealed)
	if err != nil {
		return builtArchive{}, false, err
	}
	seal, err := journalledSeal(obs.Claim)
	if err != nil {
		return builtArchive{}, false, err
	}
	if seal.N != src.FinalTXID {
		return builtArchive{}, false, fmt.Errorf("move %s: the seal is at TXID %d and the record at %d", c.g.ID, seal.N, src.FinalTXID)
	}
	a, files, why, err := c.cutArchive(ctx, src.Files.Plain)
	if err != nil {
		return builtArchive{}, false, err
	}
	if why != "" {
		_, err := c.revert(ctx, provision.RevertArchiveRefused)
		return builtArchive{}, false, err
	}
	seal.Files = files
	seal.LeaseUntil = c.bound(seal.LeaseUntil)
	if err := seal.Validate(); err != nil {
		return builtArchive{}, false, err
	}
	if _, err := hostclaim.Update(ctx, c.s, obs, func(cl *hostclaim.Claim) error {
		if cl.Op.Exec != "" || cl.Op.Journal[hostclaim.JournalEvidenceSource] != src.Evidence {
			return fmt.Errorf("%w: the seal needs exec none and this stop's evidence", hostclaim.ErrWrongState)
		}
		if err := hostclaim.RequireMoveOp(c.g.ID, hostclaim.KindMoving, hostclaim.PhaseSealed)(cl); err != nil {
			return err
		}
		return hostclaim.SetJournal(c.g.ID, map[string]string{hostclaim.JournalSeal: seal.Encode()})(cl)
	}); err != nil {
		return builtArchive{}, false, fmt.Errorf("seal the archive: %w", err)
	}
	if err := c.fault("after-archive-seal"); err != nil {
		return builtArchive{}, false, err
	}
	own, _, err := c.ownStores(ctx)
	if err != nil {
		return builtArchive{}, false, err
	}
	// G7: the bound in the ledger as well as the seal. A ledger that cannot
	// be read or written stops the phase here; the replay seals again and
	// writes it (ruling 5).
	ledgerKey := moves.LedgerKey(c.rec.Host, c.g.ID)
	ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey)
	if err != nil {
		return builtArchive{}, false, err
	}
	ledger.LeaseUntil = seal.LeaseUntil
	if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
		return builtArchive{}, false, err
	}
	return a, true, nil
}

// archived (§7.4, §7.5): the flip. Its precondition is re-read from the
// claim: the seal names the archive at the record's N, and the record's
// digests are the seal's archive; otherwise this is a replay of staged's
// tail, which is redone (never a flip on a seal without the archive). Then
// the record flips (one CAS: region, state landing, the landing deadline)
// and the claim follows: landing/checking with the target executing,
// citing the source's evidence (hostclaim.Flip). The claim half is tried
// once here; anyone who sees the record ahead writes it (the console's
// recovery pass, Plan 5).
func (c *cutRun) archived(ctx context.Context) (bool, error) {
	cfg := c.cfg()
	src := c.rec.Move.Source
	if src == nil || src.Files == nil {
		return false, fmt.Errorf("move %s: the record has no source baselines", c.g.ID)
	}
	obs, err := c.sealedClaim(ctx, src.Evidence, hostclaim.PhaseSealed)
	if err != nil {
		return false, err
	}
	seal, err := journalledSeal(obs.Claim)
	if err != nil {
		return false, err
	}
	if !seal.HasFiles() || seal.N != src.FinalTXID || !sealNamesArchive(seal, src.Files.Sealed) {
		log.Printf("moves: %s: archived without a seal naming the archive; redoing the archive", c.rec.Host)
		if _, ok, err := c.sealArchive(ctx); err != nil || !ok {
			return false, err
		}
		if obs, err = c.claim(ctx); err != nil {
			return false, err
		}
		if seal, err = journalledSeal(obs.Claim); err != nil {
			return false, err
		}
		if !seal.HasFiles() || seal.N != src.FinalTXID || !sealNamesArchive(seal, src.Files.Sealed) {
			return false, fmt.Errorf("move %s: the record's archive %q is not the one the seal names", c.g.ID, src.Files.Sealed)
		}
	}
	targetBox := obs.Claim.Op.Intent["target"]
	holder := *obs.Claim.Holder
	now := cfg.Now()
	ok, err := c.cutTransition(ctx, "the flip", provision.MoveFlip(c.g, now, now.Add(cutLandingWindow)))
	if !ok {
		return ok, err
	}
	log.Printf("moves: %s: flipped to %s", c.rec.Host, c.rec.Move.To)
	if err := c.fault("after-record-flip"); err != nil {
		return true, err
	}
	if _, err := hostclaim.Update(ctx, c.s, obs, hostclaim.Flip(c.g.ID, targetBox, holder.Stamp, holder.Generation)); err != nil {
		log.Printf("moves: %s: the claim half of the flip waits for the recovery pass: %v", c.rec.Host, err)
	}
	nudgeSerial(ctx, c.s)
	return true, nil
}

// revertClaim is the claim half of a revert (§7.4 reverting, §7.5): once
// the source has given up execution, the claim goes to reverting under
// the source with the staging lane's park queued (hostclaim.Revert). While
// the source still executes (a revert from stopping, or from stopped
// before exec went to none) there is nothing to hand back, and the claim
// stays moving: the recovery (Plan 4) starts from there. A claim already
// reverting for this move is done.
//
// The park's evidence_at is the writers' horizon (G7): the later of the
// seal's bound and this decision's time plus a signature's life and the
// skew, journalled once so a replay names the same scope.
func (w *movesWorker) revertClaim(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	id := rec.Move.ID
	obs, err := hostclaim.Read(ctx, s, rec.Host)
	if err != nil {
		return fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent || obs.Claim.Op.ID != id {
		return nil
	}
	op := obs.Claim.Op
	if op.Kind == hostclaim.KindReverting || op.Kind != hostclaim.KindMoving || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceSource] == "" {
		return nil
	}
	if obs.Claim.Holder == nil || cfg.Box == "" {
		return errors.New("the revert needs the holder and this box's label")
	}
	j, err := moves.ReadJournal(cfg.DataDir, rec.Host, id)
	if err != nil {
		return err
	}
	var bound time.Time
	if raw, ok := j.Steps[cutStepRevertBound]; ok {
		if bound, err = time.Parse(time.RFC3339Nano, raw); err != nil {
			bound = time.Time{}
		}
	}
	if bound.IsZero() {
		bound = cfg.Now().Add(moves.SignatureLife + moves.DefaultSkew).UTC()
		if seal, err := journalledSeal(obs.Claim); err == nil && seal.Bound().After(bound) {
			bound = seal.Bound().UTC()
		}
		j.Mark(cutStepRevertBound, bound.Format(time.RFC3339Nano))
		if err := j.Commit(cfg.DataDir); err != nil {
			return err
		}
	}
	scope := moves.StagingScope(rec.Host, id, obs.Claim.Holder.Incarnation, bound)
	if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Revert(id, cfg.Box, park.PlacementRegion(rec.Move.To), &scope, obligationID)); err != nil {
		return fmt.Errorf("revert the claim: %w", err)
	}
	log.Printf("moves: %s: the claim reverts to this box; the staging lane in %s is queued for its park", rec.Host, rec.Move.To)
	return nil
}

// obligationID mints a claim obligation or retirement id.
func obligationID() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		return ""
	}
	return hex.EncodeToString(b[:])
}

// regularFile reports whether path is a regular file. Absence (ENOENT) is
// false with no error; any other answer from the filesystem, an error or
// something there that is not a regular file, is an error, never absence
// (ruling 3): the cut decides coverage on it, so it must not read a
// transient fault as "the database is gone".
func regularFile(path string) (bool, error) {
	st, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("stat %s: %w", path, err)
	}
	if !st.Mode().IsRegular() {
		return false, fmt.Errorf("%s is not a regular file (%s)", path, st.Mode())
	}
	return true, nil
}

// fileSHA256 is "sha256:<hex>" of the file at path.
func fileSHA256(path string) (string, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer f.Close()
	h := sha256.New()
	if _, err := io.Copy(h, f); err != nil {
		return "", err
	}
	return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
}

// holdMoving is moveCfg.HoldMoving in production: the row held "moving"
// under op, conditional on its current hold (SetHoldIf) and under the
// host's hold lock, which reconcile's hold decision and the landing's open
// share (Task 6). A row already held moving by op is left as it is; a row
// held by another op is refused. SQLITE_BUSY while root freezes holds is
// an error the next pass retries (G30). A write that landed touches the
// stage-requested marker.
func holdMoving(reg *registry.Registry, written func(host string)) func(host, op string) error {
	return func(host, op string) error {
		var err error
		provision.WithHostHoldLock(host, func() {
			var rt *registry.Route
			if rt, err = reg.Get(host); err != nil {
				return
			}
			if rt == nil {
				err = fmt.Errorf("%s has no registry row on this box", host)
				return
			}
			if rt.HoldOp == op && rt.Status == "moving" {
				return
			}
			if rt.HoldOp != "" && rt.HoldOp != op {
				err = fmt.Errorf("%s is held %s by %s", host, rt.Status, rt.HoldOp)
				return
			}
			var held bool
			if held, err = reg.SetHoldIf(host, "moving", op, rt.HoldOp); err != nil {
				return
			}
			if !held {
				err = fmt.Errorf("%s: the row's hold changed under the cut", host)
				return
			}
			if written != nil {
				written(host)
			}
		})
		return err
	}
}

// localLTXPosition is moveCfg.LocalPosition in production: the stopped
// database's TXID, the newest level-0 file in the activator's litestream
// meta directory (lease.Held's SetMetaPath, litestream's DB.MaxLTX). No
// directory is position 0.
func localLTXPosition(dbPath string) (uint64, error) {
	dir := filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream", "ltx", "0")
	ents, err := os.ReadDir(dir)
	if errors.Is(err, os.ErrNotExist) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	var newest ltx.TXID
	for _, e := range ents {
		if _, maxTX, err := ltx.ParseFilename(e.Name()); err == nil && maxTX > newest {
			newest = maxTX
		}
	}
	return uint64(newest), nil
}

// errHolderLocation: the claim holder's location is not the activator's
// own lease prefix, so the cut cannot stop its writer (ruling T5-lease).
var errHolderLocation = errors.New("the claim holder is not at the activator's lease location")

// writerLease is the one lease object the host's writer holds: the
// activator's, at activator.HibernatePrefix(prefix, host), as
// ExecBackend.Acquire takes it. Phase 1 cuts a holder only when its
// location is exactly that (ruling T5-lease): the final checkpoint and the
// stop evidence then exclude the real writer, and a scoped location
// (instances/<host>/<incarnation>), whose lease would be a different
// object, is refused at admission. The claim's generation is still bound
// on that lease, so the replica lineage stays the holder's.
func writerLease(prefix, host string, h hostclaim.Holder) (string, error) {
	want := activator.HibernatePrefix(prefix, host)
	if strings.TrimSuffix(h.Location, "/") != want {
		return "", fmt.Errorf("%w: %s is at %q, not %q", errHolderLocation, host, h.Location, want)
	}
	return want, nil
}

// holderSupported is writerLease for this cut's box.
func (c *cutRun) holderSupported(h hostclaim.Holder) error {
	_, err := writerLease(c.cfg().OwnPrefix, c.rec.Host, h)
	return err
}

// finalCheckpoint is moveCfg.FinalCheckpoint in production (§5.3 step 5,
// G3, ruling T5-lease): the activator's own lease for host, refused for a
// holder anywhere else before any request; the replica client bound to the
// claim's lineage with check as the authority read before every replica
// request, a bound generation (BindGeneration) or the root layout
// (BindRoot) alike; one checkpoint; the lease released.
func finalCheckpoint(bucket, prefix, region string) func(ctx context.Context, host, dbPath string, h hostclaim.Holder, check func(context.Context) error) error {
	return func(ctx context.Context, host, dbPath string, h hostclaim.Holder, check func(context.Context) error) error {
		at, err := writerLease(prefix, host, h)
		if err != nil {
			return err
		}
		held, err := lease.Acquire(ctx, bucket, at, region)
		if err != nil {
			return err
		}
		defer func() {
			rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			if err := held.Release(rctx); err != nil {
				log.Printf("moves: release the checkpoint lease at %s: %v", at, err)
			}
		}()
		if g := h.Generation; g != "" && g != hostclaim.RootGeneration {
			lineage, _ := moves.ReplicaSource(h)
			err = held.BindGeneration(lineage, check)
		} else {
			err = held.BindRoot(check)
		}
		if err != nil {
			return err
		}
		_, err = held.Checkpoint(ctx, dbPath)
		return err
	}
}