package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"maps"
	"math"
	"net/http"
	"net/url"
	"os"
	"slices"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/awsinv"
	"github.com/carlosframework/platform/internal/hostfence"
	"github.com/carlosframework/platform/internal/hostresolve"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
)

// `carlos move` (spec §12, moves Phase 1 Plan 6 Task 1a): request a move of
// one instance to another region, read its state, cut, abort, and retry a
// stuck one. Operators only, through the console, with
// CARLOS_MOVES_PHASE1 on there. The verb refuses to send anything to a
// console whose move answer lacks the Phase 1 marker (preflight L4), so it
// can never start a legacy move.
//
// Waiting is the default. The wait polls the flat view every two seconds and
// reads detail=1 only when the state or the phase changes (L19). A 429, on a
// read or on a mutation, backs off to ten seconds, is said once, and never
// ends the command: the doors' rate limit answers before any handler, so the
// same call is sent again. The outcome is read only from `last`, bound to the
// id the console gave the move (L2, Review Focus 1), and its regions are that
// move's own, bound from its own answers; a region the wait never learnt is
// never guessed. Exit codes (L24): 0 done, or a command accepted with
// --no-wait (which reads nothing after the acceptance); 3 rolled back or
// aborted; 4 stuck, rolling back past its release deadline, relocating,
// ready for a manual cut, or an outcome not recorded; 5 refused (by the
// console, or before anything was sent); 1 the console not reached, its
// answer unreadable, or a 5xx; 2 a usage error.
//
// What the operator reads is this file's own copy, through moveSay (the copy
// gate's sink, movesLocalSinks), or text the console wrote, as serverText
// through saySpoken. serverText is made only in roundTrip, from the body of a
// response the HTTP client returned (movesServerText). Go's own errors are
// never shown. The flag help is said by moveFlagHelp (movesSinkAllow).

const (
	moveExitDone     = 0
	moveExitError    = 1
	moveExitUsage    = 2
	moveExitRolled   = 3
	moveExitWaiting  = 4
	moveExitRefused  = 5
	moveProtocolMark = "phase1"
)

// The sentences this file adds (copy record MV-N350 to MV-N379, pending).
const (
	moveNotEnabled  = "Moves are not enabled on this console."
	moveNoApp       = "carlos move needs --app: name the app the instance belongs to."
	moveAsRoot      = "Run carlos move as yourself, not as root. It uses your own credentials."
	moveDirectMode  = "carlos move works through a console. Unset CARLOS_DEPLOYMENT_BUCKET and CARLOS_DEPLOYMENT_DIR and run carlos auth login."
	moveNotRecorded = "This move's outcome is not recorded."
	moveLimited     = "The console is limiting requests. Checking every 10 seconds."
	moveNeedsHost   = "carlos move needs --host <host>."
	moveNeedsTo     = "carlos move needs --to <region>."
	moveNeedsRetry  = "carlos move resolve needs --retry or --declare-lost <box>."
	moveUnreachable = "Could not reach the console. Check the connection and try again."
	moveUnreadable  = "The console's answer could not be read."
	moveNotFound    = "Not found. Check --account and --app, and that you are a member of the instance's account."
	moveNoSession   = "carlos move could not sign in to a console. Run carlos auth login, or check --console, CARLOS_CONSOLE and CARLOS_SERVICE_TOKEN."
	moveStopped     = "Stopped waiting."
	moveRetryOrLost = "carlos move resolve takes --retry or --declare-lost, not both."
	moveNoAWS       = "declare-lost needs your AWS credentials to confirm the box is stopped or terminated."
	moveNoFlipTime  = "The move's flip time could not be read, so its target cannot be declared lost from here."
	moveOtherLoss   = "The console asks to confirm a loss carlos move does not know. Use carlos ops host resolve."
	moveLossChanged = "The move changed since it was read. Nothing was declared. Run the command again."
	// The offline subcommand named no verb, or one it does not have (MV-O181,
	// which replaced MV-O56 when stage became a verb).
	moveOfflineUsage = "carlos move offline needs stage, start, seal, flip, open, finish, abort or status."
)

func moveSay(out io.Writer, line string) {
	fmt.Fprintln(out, line)
}

// serverText is text the console wrote: its refusal sentence, or the stuck
// reason its detail answer carries from the record (a sentence a provision
// transition recorded). It is made only in roundTrip, from the body of a
// response the HTTP client returned, and said only through saySpoken. The
// copy gate holds both (TestMovesServerTextHasOneMaker).
type serverText struct{ spoken string }

// saySpoken says text the console wrote, as it wrote it.
func saySpoken(out io.Writer, t serverText) {
	fmt.Fprintln(out, t.spoken)
}

func runMove(args []string) {
	deps := moveDeps{session: resolveMode, euid: os.Geteuid, poll: 2 * time.Second, stdin: os.Stdin, ec2: awsinv.NewEC2Reader}
	os.Exit(moveMain(context.Background(), args, os.Stdout, deps))
}

// moveDeps are what moveMain reaches outside itself, so a test can run it
// against an httptest console.
type moveDeps struct {
	session func(consoleFlag string) (*apiSession, error)
	euid    func() int
	poll    time.Duration
	// stdin is where a declare-lost confirmation is read; ec2 is the
	// operator's own read-only EC2 seam (awsinv.NewEC2Reader).
	stdin io.Reader
	ec2   func(ctx context.Context, region string) (awsinv.ExactEC2API, error)
}

type moveArgs struct {
	verb                    string // request | status | cut | abort | resolve | offline
	offline                 string // the offline verb: stage | start | seal | flip | open | finish | abort | status
	host, to, app           string
	account, accountID      string
	console                 string
	manualCut, noWait, wait bool
	retry                   bool
	lostBox                 string // the box to declare lost (Task 1b)
}

// parseMoveArgs reads `carlos move [status|cut|abort|resolve] <flags>`. Each
// verb takes only its own flags; a flag another verb owns is a usage error.
// Usage problems are said on out and returned as an error.
func parseMoveArgs(args []string, out io.Writer) (moveArgs, error) {
	a := moveArgs{verb: "request"}
	if len(args) > 0 {
		switch args[0] {
		case "status", "cut", "abort", "resolve":
			a.verb, args = args[0], args[1:]
		case "offline":
			if len(args) < 2 || !slices.Contains(offlineVerbs, args[1]) {
				moveSay(out, moveOfflineUsage)
				return moveArgs{}, errors.New("no offline verb")
			}
			a.verb, a.offline, args = "offline", args[1], args[2:]
		}
	}
	name := "move " + a.verb
	if a.offline != "" {
		name += " " + a.offline
	}
	fs := flag.NewFlagSet(name, flag.ContinueOnError)
	var parsed strings.Builder
	fs.SetOutput(&parsed)
	fs.Usage = func() {} // the help is said by moveFlagHelp
	fs.StringVar(&a.host, "host", "", "the instance's host name")
	fs.StringVar(&a.app, "app", "", "the app the instance belongs to")
	fs.StringVar(&a.app, "a", "", "shorthand for --app")
	fs.StringVar(&a.account, "account", "", "the account the app belongs to (default: your default account)")
	fs.StringVar(&a.accountID, "account-id", "", "the account, by its id")
	fs.StringVar(&a.console, "console", "", "the console to work through (default: the only logged-in one)")
	switch a.verb {
	case "request":
		fs.StringVar(&a.to, "to", "", "the region to move the instance to")
		fs.BoolVar(&a.manualCut, "manual-cut", false, "stop when the move is ready, and cut with carlos move cut")
		fs.BoolVar(&a.noWait, "no-wait", false, "return once the console has accepted the request")
	case "status":
		fs.BoolVar(&a.wait, "wait", false, "wait for the move in flight, or the last one, to finish")
	case "cut", "abort":
		fs.BoolVar(&a.noWait, "no-wait", false, "return once the console has accepted it")
	case "offline":
		if a.offline == "start" || a.offline == "stage" {
			fs.StringVar(&a.to, "to", "", "the region to move the instance to")
		}
	case "resolve":
		fs.BoolVar(&a.retry, "retry", false, "retry a stuck move once its cause is fixed")
		fs.StringVar(&a.lostBox, "declare-lost", "", "declare the move's target box lost, once AWS shows it stopped or terminated")
		fs.BoolVar(&a.noWait, "no-wait", false, "return once the console has accepted it")
	}
	if err := fs.Parse(args); err != nil {
		moveFlagHelp(out, fs, &parsed)
		return moveArgs{}, err
	}
	if fs.NArg() > 0 {
		moveSay(out, moveNeedsHost)
		return moveArgs{}, errors.New("unexpected arguments")
	}
	switch {
	case a.host == "":
		moveSay(out, moveNeedsHost)
		return moveArgs{}, errors.New("no --host")
	case (a.verb == "request" || a.offline == "start" || a.offline == "stage") && a.to == "":
		moveSay(out, moveNeedsTo)
		return moveArgs{}, errors.New("no --to")
	case a.verb == "resolve" && a.retry && a.lostBox != "":
		moveSay(out, moveRetryOrLost)
		return moveArgs{}, errors.New("--retry and --declare-lost")
	case a.verb == "resolve" && !a.retry && a.lostBox == "":
		moveSay(out, moveNeedsRetry)
		return moveArgs{}, errors.New("no --retry or --declare-lost")
	}
	if a.verb != "status" && a.verb != "offline" {
		a.wait = !a.noWait
	}
	return a, nil
}

