package console

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"slices"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// The offline operator move's doors (docs/superpowers/plans/2026-10-20-
// offline-operator-move.md, Task 6). One hibernating instance moves between
// two regions with planned downtime, an operator at each step:
//
//	POST .../instances/{host}/move/offline/stage     {"region":"uk"}   pre-copy: -> staging
//	POST .../instances/{host}/move/offline           {"region":"uk"}   start (staging -> stopping)
//	POST .../instances/{host}/move/offline/seal                        stopping -> sealing
//	POST .../instances/{host}/move/offline/flip                        sealed -> landing
//	POST .../instances/{host}/move/offline/open                        landed -> open
//	POST .../instances/{host}/move/offline/finish                      open -> finishing -> cleared
//	POST .../instances/{host}/move/offline/abort                       back to the source
//	GET  .../instances/{host}/move/offline                             the status view
//
// Every door is moveOperatorGate (the byte-identical 404 for anyone but an
// operator's own CLI token), then MV-O20 when this console runs host claims
// or Phase 1 moves, then its own checks, then exactly one SetRecordMove
// transition and a serial bump, and answers 202 with the status view.
//
// The doors decide from evidence the console reads and never from what the
// operator says: the record, the edges files, the boxes' status reports and
// the reports each box writes under control/offline-moves/. The console
// writes nothing but the record.

// offlineMoveCapability is the edges-file capability a box publishes only
// when hibernation is configured and host claims, Phase 1 and an armed or
// enforced guard are all off (internal/host, Task 3).
const offlineMoveCapability = "offline-move-1"

// offlineStageCapability is the one a box publishes beside it when it can
// stage (internal/host movesoffline_stage.go): run the pre-copy as the
// stager and remove it on abort. A box without it treats a staging record
// as nothing to do, so the stage door refuses a region holding one.
const offlineStageCapability = "offline-stage-1"

// offlineReportFresh is how recently a box's report must be written for a
// door to believe it. Boxes rewrite their reports on every moves pass, and a
// source or sealer box every 15 s or so while its report is one a door reads
// (internal/host offlineReassertEvery), so a report older than this is from a
// box that has stopped reporting, and the door says so (MV-O178).
const offlineReportFresh = 2 * time.Minute

// Every sentence a door speaks has its row in the copy record (MV-O20 to
// MV-O31, MV-O50, MV-O51) or is an existing approved one, written out whole
// where it is used so the copy gate reads it.

// offlineName is a region code as the operator reads it: the home's
// configured name, or the word home when none is configured.
func (s *Server) offlineName(code string) string {
	if name := s.homeRegionOut(code); name != "" {
		return name
	}
	return "home"
}

// offlineDoor is what every door has by the time its own checks begin.
type offlineDoor struct {
	acct *Account
	app  string
	host string
	rec  *provision.Record
	addr string
}

// offlineGate is the front half every door shares: the move family's gate
// and operator check (a 404 for anyone else), then MV-O20.
func (s *Server) offlineGate(w http.ResponseWriter, r *http.Request) (offlineDoor, bool) {
	acct, app, host, rec, addr, ok := s.moveOperatorGate(w, r)
	if !ok {
		return offlineDoor{}, false
	}
	if s.cfg.HostClaim || s.cfg.MovesEnabled {
		jsonError(w, http.StatusConflict, "offline moves are not available while host claims or Phase 1 moves are on")
		return offlineDoor{}, false
	}
	return offlineDoor{acct: acct, app: app, host: host, rec: rec, addr: addr}, true
}

// offlineInFlight is the move of a door that acts on one: an offline move
// the record carries, else MV-O24.
func (s *Server) offlineInFlight(w http.ResponseWriter, d offlineDoor) (*provision.MoveState, bool) {
	if d.rec.Move == nil || !d.rec.Move.Offline() {
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s has no offline move in flight", d.host))
		return nil, false
	}
	return d.rec.Move, true
}

// offlineStateRefusal is MV-O25.
func offlineStateRefusal(state, host string) *offlineRefusal {
	return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("this move is %s. run carlos move offline status --host %s to see what it waits for", state, host)}
}

// offlineWaitRefusal is M-C10 with <from> filled in: a source box has not
// confirmed the stop.
func (s *Server) offlineWaitRefusal(box, from, host string) *offlineRefusal {
	return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("Waiting for %s in %s to confirm it has stopped %s. %s is offline.", box, from, host, host)}
}

// offlineStaleRefusal is MV-O178: box's report is there but older than
// offlineReportFresh. That says nothing about the box's process or its stop,
// only that the door cannot believe the report yet; a box taking part in a
// move rewrites it every few seconds, so the step is retried, not looked into
// (harness Run 9, finding 7: flip answered MV-O31 for a stale report, and an
// operator went looking for a process that was not there).
func offlineStaleRefusal(box string) *offlineRefusal {
	return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("the report from %s is not current. it reports again within a minute. try again then", box)}
}

// offlineRefusal is what a door refuses with: the status and the sentence.
type offlineRefusal struct {
	status int
	msg    string
}

func writeRefusal(w http.ResponseWriter, ref *offlineRefusal) {
	jsonError(w, ref.status, ref.msg)
}

// offlineConflict is the sentence for a transition's refusal. Never the
// error's own text: phase1Conflict passes only recorded sentences and
// answers a fixed one for anything else.
func offlineConflict(err error) string {
	msg := strings.TrimPrefix(err.Error(), provision.ErrMoveConflict.Error()+": ")
	switch {
	// The offline transitions' own recorded sentences (MV-O34 to MV-O43).
	// Each is a whole sentence or one format filled with a state, a box or a
	// region the record named, so it may go out as it is.
	case msg == "this is not an offline move",
		msg == "the record carries a different offline move",
		msg == "this is an offline move. use carlos move offline",
		strings.HasPrefix(msg, "a seal names the box that seals and the boxes that stopped the instance."),
		strings.HasPrefix(msg, "this offline move is ") && strings.Contains(msg, ". this step follows "),
		strings.HasPrefix(msg, "this offline move is ") && strings.Contains(msg, " is sealed by "),
		strings.HasPrefix(msg, "this offline move is ") && strings.HasSuffix(msg, ". it cannot be aborted once it is open"),
		strings.HasPrefix(msg, "this offline move is ") && strings.HasSuffix(msg, ". it finishes from finishing or reverting"),
		strings.HasPrefix(msg, "the record's region is ") && strings.HasSuffix(msg, " the instance changed under the move"),
		strings.HasPrefix(msg, "the instance is in ") && strings.Contains(msg, ", not "),
		// The move family's own recorded refusals, as phase1Conflict passes
		// them.
		msg == "no move in flight",
		msg == "malformed move start",
		strings.HasPrefix(msg, "a move to ") && strings.Contains(msg, " is already "),
		strings.HasPrefix(msg, "the instance is already in "),
		strings.HasPrefix(msg, "a ") && strings.HasSuffix(msg, " does not move. delete it and create one in the new region"):
		return msg
	}
	return "this move's state does not allow that. read its state and try again"
}

// ---- evidence ----

// offlineReport is one report read from the bucket, with the bytes its
// digest is taken over.
type offlineReport struct {
	moves.OfflineReport
	raw []byte
}

// offlineEvidence is every report of one move, as read, and its seal: the
// immutable record of what the sealer decided (moves.OfflineSealKey), nil
// when absent or not a valid seal of this move.
type offlineEvidence struct {
	reports []offlineReport
	seal    *offlineReport
}

func (e offlineEvidence) get(role, box string) (offlineReport, bool) {
	for _, r := range e.reports {
		if r.Role == role && r.Box == box {
			return r, true
		}
	}
	return offlineReport{}, false
}

func (e offlineEvidence) byRole(role string) []offlineReport {
	var out []offlineReport
	for _, r := range e.reports {
		if r.Role == role {
			out = append(out, r)
		}
	}
	return out
}

// readOfflineEvidence lists and reads every report of the move. A report
// that does not parse, fails moves.ValidateOfflineReport, or does not sit at
// the key its own fields name is not evidence and is skipped, which can only
// make a door refuse. A listing or a read that fails is the error.
func (s *Server) readOfflineEvidence(ctx context.Context, host, id string) (offlineEvidence, error) {
	prefix := moves.OfflineReportPrefix(host, id)
	if prefix == "" {
		return offlineEvidence{}, nil
	}
	keys, err := s.cfg.Releases.List(ctx, prefix)
	if err != nil {
		return offlineEvidence{}, err
	}
	var ev offlineEvidence
	sealKey := moves.OfflineSealKey(host, id)
	for _, key := range keys {
		if key == sealKey {
			continue // read below, as the seal and not as a report
		}
		raw, _, err := s.cfg.Releases.Get(ctx, key)
		if errors.Is(err, os.ErrNotExist) {
			continue
		}
		if err != nil {
			return offlineEvidence{}, err
		}
		var rep moves.OfflineReport
		if err := json.Unmarshal(raw, &rep); err != nil || moves.ValidateOfflineReport(rep) != nil ||
			rep.Host != host || rep.Op != id || moves.OfflineReportKey(rep.Host, rep.Op, rep.Role, rep.Box) != key {
			log.Printf("console: offline move %s: ignoring %s: not a report of this move", host, key)
			continue
		}
		ev.reports = append(ev.reports, offlineReport{OfflineReport: rep, raw: raw})
	}
	sort.Slice(ev.reports, func(i, j int) bool {
		if ev.reports[i].Role != ev.reports[j].Role {
			return ev.reports[i].Role < ev.reports[j].Role
		}
		return ev.reports[i].Box < ev.reports[j].Box
	})
	raw, _, err := s.cfg.Releases.Get(ctx, sealKey)
	switch {
	case errors.Is(err, os.ErrNotExist):
	case err != nil:
		return offlineEvidence{}, err
	default:
		var rep moves.OfflineReport
		if err := json.Unmarshal(raw, &rep); err != nil || moves.ValidateOfflineSeal(rep, host, id) != nil {
			log.Printf("console: offline move %s: ignoring %s: not the seal of this move", host, sealKey)
		} else {
			ev.seal = &offlineReport{OfflineReport: rep, raw: raw}
		}
	}
	return ev, nil
}

func offlineFresh(r moves.OfflineReport, now time.Time) bool {
	age := now.Sub(r.At)
	return !r.At.IsZero() && age <= offlineReportFresh && age >= -time.Minute
}

func offlineStopped(r moves.OfflineReport) bool { return r.Held && r.Fenced && r.Quiesced }

// offlineSealItems and offlineLandingItems are the one place the doors ask
// the manifest which items a report must carry OK: flip needs the sealer's
// report OK for every item the source side reports (Source "seal"), open the
// target's for every item the landing owns.
func offlineSealItems(r moves.OfflineReport) (bool, []string) {
	return r.SourceItemsOK()
}

func offlineLandingItems(r moves.OfflineReport) (bool, []string) {
	return r.ItemsOK(moves.OfflineOwnerLanding)
}

// offlineEvidenceMatches says whether the record's source.evidence names
// this report: the report's sha256 as hex, bare or after the report's key
// and one separator.
func offlineEvidenceMatches(evidence, key string, raw []byte) bool {
	sum := sha256.Sum256(raw)
	h := hex.EncodeToString(sum[:])
	tail, ok := strings.CutPrefix(evidence, key)
	if ok && tail != "" && strings.Contains("@: #", tail[:1]) {
		evidence = tail[1:]
	}
	return evidence == h || evidence == "sha256:"+h
}

// offlineFleet is the edges files and each box's status report, read
// together. Incomplete evidence is not an answer: ok is false.
type offlineFleet struct {
	snap    edgesSnapshot
	metrics map[string]*status.BoxMetrics
}

func (s *Server) readOfflineFleet(ctx context.Context) (offlineFleet, bool) {
	snap, err := s.liveEdges(ctx)
	if err != nil || !snap.EvidenceComplete() {
		log.Printf("console: offline move: the edges files are not complete evidence (error %v)", err)
		return offlineFleet{}, false
	}
	boxes, complete, err := status.LoadBoxesComplete(ctx, s.cfg.Releases)
	if err != nil || !complete {
		log.Printf("console: offline move: the boxes' status reports are not complete evidence (complete %v, error %v)", complete, err)
		return offlineFleet{}, false
	}
	f := offlineFleet{snap: snap, metrics: map[string]*status.BoxMetrics{}}
	for _, b := range boxes {
		f.metrics[b.Label] = b.Metrics
	}
	sort.Slice(f.snap.Files, func(i, j int) bool { return f.snap.Files[i].Box < f.snap.Files[j].Box })
	return f, true
}

// offlineCanHold says whether a box's published state lets it hold an
// ordinary instance: hibernation not configured off, and the shared pool not
// served off. Absent is unknown, and unknown could hold.
func offlineCanHold(f edgesFilePlacementView) bool {
	return (f.HibernationConfigured == nil || *f.HibernationConfigured) &&
		(f.ServesSharedPool == nil || *f.ServesSharedPool)
}

func offlineCapable(f edgesFilePlacementView) bool {
	return slices.Contains(f.Capabilities, offlineMoveCapability)
}

func offlineServes(f edgesFilePlacementView, host string) bool {
	_, ok := f.Hosts[host]
	return ok
}

// offlineClaimsOnBox is what a claims-on box has to show the start door
// (ruling 2026-10-20, point 3). Every field is published state; none is the
// operator's word. Census and SharedPool are pointers: nil is "the box does
// not say", and an unsaid fact refuses.
type offlineClaimsOnBox struct {
	HostRow    bool  // its edges file lists the host
	Census     *bool // its report says root's census.json is present
	SharedPool *bool // its edges file says whether it serves the shared pool
}

// ok: no row for the host, census.json reported present, and pools that do
// not match the record's placement (the record is in the shared pool, so a
// box that serves it matches).
func (b offlineClaimsOnBox) ok() bool {
	return !b.HostRow && b.Census != nil && *b.Census && b.SharedPool != nil && !*b.SharedPool
}

// offlineCensusReported is whether a box's status report says root's
// census.json is present. No report field carries it today, and the move
// adds none, so every claims-on box fails this condition until one does.
func offlineCensusReported(*status.BoxMetrics) *bool { return nil }

// offlineSourceBoxes are the edges files of the record's source placement
// that could hold the host, in box order.
func (f offlineFleet) boxesAt(code string) []edgesFilePlacementView {
	var out []edgesFilePlacementView
	for _, b := range f.snap.Files {
		if b.Placement == code && offlineCanHold(b) {
			out = append(out, b)
		}
	}
	return out
}

// ---- start ----