// moveFlagHelp says the flag help: what the flag package wrote while parsing
// (its error for a bad flag, if any) and its rendering of the verb's flags
// and their usage strings. It is the copy gate's one help emission
// (movesSinkAllow); the usage strings go to copy review with the CLI page.
func moveFlagHelp(out io.Writer, fs *flag.FlagSet, parsed *strings.Builder) {
	fs.SetOutput(parsed)
	fs.PrintDefaults()
	moveSay(out, strings.TrimRight(parsed.String(), "\n"))
}

// moveMain runs one `carlos move` and returns its exit code.
func moveMain(ctx context.Context, args []string, out io.Writer, d moveDeps) int {
	a, err := parseMoveArgs(args, out)
	if errors.Is(err, flag.ErrHelp) {
		return moveExitDone
	}
	if err != nil {
		return moveExitUsage
	}
	// L25: the operator's own credentials are what count (declare-lost reads
	// the cloud with them), and run as root rootaws replaces them. One rule
	// for the whole verb.
	if d.euid() == 0 {
		moveSay(out, moveAsRoot)
		return moveExitRefused
	}
	// resolveMode's own sentences are in no copy record (the general record
	// is the nearest-region plan's), so carlos move says its own.
	sess, err := d.session(a.console)
	if err != nil {
		moveSay(out, moveNoSession)
		return moveExitError
	}
	if sess == nil {
		moveSay(out, moveDirectMode)
		return moveExitRefused
	}
	acct := resolvedAccount(sess, &a.account, &a.accountID)
	app := resolvedApp(sess, acct, a.app, moveNoApp)
	m := &mover{sess: sess, out: out, host: a.host, poll: d.poll,
		path: apiPath(acct, app, "/instances/"+url.PathEscape(a.host)+"/move")}

	// The offline move is the operator's tool for a console with Phase 1 off.
	// It has no Phase 1 marker to check: its doors refuse for themselves.
	if a.verb == "offline" {
		m.path += "/offline"
		return m.offline(ctx, a)
	}

	// L4: read the status first, and send nothing to a console that does
	// not answer as Phase 1.
	view, code := m.readFlat(ctx)
	if code >= 0 {
		return code
	}
	if view.Protocol != moveProtocolMark {
		moveSay(out, moveNotEnabled)
		return moveExitRefused
	}

	switch a.verb {
	case "status":
		return m.status(ctx, view, a.wait)
	case "request":
		cut := "auto"
		if a.manualCut {
			cut = "manual"
		}
		return m.mutate(ctx, http.MethodPost, "", map[string]string{"region": a.to, "cut": cut}, a.wait, true)
	case "cut":
		return m.mutate(ctx, http.MethodPost, "/cut", struct{}{}, a.wait, false)
	case "abort":
		return m.mutate(ctx, http.MethodDelete, "", nil, a.wait, false)
	default: // resolve
		if a.lostBox != "" {
			return m.declareLost(ctx, a.lostBox, d, a.wait)
		}
		return m.mutate(ctx, http.MethodPost, "/resolve", map[string]bool{"retry": true}, a.wait, false)
	}
}

// The console's answers, as this CLI reads them (Task 2's shapes, copied
// here as the CLI convention is).
type moveFlat struct {
	State     string       `json:"state"`
	Phase     string       `json:"phase"`
	Target    string       `json:"target"`
	UpdatedAt int64        `json:"updated_at"`
	ID        string       `json:"id"`
	Last      *moveOutcome `json:"last"`
	Protocol  string       `json:"protocol"`
}

// moveOutcome is the record's last close (D11). Reason is read when the
// console's answer carries one; today's record does not, and the CLI never
// makes one up.
type moveOutcome struct {
	ID       string    `json:"id"`
	Outcome  string    `json:"outcome"`
	At       time.Time `json:"at"`
	OfflineS float64   `json:"offline_s"`
	Reason   string    `json:"reason"`
}

type moveAnswer struct {
	moveFlat
	Detail *moveDetailView `json:"detail"`
	// stuck is the journal's stuck reason as server text, set by roundTrip
	// (hasStuck when there is one).
	stuck    serverText
	hasStuck bool
	// waiting is the offline status view's "waiting": the sentence the next
	// verb would be refused with, as the console wrote it (set by roundTrip).
	WireWaiting string `json:"waiting"` // read only by roundTrip
	waiting     serverText
	hasWaiting  bool
}

type moveDetailView struct {
	From          string       `json:"from"`
	Region        string       `json:"region"`
	Move          *moveJournal `json:"move"`
	Claim         *moveClaim   `json:"claim"`
	TargetReport  *moveReport  `json:"target_report"`
	WaitingForBox string       `json:"waiting_for_box"`
}

type moveJournal struct {
	Incarnation string        `json:"incarnation"`
	ID          string        `json:"id"`
	To          string        `json:"to"`
	Cut         string        `json:"cut"`
	State       string        `json:"state"`
	CutPhase    string        `json:"cut_phase"`
	CutBudgetS  int           `json:"cut_budget_s"`
	FlippedAt   *time.Time    `json:"flipped_at"`
	Reason      string        `json:"reason"`
	WireStuck   string        `json:"stuck_reason"` // read only by roundTrip
	Rollback    *moveRollback `json:"rollback"`
}

type moveRollback struct {
	Reason          string     `json:"reason"`
	TargetPhase     string     `json:"target_phase"`
	ReleaseDeadline *time.Time `json:"release_deadline"`
}

type moveClaim struct {
	Kind  string `json:"kind"`
	MvBox string `json:"target"`
}

type moveReport struct {
	Checks map[string]struct {
		OK     bool   `json:"ok"`
		Reason string `json:"reason"`
	} `json:"checks"`
}

// mover is one invocation's conversation with the move door.
type mover struct {
	sess *apiSession
	out  io.Writer
	host string
	path string
	poll time.Duration
	// limited: a 429 was answered, so the command waits 10 seconds between
	// calls now and has said so (once).
	limited bool
}

func (m *mover) withDetail() string {
	if strings.Contains(m.path, "?") {
		return m.path + "&detail=1"
	}
	return m.path + "?detail=1"
}

// moveTransient is how many failed reads in a row the wait rides out before
// it gives up with exit 1: a console restart is not a failed move.
const moveTransient = 5

// moveReply is one response from the move door, as roundTrip read it.
type moveReply struct {
	status  int
	answer  moveAnswer // decoded, on a 2xx
	body    []byte     // the 2xx body as read (a declare-lost inspection)
	refusal serverText // the console's own sentence, on another status
	refused bool       // refusal holds one
	moved   string     // the console's new origin, on a console-shaped redirect
	failed  bool       // the console was not reached, or its answer not read
	unread  bool       // ... and it was the answer that could not be read
	stopped bool       // the context ended the call
}