// offlineSelect is the start door's box rules, from the fleet read and
// nothing else: exactly one target box, at least one source box, the
// capability on each, no claims-on box in either region that could touch the
// host, an empty target, the deployment's secrets key on the target and a
// published data bucket there.
func (s *Server) offlineSelect(ctx context.Context, f offlineFleet, host, from, to string) *offlineRefusal {
	fromName, toName := s.offlineName(from), s.offlineName(to)
	src, tgt := f.boxesAt(from), f.boxesAt(to)
	if len(tgt) != 1 {
		return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s has %d boxes. an offline move needs exactly one box in the region it moves to", toName, len(tgt))}
	}
	if len(src) == 0 {
		return &offlineRefusal{status: http.StatusBadRequest, msg: "no eligible box in " + fromName + " publishes an edges file"}
	}
	for _, b := range append(slices.Clone(src), tgt...) {
		if !offlineCapable(b) {
			return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s cannot take part in an offline move. it runs an older release or has host claims on", b.Box)}
		}
	}
	// Ruling 2026-10-20, point 3: a box in either region that runs claims, or
	// whose report does not say it does not, must publish no row for the
	// host, report census.json present and serve pools the record's
	// placement does not match. A box with the capability has vouched for
	// itself (it publishes it only with claims off) unless its report says
	// claims are on.
	for _, b := range f.snap.Files {
		if b.Placement != from && b.Placement != to {
			continue
		}
		m := f.metrics[b.Box]
		if (offlineCapable(b) && (m == nil || !m.HostClaimEnabled)) || (m != nil && !m.HostClaimEnabled) {
			continue
		}
		if !(offlineClaimsOnBox{HostRow: offlineServes(b, host), Census: offlineCensusReported(m), SharedPool: b.ServesSharedPool}).ok() {
			return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s cannot take part in an offline move. it runs an older release or has host claims on", b.Box)}
		}
	}
	t := tgt[0]
	if offlineServes(t, host) {
		return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s already serves %s. an offline move needs the region it moves to to be empty of it", t.Box, host)}
	}
	if fp := s.expectedSecretsFP(); fp == "" || t.SecretsKeyFP != fp {
		return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s does not hold this deployment's secrets key", t.Box)}
	}
	if t.DataBucket == "" {
		caps, _, _ := movePlacements(ctx, s.cfg.Releases, s.moveHomeName())
		return &offlineRefusal{status: http.StatusBadRequest, msg: "region " + toName + " cannot take a move in this deployment. it publishes no staging bucket. Regions that can: " + s.stagingNames(caps)}
	}
	return nil
}

// offlineClaimEvidence refuses a host that has, anywhere, a claim: the
// record's acquisition, control/hosts/<host>.json, a domains claim carrying
// one, or the app's admission (MV-O29); and a deployment that has begun the
// claim backfill (MV-O20). A read that fails is a 502 and never an absence.
func (s *Server) offlineClaimEvidence(ctx context.Context, d offlineDoor) *offlineRefusal {
	has := &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s has a host claim. offline moves are only for hosts without one", d.host)}
	unreadable := &offlineRefusal{status: http.StatusBadGateway, msg: "could not read the host claim"}
	if d.rec.Acquisition != "" {
		return has
	}
	obs, err := hostclaim.Read(ctx, s.cfg.Releases, d.host)
	if err != nil {
		log.Printf("console: offline move %s: claim read: %v", d.host, err)
		return unreadable
	}
	if !obs.Absent {
		return has
	}
	if c, err := domains.GetClaim(ctx, s.cfg.Releases, d.host); err == nil && c.Acquisition != "" {
		return has
	} else if err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("console: offline move %s: domains claim read: %v", d.host, err)
		return unreadable
	}
	key, err := hostops.AdmissionKey(d.acct.Sqid, d.app)
	if err != nil {
		log.Printf("console: offline move %s: admission key: %v", d.host, err)
		return unreadable
	}
	if _, _, err := s.cfg.Releases.Get(ctx, key); err == nil {
		return has
	} else if !errors.Is(err, os.ErrNotExist) {
		log.Printf("console: offline move %s: admission read: %v", d.host, err)
		return unreadable
	}
	// A plan at done is allowed; one that has begun, or that cannot be
	// verified, is not.
	switch err := hostops.CheckCreateFreeze(ctx, s.cfg.Releases, s.cfg.ReleasePub); {
	case err == nil:
	case errors.Is(err, hostops.ErrPending), errors.Is(err, hostops.ErrIdentity):
		return &offlineRefusal{status: http.StatusConflict, msg: "offline moves are not available while host claims or Phase 1 moves are on"}
	default:
		log.Printf("console: offline move %s: backfill plan read: %v", d.host, err)
		return unreadable
	}
	return nil
}

// offlineBeginChecks is what start and stage check before either writes:
// the instance's class, its claims, the region asked for, the app's
// regions and the fleet's box rules (offlineSelect). It answers the region
// code and the fleet it read, or false once it has answered the request.
// A staged move is checked again in full when it starts: the fleet and the
// claims may have changed since the stage.
func (s *Server) offlineBeginChecks(w http.ResponseWriter, r *http.Request, d offlineDoor) (string, offlineFleet, bool) {
	rec, host := d.rec, d.host
	if s.refuseSeedMove(w, rec) {
		return "", offlineFleet{}, false
	}
	if rec.Role != "" {
		jsonError(w, http.StatusBadRequest, "a replica does not move. delete it and create one in the new region")
		return "", offlineFleet{}, false
	}
	if len(rec.Replicas) > 0 {
		jsonError(w, http.StatusBadRequest, "a home with replicas does not move. remove its replicas first")
		return "", offlineFleet{}, false
	}
	if rec.Steering != "" || rec.Unit != "" {
		jsonError(w, http.StatusBadRequest, fmt.Sprintf("%s is steered or runs under a unit. offline moves cannot carry either yet", host))
		return "", offlineFleet{}, false
	}
	ctx := r.Context()
	if ref := s.offlineClaimEvidence(ctx, d); ref != nil {
		writeRefusal(w, ref)
		return "", offlineFleet{}, false
	}
	var req struct {
		Region string `json:"region"`
	}
	if !readJSON(w, r, &req) {
		return "", offlineFleet{}, false
	}
	region := strings.TrimSpace(req.Region)
	to := s.homeRegionIn(region)
	if region == "" || !provision.ValidRegion(to) {
		jsonError(w, http.StatusBadRequest, "invalid region")
		return "", offlineFleet{}, false
	}
	if to == rec.Region {
		jsonError(w, http.StatusConflict, "the instance is already in "+region)
		return "", offlineFleet{}, false
	}
	m, err := provision.GetManifest(ctx, s.cfg.Releases, d.acct.Sqid, d.app)
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("console: offline move %s: manifest: %v", host, err)
		jsonError(w, http.StatusServiceUnavailable, "could not read the app's instances manifest. try again")
		return "", offlineFleet{}, false
	}
	if m != nil && !s.moveRegionAllowed(m.Regions, to) {
		jsonError(w, http.StatusBadRequest, "region "+region+" is not one of this app's regions: "+strings.Join(m.Regions, ", "))
		return "", offlineFleet{}, false
	}
	fleet, ok := s.readOfflineFleet(ctx)
	if !ok {
		jsonError(w, http.StatusConflict, "the fleet's edges files could not all be read. try again")
		return "", offlineFleet{}, false
	}
	if to == "" || rec.Region == "" {
		// A box that sets CARLOS_REGION to the home's name is home to this
		// console and not to itself, so it cannot take a move into home, and
		// out of home the placement it publishes is not the record's.
		for _, f := range fleet.snap.Files {
			if f.Placement != "" && homePlacementCode(s.moveHomeName(), f.Placement) == "" {
				jsonError(w, http.StatusConflict, f.Box+" sets CARLOS_REGION, so it cannot take a move into the home region")
				return "", offlineFleet{}, false
			}
		}
	}
	if ref := s.offlineSelect(ctx, fleet, host, rec.Region, to); ref != nil {
		writeRefusal(w, ref)
		return "", offlineFleet{}, false
	}
	return to, fleet, true
}

// offlineStagedFor is the refusal of a start or a stage that names another
// region than the one the instance is staged for (MV-O179).
func (s *Server) offlineStagedFor(host, to string) *offlineRefusal {
	return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s is staged for %s. abort the move to move it to another region", host, s.offlineName(to))}
}

// offlineStaged is the record's offline move when it is staging.
func offlineStaged(rec *provision.Record) (*provision.MoveState, bool) {
	if mv := rec.Move; mv.Offline() && mv.State == provision.OfflineStaging {
		return mv, true
	}
	return nil, false
}

// offlineStagerGone is MV-O190: the pinned stager of a staged move no
// longer publishes an edges file in the source region. Its disappearance
// is never proof that it stopped copying, so start and seal refuse.
func (s *Server) offlineStagerGone(box, host, from string) *offlineRefusal {
	return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s, which staged %s, no longer publishes an edges file in %s. bring it back or abort the move", box, host, from)}
}

// offlineStageCoords refuses a staged move whose target publishes other
// staging coordinates than the ones the stage fixed on the record (MV-O192,
// Astra round 2, finding 2): the staged copy is in the stored bucket, and
// nothing is ever retargeted. offlineSelect has already required exactly
// one target box.
func (s *Server) offlineStageCoords(f offlineFleet, mv *provision.MoveState, host string) *offlineRefusal {
	if mv.StagedAt == nil {
		return nil
	}
	tgt := f.boxesAt(mv.To)
	if len(tgt) != 1 || tgt[0].DataBucket != mv.StageBucket || tgt[0].DataBucketRegion != mv.StageRegion {
		return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s now publishes a different data bucket or region from what was recorded when %s was staged. abort the move and stage it again", s.offlineName(mv.To), host)}
	}
	return nil
}

// offlineStagerPresent refuses a staged move whose pinned stager is not in
// the source region's fleet now (MV-O190).
func (s *Server) offlineStagerPresent(f offlineFleet, mv *provision.MoveState, host string) *offlineRefusal {
	if mv.StagedAt == nil {
		return nil
	}
	for _, b := range f.boxesAt(mv.From) {
		if b.Box == mv.Stager {
			return nil
		}
	}
	return s.offlineStagerGone(mv.Stager, host, s.offlineName(mv.From))
}

func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Request) {
	d, ok := s.offlineGate(w, r)
	if !ok {
		return
	}
	to, fleet, ok := s.offlineBeginChecks(w, r, d)
	if !ok {
		return
	}
	// A staged instance starts the move it is staged for: staging to
	// stopping, the instance offline from now. The lock is the stage's.
	if mv, staged := offlineStaged(d.rec); staged {
		if to != mv.To {
			writeRefusal(w, s.offlineStagedFor(d.host, mv.To))
			return
		}
		if ref := s.offlineStagerPresent(fleet, mv, d.host); ref != nil {
			writeRefusal(w, ref)
			return
		}
		if ref := s.offlineStageCoords(fleet, mv, d.host); ref != nil {
			writeRefusal(w, ref)
			return
		}
		s.offlineAdvance(w, r, d, provision.OfflineStartStaged(mv.ID, d.addr, time.Now()))
		return
	}
	s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {
		return provision.OfflineStart(id, d.rec.Region, to, d.addr, time.Now())
	})
}

// handleAPIInstanceOfflineStage is the pre-copy door. On a record at rest it
// creates the move in offline-staging, checked exactly as start is and
// under the same one-move lock, and the source region's stager runs one
// round on the serial bump. On a record already staging for that region it
// moves staged_at forward (OfflineRestage), so the stager runs one more
// round; each round copies only what is new (the receipts).
func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Request) {
	d, ok := s.offlineGate(w, r)
	if !ok {
		return
	}
	to, fleet, ok := s.offlineBeginChecks(w, r, d)
	if !ok {
		return
	}
	if mv, staged := offlineStaged(d.rec); staged {
		if to != mv.To {
			writeRefusal(w, s.offlineStagedFor(d.host, mv.To))
			return
		}
		if ref := s.offlineStageCoords(fleet, mv, d.host); ref != nil {
			writeRefusal(w, ref)
			return
		}
		s.offlineAdvance(w, r, d, provision.OfflineRestage(mv.ID, time.Now()))
		return
	}
	for _, b := range append(fleet.boxesAt(d.rec.Region), fleet.boxesAt(to)...) {
		if !slices.Contains(b.Capabilities, offlineStageCapability) {
			jsonError(w, http.StatusConflict, fmt.Sprintf("%s cannot stage an offline move. it runs an older release", b.Box))
			return
		}
	}
	// The stager is pinned here, for the whole move: the one box that ever
	// writes this move's staging prefix, and the one whose cleanup an abort
	// waits for (Astra round 1, finding 2). offlineSelect has already
	// required at least one source box.
	var names []string
	for _, b := range fleet.boxesAt(d.rec.Region) {
		names = append(names, b.Box)
	}
	// The staging destination is fixed here too, before any copy (Astra
	// round 2, finding 2): the one target box's data bucket and region, as
	// offlineSelect found it. A target that publishes a source box's own
	// bucket is refused: the activator there would promote the staged copy
	// over the serving replica on its next wake (MV-O191).
	tgt := fleet.boxesAt(to)[0]
	for _, b := range fleet.boxesAt(d.rec.Region) {
		if b.DataBucket == tgt.DataBucket {
			jsonError(w, http.StatusConflict, fmt.Sprintf("%s publishes the same data bucket as %s. a staged copy there would replace the serving one", s.offlineName(to), s.offlineName(d.rec.Region)))
			return
		}
	}
	st := provision.StageTarget{Stager: moves.OfflineStager(names), Bucket: tgt.DataBucket, Region: tgt.DataBucketRegion}
	s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {
		return provision.OfflineStage(id, d.rec.Region, to, d.addr, st, time.Now())
	})
}

// offlineBegin mints the move's id and writes the transition begin makes
// with it (OfflineStart or OfflineStage) under the one-move lock, then
// bumps the serial and answers 202 with the status view.
func (s *Server) offlineBegin(w http.ResponseWriter, r *http.Request, d offlineDoor, begin func(id string) func(*provision.Record) error) {
	ctx := r.Context()
	host := d.host
	id, err := moves.NewMoveID(rand.Reader)
	if err != nil {
		log.Printf("console: offline move %s: move id: %v", host, err)
		jsonError(w, http.StatusBadGateway, "could not mint a move id")
		return
	}
	start := begin(id)
	probe := *d.rec
	if err := start(&probe); err != nil {
		jsonError(w, http.StatusConflict, offlineConflict(err))
		return
	}
	// One offline move at a time: the lock is taken before the record is
	// written, and released if the write fails.
	if ref := s.offlineTakeLock(ctx, d, id); ref != nil {
		writeRefusal(w, ref)
		return
	}
	// The write is bounded well under offlineLockGrace, so the lock cannot be
	// judged stale while it is still in flight.
	wctx, cancel := context.WithTimeout(ctx, offlineStartWrite)
	defer cancel()
	got, err := s.offlineApply(wctx, d, start)
	if err != nil {
		// A failed write does not prove nothing was written (a commit whose
		// reply was lost). The lock is never released here: if the record
		// carries this move the start succeeded, and if it does not the
		// lock goes stale and frees itself.
		rctx, rcancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
		defer rcancel()
		if cur, rerr := provision.GetRecord(rctx, s.cfg.Releases, d.acct.Sqid, d.app, host); rerr == nil && cur.Move.Offline() && cur.Move.ID == id {
			got, err = cur, nil
		}
	}
	if err != nil {
		s.offlineWriteFailed(w, d, err)
		return
	}
	s.bumpSerial(ctx)
	writeJSON(w, http.StatusAccepted, s.offlineViewOf(ctx, got))
}