// roundTrip sends one request to the move door and reads its response. It is
// the one maker of serverText (movesServerText): the refusal's "error"
// sentence, and the stuck reason of a detail answer, read from the body of
// the response the HTTP client returned.
func (m *mover) roundTrip(ctx context.Context, method, path string, in any) moveReply {
	var body io.Reader
	if in != nil {
		raw, err := json.Marshal(in)
		if err != nil {
			return moveReply{failed: true, unread: true}
		}
		body = bytes.NewReader(raw)
	}
	req, err := http.NewRequestWithContext(ctx, method, m.sess.c.base+path, body)
	if err != nil {
		return moveReply{failed: true}
	}
	if in != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	if m.sess.c.token != "" {
		req.Header.Set("Authorization", "Bearer "+m.sess.c.token)
	}
	res, err := m.sess.c.httpc.Do(req)
	if err != nil {
		return moveReply{failed: true, stopped: ctx.Err() != nil}
	}
	defer res.Body.Close()
	r := moveReply{status: res.StatusCode}
	if moved := consoleMoved(m.sess.c.base, res.StatusCode, res.Header.Get("Location")); moved != nil {
		r.moved = moved.To
		return r
	}
	raw, err := io.ReadAll(io.LimitReader(res.Body, 1<<20))
	if err != nil {
		r.failed = true
		return r
	}
	if res.StatusCode/100 == 2 {
		r.body = raw
		if json.Unmarshal(raw, &r.answer) != nil {
			r.failed, r.unread = true, true
			return r
		}
		if d := r.answer.Detail; d != nil && d.Move != nil && d.Move.WireStuck != "" {
			r.answer.stuck, r.answer.hasStuck = serverText{spoken: d.Move.WireStuck}, true
		}
		if r.answer.WireWaiting != "" {
			r.answer.waiting, r.answer.hasWaiting = serverText{spoken: r.answer.WireWaiting}, true
		}
		return r
	}
	var e struct {
		Error string `json:"error"`
	}
	if json.Unmarshal(raw, &e) == nil && e.Error != "" {
		r.refusal, r.refused = serverText{spoken: e.Error}, true
	}
	return r
}

// flat is the reply's flat view.
func (r moveReply) flat() moveFlat {
	return r.answer.moveFlat
}

// sayReply says why a reply ended the command: the console's own sentence
// when it gave one, else this file's.
func (m *mover) sayReply(r moveReply) {
	switch {
	case r.moved != "":
		moveSay(m.out, fmt.Sprintf("This console has moved to %s. Run carlos auth login --console %s, then run the command again.", r.moved, r.moved))
	case r.stopped:
		moveSay(m.out, moveStopped)
	case r.unread:
		moveSay(m.out, moveUnreadable)
	case r.failed:
		moveSay(m.out, moveUnreachable)
	case r.refused:
		saySpoken(m.out, r.refusal)
	case r.status == http.StatusNotFound:
		moveSay(m.out, moveNotFound)
	default:
		moveSay(m.out, fmt.Sprintf("The console answered %d and gave no reason.", r.status))
	}
}

// limit notes a 429: from now on the command waits 10 seconds between calls,
// and says so the first time.
func (m *mover) limit() {
	if !m.limited {
		m.limited = true
		moveSay(m.out, moveLimited)
	}
}

// get reads one answer, riding out 429s (backing off, said once) and a few
// transient failures. code is -1 on success, else the exit code to return.
func (m *mover) get(ctx context.Context, path string) (moveAnswer, int) {
	r, code := m.getReply(ctx, path)
	return r.answer, code
}

// getReply is get with the whole reply, for a reader of the raw body.
func (m *mover) getReply(ctx context.Context, path string) (moveReply, int) {
	failures := 0
	for {
		r := m.roundTrip(ctx, http.MethodGet, path, nil)
		ok := !r.failed && r.moved == ""
		switch {
		case ok && r.status == http.StatusOK:
			return r, -1
		case r.moved != "" || r.stopped:
			m.sayReply(r)
			return moveReply{}, moveExitError
		case ok && r.status == http.StatusTooManyRequests:
			m.limit()
		case ok && r.status >= 400 && r.status < 500:
			m.sayReply(r)
			return moveReply{}, moveExitRefused
		default:
			failures++
			if failures >= moveTransient {
				m.sayReply(r)
				return moveReply{}, moveExitError
			}
		}
		if !m.sleep(ctx) {
			moveSay(m.out, moveStopped)
			return moveReply{}, moveExitError
		}
	}
}

func (m *mover) interval() time.Duration {
	if m.limited {
		return 5 * m.poll // 10 s at the production 2 s poll
	}
	return m.poll
}