// offlineStartWrite bounds the start door's record write.
const offlineStartWrite = 30 * time.Second

// offlineApply is one transition through the record writer.
func (s *Server) offlineApply(ctx context.Context, d offlineDoor, mut func(*provision.Record) error) (*provision.Record, error) {
	return provision.SetRecordMove(ctx, s.cfg.Releases, d.acct.Sqid, d.app, d.host, mut)
}

// offlineWriteFailed answers a transition's failure: a refusal with its
// recorded sentence, anything else as unwritten.
func (s *Server) offlineWriteFailed(w http.ResponseWriter, d offlineDoor, err error) {
	if errors.Is(err, provision.ErrMoveConflict) {
		jsonError(w, http.StatusConflict, offlineConflict(err))
		return
	}
	log.Printf("console: offline move %s/%s/%s: %v", d.acct.Sqid, d.app, d.host, err)
	jsonError(w, http.StatusBadGateway, "could not write the move. try again")
}

// offlineWrite applies one transition and bumps the serial, or answers the
// refusal itself. When the move it leaves is cleared (finish, or the finish
// that ends an abort) it releases the lock.
func (s *Server) offlineWrite(w http.ResponseWriter, r *http.Request, d offlineDoor, mut func(*provision.Record) error) (*provision.Record, bool) {
	ctx := r.Context()
	rec, err := s.offlineApply(ctx, d, mut)
	if err != nil {
		s.offlineWriteFailed(w, d, err)
		return nil, false
	}
	s.bumpSerial(ctx)
	if rec.Move == nil && d.rec.Move != nil {
		s.offlineReleaseLock(ctx, d.host, d.rec.Move.ID)
	}
	return rec, true
}

// offlineAdvance applies one transition and answers 202 with the status
// view of the record it left.
func (s *Server) offlineAdvance(w http.ResponseWriter, r *http.Request, d offlineDoor, mut func(*provision.Record) error) {
	if rec, ok := s.offlineWrite(w, r, d, mut); ok {
		writeJSON(w, http.StatusAccepted, s.offlineViewOf(r.Context(), rec))
	}
}

// offlineLockGrace is how long a lock is believed whatever its record says:
// the start door takes the lock, then writes the record.
const offlineLockGrace = 2 * time.Minute

// offlineTakeLock takes control/offline-moves/active.json for this host and
// move, create-only. A lock already there is live while the record it names
// still carries that move, or is younger than offlineLockGrace (MV-O30 naming
// its host); otherwise it is stale (a crashed start, a failed release) and is
// replaced under its etag. A
// lock that cannot be read is not guessed at.
func (s *Server) offlineTakeLock(ctx context.Context, d offlineDoor, id string) *offlineRefusal {
	unreadable := &offlineRefusal{status: http.StatusBadGateway, msg: "could not read the instance"}
	body, err := json.Marshal(moves.OfflineActive{Host: d.host, Op: id, Account: d.acct.Sqid, App: d.app, At: time.Now().UTC()})
	if err != nil {
		return unreadable
	}
	_, err = s.cfg.Releases.Put(ctx, moves.OfflineActiveKey, body, "")
	if err == nil {
		return nil
	}
	if !errors.Is(err, release.ErrCASConflict) {
		log.Printf("console: offline move %s: taking the lock: %v", d.host, err)
		return unreadable
	}
	raw, etag, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey)
	var held moves.OfflineActive
	if err != nil || json.Unmarshal(raw, &held) != nil || held.Host == "" {
		log.Printf("console: offline move %s: reading the lock: %v", d.host, err)
		return unreadable
	}
	busy := &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s is moving offline. move one instance at a time", held.Host)}
	// A lock is taken before its record is written, so a record without the
	// move is only a stale lock once the holder has had time to write it.
	if time.Since(held.At) < offlineLockGrace {
		return busy
	}
	other, err := provision.GetRecord(ctx, s.cfg.Releases, held.Account, held.App, held.Host)
	switch {
	case errors.Is(err, os.ErrNotExist):
	case err != nil:
		log.Printf("console: offline move %s: reading the lock holder %s: %v", d.host, held.Host, err)
		return unreadable
	case other.Move.Offline() && other.Move.ID == held.Op:
		return busy
	}
	if _, err := s.cfg.Releases.Put(ctx, moves.OfflineActiveKey, body, etag); err != nil {
		if errors.Is(err, release.ErrCASConflict) {
			return busy
		}
		log.Printf("console: offline move %s: replacing a stale lock: %v", d.host, err)
		return unreadable
	}
	return nil
}

// offlineReleaseLock deletes the lock under the etag it read, when it is
// this move's. A failure is logged and left: the next start finds the lock
// stale (its record no longer carries the move) and replaces it.
func (s *Server) offlineReleaseLock(ctx context.Context, host, id string) {
	ctx = context.WithoutCancel(ctx)
	raw, etag, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey)
	if errors.Is(err, os.ErrNotExist) {
		return
	}
	var held moves.OfflineActive
	if err != nil || json.Unmarshal(raw, &held) != nil {
		log.Printf("console: offline move %s: reading the lock to release it: %v", host, err)
		return
	}
	if held.Host != host || held.Op != id {
		return
	}
	if err := release.DeleteIfMatch(ctx, s.cfg.Releases, moves.OfflineActiveKey, etag); err != nil {
		log.Printf("console: offline move %s: releasing the lock: %v", host, err)
	}
}

// ---- the step each state waits for ----

// The verbs.
const (
	offlineVerbStart  = "start"
	offlineVerbSeal   = "seal"
	offlineVerbFlip   = "flip"
	offlineVerbOpen   = "open"
	offlineVerbFinish = "finish"
)

// offlineVerbOf is the one door a state is waiting on. sealing and landing
// wait for a box, not for the operator.
func offlineVerbOf(state string) string {
	switch state {
	case provision.OfflineStaging:
		return offlineVerbStart
	case provision.OfflineStopping:
		return offlineVerbSeal
	case provision.OfflineSealed:
		return offlineVerbFlip
	case provision.OfflineLanded:
		return offlineVerbOpen
	case provision.OfflineOpen, provision.OfflineFinishing, provision.OfflineReverting:
		return offlineVerbFinish
	}
	return ""
}

// offlineStep is a door's checks passed: the transition to write.
type offlineStep struct {
	mut func(*provision.Record) error
}

// offlineOpenMut is the open transition. The one place the door names it.
func offlineOpenMut(id string, now time.Time) func(*provision.Record) error {
	return provision.OfflineOpenTransition(id, now)
}