// sleep waits one interval; false when the context ended first.
func (m *mover) sleep(ctx context.Context) bool {
	t := time.NewTimer(m.interval())
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

func (m *mover) readFlat(ctx context.Context) (moveFlat, int) {
	a, code := m.get(ctx, m.path)
	return a.moveFlat, code
}

func (m *mover) readDetail(ctx context.Context) (moveAnswer, int) {
	a, code := m.get(ctx, m.withDetail())
	if code < 0 && a.Detail == nil {
		a.Detail = &moveDetailView{}
	}
	return a, code
}

// send sends one mutation, sending the same call again after the back-off
// while the doors' rate limit answers 429 (fix round 1, ruling 1: the limit
// answers before any handler, so nothing was done).
func (m *mover) send(ctx context.Context, method, tail string, body any) moveReply {
	for {
		r := m.roundTrip(ctx, method, m.path+tail, body)
		if r.failed || r.moved != "" || r.status != http.StatusTooManyRequests {
			return r
		}
		m.limit()
		if !m.sleep(ctx) {
			return moveReply{failed: true, stopped: true}
		}
	}
}

// mutate sends one move mutation. A refusal is the console's own sentence
// and exit 5. On acceptance the answer names the move (Task 2). With
// --no-wait that answer is all: its state is said, and nothing more is read
// (fix round 2, ruling 2). Else the move's detail, read once now, binds its
// regions, a request says M-C1, and the wait follows the move's id.
func (m *mover) mutate(ctx context.Context, method, tail string, body any, wait, request bool) int {
	return m.accepted(ctx, m.send(ctx, method, tail, body), wait, request)
}

// accepted ends a mutation from the console's reply: a refusal or a failure
// is said and ends it; an acceptance is said (--no-wait) or waited on.
func (m *mover) accepted(ctx context.Context, r moveReply, wait, request bool) int {
	if code := m.refusal(r); code >= 0 {
		return code
	}
	view := r.flat()
	if view.ID == "" || view.Protocol != moveProtocolMark {
		moveSay(m.out, moveNotEnabled)
		return moveExitRefused
	}
	if !wait {
		moveSay(m.out, fmt.Sprintf("Move %s of %s is %s.", view.ID, m.host, view.State))
		return moveExitDone
	}
	w := newMoveWatch(m, view.ID)
	w.toRegion = view.Target
	a, code := m.readDetail(ctx)
	if code >= 0 {
		return code
	}
	w.learn(a)
	if request && w.first("start") {
		moveSay(m.out, moves.MoveStartingLine(m.host, w.fromRegion, w.toRegion))
	}
	return w.run(ctx)
}

// status answers `carlos move status`: without --wait, one line for the
// move in flight (and what the wait would say of its state), or the last
// move's outcome; with --wait, the wait for the move in flight, or the
// outcome of the last one when nothing is in flight.
func (m *mover) status(ctx context.Context, view moveFlat, wait bool) int {
	id := view.ID
	if id == "" && view.Last != nil {
		id = view.Last.ID
	}
	if id == "" {
		moveSay(m.out, fmt.Sprintf("No move is in flight for %s.", m.host))
		return moveExitDone
	}
	w := newMoveWatch(m, id)
	if wait {
		return w.run(ctx)
	}
	a, code := m.readDetail(ctx)
	if code >= 0 {
		return code
	}
	if a.ID == id {
		moveSay(m.out, fmt.Sprintf("Move %s of %s is %s.", id, m.host, a.State))
	}
	// The state's own sentence, and its exit code when it is one the wait
	// would end on (an outcome, stuck, waiting); else 0.
	if code := w.observe(a); code >= 0 {
		return code
	}
	return moveExitDone
}

// moveWatch is the plan's moveWait: it follows one move, by id, and prints
// each sentence once.
type moveWatch struct {
	m                    *mover
	id                   string
	fromRegion, toRegion string // this move's, from its own answers only
	said                 map[string]bool
	lastKey              string
	last                 moveAnswer // the detail read last
}

func newMoveWatch(m *mover, id string) *moveWatch {
	return &moveWatch{m: m, id: id, said: map[string]bool{}}
}

// first reports whether key's sentence is due: true the first time only.
func (w *moveWatch) first(key string) bool {
	if w.said[key] {
		return false
	}
	w.said[key] = true
	return true
}

// learn binds the move's regions from an answer about this move, and only
// such an answer: another move's record never names this one's regions.
func (w *moveWatch) learn(a moveAnswer) {
	if a.ID != w.id {
		return
	}
	if a.Detail.From != "" {
		w.fromRegion = a.Detail.From
	}
	if a.Target != "" {
		w.toRegion = a.Target
	}
}

// run polls until this move ends or needs an operator, and returns the exit
// code. The detail is read only when the state or the phase changes (L19);
// between reads, the detail already read is looked at again, so a release
// deadline that passes while nothing changes is still seen. The outcome is
// never inferred from a state seen: it is last, bound to this move's id.
func (w *moveWatch) run(ctx context.Context) int {
	for first := true; ; first = false {
		if !first && !w.m.sleep(ctx) {
			moveSay(w.m.out, moveStopped)
			return moveExitError
		}
		view, code := w.m.readFlat(ctx)
		if code >= 0 {
			return code
		}
		a := w.last
		if key := view.State + "|" + view.Phase + "|" + view.ID; key != w.lastKey {
			if a, code = w.m.readDetail(ctx); code >= 0 {
				return code
			}
			w.lastKey, w.last = key, a
		}
		if code := w.observe(a); code >= 0 {
			return code
		}
	}
}

// observe prints what one detail answer says of this move, and returns an
// exit code when the wait ends here (else -1).
func (w *moveWatch) observe(a moveAnswer) int {
	w.learn(a)
	host, from, to := w.m.host, w.fromRegion, w.toRegion
	if a.ID != w.id {
		// This move is not in flight: its outcome is last, or it is not
		// recorded (Review Focus 1).
		return w.outcome(a)
	}
	j := a.Detail.Move
	if j == nil {
		j = &moveJournal{}
	}
	if w.first("start") {
		moveSay(w.m.out, moves.MoveStartingLine(host, from, to))
	}
	switch a.State {
	case "preparing":
		if w.first("preparing") {
			moveSay(w.m.out, fmt.Sprintf("Copying data. %s keeps serving while this runs.", host))
		}
		if r := a.Detail.TargetReport; r != nil {
			if w.first("checking") {
				moveSay(w.m.out, fmt.Sprintf("Checking that %s can run %s.", to, host))
			}
			ids := make([]string, 0, len(r.Checks))
			for id := range r.Checks {
				ids = append(ids, id)
			}
			slices.Sort(ids)
			for _, id := range ids {
				if c := r.Checks[id]; !c.OK && c.Reason != "" {
					if w.first("check:" + c.Reason) {
						moveSay(w.m.out, fmt.Sprintf("%s cannot take this move yet: %s.", to, c.Reason))
					}
				}
			}
		}
	case "ready":
		// A cut the source refused (the cut guard, a replica layout it
		// cannot cut) is back at ready with the reason recorded, unpinned
		// until the cause is gone and a repin clears the reason. Until then
		// it is not ready to cut, so M-C12 is not said (fix round 1).
		if j.Reason != "" {
			if w.first("refused:" + j.Reason) {
				moveSay(w.m.out, fmt.Sprintf("The cut did not start: %s.", j.Reason))
			}
			if j.Cut == "manual" {
				return moveExitWaiting
			}
			break
		}
		if j.Cut == "manual" {
			moveSay(w.m.out, fmt.Sprintf("Ready to cut. Run carlos move cut --host %s when you are ready.", host))
			return moveExitWaiting
		}
	case "cutting":
		if j.CutPhase != "" && j.CutPhase != "requested" && j.CutBudgetS > 0 {
			if w.first("cut") {
				moveSay(w.m.out, moves.CutStartingLine(to, host, moves.CutBudgetText(j.CutBudgetS)))
			}
		}
	case "landing", "committed":
		if w.first("landing") {
			moveSay(w.m.out, moves.FlippedLine(from, to))
		}
	case "rolling_back", "reverting":
		reason := j.Reason
		if j.Rollback != nil && j.Rollback.Reason != "" {
			reason = j.Rollback.Reason
		}
		if reason != "" {
			if w.first("rollback") {
				moveSay(w.m.out, moves.MovingBackLine(to, host, reason, from))
			}
		}
		if rb := j.Rollback; rb != nil && rb.ReleaseDeadline != nil && time.Now().After(*rb.ReleaseDeadline) && rb.TargetPhase == "releasing" {
			box := "the target box"
			if c := a.Detail.Claim; c != nil && c.MvBox != "" {
				box = c.MvBox
			}
			moveSay(w.m.out, fmt.Sprintf("Waiting for %s in %s to confirm it has stopped %s. %s is offline.", box, to, host, host))
			return moveExitWaiting
		}
	case "stuck":
		moveSay(w.m.out, fmt.Sprintf("%s could not start in %s or in %s. It is offline. Run carlos move resolve --host %s --retry once the cause is fixed.", host, to, from, host))
		if a.hasStuck {
			saySpoken(w.m.out, a.stuck)
		}
		return moveExitWaiting
	case "relocating":
		if region := a.Detail.WaitingForBox; region != "" {
			moveSay(w.m.out, fmt.Sprintf("%s is waiting for a box in %s. Its last box was terminated before the move finished. Its data up to the last completed upload is kept in %s. It will start when a box joins the region, and any writes after that upload will be reported then.", host, region, region))
			return moveExitWaiting
		}
		moveSay(w.m.out, fmt.Sprintf("%s is relocating after its box was terminated. This move will not finish.", host))
		return moveExitWaiting
	}
	return -1
}

// outcome is the end of a move that is no longer in flight: last, if it is
// this move's, else not recorded. Its regions are the move's own, as the
// wait learnt them; one it never learnt is never filled in from the record
// (fix round 2, ruling 3). Without its region a sentence that names it gives
// way: M-C9 alone for M-C8 (whose destination is unknown), and the outcome
// line for M-C7, M-C9 and the aborted line.
func (w *moveWatch) outcome(a moveAnswer) int {
	host := w.m.host
	l := a.Last
	if l == nil || l.ID != w.id {
		moveSay(w.m.out, moveNotRecorded)
		return moveExitWaiting
	}
	from, to := w.fromRegion, w.toRegion
	switch l.Outcome {
	case "done":
		if to == "" {
			return w.outcomeLine(l, moveExitDone)
		}
		moveSay(w.m.out, moves.MovedLine(host, to, int(math.Round(l.OfflineS))))
		return moveExitDone
	case "rolled_back":
		if from == "" {
			return w.outcomeLine(l, moveExitRolled)
		}
		// M-C8 for a rollback the wait missed only when the outcome carries
		// its reason (ruling 4), and only with its destination known.
		if l.Reason != "" && to != "" && w.first("rollback") {
			moveSay(w.m.out, moves.MovingBackLine(to, host, l.Reason, from))
		}
		moveSay(w.m.out, moves.MovedBackLine(host, from))
		return moveExitRolled
	case "aborted":
		if from == "" {
			return w.outcomeLine(l, moveExitRolled)
		}
		moveSay(w.m.out, fmt.Sprintf("The move was aborted. %s stays in %s.", host, from))
		return moveExitRolled
	}
	moveSay(w.m.out, moveNotRecorded)
	return moveExitWaiting
}

// outcomeLine says the outcome without a region, when the wait never learnt
// the move's own.
func (w *moveWatch) outcomeLine(l *moveOutcome, code int) int {
	moveSay(w.m.out, fmt.Sprintf("Move %s of %s is %s.", l.ID, w.m.host, l.Outcome))
	return code
}

// declareLost is carlos move resolve --declare-lost <box> (Task 1b): claim
// Plan 10's move door, inspect then declare, through declareHostLoss, the
// half carlos ops host resolve shares. The operator types the host name to
// confirm (Paul's ruling of 2026-10-10); the declaration carries the
// challenge's own ID. M-C18 is said for a move-target-lost inspection, with
// the move's flip time from detail=1; a destroyed-local-only one has its own
// sentence. The box is read through the operator's own EC2 credentials,
// never the console's.
func (m *mover) declareLost(ctx context.Context, box string, d moveDeps, wait bool) int {
	a, code := m.readDetail(ctx)
	if code >= 0 {
		return code
	}
	read := a.Detail.Move // the move M-C18 describes
	var flipped *time.Time
	if read != nil {
		flipped = read.FlippedAt
	}
	refused := -1 // the exit code a refusal already said, if any
	stdin := bufio.NewReader(d.stdin)
	door := hostLossDoor{
		inspect: func(ctx context.Context) (hostresolve.Inspection, error) {
			var in hostresolve.Inspection
			r := m.send(ctx, http.MethodPost, "/resolve", map[string]string{"declare_lost": box})
			if code := m.refusal(r); code >= 0 {
				refused = code
				return in, errHostLossNotConfirmed
			}
			if err := json.Unmarshal(r.body, &in); err != nil {
				moveSay(m.out, moveUnreadable)
				refused = moveExitError
				return in, err
			}
			return in, nil
		},
		confirm: func(in hostresolve.Inspection) bool {
			switch in.Details.Consequences.Transition {
			case "move-target-lost":
				// M-C18 describes the move read above: the challenge must
				// name that move, and its incarnation when both carry one
				// (fix round 1, finding 1). Otherwise nothing is shown or
				// declared, and nothing is tried again.
				if read == nil || read.ID != in.Details.Op.ID ||
					(read.Incarnation != "" && in.Details.HolderIncarnation != "" && read.Incarnation != in.Details.HolderIncarnation) {
					moveSay(m.out, moveLossChanged)
					return false
				}
				if flipped == nil {
					moveSay(m.out, moveNoFlipTime)
					return false
				}
				moveSay(m.out, fmt.Sprintf("This declares %s lost. Anything %s wrote on %s after %s is lost. Type the host name to confirm.", box, m.host, box, flipped.UTC().Format("2006-01-02 15:04:05 UTC")))
			case "destroyed-local-only":
				moveSay(m.out, fmt.Sprintf("This records that %s is gone with its disks. %s keeps nothing there that it still needs. Type the host name to confirm.", box, m.host))
			default:
				moveSay(m.out, moveOtherLoss)
				return false
			}
			line, _ := stdin.ReadString('\n')
			if strings.TrimSpace(line) != m.host {
				moveSay(m.out, fmt.Sprintf("That is not %s. Nothing was declared.", m.host))
				return false
			}
			return true
		},
		reader: d.ec2,
		now:    time.Now,
	}
	var declared moveReply
	door.declare = func(ctx context.Context, sub hostfence.Submission) error {
		declared = m.send(ctx, http.MethodPost, "/resolve", map[string]any{"declare_lost": box, "submission": sub})
		return nil
	}
	stage, _ := declareHostLoss(ctx, door)
	switch {
	case refused >= 0:
		return refused
	case stage == hostLossConfirm:
		return moveExitRefused
	case stage == hostLossReader:
		moveSay(m.out, moveNoAWS)
		return moveExitRefused
	case stage == hostLossObserve:
		moveSay(m.out, fmt.Sprintf("AWS does not show %s stopped or terminated, so it cannot be declared lost.", box))
		return moveExitRefused
	}
	return m.accepted(ctx, declared, wait, false)
}

// refusal says a reply that is not an acceptance and returns its exit code,
// or -1 for a 2xx.
func (m *mover) refusal(r moveReply) int {
	switch {
	case r.failed || r.moved != "":
		m.sayReply(r)
		return moveExitError
	case r.status >= 400 && r.status < 500:
		m.sayReply(r)
		return moveExitRefused
	case r.status < 200 || r.status > 299:
		m.sayReply(r)
		return moveExitError
	}
	return -1
}

// ---- carlos move offline ----
//
// An operator moves one instance with planned downtime, one step at a time:
// start, seal, flip, open, finish, abort, and status (offline-move plan,
// Task 7). stage comes before start when the operator wants the offline
// window short: it copies the instance's data to the target while it keeps
// serving, and can be run again. Each verb posts to its door, then polls the status door every
// interval until the step it started is done or has failed, and says which
// verb comes next. It never runs the next verb by itself. The exit codes are
// the family's: 0 the step is done, 3 the move was aborted, 4 the step failed
// or the move waits for the operator, 5 the console refused, 1 an error.
// Refusals are the console's own sentences (serverText); the CLI invents none.

// offlineVerbs are the eight verbs after `carlos move offline`.
var offlineVerbs = []string{"stage", "start", "seal", "flip", "open", "finish", "abort", "status"}

// offlineView is the offline status view (GET) and the 202 every mutating
// door answers with.
type offlineView struct {
	Host     string   `json:"host"`
	State    string   `json:"state"`
	ID       string   `json:"id"`
	From     string   `json:"from"`
	To       string   `json:"to"`
	Sealer   string   `json:"sealer"`
	Boxes    []string `json:"boxes"`
	Next     string   `json:"next"`
	CanAbort bool     `json:"can_abort"`
	// NeverOffline: staged and never started, so an abort only removes
	// the staged copy.
	NeverOffline bool `json:"never_offline"`
	// StagedAt is the latest stage request (Unix nanoseconds), which
	// Copying.Answered equals once the stager has run its round.
	StagedAt int64 `json:"staged_at"`
	// Failed: a box's report says the step cannot finish (the status view's
	// own flag; the CLI says MV-O6 or MV-O9 for it).
	Failed  bool                `json:"failed"`
	Left    []offlineLeftView   `json:"left"`
	Reports []offlineReportView `json:"reports"`
	Items   []offlineItemView   `json:"items"`
	Last    *moveOutcome        `json:"last"`
	// Copying is the pre-copy's progress (the console's stage), nil when no
	// box has staged this move.
	Copying *offlineCopying `json:"stage"`
}

// offlineCopying is the status view's stage: the stager's last round. The
// fields are named apart from the tree's for the copy gate; Trouble is the
// box's own diagnostic, not copy.
type offlineCopying struct {
	Stager     string `json:"box"`
	Phase      string `json:"phase"`
	RoundAt    int64  `json:"at"`
	Total      int    `json:"replica_objects"`
	TotalBytes int64  `json:"replica_bytes"`
	Done       int    `json:"staged_objects"`
	DoneBytes  int64  `json:"staged_bytes"`
	Trouble    string `json:"error"`
	Answered   int64  `json:"request"`
}

// megabytes is a byte count as a progress line says it.
func megabytes(n int64) string { return fmt.Sprintf("%.1f", float64(n)/1e6) }

// sayStaged says the stage's last round: how much of the replica the
// target holds and when (MV-O183), that the round failed (MV-O185), or
// that none has run yet (MV-O186).
func (w *offlineRun) sayStaged(v offlineView) {
	host, c := w.m.host, v.Copying
	if c == nil {
		moveSay(w.m.out, fmt.Sprintf("Waiting for the first staging round of %s.", host))
		return
	}
	if c.Phase == moves.OfflinePhaseFailed {
		moveSay(w.m.out, fmt.Sprintf("The last staging round failed: %s. Run carlos move offline stage --host %s --to %s to try again.", c.Trouble, host, v.To))
		return
	}
	at := time.Unix(c.RoundAt, 0).UTC().Format("2006-01-02 15:04 UTC")
	moveSay(w.m.out, fmt.Sprintf("Staged %d of %d objects for %s in %s (%s of %s MB) at %s.", c.Done, c.Total, host, v.To, megabytes(c.DoneBytes), megabytes(c.TotalBytes), at))
}

// sayStartWhenReady is MV-O184: what comes after a stage.
func (w *offlineRun) sayStartWhenReady(v offlineView) {
	host := w.m.host
	moveSay(w.m.out, fmt.Sprintf("%s keeps serving. Run carlos move offline start --host %s --to %s when you are ready to take it offline.", host, host, v.To))
}

type offlineReportView struct {
	Box      string                    `json:"box"`
	Role     string                    `json:"role"`
	Phase    string                    `json:"phase"`
	Fresh    bool                      `json:"fresh"`
	Verdicts map[string]offlineVerdict `json:"items"`
}

// offlineLeftView is a path a source box could not park and left in place,
// with the reason it gave (a box diagnostic, not copy).
type offlineLeftView struct {
	Holder  string `json:"box"`
	Where   string `json:"path"`
	Because string `json:"reason"`
}

// offlineVerdict is one check of a report; its names are apart from the
// tree's for the copy gate (see offlineItemView).
type offlineVerdict struct {
	Passed bool   `json:"ok"`
	Why    string `json:"reason"`
}

// The fields are named apart from the rest of the tree's (Detail, ID): the
// copy gate reads a field by its name across the scanned files.
type offlineItemView struct {
	Slug       string `json:"id"`
	Owner      string `json:"owner"`
	Check      string `json:"status"`
	Diagnostic string `json:"detail"`
}

// offlineReply is one answer of the offline doors: the view, and the sentence
// the console says the next verb waits for (server text).
type offlineReply struct {
	v          offlineView
	waiting    serverText
	hasWaiting bool
}

// report is the box's report of a role, if it wrote one.
func (v offlineView) report(role, box string) *offlineReportView {
	for i := range v.Reports {
		if v.Reports[i].Role == role && v.Reports[i].Box == box {
			return &v.Reports[i]
		}
	}
	return nil
}

// failed names the items an owner's report failed, by the box's own
// diagnostic (or the item's name when it gave none), joined. Empty when none
// failed.
func (v offlineView) failed(owner string) string {
	var parts []string
	for _, it := range v.Items {
		if it.Owner != owner || it.Check != "failed" {
			continue
		}
		if it.Diagnostic == "" {
			parts = append(parts, it.Slug)
			continue
		}
		parts = append(parts, it.Diagnostic)
	}
	return strings.Join(parts, "; ")
}

// failedReport is the role's report when it says it failed.
func (v offlineView) failedReport(role string) *offlineReportView {
	for i := range v.Reports {
		if v.Reports[i].Role == role && v.Reports[i].Phase == moves.OfflinePhaseFailed {
			return &v.Reports[i]
		}
	}
	return nil
}

// reason is what a failed report says: its failing checks, else its phase.
func (r offlineReportView) reason() string {
	var parts []string
	for _, id := range slices.Sorted(maps.Keys(r.Verdicts)) {
		c := r.Verdicts[id]
		switch {
		case c.Passed:
		case c.Why != "":
			parts = append(parts, c.Why)
		default:
			parts = append(parts, id)
		}
	}
	if len(parts) > 0 {
		return strings.Join(parts, "; ")
	}
	return r.Phase
}

// The roles a failed step can be: the sealer sealing, the target landing,
// or a source box cleaning up after the open.
const (
	failSealer = "sealer"
	failTarget = "target"
	failSource = "source"
)

// failure says whether a box has failed its step, which role it was and why:
// the items the role's report failed, else what its failed report says, else
// the word failed. Before the open it is the target landing, else the sealer
// sealing (both at once go by the state the move is in). Once the move is
// open or finishing a failed report of any role counts, source boxes included;
// while it reverts only a failed source box does, since a target or sealer
// that failed earlier is being put back.
func (v offlineView) failure() (kind, reason string, ok bool) {
	switch v.State {
	case provision.OfflineOpen, provision.OfflineFinishing, provision.OfflineReverting:
		roles := []string{moves.OfflineRoleSealer, moves.OfflineRoleTarget, moves.OfflineRoleSource}
		if v.State == provision.OfflineReverting {
			roles = roles[2:]
		}
		for _, role := range roles {
			if r := v.failedReport(role); r != nil {
				return role, r.reason(), true
			}
		}
		return "", "", false
	}
	sealing := v.State == provision.OfflineStopping || v.State == provision.OfflineSealing || v.State == provision.OfflineSealed
	sealReason, landReason := v.failed(moves.OfflineOwnerSeal), v.failed(moves.OfflineOwnerLanding)
	sealRep, landRep := v.failedReport(moves.OfflineRoleSealer), v.failedReport(moves.OfflineRoleTarget)
	sealFailed := sealRep != nil || (sealing && sealReason != "")
	landFailed := landRep != nil || (!sealing && landReason != "")
	switch {
	case sealFailed && (sealing || !landFailed):
		if sealReason == "" && sealRep != nil {
			sealReason = sealRep.reason()
		}
		return failSealer, sealReason, true
	case landFailed:
		if landReason == "" && landRep != nil {
			landReason = landRep.reason()
		}
		return failTarget, landReason, true
	case v.Failed:
		if sealing {
			return failSealer, moves.OfflinePhaseFailed, true
		}
		return failTarget, moves.OfflinePhaseFailed, true
	}
	return "", "", false
}

// sayFailure says a failed step: MV-O6 for the target before the open,
// MV-O9 for the sealer or a target once open (abort refuses there), MV-O65
// for a source box.
func (w *offlineRun) sayFailure(v offlineView) bool {
	kind, reason, ok := v.failure()
	if !ok {
		return false
	}
	host := w.m.host
	if kind == failTarget && (v.State == provision.OfflineOpen || v.State == provision.OfflineFinishing) {
		// Once the move is open, abort refuses (MV-O27), so MV-O6's advice
		// would send the operator to a verb that says no; MV-O9 says it.
		kind = failSealer
	}
	switch kind {
	case failTarget:
		moveSay(w.m.out, fmt.Sprintf("%s could not take %s: %s. Run carlos move offline abort --host %s to move it back.", v.To, host, reason, host))
	case failSource:
		for _, r := range v.Reports {
			if r.Role == moves.OfflineRoleSource && r.Phase == moves.OfflinePhaseFailed {
				moveSay(w.m.out, fmt.Sprintf("%s could not clean up %s: %s. Run carlos move offline status --host %s for each item.", r.Box, host, reason, host))
				return true
			}
		}
	default:
		moveSay(w.m.out, fmt.Sprintf("The move could not finish a step: %s. Run carlos move offline status --host %s for each item.", reason, host))
	}
	return true
}

// offlineRun is one `carlos move offline` verb's conversation: the move it
// follows, as the console's answers named it, and what has been said.
type offlineRun struct {
	m            *mover
	id, from, to string
	said         map[string]bool
	lastWaiting  serverText
}

func (w *offlineRun) learn(v offlineView) {
	if v.ID != "" && (w.id == "" || w.id == v.ID) {
		w.id, w.from, w.to = v.ID, v.From, v.To
	}
}

func (w *offlineRun) once(key string) bool {
	if w.said[key] {
		return false
	}
	w.said[key] = true
	return true
}

// read decodes a 2xx reply from an offline door.
func (w *offlineRun) read(r moveReply) (offlineReply, int) {
	var v offlineView
	if json.Unmarshal(r.body, &v) != nil {
		moveSay(w.m.out, moveUnreadable)
		return offlineReply{}, moveExitError
	}
	w.learn(v)
	return offlineReply{v: v, waiting: r.answer.waiting, hasWaiting: r.answer.hasWaiting}, -1
}

// poll reads the status door.
func (w *offlineRun) poll(ctx context.Context) (offlineReply, int) {
	r, code := w.m.getReply(ctx, w.m.path)
	if code >= 0 {
		return offlineReply{}, code
	}
	return w.read(r)
}

// post sends a verb's door. A refusal is said as the console wrote it.
func (w *offlineRun) post(ctx context.Context, tail string, body any) (offlineReply, int) {
	r := w.m.send(ctx, http.MethodPost, tail, body)
	if code := w.m.refusal(r); code >= 0 {
		return offlineReply{}, code
	}
	return w.read(r)
}

// follow polls until step says the verb's step has ended, and returns the
// exit code it gave.
func (w *offlineRun) follow(ctx context.Context, cur offlineReply, step func(offlineReply) (int, bool)) int {
	for {
		if code, done := step(cur); done {
			return code
		}
		if !w.m.sleep(ctx) {
			moveSay(w.m.out, moveStopped)
			return moveExitError
		}
		var code int
		if cur, code = w.poll(ctx); code >= 0 {
			return code
		}
	}
}

// sayWaiting says the console's sentence for what the next verb waits for,
// each time it changes.
func (w *offlineRun) sayWaiting(r offlineReply) {
	if r.hasWaiting && r.waiting != w.lastWaiting {
		saySpoken(w.m.out, r.waiting)
		w.lastWaiting = r.waiting
	}
}

// sayCleaning names, once each, the boxes the finish waits for.
func (w *offlineRun) sayCleaning(v offlineView) {
	var source, sealer string
	switch v.State {
	case provision.OfflineOpen:
		source = moves.OfflinePhaseParked
	case provision.OfflineFinishing:
		source, sealer = moves.OfflinePhaseGone, moves.OfflinePhaseLifted
	case provision.OfflineReverting:
		source, sealer = moves.OfflinePhaseReverted, moves.OfflinePhaseReverted
	default:
		return
	}
	var boxes []string
	for _, b := range v.Boxes {
		if r := v.report(moves.OfflineRoleSource, b); r == nil || r.Phase != source {
			boxes = append(boxes, b)
		}
	}
	if sealer != "" && v.Sealer != "" {
		if r := v.report(moves.OfflineRoleSealer, v.Sealer); r == nil || r.Phase != sealer {
			boxes = append(boxes, v.Sealer)
		}
	}
	for _, b := range boxes {
		if w.once("clean:" + v.State + ":" + b) {
			moveSay(w.m.out, fmt.Sprintf("Waiting for %s to clean up %s.", b, w.m.host))
		}
	}
}

// ended says the end of the move this run followed, from the console's
// `last`, which counts only when it is that move's own.
func (w *offlineRun) ended(v offlineView) int {
	host := w.m.host
	l := v.Last
	if w.id == "" || l == nil || l.ID != w.id {
		moveSay(w.m.out, moveNotRecorded)
		return moveExitWaiting
	}
	switch l.Outcome {
	case "done":
		if w.to == "" {
			offlineOutcomeLine(w.m, l)
			return moveExitDone
		}
		moveSay(w.m.out, moves.MovedLine(host, w.to, int(math.Round(l.OfflineS))))
		return moveExitDone
	case "aborted":
		if w.from == "" {
			offlineOutcomeLine(w.m, l)
			return moveExitRolled
		}
		moveSay(w.m.out, moves.MovedBackLine(host, w.from))
		return moveExitRolled
	}
	moveSay(w.m.out, moveNotRecorded)
	return moveExitWaiting
}

// offlineOutcomeLine says an outcome without the regions: the move, the host
// and how it ended.
func offlineOutcomeLine(m *mover, l *moveOutcome) {
	moveSay(m.out, fmt.Sprintf("Move %s of %s is %s.", l.ID, m.host, l.Outcome))
}

// served reports that the target serves the instance: its report says
// opened, the move has gone on to finishing, or the next step is finish.
func (w *offlineRun) served(v offlineView) bool {
	if v.State == provision.OfflineFinishing || v.Next == "finish" {
		return true
	}
	for _, r := range v.Reports {
		if r.Role == moves.OfflineRoleTarget && r.Phase == moves.OfflinePhaseOpened {
			return true
		}
	}
	return false
}

// sayServed is MV-O7, said once.
func (w *offlineRun) sayServed(v offlineView) {
	if w.once("served") {
		host := w.m.host
		moveSay(w.m.out, fmt.Sprintf("%s is served from %s. Run carlos move offline finish --host %s once %s has cleaned up.", host, v.To, host, v.From))
	}
}

// headline says where a move in flight stands, in the verbs' own sentences.
func (w *offlineRun) headline(r offlineReply) {
	v, host := r.v, w.m.host
	if v.State == provision.OfflineStaging {
		moveSay(w.m.out, fmt.Sprintf("Staging %s in %s. %s keeps serving while the data is copied.", host, v.To, host))
		w.sayStaged(v)
		if v.Next == "start" {
			w.sayStartWhenReady(v)
			return
		}
		w.sayWaiting(r)
		return
	}
	moveSay(w.m.out, moves.MoveStartingLine(host, v.From, v.To))
	if w.sayFailure(v) {
		return
	}
	switch v.State {
	case provision.OfflineStopping:
		if v.Next == "seal" {
			moveSay(w.m.out, fmt.Sprintf("Every box in %s has stopped %s. Run carlos move offline seal --host %s to copy it.", v.From, host, host))
			return
		}
		w.sayWaiting(r)
	case provision.OfflineSealing:
		moveSay(w.m.out, fmt.Sprintf("Copying %s to %s.", host, v.To))
	case provision.OfflineSealed:
		if v.Next == "flip" {
			moveSay(w.m.out, fmt.Sprintf("Copied and checked. Run carlos move offline flip --host %s to start it in %s.", host, v.To))
			return
		}
		w.sayWaiting(r)
	case provision.OfflineLanding:
		moveSay(w.m.out, moves.FlippedLine(v.From, v.To))
	case provision.OfflineLanded:
		if v.Next == "open" {
			moveSay(w.m.out, fmt.Sprintf("%s holds everything %s needs. Nothing is served yet. Run carlos move offline open --host %s to serve it from %s.", v.To, host, host, v.To))
			return
		}
		w.sayWaiting(r)
	case provision.OfflineOpen, provision.OfflineFinishing:
		if w.served(v) {
			w.sayServed(v)
		} else {
			w.sayWaiting(r)
		}
		w.sayCleaning(v)
	case provision.OfflineReverting:
		moveSay(w.m.out, fmt.Sprintf("Moving %s back to %s.", host, v.From))
		w.sayCleaning(v)
		if v.Next == "finish" {
			moveSay(w.m.out, fmt.Sprintf("Every box has put %s back in %s. Run carlos move offline finish --host %s to end the move.", host, v.From, host))
		}
	}
}

// offline runs one `carlos move offline` verb.
func (m *mover) offline(ctx context.Context, a moveArgs) int {
	w := &offlineRun{m: m, said: map[string]bool{}}
	switch a.offline {
	case "status":
		return w.runStatus(ctx)
	case "stage":
		return w.runStage(ctx, a.to)
	case "start":
		return w.runStart(ctx, a.to)
	case "seal":
		return w.runSeal(ctx)
	case "flip":
		return w.runFlip(ctx)
	case "open":
		return w.runOpen(ctx)
	case "finish":
		return w.runFinish(ctx)
	default: // abort
		return w.runAbort(ctx)
	}
}

// stray is a view the verb's step did not expect: a move that is no longer
// in flight says how it ended; any other state is said where it stands, and
// the operator decides.
func (w *offlineRun) stray(r offlineReply) int {
	if r.v.State == "none" {
		return w.ended(r.v)
	}
	w.headline(r)
	return moveExitWaiting
}

func (w *offlineRun) runStatus(ctx context.Context) int {
	r, code := w.poll(ctx)
	if code >= 0 {
		return code
	}
	v, host := r.v, w.m.host
	if v.State == "none" {
		if v.Last != nil {
			offlineOutcomeLine(w.m, v.Last)
		} else {
			moveSay(w.m.out, fmt.Sprintf("No move is in flight for %s.", host))
		}
		return moveExitDone
	}
	w.headline(r)
	if v.State == provision.OfflineStaging {
		// Nothing of the manifest is checked before the start, so the
		// items would all say waiting. The stage's own line said it.
		if v.Copying != nil && v.Copying.Phase == moves.OfflinePhaseFailed {
			return moveExitWaiting
		}
		return moveExitDone
	}
	failed := false
	for _, it := range v.Items {
		if it.Diagnostic == "" {
			moveSay(w.m.out, fmt.Sprintf("Item %s is %s.", it.Slug, it.Check))
		} else {
			moveSay(w.m.out, fmt.Sprintf("Item %s is %s: %s.", it.Slug, it.Check, it.Diagnostic))
		}
		failed = failed || it.Check == "failed"
	}
	for _, rep := range v.Reports {
		if rep.Fresh {
			moveSay(w.m.out, fmt.Sprintf("%s is %s as the %s.", rep.Box, rep.Phase, rep.Role))
		} else {
			moveSay(w.m.out, fmt.Sprintf("%s is %s as the %s. Its report is not current.", rep.Box, rep.Phase, rep.Role))
		}
	}
	for _, l := range v.Left {
		moveSay(w.m.out, fmt.Sprintf("%s left %s in place: %s.", l.Holder, l.Where, l.Because))
	}
	if _, _, stepFailed := v.failure(); failed || v.Failed || stepFailed {
		return moveExitWaiting
	}
	return moveExitDone
}

// runStage posts the stage door, then waits for the round that answers the
// stage request the post made (or a later one's, if another terminal ran
// stage meanwhile).
func (w *offlineRun) runStage(ctx context.Context, to string) int {
	r, code := w.post(ctx, "/stage", map[string]string{"region": to})
	if code >= 0 {
		return code
	}
	host := w.m.host
	if r.v.State != provision.OfflineStaging {
		return w.stray(r)
	}
	moveSay(w.m.out, fmt.Sprintf("Staging %s in %s. %s keeps serving while the data is copied.", host, r.v.To, host))
	asked := r.v.StagedAt
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		if r.v.State != provision.OfflineStaging {
			return w.stray(r), true
		}
		c := r.v.Copying
		if c == nil || c.Answered < asked || c.Answered != r.v.StagedAt {
			return 0, false
		}
		w.sayStaged(r.v)
		if c.Phase == moves.OfflinePhaseFailed {
			return moveExitWaiting, true
		}
		w.sayStartWhenReady(r.v)
		return moveExitDone, true
	})
}

func (w *offlineRun) runStart(ctx context.Context, to string) int {
	r, code := w.post(ctx, "", map[string]string{"region": to})
	if code >= 0 {
		return code
	}
	host := w.m.host
	moveSay(w.m.out, moves.MoveStartingLine(host, r.v.From, r.v.To))
	moveSay(w.m.out, fmt.Sprintf("%s is offline from now until the move opens or is aborted.", host))
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		switch r.v.State {
		case provision.OfflineStopping:
			if r.v.Next == "seal" {
				moveSay(w.m.out, fmt.Sprintf("Every box in %s has stopped %s. Run carlos move offline seal --host %s to copy it.", r.v.From, host, host))
				return moveExitDone, true
			}
			w.sayWaiting(r)
			return 0, false
		}
		return w.stray(r), true
	})
}

func (w *offlineRun) runSeal(ctx context.Context) int {
	r, code := w.post(ctx, "/seal", struct{}{})
	if code >= 0 {
		return code
	}
	host := w.m.host
	moveSay(w.m.out, fmt.Sprintf("Copying %s to %s.", host, r.v.To))
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		switch r.v.State {
		case provision.OfflineSealing, provision.OfflineSealed:
			if w.sayFailure(r.v) {
				return moveExitWaiting, true
			}
			if r.v.State == provision.OfflineSealed {
				if r.v.Next == "flip" {
					moveSay(w.m.out, fmt.Sprintf("Copied and checked. Run carlos move offline flip --host %s to start it in %s.", host, r.v.To))
					return moveExitDone, true
				}
				w.sayWaiting(r)
			}
			return 0, false
		}
		return w.stray(r), true
	})
}