// offlineStepFor runs the checks of the verb the move's state waits for,
// from the evidence. A refusal is the sentence the door answers and the
// status view shows as what the move waits for.
func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provision.MoveState, ev offlineEvidence, now time.Time) (offlineStep, *offlineRefusal) {
	host, from := d.host, s.offlineName(mv.From)
	switch mv.State {
	case provision.OfflineStaging:
		// What start would check of the fleet now; its request checks
		// (the class, the claims, the region) are the start door's.
		fleet, ok := s.readOfflineFleet(ctx)
		if !ok {
			return offlineStep{}, &offlineRefusal{status: http.StatusConflict, msg: "the fleet's edges files could not all be read. try again"}
		}
		if ref := s.offlineSelect(ctx, fleet, host, mv.From, mv.To); ref != nil {
			return offlineStep{}, ref
		}
		if ref := s.offlineStagerPresent(fleet, mv, host); ref != nil {
			return offlineStep{}, ref
		}
		if ref := s.offlineStageCoords(fleet, mv, host); ref != nil {
			return offlineStep{}, ref
		}
		return offlineStep{mut: provision.OfflineStartStaged(mv.ID, d.addr, now)}, nil
	case provision.OfflineStopping:
		fleet, ok := s.readOfflineFleet(ctx)
		if !ok {
			return offlineStep{}, &offlineRefusal{status: http.StatusConflict, msg: "the fleet's edges files could not all be read. try again"}
		}
		sealer, boxes, ref := s.offlineSealCheck(fleet, ev, mv, host, from, now)
		if ref != nil {
			return offlineStep{}, ref
		}
		return offlineStep{mut: provision.OfflineSeal(mv.ID, sealer, boxes)}, nil
	case provision.OfflineSealed:
		if ref := s.offlineFlipCheck(ev, mv, host, from, now); ref != nil {
			return offlineStep{}, ref
		}
		return offlineStep{mut: provision.OfflineFlip(mv.ID)}, nil
	case provision.OfflineLanded:
		if ref := s.offlineOpenCheck(ev, mv, host, from, now); ref != nil {
			return offlineStep{}, ref
		}
		return offlineStep{mut: offlineOpenMut(mv.ID, now)}, nil
	case provision.OfflineOpen:
		if ref := offlineFinishCheck(ev, mv, host, moves.OfflinePhaseOpened, "", moves.OfflinePhaseParked, mv.Boxes); ref != nil {
			return offlineStep{}, ref
		}
		return offlineStep{mut: provision.OfflineFinishingTransition(mv.ID)}, nil
	case provision.OfflineFinishing:
		if ref := offlineFinishCheck(ev, mv, host, "", moves.OfflinePhaseLifted, moves.OfflinePhaseGone, mv.Boxes); ref != nil {
			return offlineStep{}, ref
		}
		return offlineStep{mut: provision.OfflineFinish(mv.ID, now)}, nil
	case provision.OfflineReverting:
		// Every box the seal named, and the sealer. A move aborted before
		// the seal named neither: the boxes are the source region's as the
		// edges files publish them now, and there is no sealer (it never
		// took a tombstone).
		boxes, sealerPhase := mv.Boxes, moves.OfflinePhaseReverted
		if mv.Sealer == "" {
			fleet, ok := s.readOfflineFleet(ctx)
			if !ok {
				return offlineStep{}, &offlineRefusal{status: http.StatusConflict, msg: "the fleet's edges files could not all be read. try again"}
			}
			boxes, sealerPhase = nil, ""
			for _, b := range fleet.boxesAt(mv.From) {
				boxes = append(boxes, b.Box)
			}
		}
		if ref := offlineFinishCheck(ev, mv, host, "", sealerPhase, moves.OfflinePhaseReverted, boxes); ref != nil {
			return offlineStep{}, ref
		}
		// A staged move also waits for its pinned stager to say the staged
		// copy is gone, whatever its last round reported (Astra round 1,
		// findings 1 and 2).
		if mv.StagedAt != nil {
			if rep, ok := ev.get(moves.OfflineRoleStager, mv.Stager); !ok || rep.Phase != moves.OfflinePhaseReverted {
				return offlineStep{}, offlineStateRefusal(mv.State, host)
			}
		}
		return offlineStep{mut: provision.OfflineFinish(mv.ID, now)}, nil
	}
	return offlineStep{}, offlineStateRefusal(mv.State, host)
}

// offlineSealCheck: every source box reports this move held, fenced and
// quiesced, and its edges file no longer lists the host. It names the sealer
// and the boxes the later doors wait for.
func (s *Server) offlineSealCheck(f offlineFleet, ev offlineEvidence, mv *provision.MoveState, host, from string, now time.Time) (sealer string, boxes []string, ref *offlineRefusal) {
	src := f.boxesAt(mv.From)
	if len(src) == 0 {
		return "", nil, &offlineRefusal{status: http.StatusBadRequest, msg: "no eligible box in " + from + " publishes an edges file"}
	}
	// A staged move's pinned stager is part of the barrier whatever the
	// fleet says (Astra round 2, finding 1): it must still be in the
	// source region's fleet (so it is one of the boxes below, and of the
	// boxes the seal names), and it must have said, after the move left
	// staging and under its pass lock, that no stage round is in flight
	// and none will start (its stager report, phase stopped, fresh). The
	// staging destination must still be the one the stage fixed.
	if mv.StagedAt != nil {
		if ref := s.offlineStagerPresent(f, mv, host); ref != nil {
			return "", nil, ref
		}
		if ref := s.offlineStageCoords(f, mv, host); ref != nil {
			return "", nil, ref
		}
		ack, ok := ev.get(moves.OfflineRoleStager, mv.Stager)
		if ok && ack.Phase == moves.OfflinePhaseStopped && !offlineFresh(ack.OfflineReport, now) {
			return "", nil, offlineStaleRefusal(mv.Stager)
		}
		if !ok || ack.Phase != moves.OfflinePhaseStopped {
			return "", nil, s.offlineWaitRefusal(mv.Stager, from, host)
		}
	}
	var data []string
	for _, b := range src {
		rep, ok := ev.get(moves.OfflineRoleSource, b.Box)
		if ok && !offlineFresh(rep.OfflineReport, now) {
			return "", nil, offlineStaleRefusal(b.Box)
		}
		if !ok || !offlineStopped(rep.OfflineReport) || offlineServes(b, host) {
			return "", nil, s.offlineWaitRefusal(b.Box, from, host)
		}
		boxes = append(boxes, b.Box)
		if rep.LocalDB || rep.Files || rep.Env {
			data = append(data, b.Box)
		}
	}
	for _, b := range boxes {
		if rep, _ := ev.get(moves.OfflineRoleSource, b); rep.Sidecar {
			return "", nil, &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s keeps a sidecar file for %s. instances with sidecars cannot move yet", b, host)}
		}
	}
	if len(data) > 1 {
		return "", nil, &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s keeps local data on more than one box: %s. remove the stale copy first", host, strings.Join(data, ", "))}
	}
	sealer = boxes[0]
	if len(data) == 1 {
		sealer = data[0]
	}
	return sealer, boxes, nil
}

// offlineSealHolds says the record's source.evidence names the seal on disk,
// byte for byte. The seal is written once, so the hash holds for the whole
// move whatever the sealer's living report does each pass.
func offlineSealHolds(ev offlineEvidence, mv *provision.MoveState, host string) bool {
	return ev.seal != nil && ev.seal.Box == mv.Sealer && mv.Source != nil &&
		offlineEvidenceMatches(mv.Source.Evidence, moves.OfflineSealKey(host, mv.ID), ev.seal.raw)
}

// offlineFlipCheck: the seal is the one the record names and every item the
// source side reports is OK in it; every source box has stopped and left no
// process or layout journal, on a report this fresh.
func (s *Server) offlineFlipCheck(ev offlineEvidence, mv *provision.MoveState, host, from string, now time.Time) *offlineRefusal {
	if !offlineSealHolds(ev, mv, host) {
		return offlineStateRefusal(mv.State, host)
	}
	if items, _ := offlineSealItems(ev.seal.OfflineReport); !items {
		return offlineStateRefusal(mv.State, host)
	}
	for _, box := range mv.Boxes {
		rep, ok := ev.get(moves.OfflineRoleSource, box)
		if !ok {
			return s.offlineWaitRefusal(box, from, host)
		}
		if !offlineFresh(rep.OfflineReport, now) {
			return offlineStaleRefusal(box)
		}
		if !offlineStopped(rep.OfflineReport) {
			return s.offlineWaitRefusal(box, from, host)
		}
		if rep.Procs != 0 || rep.LayoutJournal {
			return processRefusal(box, host)
		}
	}
	return nil
}

func processRefusal(box, host string) *offlineRefusal {
	return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s still runs a process for %s. wait for it to stop, then flip", box, host)}
}

// offlineOpenCheck is the re-check of the stop: the target's landing report
// has every landing item OK, the seal is still the one the record names, the
// sealer's living report is from this pass and holds the tombstone for this
// move, and every source box is still held, fenced and quiesced.
func (s *Server) offlineOpenCheck(ev offlineEvidence, mv *provision.MoveState, host, from string, now time.Time) *offlineRefusal {
	tgt := ev.byRole(moves.OfflineRoleTarget)
	if len(tgt) != 1 {
		return offlineStateRefusal(mv.State, host)
	}
	if items, _ := offlineLandingItems(tgt[0].OfflineReport); !items {
		return offlineStateRefusal(mv.State, host)
	}
	if !offlineSealHolds(ev, mv, host) {
		return offlineStateRefusal(mv.State, host)
	}
	seal, ok := ev.get(moves.OfflineRoleSealer, mv.Sealer)
	if ok && !offlineFresh(seal.OfflineReport, now) {
		return offlineStaleRefusal(mv.Sealer)
	}
	if !ok || seal.TombstoneOwner != moves.TombstoneOwner(mv.ID) || !seal.TombstoneUntil.After(now) {
		return offlineStateRefusal(mv.State, host)
	}
	for _, box := range mv.Boxes {
		rep, ok := ev.get(moves.OfflineRoleSource, box)
		if ok && !offlineFresh(rep.OfflineReport, now) {
			return offlineStaleRefusal(box)
		}
		if !ok || !offlineStopped(rep.OfflineReport) {
			return s.offlineWaitRefusal(box, from, host)
		}
	}
	return nil
}

// offlineFinishCheck: the target's phase (when targetPhase is set), the
// sealer's phase (when sealerPhase is set) and every source box's phase,
// each as the box's own report says. Otherwise MV-O25. Open to finishing
// wants the target opened and every source parked; finishing to cleared the
// sealer lifted and every source gone; reverting to cleared the sealer and
// every source reverted (the target is not asked: a target box that died
// after the flip must not hold the abort).
func offlineFinishCheck(ev offlineEvidence, mv *provision.MoveState, host, targetPhase, sealerPhase, sourcePhase string, boxes []string) *offlineRefusal {
	refuse := offlineStateRefusal(mv.State, host)
	if targetPhase != "" {
		tgt := ev.byRole(moves.OfflineRoleTarget)
		if len(tgt) != 1 || tgt[0].Phase != targetPhase {
			return refuse
		}
	}
	if sealerPhase != "" {
		if rep, ok := ev.get(moves.OfflineRoleSealer, mv.Sealer); !ok || rep.Phase != sealerPhase {
			return refuse
		}
	}
	if len(boxes) == 0 {
		return refuse
	}
	for _, box := range boxes {
		rep, ok := ev.get(moves.OfflineRoleSource, box)
		if !ok || rep.Phase != sourcePhase || !offlineRetiredTerminal(rep.Retired) {
			return refuse
		}
	}
	return nil
}

// offlineRetiredTerminal says every path a parked report lists is settled:
// parked, absent, or left with the reason it could not be parked (a
// root-owned directory, a view mount). A left path is terminal; the move
// goes on and the status view shows it. An unknown action, or a left with no
// reason, is not a settled path.
func offlineRetiredTerminal(retired []moves.RetiredPath) bool {
	for _, p := range retired {
		switch p.Action {
		case moves.RetiredParked, moves.RetiredAbsent:
		case moves.RetiredLeft:
			if p.Reason == "" {
				return false
			}
		default:
			return false
		}
		if p.Path == "" {
			return false
		}
	}
	return true
}

// offlineVerbDoor is the shape of seal, flip, open and finish: the gate, an
// offline move in flight, the state the verb follows, the evidence, the
// verb's checks, one transition.
func (s *Server) offlineVerbDoor(verb string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		d, ok := s.offlineGate(w, r)
		if !ok {
			return
		}
		mv, ok := s.offlineInFlight(w, d)
		if !ok {
			return
		}
		if offlineVerbOf(mv.State) != verb {
			writeRefusal(w, offlineStateRefusal(mv.State, d.host))
			return
		}
		ctx := r.Context()
		ev, err := s.readOfflineEvidence(ctx, d.host, mv.ID)
		if err != nil {
			log.Printf("console: offline move %s: reports: %v", d.host, err)
			jsonError(w, http.StatusBadGateway, "could not read the move's reports. try again")
			return
		}
		step, ref := s.offlineStepFor(ctx, d, mv, ev, time.Now())
		if ref != nil {
			writeRefusal(w, ref)
			return
		}
		s.offlineAdvance(w, r, d, step.mut)
	}
}

func (s *Server) handleAPIInstanceOfflineAbort(w http.ResponseWriter, r *http.Request) {
	d, ok := s.offlineGate(w, r)
	if !ok {
		return
	}
	mv, ok := s.offlineInFlight(w, d)
	if !ok {
		return
	}
	switch mv.State {
	case provision.OfflineOpen, provision.OfflineFinishing:
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s is already served from %s. move it back with a new move", d.host, s.offlineName(mv.To)))
		return
	case provision.OfflineReverting:
		// Already reverting: nothing to write.
		writeJSON(w, http.StatusAccepted, s.offlineViewOf(r.Context(), d.rec))
		return
	}
	s.offlineAdvance(w, r, d, provision.OfflineAbort(mv.ID))
}

// ---- status ----

// offlineReportView is a report as the status view carries it: every field
// the box wrote, and whether it is current enough for a door to believe.
type offlineReportView struct {
	moves.OfflineReport
	Fresh bool `json:"fresh"`
}

// offlineLeftView is a path a source box could not park and left in place,
// with the reason it gave (the parked report's inventory).
type offlineLeftView struct {
	Box    string `json:"box"`
	Path   string `json:"path"`
	Reason string `json:"reason"`
}

// offlineItemView is one row of the manifest with the latest check of it.
// Status is ok, waiting, failed or unchecked (no report decides the item).
type offlineItemView struct {
	ID          string `json:"id"`
	Item        string `json:"item"`
	Disposition string `json:"disposition"`
	Owner       string `json:"owner"`
	Source      string `json:"source,omitempty"`
	Status      string `json:"status"`
	Detail      string `json:"detail,omitempty"`
}

// offlineView is the status answer, and the 202 every mutating door gives.
// State is "none" when no offline move is in flight (Last then carries the
// outcome of the one that ended). Next is the verb that moves the move on
// right now; empty while a box has work to do. Waiting is the sentence that
// verb would be refused with. CanAbort says abort is legal. Failed says the
// sealer or the target reported it could not do its step: abort is the way on.
type offlineView struct {
	Host        string   `json:"host"`
	State       string   `json:"state"`
	ID          string   `json:"id,omitempty"`
	From        string   `json:"from,omitempty"`
	To          string   `json:"to,omitempty"`
	Sealer      string   `json:"sealer,omitempty"`
	Stager      string   `json:"stager,omitempty"`
	Boxes       []string `json:"boxes,omitempty"`
	RequestedBy string   `json:"requested_by,omitempty"`
	RequestedAt int64    `json:"requested_at,omitempty"`
	UpdatedAt   int64    `json:"updated_at,omitempty"`
	OpenedAt    int64    `json:"opened_at,omitempty"`
	Next        string   `json:"next,omitempty"`
	Waiting     string   `json:"waiting,omitempty"`
	CanAbort    bool     `json:"can_abort,omitempty"`
	// NeverOffline: the move was staged and never started, so its
	// instance has kept serving throughout (an abort from staging).
	NeverOffline bool `json:"never_offline,omitempty"`
	// StagedAt is the record's staged_at in Unix nanoseconds: the latest
	// stage request. Stage.Request equals it once the stager has answered
	// that request.
	StagedAt int64                  `json:"staged_at,omitempty"`
	Failed   bool                   `json:"failed,omitempty"`
	Left     []offlineLeftView      `json:"left,omitempty"`
	Stage    *offlineStageView      `json:"stage,omitempty"`
	Reports  []offlineReportView    `json:"reports,omitempty"`
	Items    []offlineItemView      `json:"items,omitempty"`
	Last     *provision.MoveOutcome `json:"last,omitempty"`
}