func (w *offlineRun) runFlip(ctx context.Context) int {
	r, code := w.post(ctx, "/flip", struct{}{})
	if code >= 0 {
		return code
	}
	host := w.m.host
	moveSay(w.m.out, moves.FlippedLine(r.v.From, r.v.To))
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		switch r.v.State {
		case provision.OfflineLanding, provision.OfflineLanded:
			if w.sayFailure(r.v) {
				return moveExitWaiting, true
			}
			if r.v.State == provision.OfflineLanded {
				if r.v.Next == "open" {
					moveSay(w.m.out, fmt.Sprintf("%s holds everything %s needs. Nothing is served yet. Run carlos move offline open --host %s to serve it from %s.", r.v.To, host, host, r.v.To))
					return moveExitDone, true
				}
				w.sayWaiting(r)
			}
			return 0, false
		}
		return w.stray(r), true
	})
}

func (w *offlineRun) runOpen(ctx context.Context) int {
	r, code := w.post(ctx, "/open", struct{}{})
	if code >= 0 {
		return code
	}
	// The line is said once the target serves (its report says opened, or the
	// next step is finish), not when the door only accepted the step.
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		if r.v.State == provision.OfflineOpen {
			if w.sayFailure(r.v) {
				return moveExitWaiting, true
			}
			if w.served(r.v) {
				w.sayServed(r.v)
			} else {
				w.sayWaiting(r)
			}
			if r.v.Next == "finish" {
				return moveExitDone, true
			}
			if w.sayFailure(r.v) {
				return moveExitWaiting, true
			}
			w.sayCleaning(r.v)
			return 0, false
		}
		return w.stray(r), true
	})
}