// offlineStageView is the pre-copy's progress, from the latest stager
// report: when its last round ended, the replica as that round listed it,
// how much of it the target's staging prefix holds, and what the round
// copied. Phase failed carries the box's own diagnostic in Error. Stager
// reports are rewritten each moves pass, not every few seconds, so their
// age is shown and never judged.
type offlineStageView struct {
	Box            string `json:"box"`
	Phase          string `json:"phase"`
	At             int64  `json:"at"`
	ReplicaObjects int    `json:"replica_objects"`
	ReplicaBytes   int64  `json:"replica_bytes"`
	StagedObjects  int    `json:"staged_objects"`
	StagedBytes    int64  `json:"staged_bytes"`
	RoundObjects   int    `json:"round_objects"`
	RoundBytes     int64  `json:"round_bytes"`
	Error          string `json:"error,omitempty"`
	// Request is the staged_at (Unix nanoseconds) the round answered.
	Request int64 `json:"request,omitempty"`
}

// offlineStageOf is the stage view of the newest stager report, nil when no
// box has staged this move.
func offlineStageOf(ev offlineEvidence) *offlineStageView {
	var latest *offlineReport
	for _, r := range ev.byRole(moves.OfflineRoleStager) {
		if latest == nil || r.At.After(latest.At) {
			latest = &r
		}
	}
	if latest == nil {
		return nil
	}
	r := latest.OfflineReport
	v := &offlineStageView{Box: r.Box, Phase: r.Phase, At: r.At.Unix(),
		ReplicaObjects: r.ReplicaObjects, ReplicaBytes: r.ReplicaBytes, StagedObjects: r.StagedObjects, StagedBytes: r.StagedBytes,
		RoundObjects: r.RoundObjects, RoundBytes: r.RoundBytes, Error: r.StageError}
	if !r.StageRequest.IsZero() {
		v.Request = r.StageRequest.UnixNano()
	}
	return v
}

// offlineViewOf builds the status view of a record. Evidence it cannot read
// leaves the reports out and says so in Waiting; it never fails the answer.
func (s *Server) offlineViewOf(ctx context.Context, rec *provision.Record) offlineView {
	v := offlineView{Host: rec.Host, State: "none", Last: rec.LastMove}
	mv := rec.Move
	if mv == nil || !mv.Offline() {
		return v
	}
	now := time.Now()
	v.State, v.ID = mv.State, mv.ID
	v.From, v.To = s.offlineName(mv.From), s.offlineName(mv.To)
	v.Sealer, v.Boxes, v.Stager = mv.Sealer, mv.Boxes, mv.Stager
	v.NeverOffline = !mv.StartedOffline()
	if mv.StagedAt != nil {
		v.StagedAt = mv.StagedAt.UnixNano()
	}
	v.RequestedBy, v.RequestedAt, v.UpdatedAt = mv.RequestedBy, mv.RequestedAt.Unix(), mv.UpdatedAt.Unix()
	if mv.OpenedAt != nil {
		v.OpenedAt = mv.OpenedAt.Unix()
	}
	switch mv.State {
	case provision.OfflineStaging, provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded:
		v.CanAbort = true
	}
	ev, err := s.readOfflineEvidence(ctx, rec.Host, mv.ID)
	if err != nil {
		log.Printf("console: offline move %s: reading the reports: %v", rec.Host, err)
		v.Waiting = "could not read the move's reports. try again"
		return v
	}
	for _, r := range ev.reports {
		v.Reports = append(v.Reports, offlineReportView{OfflineReport: r.OfflineReport, Fresh: offlineFresh(r.OfflineReport, now)})
	}
	v.Items = offlineItems(ev, mv)
	v.Stage = offlineStageOf(ev)
	for _, r := range ev.byRole(moves.OfflineRoleSource) {
		for _, p := range r.Retired {
			if p.Action == moves.RetiredLeft {
				v.Left = append(v.Left, offlineLeftView{Box: r.Box, Path: p.Path, Reason: p.Reason})
			}
		}
	}
	if verb := offlineVerbOf(mv.State); verb != "" {
		d := offlineDoor{host: rec.Host}
		if _, ref := s.offlineStepFor(ctx, d, mv, ev, now); ref != nil {
			v.Waiting = ref.msg
		} else {
			v.Next = verb
		}
	}
	// A box that says it failed (the sealer, or the target that cannot
	// land) means abort is the way on. The CLI says it with MV-O6.
	if v.CanAbort {
		for _, r := range ev.reports {
			if r.Phase == moves.OfflinePhaseFailed && (r.Role == moves.OfflineRoleSealer || r.Role == moves.OfflineRoleTarget) {
				v.Failed = true
			}
		}
	}
	return v
}

// offlineItems is the manifest with each item's latest check, from the
// report of the step that owns it: the sealer's for a seal item, the
// target's landing report for a landing item. A check present and not OK is
// failed when the report that holds it says failed, else waiting (the box
// retries it). An item no report decides is unchecked, and a door-owned one
// was decided at start.
func offlineItems(ev offlineEvidence, mv *provision.MoveState) []offlineItemView {
	owners := map[string][]offlineReport{moves.OfflineOwnerLanding: ev.byRole(moves.OfflineRoleTarget)}
	if r, ok := ev.get(moves.OfflineRoleSealer, mv.Sealer); ok {
		owners[moves.OfflineOwnerSeal] = []offlineReport{r}
	}
	if ev.seal != nil {
		// The seal is what the sealer decided; the living report is the
		// latest word on the same items, and is asked first.
		owners[moves.OfflineOwnerSeal] = append(owners[moves.OfflineOwnerSeal], *ev.seal)
	}
	out := make([]offlineItemView, 0, len(moves.OfflineManifest))
	for _, it := range moves.OfflineManifest {
		row := offlineItemView{ID: it.ID, Item: it.Item, Disposition: it.Disposition, Owner: it.Owner, Source: it.Source, Status: "waiting"}
		switch it.Owner {
		case moves.OfflineOwnerDoor:
			row.Status = "ok"
		case moves.OfflineOwnerNone:
			row.Status = "unchecked"
		default:
			for _, r := range owners[it.Owner] {
				c, found := r.Items[it.ID]
				if !found {
					continue
				}
				row.Detail = c.Reason
				switch {
				case c.OK:
					row.Status = "ok"
				case r.Phase == moves.OfflinePhaseFailed:
					row.Status = "failed"
				}
				break
			}
		}
		out = append(out, row)
	}
	return out
}

func (s *Server) handleAPIInstanceOfflineStatus(w http.ResponseWriter, r *http.Request) {
	d, ok := s.offlineGate(w, r)
	if !ok {
		return
	}
	writeJSON(w, http.StatusOK, s.offlineViewOf(r.Context(), d.rec))
}