// finish posts until the move is cleared. The first post (from open) starts
// the clean-up and the console answers offline-finishing; once the boxes have
// cleaned up and the tombstone is lifted, the status shows the next verb is
// finish again and the second post ends the move. A move reverting from an
// abort needs only the one.
func (w *offlineRun) runFinish(ctx context.Context) int {
	cur, code := w.poll(ctx)
	if code >= 0 {
		return code
	}
	for {
		// Wait until the door would accept: the boxes' clean-up shows in the
		// status before the post, so the door is never asked early.
		switch cur.v.State {
		case provision.OfflineOpen, provision.OfflineFinishing, provision.OfflineReverting:
			if cur.v.Next != "finish" {
				code := w.follow(ctx, cur, func(r offlineReply) (int, bool) {
					switch r.v.State {
					case provision.OfflineOpen, provision.OfflineFinishing, provision.OfflineReverting:
						if r.v.Next == "finish" {
							cur = r
							return -1, true
						}
						if w.sayFailure(r.v) {
							return moveExitWaiting, true
						}
						w.sayCleaning(r.v)
						return 0, false
					}
					cur = r
					return -1, true
				})
				if code >= 0 {
					return code
				}
			}
		}
		prev := cur.v.State
		r, code := w.post(ctx, "/finish", struct{}{})
		if code >= 0 {
			return code
		}
		if r.v.State == "none" {
			return w.ended(r.v)
		}
		if r.v.State == prev {
			return w.stray(r)
		}
		cur = r
	}
}

func (w *offlineRun) runAbort(ctx context.Context) int {
	r, code := w.post(ctx, "/abort", struct{}{})
	if code >= 0 {
		return code
	}
	// A move aborted while staging was never offline: there is nothing to
	// move back, only the staged copy to remove.
	if r.v.NeverOffline {
		return w.abortStaged(ctx, r)
	}
	host := w.m.host
	moveSay(w.m.out, fmt.Sprintf("Moving %s back to %s.", host, r.v.From))
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		if r.v.State == provision.OfflineReverting {
			if r.v.Next == "finish" {
				moveSay(w.m.out, fmt.Sprintf("Every box has put %s back in %s. Run carlos move offline finish --host %s to end the move.", host, r.v.From, host))
				return moveExitRolled, true
			}
			w.sayCleaning(r.v)
			return 0, false
		}
		return w.stray(r), true
	})
}

// abortStaged follows an abort from staging: the stager removes the staged
// copy, every source box reports, and the operator runs finish.
func (w *offlineRun) abortStaged(ctx context.Context, r offlineReply) int {
	host := w.m.host
	moveSay(w.m.out, fmt.Sprintf("Removing the staged copy of %s from %s.", host, r.v.To))
	return w.follow(ctx, r, func(r offlineReply) (int, bool) {
		if r.v.State == provision.OfflineReverting {
			if r.v.Next == "finish" {
				moveSay(w.m.out, fmt.Sprintf("The staged copy of %s is removed. Run carlos move offline finish --host %s to end the move.", host, host))
				return moveExitRolled, true
			}
			return 0, false
		}
		return w.stray(r), true
	})
}