// Package provision owns instance provisioning as a control-plane data
// write: the bucket record and app manifest shapes, their validation, the
// store operations the console's instance verbs use, and the route
// conventions shared with `carlos add`. The reconcile pass that turns
// records into routes lives in reconcile.go (Task 3). Spec:
// docs/superpowers/specs/2026-08-04-instance-provisioning-design.md.
package provision

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/controlname"
	"github.com/carlosframework/platform/internal/controlsock"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"golang.org/x/crypto/bcrypt"
)

// Record is the wire shape at apps/<sqid>/<app>/instances/<host>.json — one
// per provisioned instance, create-only so a lost race is a 409 and never a
// silent overwrite. The body must agree with its own key path (Host/App/
// Account vs the key's segments); GetRecord and ListRecords enforce that, a
// record that lies about identity is never guessed at (design §3.2).
type Record struct {
	// RecordSeq is advanced by every record writer once present; absent is legacy zero.
	RecordSeq   uint64 `json:"record_seq,omitempty"`
	Acquisition string `json:"acq,omitempty"`
	// Maintenance is preserved by ordinary writers; only a future guarded
	// maintenance transition may install, replace, or clear it.
	Maintenance MaintenanceProjection `json:"maintenance,omitempty"`
	Schema      int                   `json:"schema"` // 1
	Host        string                `json:"host"`
	Slug        string                `json:"slug"`
	App         string                `json:"app"`
	Account     string                `json:"account"`
	Pool        string                `json:"pool,omitempty"`
	Region      string                `json:"region,omitempty"`
	Channel     string                `json:"channel"`
	CreatedAt   time.Time             `json:"created_at"`
	CreatedBy   string                `json:"created_by,omitempty"`

	// Move is an in-flight residence move (issue #331, instance-move spec):
	// present only between move-prepare and the cut's terminal flip, absent
	// on every record at rest. Additive + omitempty — pre-move records parse
	// unchanged. Written ONLY through SetRecordMove's guarded transitions.
	Move *MoveState `json:"move,omitempty"`

	// LastMove is the terminal outcome the close CAS leaves as it clears
	// Move (index D11), so a CLI that missed the intermediate states still
	// reads how the move it requested ended. Written by the guarded close
	// transitions only (MoveGuardedAbortComplete, and the done and
	// rolled_back closes the later plans add). Additive and omitempty:
	// absent on every record at rest on main.
	LastMove *MoveOutcome `json:"last_move,omitempty"`

	// ActiveRun is the instance's `carlos run`, if any (run spec §5.1):
	// queued, picked or cancelling while open, then done, cancelled or
	// expired until the console clears it. Runs, moves and imports all
	// change this one object, so they serialize on its ETag. Ordinary
	// writers carry it through unchanged; only the Run* transitions
	// (run.go) write it. Additive and omitempty.
	ActiveRun *RunState `json:"run,omitempty"`

	// RunKeys is the instance's recent keyed run admissions (service runs
	// spec § 3): each Idempotency-Key, scoped to its caller and hashed,
	// with the run it admitted. RunAdmit writes it in the same
	// compare-and-swap as the admission, so admission and dedupe are one
	// write; every other writer carries it through unchanged. Additive and
	// omitempty. A writer built before it drops it, which is why keyed
	// creates wait for CARLOS_RUN_KEYS=on.
	RunKeys []RunKey `json:"run_keys,omitempty"`

	// Seed is where this instance's database started (seeded-instances spec
	// §3.2): a sibling's replica (--from) or an uploaded snapshot (import).
	// Additive and omitempty. Like Maintenance, it is preserved by ordinary
	// writers and changed only through the named seed transitions (seed.go);
	// it stays on the record after placement as provenance.
	Seed *SeedSource `json:"seed,omitempty"`

	// --- shape fields (2026-08-10 route-records design §3) --------------
	// Every one is omitempty with a zero value that reproduces the mint
	// reconcile hardcoded before they existed: a record written by an older
	// console still materializes a byte-identical route.

	// Kind is the route class: "instance" | "service" | "static". Empty
	// means "instance", which is what reconcile minted unconditionally
	// before this field existed.
	Kind string `json:"kind,omitempty"`
	// Addr is a TCP upstream as host:port, for whole-app tenants whose
	// process the platform does not spawn (kass, keymail, titogo, the
	// consoles). Set means the route proxies to it and derives no socket.
	Addr string `json:"addr,omitempty"`
	// Unit is the systemd unit that owns this route's process, so adoption
	// can cycle it (console.service — issue #110). See registry.Route.Unit
	// for why a typo here is a route whose adoption silently restarts
	// nothing; ValidateShape is the door that refuses one.
	Unit string `json:"unit,omitempty"`
	// Hibernate is a POINTER because absent and false are different facts:
	// every record already in a production bucket has no such key, and a
	// plain bool would read those as false and un-hibernate the fleet on
	// the first pass after deploy. nil = HibernateOrDefault's derivation.
	Hibernate *bool `json:"hibernate,omitempty"`
	// Environment is the named config bundle this route binds to ("" = the
	// app's default bundle alone). The second member-facing axis of
	// `carlos route`, and owned by this record for the same reason Channel
	// is (design §5).
	Environment string `json:"environment,omitempty"`
	// StatusNamed opts this app into being named on the public status page.
	// A POINTER for the same reason Hibernate is: this column is CONVERGED,
	// so absent must mean "this record does not speak to the status page" —
	// a plain bool would read every legacy record (and every record whose
	// writer left the field out) as an explicit false and silently opt out a
	// route an operator opted in with `carlos add -status-named`. nil =
	// StatusNamedOrDefault's answer.
	StatusNamed *bool `json:"status_named,omitempty"`

	// --- multi-upstream fields (2026-08-12, HA design, issue #152) --------

	// Addrs is a route's upstream set, host:port each, for a tenant fronting
	// more than one process. Mutually exclusive with Addr — a record picks
	// one shape and ValidateShape refuses both set. Empty means "no
	// upstreams of its own" (AddrsOrDefault folds Addr in for the legacy
	// single-upstream case).
	Addrs []string `json:"addrs,omitempty"`
	// Health is the route's HTTP probe config. nil means unset;
	// HealthOrDefault derives the answer (probed once Addrs names more than
	// one upstream, unprobed otherwise — today's behavior for every record
	// that predates this field). Wire key is "probe", not "health": the
	// console's instance-list API already ships "health" for a computed
	// one-word status string (internal/console/apiinstances.go's
	// instanceListRecord.Health) that a deployed CLI parses — that surface
	// is shipped and keeps the name, this field is new and yields it.
	Health *Health `json:"probe,omitempty"`

	// --- steering fields (2026-08-23, multi-region ingress design Part 4) --

	// Steering is this route's DNS steering opt-in: "" (default — today's
	// static per-host answer, and what every record written before this
	// field existed reads as) or "latency" (a Route53 latency record per
	// armed edge, converged by the steering housekeeper duty — Tasks 4-6 of
	// the multi-region-steering plan). A STRING, deliberately, not a bool:
	// Part 4 names other steering policies as future work, and a string
	// leaves room for a new value to land without another wire-shape break.
	// ValidSteering is the door.
	//
	// SINGLE-BOX INERT as of this field's introduction (plan Task 2):
	// nothing reads it yet. It reaches the registry row (routeFor,
	// convergeOwned) and nothing further — the console door (Task 3) is
	// what lets a member set it, and the desired-state builder and Route53
	// converger (Tasks 4-6) are what give a non-empty value any effect, and
	// even then only once a deployment arms CARLOS_STEERING_ZONE_ID.
	Steering string `json:"steering,omitempty"`

	// Replica is NOT a separate record field: Role carries the role
	// vocabulary alone (ValidRole's door), and every reader derives the
	// boolean from it. A second spelling of the same fact is two chances
	// to disagree.
	//
	// Gate is the bcrypt hash of the host's shared password ("" = open,
	// the value every record written before this field existed decodes
	// to). The ONLY place the secret material lives in the bucket; the
	// plaintext crosses the console once, at the door that hashes it
	// (internal/gate.Hash), and is never stored, logged or echoed — the
	// console's record views shadow this field (apiinstances.go). A
	// string, not a pointer: absent and "" are the same fact, an open
	// host. Converged into routes.gate by Reconcile's owned-converge arm
	// (reconcile.go) and read by the edge on every request. Spec docs/superpowers/specs/2026-09-07-password-gate-design.md.
	Gate string `json:"gate,omitempty"`

	// --- replica fields (carlos/platform discussion #162, edge checkout) ---

	// Role is what this instance IS within its app: "" (the default every
	// record written before this field existed decodes to) is the serving
	// home — the instance the account's hosts resolve to, the one whose
	// data is authoritative. "replica" is a SECOND instance of the same
	// account, holding a one-way mirror of it, placed near the buyers: an
	// edge-checkout shard selling against a capacity lease from home and
	// draining orders back in batches (titogo edge-checkout design
	// record, 2026-09-21). The role is NOT a routing fact by itself — it
	// takes effect through Replicas (below) and the reconciler's mint
	// gate — and ValidRole is the door.
	//
	// Written only by the create door that also names the placement; no
	// PATCH axis flips a record into or out of the replica role in place
	// (delete and re-create is the supported transition), because a role
	// flip is a data-plane event — which database is authoritative — not
	// a shape edit.
	Role string `json:"role,omitempty"`

	// Replicas names the placement codes an account's replicas live in:
	// the home record's own opt-in, "ie,uk" style, deduplicated,
	// canonically sorted and validated against ValidRegion. Empty on
	// every record that predates the field = no replicas anywhere =
	// today's routing, byte for byte. It is CONVERGED (Reconcile writes
	// the record's value over the row's on every pass, through
	// SetReplicas — the one deliberate exception to that arm's
	// never-widen rule, because replica placement is the bucket's
	// decision to own: a stale placement left behind after the set
	// shrank would keep a retired region's replica on the account's
	// hosts forever), published by the edges file, honoured by the
	// steering converger — and never read by the edge's own dispatch,
	// which never hairpins a replica.
	Replicas []string `json:"replicas,omitempty"`

	// --- update settings (instance update settings spec §3.1, §4.2) -----
	// Every one is omitempty with "never set" as its zero value, so a record
	// written before them decodes with none and reconcile resolves it to
	// auto: today's adoption. Set only through the console's /updates doors.

	// Updates is "auto", "manual", or "" (never set: the app default).
	Updates string `json:"updates,omitempty"`
	// Major holds the instance on one semver major of stable. nil: never set.
	Major *int `json:"major,omitempty"`
	// Pin is the exact build to run. It must have a promotion record.
	Pin string `json:"pin,omitempty"`
	// InstallRequest is "Update now": an id that only ever increases. Root
	// resolves and completes it (plan 4); the console never chooses a build.
	InstallRequest *InstallRequest `json:"install_request,omitempty"`
}

// Health is the per-route probe config (HA spec Part II). Zero value =
// unset; WithDefaults fills in the platform's defaults for whatever the
// caller left at zero.
type Health struct {
	Path       string `json:"path,omitempty"`        // default "/"
	Status     int    `json:"status,omitempty"`      // default 200
	HostHeader string `json:"host_header,omitempty"` // default: the route host
	IntervalMS int    `json:"interval_ms,omitempty"` // default 2000
	FailAfter  int    `json:"fail_after,omitempty"`  // default 3
	RiseAfter  int    `json:"rise_after,omitempty"`  // default 2
}

// WithDefaults fills the five zero-valued fields above with the platform's
// defaults. HostHeader is deliberately left as-is (including empty) — the
// edge substitutes the route's own host for an unset HostHeader, and that
// substitution needs to see "unset", not a value this method invented.
func (h Health) WithDefaults() Health {
	if h.Path == "" {
		h.Path = "/"
	}
	if h.Status == 0 {
		h.Status = 200
	}
	if h.IntervalMS == 0 {
		h.IntervalMS = 2000
	}
	if h.FailAfter == 0 {
		h.FailAfter = 3
	}
	if h.RiseAfter == 0 {
		h.RiseAfter = 2
	}
	return h
}

// Manifest is the wire shape at apps/<sqid>/<app>/instances.json — the app's
// opt-in to instance provisioning and the policy every record is checked
// against (host suffix, region allow-list, seed argv). Absence means the
// feature does not exist for that app, the same convention litestream's own
// manifest uses (design §3.1).
type Manifest struct {
	Schema  int      `json:"schema"` // 1
	App     string   `json:"app"`
	Domain  string   `json:"domain"`
	Pool    string   `json:"pool,omitempty"`
	Channel string   `json:"channel,omitempty"`
	Regions []string `json:"regions,omitempty"`
	Seed    []string `json:"seed,omitempty"`

	// PrivateStorage is the app's standing policy that every instance
	// created under this manifest gets a config environment of its own and
	// an object store bound to that environment (issue #412, spec
	// 2026-09-22-instance-private-storage-design.md §5). Declared once, by
	// a signed-in person, at `carlos instances enable --private-storage`;
	// spent by the create door, which an `instance`-role service credential
	// can already reach. That split is the point: the policy widens no
	// role, so a signup path provisions a tenant's bucket without ever
	// holding the power to create one.
	//
	// A plain bool, not a pointer, and the difference from Record's
	// Hibernate and StatusNamed is worth stating because their reason does
	// not apply here. Those are CONVERGED columns — a box re-reads them
	// every pass, so "absent" has to be distinguishable from an explicit
	// false or the first pass after deploy acts on a value nobody wrote.
	// This is read at one door, at create time, and false IS the behaviour
	// every manifest in a production bucket already has. There is nothing
	// for absence to mean that false does not already mean.
	PrivateStorage bool `json:"private_storage,omitempty"`

	// Moves is the app's move declaration (move-any-app design §7.3 S1):
	// the readiness path a probe requests. Absent means "/".
	Moves *MovesManifest `json:"moves,omitempty"`

	// Updates is the app's instance update policy (spec §3.2). nil: no
	// instance may change anything itself, and the defaults are auto, this
	// manifest's Channel, and no major hold.
	Updates *UpdatePolicy `json:"updates,omitempty"`
}

type MovesManifest struct {
	ReadyPath string `json:"ready_path,omitempty"`
}

// ReadyPathOrDefault is the probe's path (§7.7).
func (m Manifest) ReadyPathOrDefault() string {
	if m.Moves != nil && m.Moves.ReadyPath != "" {
		return m.Moves.ReadyPath
	}
	return "/"
}

// ManifestKey, RecordsPrefix and RecordKey are the bucket key conventions
// for a provisioned app — beside release.AppPrefix's channels/ and
// releases/, matching the ledgers.json precedent of a per-app manifest
// sitting beside the collection it governs.
func ManifestKey(acct, app string) string   { return release.AppPrefix(acct, app) + "instances.json" }
func RecordsPrefix(acct, app string) string { return release.AppPrefix(acct, app) + "instances/" }
func RecordKey(acct, app, host string) string {
	return RecordsPrefix(acct, app) + host + ".json"
}

// ParseRecordKey reports whether key is an account-qualified instance
// record (apps/<sqid>/<app>/instances/<host>.json) known to belong to a
// materialized account, and if so its account, app and host. Mirrors
// domains.appKeyOwner in miniature: known guards against a key whose
// account has not materialized yet, and app is refused when it collides
// with the release plane's own words ("channels", "releases") — the same
// two-namespace collision appKeyOwner documents. A key with anything but
// exactly one non-empty segment after "instances/" is not a record: the
// manifest (instances.json, no trailing segment) and any nested path both
// refuse.
func ParseRecordKey(key string, known map[string]bool) (acct, app, host string, ok bool) {
	rest, ok := strings.CutPrefix(key, "apps/")
	if !ok {
		return "", "", "", false
	}
	acct, rest, ok = strings.Cut(rest, "/")
	if !ok || !known[acct] {
		return "", "", "", false
	}
	app, rest, ok = strings.Cut(rest, "/")
	if !ok || app == "" || app == "channels" || app == "releases" {
		return "", "", "", false
	}
	seg, rest, ok := strings.Cut(rest, "/")
	if !ok || seg != "instances" {
		return "", "", "", false
	}
	if rest == "" || strings.Contains(rest, "/") {
		return "", "", "", false
	}
	host, ok = strings.CutSuffix(rest, ".json")
	if !ok || host == "" {
		return "", "", "", false
	}
	return acct, app, host, true
}

// ValidHost reports whether host is exactly one label followed by
// "."+domain. This is path-traversal defense stated as such: the host
// becomes a bucket key segment (RecordKey) and, on the box, the
// <host>.sock / <host>.db filesystem path segments — the label regex is
// what keeps a hostile host from ever reaching either as anything but an
// opaque, safe path component.
func ValidHost(host, domain string) bool {
	if domain == "" || len(host) > 253 {
		return false
	}
	label, ok := strings.CutSuffix(host, "."+domain)
	if !ok || label == "" {
		return false
	}
	return controlname.ValidLabel(label)
}

// ValidHostShape checks host's shape alone, with no domain to validate a
// suffix against — for doors that must still refuse a hostile host after
// its app's manifest (and so its domain) is gone: a delete, for instance,
// which must work even if the app disabled instances since the record was
// created.
func ValidHostShape(host string) bool { return controlname.ValidHostShape(host) }

// KindOrDefault resolves the record's route class. "" means "instance" —
// the class reconcile minted unconditionally before Kind existed, and so
// the only value that keeps a legacy record materializing unchanged.
func (r Record) KindOrDefault() string {
	if r.Kind == "" {
		return "instance"
	}
	return r.Kind
}

// DerivesPaths reports whether this record's route builds filesystem paths
// out of its host — the <host>.sock and <host>.db segments. It is the hinge
// two separate rules turn on: which host validator applies (ValidRecordHost:
// a host that becomes a path segment stays suffix-pinned, one that does not
// needs only a valid shape), and what hibernation defaults to (only a route
// with a socket has a process for the activator to wake).
func (r Record) DerivesPaths() bool {
	return r.KindOrDefault() == "instance" && len(r.AddrsOrDefault()) == 0
}

// AddrsOrDefault resolves the record's upstream set: Addrs if set, else the
// single-element fold of the legacy Addr field, else nil. Every caller that
// used to branch on Addr != "" reads this instead, so a record minted
// before Addrs existed proxies to exactly the upstream it always did.
func (r Record) AddrsOrDefault() []string {
	if len(r.Addrs) > 0 {
		return r.Addrs
	}
	if r.Addr != "" {
		return []string{r.Addr}
	}
	return nil
}

// HealthOrDefault resolves the record's probe config: Health if set, else
// the platform defaults once AddrsOrDefault names more than one upstream
// (a route fronting several processes needs to know which ones are up),
// else nil. A single-upstream record — Addrs unset, Addr alone or one
// element of Addrs — stays unprobed, the behavior every record had before
// Health existed.
func (r Record) HealthOrDefault() *Health {
	if r.Health != nil {
		return r.Health
	}
	if len(r.AddrsOrDefault()) > 1 {
		h := Health{}.WithDefaults()
		return &h
	}
	return nil
}

// HibernateOrDefault resolves the tri-state. An explicit value wins. Absent
// derives: a socket-backed instance hibernates (the activator model, and
// what reconcile minted unconditionally before this field existed, so every
// legacy record is unchanged), and everything else does not — a static
// route has no process to wake, and an addr-backed tenant is always on.
func (r Record) HibernateOrDefault() bool {
	if r.Hibernate != nil {
		return *r.Hibernate
	}
	return r.DerivesPaths()
}

// StatusNamedOrDefault resolves the tri-state. An explicit value wins.
// Absent is false — the status page's naming is an opt-in, and false is what
// reconcile minted unconditionally before this field existed, so every legacy
// record is unchanged. Absent and false are still different facts, and the
// difference is load-bearing at the converge site: absent leaves the row's
// column alone (an operator's `carlos add -status-named` survives), while an
// explicit false clears it.
func (r Record) StatusNamedOrDefault() bool {
	if r.StatusNamed != nil {
		return *r.StatusNamed
	}
	return false
}

// ValidKind pins the record's route-class vocabulary to the set `carlos
// add` documents. "status" is deliberately absent: the fleet status page's
// route is the platform's own, not an app's, and nothing should be able to
// mint one from an app-scoped record.
func ValidKind(kind string) bool {
	switch kind {
	case "", "instance", "service", "static":
		return true
	}
	return false
}

// ValidSteering pins the record's DNS-steering vocabulary (multi-region
// ingress design Part 4): "" (no opinion — a route's host resolves exactly
// as it does today) or "latency" (opt into a Route53 latency record per
// armed edge). Case-sensitive and closed, the same discipline ValidKind
// applies to its own vocabulary — "Latency" or "geo" are refused at this
// door rather than discovered as a steering record the Route53 converger
// (Tasks 4-6) silently ignores.
func ValidSteering(s string) bool {
	switch s {
	case "", "latency":
		return true
	}
	return false
}

// ValidRole pins the record's role vocabulary (#162): "" (the serving
// home — every record written before the field existed decodes to it,
// which is what makes the field additive) or "replica" (a one-way-mirror
// instance of the same account, selling near the buyer). Case-sensitive
// and closed, ValidSteering's discipline: a value a future reader has
// never heard of is refused where it is typed rather than discovered as a
// row nothing routes for.
func ValidRole(s string) bool {
	switch s {
	case "", "replica":
		return true
	}
	return false
}

// MaxReplicas caps the placement set one record may name. See validReplicas.
const MaxReplicas = 8

// ValidReplicas normalizes and validates a replica placement set: trim,
// drop empties, dedupe, sort (the canonical order every comma-joined
// column here keeps), and check each element against ValidRegion — the
// same vocabulary a record's own Region speaks, so "is this placement
// armed" stays one question with one answer. Exported because the console's
// create door canonicalizes the request body through the SAME derivation
// the reconciler re-checks at (neither side trusts the other about what is
// in the bucket). The count cap is the residence rule made concrete
// (design record § Residence): replicas are per-account opt-in constrained
// by the account's residence policy, not "sprinkle the fleet" — a number a
// deployment can raise deliberately beats an unbounded set nobody sized.
func ValidReplicas(regions []string) ([]string, error) {
	if len(regions) > MaxReplicas {
		return nil, fmt.Errorf("replicas: %d placements, at most %d — replicas hold order data in every region named until a drain confirms, so each one is a residence decision, not a cache hint", len(regions), MaxReplicas)
	}
	seen := make(map[string]bool, len(regions))
	out := make([]string, 0, len(regions))
	for _, r := range regions {
		r = strings.TrimSpace(r)
		if r == "" {
			continue
		}
		if !ValidRegion(r) {
			return nil, fmt.Errorf("replica placement %q: want a placement code (e.g. ie, uk)", r)
		}
		if seen[r] {
			continue
		}
		seen[r] = true
		out = append(out, r)
	}
	sort.Strings(out)
	return out, nil
}

// MaxUpstreams caps the Addrs set one record may name. The prober probes a
// host's upstreams SEQUENTIALLY once per interval (internal/edge/prober.go),
// and every scale figure the multi-upstream design rests on — probe traffic,
// per-tick wall time, the dashboard's per-upstream table — was drawn for a
// handful of backends behind one route, not hundreds. The cap is stated at
// the door so an oversized set is refused where it is typed, rather than
// discovered as a prober tick that no longer fits inside its own interval.
const MaxUpstreams = 16

// ValidateShape checks the §3 fields at the door, before a record is written
// or materialized. The rails live at the door, exactly as they do for
// registry.SetCaps and SetChannel — Upsert validates no column at all, so a
// value that reaches the row unvalidated is discovered months later as an
// adopt pass that restarted nothing (the `-backing unti` incident, cli.go).
func ValidateShape(rec Record) error {
	if !ValidKind(rec.Kind) {
		return fmt.Errorf("kind %q: want instance, service or static (empty means instance)", rec.Kind)
	}
	if !ValidRole(rec.Role) {
		return fmt.Errorf("role %q: want \"\" or \"replica\"", rec.Role)
	}
	if !ValidSteering(rec.Steering) {
		return fmt.Errorf("steering %q: want \"\" or \"latency\"", rec.Steering)
	}
	// A replica is a process holding a mirror of live data — never a file
	// tree, never a foreign upstream the platform did not spawn. Pinning
	// the class here (not only at the reconciler's mint gate) keeps the rail
	// holding no matter which writer produced the record — and the ADDR half
	// matters as much as the kind half: a replica with a public addr would
	// pass everywhereClass's dial test and mint on EVERY non-fleet box, the
	// residence rule's second door.
	if rec.Role == "replica" && rec.KindOrDefault() != "instance" {
		return errors.New("role replica with kind " + rec.KindOrDefault() + ": a replica is an instance of the account — a process the activator spawns holding a mirror of its data — drop kind or drop role")
	}
	if rec.Role == "replica" && len(rec.AddrsOrDefault()) > 0 {
		return errors.New("role replica with addr: a replica's process is the platform's own, spawned by the activator on the box its placement names — an upstream would make it an everywhere-class route and mint it on every box — drop the addr or drop role")
	}
	for _, r := range rec.Replicas {
		r = strings.TrimSpace(r)
		if r != "" && !ValidRegion(r) {
			return fmt.Errorf("replica placement %q: want a placement code (e.g. ie, uk)", r)
		}
	}
	if len(rec.Replicas) > MaxReplicas {
		return fmt.Errorf("replicas: %d placements, at most %d", len(rec.Replicas), MaxReplicas)
	}
	if rec.Gate != "" {
		if _, err := bcrypt.Cost([]byte(rec.Gate)); err != nil {
			return errors.New("gate: want a bcrypt hash — the console hashes the password; a record never carries it in clear")
		}
	}
	if rec.Addr != "" {
		if err := validAddr(rec.Addr); err != nil {
			return err
		}
	}
	if rec.Addr != "" && len(rec.Addrs) > 0 {
		return errors.New("addr and addrs both set: a record picks one upstream shape — the single legacy addr or the addrs set — drop one of the two")
	}
	if len(rec.Addrs) > MaxUpstreams {
		return fmt.Errorf("addrs: %d upstreams, at most %d: the prober walks a host's upstreams one after another every tick, and the whole design is sized for a handful of backends per route, not hundreds — split the fleet across routes instead", len(rec.Addrs), MaxUpstreams)
	}
	if len(rec.Addrs) > 0 {
		seen := make(map[string]bool, len(rec.Addrs))
		for _, a := range rec.Addrs {
			if err := validAddrsElement(a); err != nil {
				return err
			}
			if seen[a] {
				return fmt.Errorf("addrs: duplicate upstream %q", a)
			}
			seen[a] = true
		}
	}
	if rec.Unit != "" && !registry.ValidUnitName(rec.Unit) {
		return fmt.Errorf("unit %q: want a systemd service unit name (e.g. console.service)", rec.Unit)
	}
	if rec.Environment != "" && !validEnvironmentName(rec.Environment) {
		return fmt.Errorf("environment %q: want [a-z0-9-]{1,32}", rec.Environment)
	}
	if rec.KindOrDefault() == "static" {
		if len(rec.AddrsOrDefault()) > 0 {
			return errors.New("kind static with addr: a static route serves a file tree and proxies to nothing — drop one of the two")
		}
		if rec.Unit != "" {
			return errors.New("kind static with unit: a static route has no process for a unit to own — drop one of the two")
		}
		if rec.Hibernate != nil && *rec.Hibernate {
			return errors.New("kind static with hibernate: a static route has no process to wake, so arming the activator for it would wedge the host — drop hibernate")
		}
	}
	if rec.KindOrDefault() == "service" && len(rec.AddrsOrDefault()) == 0 {
		return errors.New("kind service with no addr: the platform only proxies to a service, so the record must name the upstream it proxies to")
	}
	// Same wedge as static+hibernate, arriving from the other side. Addr does
	// NOT exclude a route from activator management: internal/edge's gate is
	// `Kind == "instance" && Hibernate && !viaTunnel`, so an addr-backed
	// instance flagged for hibernation is armed for an exec-child wake it has
	// no socket for and whose process the platform never spawned.
	// HibernateOrDefault's own derivation already states the invariant ("an
	// addr-backed tenant is always on"); this is the door that enforces it.
	if len(rec.AddrsOrDefault()) > 0 && rec.Hibernate != nil && *rec.Hibernate {
		return errors.New("addr with hibernate: an addr-backed route proxies to a process the platform does not spawn and has no socket to wake, so arming the activator for it would wedge the host — drop hibernate")
	}
	return nil
}

// validAddr checks one upstream's shape: host:port, with the port an
// integer in 1..65535. This is the LEGACY single Addr's rail, unchanged:
// records carrying it exist in production buckets today, some of them naming
// their backend by hostname, so nothing here may grow stricter. The new Addrs
// set gets the stricter rail instead — see validAddrsElement.
func validAddr(a string) error {
	h, p, err := net.SplitHostPort(a)
	if err != nil || h == "" || p == "" {
		return fmt.Errorf("addr %q: want a TCP upstream as host:port (e.g. 127.0.0.1:8090)", a)
	}
	// Parse port as integer and validate range 1..65535. net.SplitHostPort
	// accepts garbage ports (e.g. "127.0.0.1:abc") — we need to refuse them
	// here at the door, not discover them months later as a route that never
	// connects (the -backing unti class of incident).
	portNum, err := strconv.ParseUint(p, 10, 16)
	if err != nil || portNum == 0 {
		return fmt.Errorf("addr %q: port must be an integer in the range 1..65535", a)
	}
	// 169.254.0.0/16 — IPv4 link-local, which on every cloud box the platform
	// runs on is the instance metadata service (169.254.169.254) and its
	// neighbours. An upstream is a URL the edge fetches on the caller's
	// behalf, so a route pointed there is an SSRF into the box's own
	// credential endpoint, dressed as an ordinary proxy. Refused at the same
	// door the port range is, so the record can never reach a box.
	//
	// This rail is DELIBERATELY minimal (ruling 2026-08-12): loopback and
	// RFC1918 stay allowed, because that is where every legitimate addr-backed
	// route on the fleet points today (127.0.0.1:8090 consoles, private-subnet
	// tenants). The wider question — what a MEMBER may state — was ruled on
	// in discussion 50 (GitHub #170) and is answered at the console's doors
	// (refuseMemberUpstreams), not here; do not widen this check into an
	// answer for it.
	if ip := net.ParseIP(h); ip != nil && ip.To4() != nil && ip.IsLinkLocalUnicast() {
		return fmt.Errorf("addr %q: 169.254.0.0/16 is link-local (the cloud metadata service lives there) — an upstream must be a real backend address", a)
	}
	return nil
}

// validAddrsElement is validAddr plus the two rails the NEW Addrs set carries
// and the legacy Addr cannot (wire-compat: no record in any bucket carries
// Addrs yet, so this field's rules can be written the way they should have
// been all along, while Addr keeps exactly the checks it shipped with).
//
//  1. The host must be an IP LITERAL. validAddr's link-local refusal only
//     looks at a parsed IP, so a NAME skips it entirely and
//     "md.attacker.example:80" resolving to 169.254.169.254 walks straight
//     through the very rail that exists to stop it. A name cannot be
//     validated at the door at all, in fact: the platform dials these
//     upstreams itself, and whatever a name resolved to when the record was
//     written is not what it must resolve to when the edge connects — the
//     target is free to move after validation. An address that is checked
//     here and dialled there has to be the same address.
//  2. Link-local is refused in BOTH families, unicast and multicast:
//     169.254.0.0/16 AND fe80::/10 (plus ff02::/16), because a box reaches
//     its metadata service over IPv6 as readily as IPv4 and "[fe80::1]:80"
//     is the same SSRF wearing the other family's clothes.
//
// Everything else validAddr allows stays allowed here — loopback and RFC1918
// are where every legitimate addr-backed route on the fleet points today
// (127.0.0.1:8090 consoles, private-subnet tenants). This is the metadata
// rail closing its own holes, not a widening of it. Whether a MEMBER may
// state an upstream at all is the console's call: refuseMemberUpstreams.
func validAddrsElement(a string) error {
	if err := validAddr(a); err != nil {
		return err
	}
	h, _, err := net.SplitHostPort(a)
	if err != nil {
		return fmt.Errorf("addr %q: want a TCP upstream as host:port (e.g. 127.0.0.1:8090)", a)
	}
	ip := net.ParseIP(h)
	if ip == nil {
		return fmt.Errorf("addrs: upstream %q names a host, not an IP: the platform dials these addresses itself, and a name can resolve somewhere else after this check passes — state the backend's IP address as ip:port (e.g. 10.0.0.1:3000)", a)
	}
	if ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() {
		return fmt.Errorf("addrs: upstream %q is link-local (169.254.0.0/16 and fe80::/10, where the cloud metadata service lives) — an upstream must be a real backend address", a)
	}
	return nil
}

// ValidRecordHost applies the host rail this record actually needs (design
// §3.2). ValidHost is path-traversal defence: it is what keeps a hostile
// host from reaching the <host>.sock and <host>.db path segments as anything
// but an opaque, safe component, and it stays exactly where that danger is.
// A record that derives no filesystem path has no such exposure — its host
// reaches a bucket key and the edge's routing table only — so it needs the
// shape check alone, which is what lets an apex (carloku.com) or a console's
// own hostname be expressed at all.
func ValidRecordHost(rec Record, domain string) bool {
	if rec.DerivesPaths() {
		return ValidHost(rec.Host, domain)
	}
	return ValidHostShape(rec.Host)
}

// regionPoolRe is the shared charset for a region or pool code: lowercase
// a-z, 0-9, hyphen, up to 16 characters. Empty is deliberately valid — the
// unconstrained region and the shared (default) pool are both "".
var regionPoolRe = regexp.MustCompile(`^[a-z0-9-]{0,16}$`)

func ValidRegion(region string) bool { return regionPoolRe.MatchString(region) }

// replicaSet renders a record's replica placement set for the wire column
// that carries it (registry.Route.Replicas): comma-joined placement codes in
// canonical (sorted, deduplicated, ValidRegion-checked at the door) order,
// "" = none. routeFor and convergeOwned both call this — one derivation,
// two writers, the upstreamCols rule: a second, independent rendering is how
// a route minted correctly comes to be "converged" onto a different value
// every pass, forever.
func replicaSet(regions []string) string {
	norm, err := ValidReplicas(regions)
	if err != nil {
		// The door has already checked the set; a record that slipped past
		// it (a hand-written bucket body — the reconcile pass reads those)
		// still renders DETERMINISTICALLY, never as two different columns
		// on two boxes. Sorted, deduped, best-effort — and LOGGED, so a
		// hand-edited body is visible rather than silently normalized.
		log.Printf("provision: replicas set failed validation (%v) — rendering the sorted, deduplicated remainder", err)
		seen := map[string]bool{}
		var keep []string
		for _, r := range regions {
			if r = strings.TrimSpace(r); r != "" && !seen[r] {
				seen[r] = true
				keep = append(keep, r)
			}
		}
		norm = keep
		sort.Strings(norm)
	}
	return strings.Join(norm, ",")
}
func ValidPool(pool string) bool { return regionPoolRe.MatchString(pool) }

// envNameRe is appconfig.ValidEnvironmentName's own shape (pipelines design
// §4a), spelled out again here rather than imported: appconfig already
// imports this package (LoadMerged reads a route's environment column),
// so the reverse import would be a cycle. A record's Environment column
// and a bundle's environment name are the SAME vocabulary; if the shape
// in appconfig/environments.go ever changes, this copy must change with
// it. It exists only because of the import direction, not because the
// rule is provision's to define.
var envNameRe = regexp.MustCompile(`^[a-z0-9-]{1,32}$`)

// validEnvironmentName is appconfig.ValidEnvironmentName's check, duplicated
// for the reason envNameRe's comment gives: "default" names the implicit
// base bundle, never a named overlay, and "" is not a record's problem to
// validate here (the field is optional; the caller above already gates on
// rec.Environment != "").
func validEnvironmentName(name string) bool {
	return name != "default" && envNameRe.MatchString(name)
}

// GetManifest reads app's instances manifest. Not-found stays
// os.ErrNotExist-wrapped (absence IS the "instances disabled" answer, not
// an error condition the caller must special-case beyond errors.Is). A
// schema this binary doesn't understand is refused rather than silently
// truncated by json.Unmarshal (appconfig/store.go's precedent) — acting on
// a newer manifest would silently discard whatever the unknown fields meant.
func GetManifest(ctx context.Context, s release.Store, acct, app string) (*Manifest, error) {
	m, _, err := GetManifestETag(ctx, s, acct, app)
	return m, err
}

// GetManifestETag is GetManifest carrying the etag its bytes were read
// under — for a caller that must write back GUARDED against what it read
// (PutManifestIfMatch), not against whatever is there at write time.
func GetManifestETag(ctx context.Context, s release.Store, acct, app string) (*Manifest, string, error) {
	raw, etag, err := s.Get(ctx, ManifestKey(acct, app))
	if err != nil {
		return nil, "", err
	}
	var m Manifest
	if err := json.Unmarshal(raw, &m); err != nil {
		return nil, "", fmt.Errorf("provision: manifest %s/%s: %w", acct, app, err)
	}
	if m.Schema > 1 {
		return nil, "", fmt.Errorf("provision: manifest %s/%s: schema %d is newer than this binary understands — upgrade carlos", acct, app, m.Schema)
	}
	return &m, etag, nil
}

// PutManifest publishes app's instances manifest — a CAS upsert (read
// etag, put, one retry on a lost race), the WriteAccountManifest shape,
// because unlike a record this object is meant to be overwritten as an
// operator adjusts policy, not created once.
func claimRawPutManifest(ctx context.Context, s release.Store, acct, app string, m Manifest) error {
	key := ManifestKey(acct, app)
	body, err := json.Marshal(m)
	if err != nil {
		return err
	}
	put := func() error {
		etag := ""
		if _, cur, err := s.Get(ctx, key); err == nil {
			etag = cur
		} else if !errors.Is(err, os.ErrNotExist) {
			return err
		}
		_, err := s.Put(ctx, key, body, etag)
		return err
	}
	if err := put(); errors.Is(err, release.ErrCASConflict) {
		return put() // one retry on a lost race; a second loss surfaces
	} else if err != nil {
		return err
	}
	return nil
}

// PutManifestIfMatch writes app's instances manifest guarded by the etag
// the caller READ it under — "" meaning it read nothing, which makes the
// write create-only. Unlike PutManifest's upsert (which re-reads inside and
// retries, so the last writer always wins) a lost race here surfaces as
// release.ErrCASConflict for the caller to answer: the console's
// member-facing enable door uses this because its domain-conflict check is
// only honest if nothing can slip in between the check and the write
// (review 2026-08-06 — an operator's store-mode manifest landing in that
// window must never be silently re-homed).
func claimRawPutManifestIfMatch(ctx context.Context, s release.Store, acct, app string, m Manifest, etag string) error {
	body, err := json.Marshal(m)
	if err != nil {
		return err
	}
	_, err = s.Put(ctx, ManifestKey(acct, app), body, etag)
	return err
}

// CreateRecord writes a new instance record, create-only: a lost race
// against a second signup for the same host comes back wrapping
// release.ErrCASConflict, never a silent overwrite of the winner's record.
func CreateRecord(ctx context.Context, s release.Store, rec Record) error {
	if len(rec.Maintenance) != 0 {
		return ErrMaintenanceProjection
	}
	body, err := json.Marshal(rec)
	if err != nil {
		return err
	}
	_, _, err = WriteRecord(ctx, s, RecordKey(rec.Account, rec.App, rec.Host), []byte(`{}`), body, "")
	return err
}

// SetRecordRegion repoints an existing instance record's Region field — the
// one field a record is expected to change AFTER creation. Unlike
// CreateRecord (create-only: a lost race is a permanent 409) this is a
// narrow read-modify-write CAS — etag from Get, Put, one retry on a lost
// race — mirroring PutManifest's own retry shape, because unlike a record's
// identity (Host/App/Account, checked at CREATE and never touched again)
// Region is meant to be repointed in place.
//
// This is placement's (console/fleets.go, customer-fleets plan) write onto
// instances that already exist: "place" walks an app's already-provisioned
// records and calls this for each, so a fleet or platform placement change
// governs the NEXT reconcile pass for every resident instance, not only the
// next one created. Without it, `carlos apps place` would only ever change
// where FUTURE instances land — the app's existing, already-parked ones
// would keep the region they were created under forever, silently exempt
// from the placement a caller was just told had taken effect.
//
// Not-found is returned verbatim (os.ErrNotExist-wrapped): a caller walking
// a key list that has gone stale between listing and this call should see
// the record is gone, not a manufactured success for a write that touched
// nothing.
func SetRecordRegion(ctx context.Context, s release.Store, acct, app, host, region string) error {
	_, err := updateRecord(ctx, s, acct, app, host, func(rec *Record) error {
		if rec.Region != region {
			if err := seedMoveRefusal(rec); err != nil {
				return err
			}
		}
		rec.Region = region
		return nil
	})
	return err
}

// updateRecord is the one read-CAS-put-one-retry scaffold under both
// in-place record writers (SetRecordRegion, SetRecordMove): re-reading and
// re-applying a narrow mutation on a lost race changes nothing about what
// it meant. Returns the record as written.
func updateRecord(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) (*Record, error) {
	return updateRecordSeeding(ctx, s, acct, app, host, mut, false)
}

// updateRecordSeeding is updateRecord's scaffold. seedWrite is true only for
// the seed transitions: every other caller is refused a change to Seed.
func updateRecordSeeding(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error, seedWrite bool) (*Record, error) {
	key := RecordKey(acct, app, host)
	update := func() (*Record, error) {
		raw, etag, err := s.Get(ctx, key)
		if err != nil {
			return nil, err
		}
		rec, err := DecodeRecord(raw)
		if err != nil {
			return nil, fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err)
		}
		if rec.Account != acct || rec.App != app || rec.Host != host {
			return nil, ErrMaintenanceProjection
		}
		before := rec
		before.Maintenance = append(MaintenanceProjection(nil), rec.Maintenance...)
		before.Seed = cloneSeed(rec.Seed) // deep: mut edits rec.Seed through its pointer
		if err := mut(&rec); err != nil {
			if errors.Is(err, errRecordUnchanged) {
				return &rec, nil // the record already says it: no write, no sequence advance
			}
			return nil, err
		}
		if !protectedFieldsEqual(before, rec) {
			return nil, ErrMaintenanceProjection
		}
		if !seedWrite && !seedEqual(before.Seed, rec.Seed) {
			return nil, ErrSeedProtected
		}
		body, err := json.Marshal(rec)
		if err != nil {
			return nil, err
		}
		_, seq, err := WriteRecord(ctx, s, key, raw, body, etag)
		if err != nil {
			return nil, err
		}
		rec.RecordSeq = seq
		return &rec, nil
	}
	rec, err := update()
	if errors.Is(err, release.ErrCASConflict) {
		return update() // one retry on a lost race; a second loss surfaces
	}
	return rec, err
}

// MoveState is the record's move journal (move-any-app design §7.2). The
// claim carries the cross-operation serialization; this field carries one
// move's own phases, because only a move reads them. Every field after
// UpdatedAt is additive and omitempty: a v0.22 console reads the ones it
// knows and refuses to abort any state it does not (G16).
type MoveState struct {
	// ID is the move id (mv_…), Incarnation the holder incarnation the
	// claim named when the move began. Both guard every box CAS (§7.2).
	ID          string `json:"id,omitempty"`
	Incarnation string `json:"incarnation,omitempty"`
	From        string `json:"from,omitempty"`
	To          string `json:"to"`
	// Cut is "auto" or "manual" (§7.1).
	Cut         string     `json:"cut,omitempty"`
	State       string     `json:"state"`
	RequestedBy string     `json:"requested_by,omitempty"`
	RequestedAt time.Time  `json:"requested_at"`
	ReadyAt     *time.Time `json:"ready_at,omitempty"`
	// UpdatedAt is when the state last changed — stamped by SetRecordMove
	// itself on every transition that leaves a move in place, so the
	// console's flat move view can carry one freshness stamp without each
	// transition re-implementing it.
	UpdatedAt time.Time `json:"updated_at"`

	// Inputs are the versions pinned at readiness and re-checked at
	// admission (§7.3 T3, §7.4 admitted, INV-6).
	Inputs *MoveInputs `json:"inputs,omitempty"`

	CutPhase   string     `json:"cut_phase,omitempty"`
	CutAt      *time.Time `json:"cut_at,omitempty"`
	CutBudgetS int        `json:"cut_budget_s,omitempty"`
	// CutAttempts is the number of probe attempts the admission's T11
	// pricing admitted (the target report's AttemptsAdmitted, H10). The
	// landing's probe reads it; absent refuses the probe.
	CutAttempts int         `json:"cut_attempts,omitempty"`
	CutDeadline *time.Time  `json:"cut_deadline,omitempty"`
	Source      *MoveSource `json:"source,omitempty"`

	FlippedAt       *time.Time          `json:"flipped_at,omitempty"`
	LandingDeadline *time.Time          `json:"landing_deadline,omitempty"`
	LandingPhase    string              `json:"landing_phase,omitempty"`
	Landing         *MoveLandingJournal `json:"landing,omitempty"`

	CommittedAt    *time.Time `json:"committed_at,omitempty"`
	OpenedAt       *time.Time `json:"opened_at,omitempty"`
	SourceParkedAt *time.Time `json:"source_parked_at,omitempty"`

	// ReopenedAt is when a revert's source answered its probe again
	// (rev_phase probed, ruling J26): a revert has no Rollback journal, and
	// MoveRevertClose counts the offline time from CutAt to this stamp.
	ReopenedAt *time.Time `json:"reopened_at,omitempty"`
	// RecoveryInputs are the inputs a revert's probe rendered (Plan 4 Task
	// 6; §7.7, J18), recorded before the render by MoveRecordRevertInputs;
	// never compared with Inputs. A rollback keeps its own in
	// Rollback.RecoveryInputs.
	RecoveryInputs *MoveInputs `json:"recovery_inputs,omitempty"`

	Rollback *MoveRollback `json:"rollback,omitempty"`
	Reason   string        `json:"reason,omitempty"`
	// StuckReason is why a stuck move waits for an operator (§7.8 stuck),
	// a recorded sentence that MoveMarkStuck writes and MoveRetry clears.
	// It is a field of its own so that Reason keeps the move's cause (a
	// revert's, a landing's mark) and Rollback.Reason the rollback's
	// through a stuck and a retry.
	StuckReason string `json:"stuck_reason,omitempty"`
	// RecoveryRetryEpoch advances only on an operator retry. Recovery
	// probe budgets use it so restarts and replays keep their spent starts.
	RecoveryRetryEpoch uint64 `json:"recovery_retry_epoch,omitempty"`

	// Relocation is the durable pre-flip conversion decision
	// (move_relocation.go). It never installs anything.
	Relocation *MoveRelocationDecision `json:"relocation,omitempty"`

	// Kind, Sealer and Boxes belong to the offline operator move
	// (offline_move.go). All three are additive and omitempty: a journal
	// written before them parses and re-encodes byte for byte. Kind is
	// "offline" for that move and empty for every other.
	Kind string `json:"kind,omitempty"`
	// Sealer is the one box that seals the move, named at seal.
	Sealer string `json:"sealer,omitempty"`
	// Boxes are the source-region boxes that had to confirm the stop, named
	// at seal.
	Boxes []string `json:"boxes,omitempty"`
	// StagedAt is when an offline move was staged (OfflineStage): created
	// in offline-staging, with the instance still serving. RequestedAt is
	// then moved to the start, when the instance went offline, so a staged
	// move's offline time counts from the start like any other's; and a
	// move aborted before its start was never offline (OfflineFinish).
	// Additive and omitempty like the three above.
	StagedAt *time.Time `json:"staged_at,omitempty"`
	// Stager is the one source box that pre-copies a staged move, named
	// by the stage door and never changed (Astra round 1, finding 2): no
	// other box ever writes the staging prefix for this move, and an abort
	// of a staged move finishes only once this box has removed the staged
	// copy. Additive and omitempty.
	Stager string `json:"stager,omitempty"`
	// StageBucket and StageRegion are the staged copy's destination, the
	// target's data bucket and its region as the stage door read them,
	// fixed before any copy (Astra round 2, finding 2). Every stage round,
	// the seal's hand-over and an abort's cleanup use them; a target that
	// later publishes other coordinates is refused, never followed.
	// Additive and omitempty.
	StageBucket string `json:"stage_bucket,omitempty"`
	StageRegion string `json:"stage_region,omitempty"`
}

// MoveInputs (what the target was checked against, §7.3 T3) is NOT
// defined here: it is main's internal/provision/moveinputs_contract.go
// (claim Plan 6, d649d8d4), the shared contract this journal references.

// MoveSource is what the source recorded at `stopped` (§7.4): the
// baselines every recovery compares against.
type MoveSource struct {
	Replication  string           `json:"replication,omitempty"`
	NextDue      int64            `json:"next_due,omitempty"`
	FinalTXID    uint64           `json:"final_txid,omitempty"`
	DBSHA256     string           `json:"db_sha256,omitempty"`
	Files        *MoveFilesDigest `json:"files,omitempty"`
	TimersSHA256 string           `json:"timers_sha256,omitempty"`
	Evidence     string           `json:"evidence,omitempty"`
	// Release and EnvDigest are the offline move's: the build the source
	// ran and the portable digest of its launch environment, recorded at
	// seal. Additive and omitempty.
	Release   string `json:"release,omitempty"`
	EnvDigest string `json:"env_digest,omitempty"`
}

type MoveFilesDigest struct {
	Plain   string `json:"plain,omitempty"`
	Sealed  string `json:"sealed,omitempty"`
	Bytes   int64  `json:"bytes,omitempty"`
	Entries int    `json:"entries,omitempty"`
}

// MoveLandingJournal is the landing's own record (§7.2 "landing"). Named
// so because MoveLanding is the state constant below.
type MoveLandingJournal struct {
	InstalledDBSHA256 string `json:"installed_db_sha256,omitempty"`
	Attempts          int    `json:"attempts,omitempty"`
}

// MoveOutcome is the terminal result the close CAS leaves on the record
// as it clears Move (index D11), so a CLI that missed the intermediate
// states still reads the outcome of the move it requested.
type MoveOutcome struct {
	ID       string    `json:"id"`
	Outcome  string    `json:"outcome"` // done | rolled_back | aborted
	At       time.Time `json:"at"`
	OfflineS float64   `json:"offline_s,omitempty"`
}

// MoveRollback is §7.8's journal.
type MoveRollback struct {
	ID              string     `json:"id,omitempty"`
	RequestedAt     *time.Time `json:"requested_at,omitempty"`
	Reason          string     `json:"reason,omitempty"`
	TargetPhase     string     `json:"target_phase,omitempty"`
	Dirty           bool       `json:"dirty,omitempty"`
	Evidence        string     `json:"evidence,omitempty"`
	SourcePhase     string     `json:"source_phase,omitempty"`
	RecoveredAt     *time.Time `json:"recovered_at,omitempty"`
	OpenedAt        *time.Time `json:"opened_at,omitempty"`
	TargetParkedAt  *time.Time `json:"target_parked_at,omitempty"`
	ReleaseDeadline *time.Time `json:"release_deadline,omitempty"`
	// StuckAt is the source phase a stuck rollback stopped in (§7.8 stuck);
	// MoveRetry returns the source to it, so a verification mismatch is
	// verified again and never bypassed (INV-4). Empty unless stuck.
	StuckAt string `json:"stuck_at,omitempty"`
	// LossRef is the declared loss's record key, or the termination fence
	// record key (§7.10, spec rev 7.19), and LostAt when the record took
	// it. Written once, by MoveRollbackTargetLost.
	LossRef string     `json:"loss_ref,omitempty"`
	LostAt  *time.Time `json:"lost_at,omitempty"`
	// RecoveryInputs are the inputs a recovery install rendered itself
	// (§7.7, J2-R2-5); never compared with Inputs.
	RecoveryInputs *MoveInputs `json:"recovery_inputs,omitempty"`
}

// Legacy reports a move journal written before ids existed (a v0.22
// console). The guarded transitions below refuse it: a legacy record is
// driven only by the legacy transitions (MovePromoteReady, MoveAbort,
// MoveAbortComplete and the rest). The new door routes a Legacy() record
// to the legacy MoveAbort, and the new worker completes a legacy aborting
// record through the legacy code path only.
func (m *MoveState) Legacy() bool { return m != nil && m.ID == "" }

// The move states. There is no "done" state: the close CAS clears Move, so
// a record at rest never says anything about moves.
const (
	MovePreparing = "preparing"
	MoveReady     = "ready"
	MoveCutting   = "cutting"
	// MoveAborting is abort's in-flight half (review round 1): the console
	// marks it, the SOURCE box deletes the staged copy from the target
	// bucket and then clears Move — because a staged replica left behind by
	// a bare clear would win a LATER landing on that box (staging-wins,
	// backend_exec.go's promoteStagedReplica) with data from an abandoned
	// move.
	MoveAborting    = "aborting"
	MoveReverting   = "reverting"
	MoveLanding     = "landing"
	MoveCommitted   = "committed"
	MoveRollingBack = "rolling_back"
	MoveStuck       = "stuck"
)

// Cut phases (§7.4), landing phases (§7.6), reverting phases (§7.4) and
// rollback phases (§7.8). Each is written before its side effects.
const (
	CutRequested = "requested"
	CutAdmitted  = "admitted"
	CutStopping  = "stopping"
	CutStopped   = "stopped"
	CutStaged    = "staged"
	CutArchived  = "archived"
	CutFlipped   = "flipped"

	LandChecking   = "checking"
	LandParking    = "parking"
	LandBuilding   = "building"
	LandPublished  = "published"
	LandLocal      = "local"
	LandConfigured = "configured"
	LandRestored   = "restored"
	LandProbing    = "probing"
	LandProbed     = "probed"

	RevEvidence = "evidence"
	RevVerify   = "verify"
	RevProbing  = "probing"
	RevProbed   = "probed"
	RevOpened   = "opened"

	RbTargetReleasing = "releasing"
	RbTargetReleased  = "released"
	RbTargetReturning = "returning"
	RbTargetReturned  = "returned"
	RbTargetCleaning  = "cleaning"
	RbTargetCleaned   = "cleaned"
	// RbTargetLost is the declared-loss decision (§7.10), a phase of its
	// own: no forward step leaves it and the source's waiting never takes
	// it. Only MoveRollbackLossCommitted moves it on, to released.
	RbTargetLost = "lost"

	RbSourceWaiting    = "waiting"
	RbSourceClaimed    = "claimed"
	RbSourceRestoring  = "restoring"
	RbSourceVerified   = "verified"
	RbSourceParking    = "parking"
	RbSourceBuilding   = "building"
	RbSourcePublished  = "published"
	RbSourceLocal      = "local"
	RbSourceConfigured = "configured"
	RbSourceRestored   = "restored"
	RbSourceProbing    = "probing"
	RbSourceProbed     = "probed"
	RbSourceRecovered  = "recovered"
	RbSourceOpened     = "opened"
)

// ErrMoveConflict is a transition refused because the record's current move
// state does not admit it — a second prepare while one is in flight, a cut
// before ready, an abort after the point of no return. The console maps it
// to 409; the box treats it as "someone else already advanced this" and
// re-reads.
var ErrMoveConflict = errors.New("provision: move state conflict")

// The move refusals that name no value (copy record MV-N12, MV-N14, MV-N15,
// MV-N40, MV-N41 and, for the legacy abort, MV-N45). Each is a whole recorded
// sentence, so the copy gate can read it; the ones that name a value (MV-N13,
// MV-N42, MV-N43, MV-N44) are written in place as one format.
const (
	refusedNoMove          = "no move in flight"
	refusedNoJournaledMove = "no journaled move in flight"
	refusedMalformedStart  = "malformed move start"
	refusedPastNoReturn    = "past the point of no return. the cut completes forward or moves back on its own"
	refusedCutNotReady     = "cut is only legal from ready"
	refusedAlreadyAborting = "already aborting"
)

// SetRecordMove applies one guarded move transition: the same narrow
// read-CAS-write with one retry that SetRecordRegion uses (and for the same
// reason — re-reading and re-applying a transition on a lost race changes
// nothing about what it meant; an ILLEGAL transition after the re-read is
// ErrMoveConflict, which is the answer, not a race). mut sees the freshly
// read record and either mutates it or returns the error that names why it
// can't.
func SetRecordMove(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) (*Record, error) {
	return updateRecord(ctx, s, acct, app, host, func(rec *Record) error {
		if err := mut(rec); err != nil {
			return err
		}
		if rec.Move != nil {
			rec.Move.UpdatedAt = time.Now().UTC()
		}
		return nil
	})
}

// The five transitions, each a mut for SetRecordMove. They are the ONLY
// writers of Record.Move.

// MoveRequest starts a move to region `to`. Refused when one is already in
// flight — the caller aborts first or waits.
func MoveRequest(to string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := seedMoveRefusal(rec); err != nil {
			return err
		}
		if rec.ActiveRun.Open() {
			return &MoveRunOpenError{Run: *rec.ActiveRun}
		}
		if rec.Move != nil {
			return fmt.Errorf("%w: a move to %s is already %s", ErrMoveConflict, rec.Move.To, rec.Move.State)
		}
		if rec.Region == to {
			return fmt.Errorf("%w: the instance is already in %s", ErrMoveConflict, to)
		}
		// A replica never gets a route on a platform box (reconcile's mint
		// gate), so no box owns its move and it would wedge in preparing
		// (G14). Checked here as well as at the console's door because
		// records have other writers.
		if rec.Role != "" {
			return fmt.Errorf("%w: a %s does not move. delete it and create one in the new region", ErrMoveConflict, rec.Role)
		}
		rec.Move = &MoveState{To: to, State: MovePreparing, RequestedAt: now.UTC()}
		return nil
	}
}

// MoveStart is what the request door decided (§7.1, §7.2): the id it
// minted, the incarnation the claim CAS named, the cut mode and who asked.
type MoveStart struct {
	ID, Incarnation, From, To, Cut, By string
	Now                                time.Time
}

// MoveStartTransition starts a move. Refused when one is in flight, when
// the target is the current region, for a replica (a replica never gets a
// route on a platform box, G14), and for any malformed start: a move
// without an id has no journal to run.
func MoveStartTransition(p MoveStart) func(*Record) error {
	return func(rec *Record) error {
		if err := seedMoveRefusal(rec); err != nil {
			return err
		}
		if rec.ActiveRun.Open() {
			return &MoveRunOpenError{Run: *rec.ActiveRun}
		}
		if rec.Move != nil {
			return fmt.Errorf("%w: %s", ErrMoveConflict, fmt.Sprintf("a move to %s is already %s", rec.Move.To, rec.Move.State))
		}
		if !moveIDShape.MatchString(p.ID) || len(p.Incarnation) != 64 || (p.Cut != "auto" && p.Cut != "manual") || p.Now.IsZero() {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)
		}
		if rec.Region == p.To || p.From != rec.Region {
			return fmt.Errorf("%w: the instance is already in %s", ErrMoveConflict, rec.Region)
		}
		if rec.Role != "" {
			return fmt.Errorf("%w: %s", ErrMoveConflict, fmt.Sprintf("a %s does not move. delete it and create one in the new region", rec.Role))
		}
		rec.Move = &MoveState{ID: p.ID, Incarnation: p.Incarnation, From: p.From, To: p.To, Cut: p.Cut, State: MovePreparing, RequestedBy: p.By, RequestedAt: p.Now.UTC()}
		return nil
	}
}

// moveIDShape is a move id: "mv_" and 26 lowercase base32 characters.
//
// Twin of moves.opIDShape (internal/moves/ids.go), which also accepts the
// rollback prefix: the copy is forced, because moves imports provision, so
// provision cannot import moves. Keep the two in step.
var moveIDShape = regexp.MustCompile(`^mv_[a-z2-7]{26}$`)

// rollbackIDShape is a rollback id: "rb_" and 26 lowercase base32
// characters, the other half of moves.opIDShape (moves.NewRollbackID mints
// it). Twin of moves.ValidOpID's rollback form, for the same reason as
// moveIDShape; moves' tests hold the two to each other.
var rollbackIDShape = regexp.MustCompile(`^rb_[a-z2-7]{26}$`)

// ValidRollbackID reports a rollback id (rollbackIDShape).
func ValidRollbackID(id string) bool { return rollbackIDShape.MatchString(id) }

// MoveGuard is what every Phase 1 transition carries (§7.2): the move id
// and the holder incarnation. A write whose guard does not match the
// record's move is a late write for another move or another incarnation
// and is refused, so a delayed round for move A can never advance a
// replacement move B that happens to be in the same state.
type MoveGuard struct{ ID, Incarnation string }

func (g MoveGuard) check(rec *Record) error {
	if rec.Move == nil || rec.Move.Legacy() {
		return fmt.Errorf("%w: %s", ErrMoveConflict, refusedNoJournaledMove)
	}
	// An offline move has its own transitions and no incarnation. A guard
	// with an empty incarnation must not reach it: kinds never cross.
	if rec.Move.Offline() {
		return fmt.Errorf("%w: %s", ErrMoveConflict, refusedOfflineKind)
	}
	if rec.Move.ID != g.ID || rec.Move.Incarnation != g.Incarnation {
		return fmt.Errorf("%w: %s", ErrMoveConflict, fmt.Sprintf("the record's move is %s for incarnation %s", rec.Move.ID, rec.Move.Incarnation))
	}
	return nil
}

// MoveGuardedPromoteReady is the new worker's preparing → ready. It pins
// the inputs the target was checked against onto the journal at readiness
// (§7.2, §7.4), so the admission re-check has something to compare with.
func MoveGuardedPromoteReady(g MoveGuard, now time.Time, in MoveInputs) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if err := MovePromoteReady(now)(rec); err != nil {
			return err
		}
		rec.Move.Inputs = pinInputs(in)
		return nil
	}
}

// MoveGuardedCut is the cut's entry (the console's cut door, or the source
// for cut == "auto"): ready → cutting/requested.
func MoveGuardedCut(g MoveGuard) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveReady {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedCutNotReady)
		}
		// Spec §7.4 pins at readiness: a ready move with no pinned inputs
		// (after a back-to-ready, until the repin) has nothing for the
		// admission to compare with, so it is not cut.
		if rec.Move.Inputs == nil {
			return fmt.Errorf("%w: the move has no pinned inputs to cut against", ErrMoveConflict)
		}
		rec.Move.State, rec.Move.CutPhase = MoveCutting, CutRequested
		return nil
	}
}

// pinInputs copies in, BundleETags map included, so the journal never
// aliases a caller's map.
func pinInputs(in MoveInputs) *MoveInputs {
	pinned := in
	if in.BundleETags != nil {
		pinned.BundleETags = make(map[string]string, len(in.BundleETags))
		for k, v := range in.BundleETags {
			pinned.BundleETags[k] = v
		}
	}
	return &pinned
}

// MoveGuardedRepin pins fresh inputs on a ready move that has none: the
// state MoveBackToReady leaves after the admission found the pin stale.
// Legal in ready with Inputs == nil only; a pin already there is never
// overwritten (the pin is what the admission compares with).
func MoveGuardedRepin(g MoveGuard, in MoveInputs) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveReady || rec.Move.Inputs != nil {
			return fmt.Errorf("%w: inputs are re-pinned only on a ready move that has none", ErrMoveConflict)
		}
		rec.Move.Inputs = pinInputs(in)
		// A refusal MoveBackToReadyRefused recorded ends here (fix round
		// 1): a move is pinned again only once its cause is gone, and from
		// then on `carlos move status` says it is ready to cut.
		rec.Move.Reason = ""
		return nil
	}
}

// MoveGuardedAbort: legal in preparing, ready and cutting/requested (§7.11).
func MoveGuardedAbort(g MoveGuard) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		switch rec.Move.State {
		case MovePreparing, MoveReady:
		case MoveCutting:
			if rec.Move.CutPhase != CutRequested {
				return fmt.Errorf("%w: %s", ErrMoveConflict, refusedPastNoReturn)
			}
		case MoveAborting:
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedAlreadyAborting)
		default:
			return fmt.Errorf("%w: %s", ErrMoveConflict, fmt.Sprintf("a move in state \"%s\" cannot be aborted", quotedInner(rec.Move.State)))
		}
		rec.Move.State = MoveAborting
		return nil
	}
}

// MoveGuardedAbortComplete clears an aborting move and records the outcome.
func MoveGuardedAbortComplete(g MoveGuard, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveAborting {
			return fmt.Errorf("%w: abort-complete is only legal from aborting", ErrMoveConflict)
		}
		rec.LastMove = &MoveOutcome{ID: g.ID, Outcome: "aborted", At: now.UTC()}
		rec.Move = nil
		return nil
	}
}

// cutOrder is the cut's phase order up to archived. flipped is not a step
// of it: the flip is MoveFlip's, one record write that also moves the
// region, so MoveCutPhase never writes it.
var cutOrder = []string{CutRequested, CutAdmitted, CutStopping, CutStopped, CutStaged, CutArchived}

var landingOrder = []string{LandChecking, LandParking, LandBuilding, LandPublished, LandLocal, LandConfigured, LandRestored, LandProbing, LandProbed}

// LandingOrder is the landing's phase order (§7.6), a copy: the claim's
// op.phase mirrors landing_phase one step at a time, and a worker that
// finds the claim behind the record walks it forward along this order
// (§7.5, the landing's syncClaimPhase).
func LandingOrder() []string { return append([]string(nil), landingOrder...) }

func nextIn(order []string, from, to string) bool {
	for i := 0; i+1 < len(order); i++ {
		if order[i] == from && order[i+1] == to {
			return true
		}
	}
	return false
}

// MoveCutPhase advances cut_phase one step, in order, under the guard
// (§7.4). stamp may fill the fields the phase records; it runs after the
// checks, on the journal the step has just advanced.
func MoveCutPhase(g MoveGuard, from, to string, stamp func(*MoveState)) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveCutting || rec.Move.CutPhase != from || !nextIn(cutOrder, from, to) {
			return fmt.Errorf("%w: cut is %s/%s, step expects %s to %s", ErrMoveConflict, rec.Move.State, rec.Move.CutPhase, from, to)
		}
		rec.Move.CutPhase = to
		if stamp != nil {
			stamp(rec.Move)
		}
		return nil
	}
}

// MoveAdmit is §7.4 requested to admitted: the budget, the absolute
// deadline and the probe attempts T11 admitted (H10), in one write. It pins nothing: the inputs were pinned at
// readiness (MoveGuardedPromoteReady, or MoveGuardedRepin after a back to
// ready), and the source compares the current inputs with that pin before
// this CAS. A cut with no pin is refused.
func MoveAdmit(g MoveGuard, now time.Time, budgetS int, deadline time.Time, attempts int) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.Inputs == nil {
			return fmt.Errorf("%w: the move has no pinned inputs to admit against", ErrMoveConflict)
		}
		if attempts < 0 {
			return fmt.Errorf("%w: a negative count of probe attempts", ErrMoveConflict)
		}
		return MoveCutPhase(g, CutRequested, CutAdmitted, func(m *MoveState) {
			t, d := now.UTC(), deadline.UTC()
			m.CutAt, m.CutBudgetS, m.CutDeadline, m.CutAttempts = &t, budgetS, &d, attempts
			// A refusal an earlier cut recorded (MoveBackToReadyRefused) is
			// over once a cut admits.
			m.Reason = ""
		})(rec)
	}
}

// RefusedReplicaLayout is the reason MoveBackToReadyRefused records when
// the claim holder's location is not the activator's own lease prefix
// (moves Task 5 fix round 1, ruling T5-lease): a scoped location,
// instances/<host>/<incarnation>, whose replica the cut cannot stop under
// the writer's lease. Moving such a host waits for a later plan. A
// recorded sentence (copy record MV-N167), like the Revert reasons.
const RefusedReplicaLayout = "the instance keeps its data under a per-incarnation path, which moves cannot cut yet"

// The cut guard's refusals (Paul's ruling, 2026-10-11), recorded by
// MoveBackToReadyRefused before anything stops. RefusedTargetUnarmed: the
// target's latest report does not state that its box can land the move and
// roll it back (moves.TargetReport.Armed; a report from an older box never
// does). RefusedSourceUnarmed: the source box cannot run the recovery a
// failed cut would need. Recorded sentences (copy record MV-N441, MV-N442).
const (
	RefusedTargetUnarmed = "the target box cannot land the move or roll it back yet"
	RefusedSourceUnarmed = "the source box cannot recover the move if the cut fails"
)

// MoveBackToReadyRefused is MoveBackToReady with the reason recorded on the
// move, for a refusal that stands until something outside the move
// changes (the operator reads it in `carlos move status`). reason must be
// a recorded constant; any other text is refused and nothing is written.
func MoveBackToReadyRefused(g MoveGuard, reason string) func(*Record) error {
	return func(rec *Record) error {
		switch reason {
		case RefusedReplicaLayout, RefusedTargetUnarmed, RefusedSourceUnarmed:
		default:
			return fmt.Errorf("provision: refusal reason %q has no recorded sentence", reason)
		}
		if err := MoveBackToReady(g)(rec); err != nil {
			return err
		}
		// Each reason is written as its constant, so the copy gate reads
		// the recorded sentence (as MoveRevert writes its reasons).
		switch reason {
		case RefusedTargetUnarmed:
			rec.Move.Reason = RefusedTargetUnarmed
		case RefusedSourceUnarmed:
			rec.Move.Reason = RefusedSourceUnarmed
		default:
			rec.Move.Reason = RefusedReplicaLayout
		}
		return nil
	}
}

// MoveBackToReady undoes an admission whose checks are no longer fresh or
// whose inputs moved (§7.4 admitted). Legal from requested or admitted
// only: nothing has stopped yet. It clears the cut fields and the pinned
// inputs; the next readiness round pins fresh ones (MoveGuardedRepin).
func MoveBackToReady(g MoveGuard) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveCutting || (rec.Move.CutPhase != CutRequested && rec.Move.CutPhase != CutAdmitted) {
			return fmt.Errorf("%w: back to ready is legal from requested or admitted only", ErrMoveConflict)
		}
		m := rec.Move
		m.State, m.CutPhase, m.CutAt, m.CutBudgetS, m.CutDeadline, m.Inputs = MoveReady, "", nil, 0, nil, nil
		return nil
	}
}

// MoveRecordSource writes the stopped source's baselines (§7.4 stopped).
func MoveRecordSource(g MoveGuard, src MoveSource) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveCutting || rec.Move.CutPhase != CutStopped {
			return fmt.Errorf("%w: source digests are recorded at stopped", ErrMoveConflict)
		}
		s := src
		if src.Files != nil {
			f := *src.Files
			s.Files = &f
		}
		rec.Move.Source = &s
		return nil
	}
}

// MoveFlip is the flip (§7.4, §7.5): region = to, state = landing, the
// landing deadline and landing_phase = checking, in one record write. The
// claim follows (hostclaim.Flip).
//
// It refuses any state but cutting/archived, whoever holds the guard. So a
// record decision written first, such as the C7 conversion's pre-flip guard
// or a revert, wins over a flip already in flight: the late flip finds the
// state changed and is refused, and the region is untouched.
func MoveFlip(g MoveGuard, now, landingDeadline time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveCutting || rec.Move.CutPhase != CutArchived {
			return fmt.Errorf("%w: the flip follows archived", ErrMoveConflict)
		}
		t, d := now.UTC(), landingDeadline.UTC()
		rec.Region = rec.Move.To
		m := rec.Move
		m.State, m.CutPhase, m.FlippedAt, m.LandingDeadline, m.LandingPhase = MoveLanding, CutFlipped, &t, &d, LandChecking
		return nil
	}
}

// MoveLandingPhase advances landing_phase one step, in order, under the
// guard (§7.6). stamp runs after the checks.
func MoveLandingPhase(g MoveGuard, from, to string, stamp func(*MoveState)) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveLanding || rec.Move.LandingPhase != from || !nextIn(landingOrder, from, to) {
			return fmt.Errorf("%w: landing is %s/%s, step expects %s to %s", ErrMoveConflict, rec.Move.State, rec.Move.LandingPhase, from, to)
		}
		rec.Move.LandingPhase = to
		if stamp != nil {
			stamp(rec.Move)
		}
		return nil
	}
}

// The reasons a revert may record. MoveState.Reason reaches an operator
// (`carlos move status`), so it is a recorded sentence (copy record MV-N158
// onward), never error text; the copy gate reads every write to a field
// named Reason and cannot read a parameter. A later task that needs another
// reason adds a constant, a case below and a row in the copy record.
const (
	RevertStopEvidenceBudget = "stop evidence budget"
	// RevertCutDeadline: a phase after stopping found cut_deadline passed.
	RevertCutDeadline = "the cut ran past its deadline"
	// RevertDeferredRounds: three final mirror rounds found the source
	// changing under them, or moved more than the priced bar.
	RevertDeferredRounds = "the final copy kept finding changes at the source"
	// RevertArchiveRefused: the stopped files directory cannot be archived
	// (a special file, over the cap, too many entries, a path too long).
	RevertArchiveRefused = "the stopped files could not be archived"
)

// MoveRevert starts the pre-flip recovery (§7.4 reverting): from a cut
// phase that has begun stopping the source. Before that, back to ready is
// the exit; after the flip, a rollback is. reason must be one of the
// Revert constants above: any other text is refused, and nothing is written.
func MoveRevert(g MoveGuard, reason string) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveCutting {
			return fmt.Errorf("%w: revert is a cut's recovery", ErrMoveConflict)
		}
		switch rec.Move.CutPhase {
		case CutStopping, CutStopped, CutStaged, CutArchived:
		default:
			return fmt.Errorf("%w: revert is legal from stopping to archived only", ErrMoveConflict)
		}
		switch reason {
		case RevertStopEvidenceBudget:
			rec.Move.Reason = RevertStopEvidenceBudget
		case RevertCutDeadline:
			rec.Move.Reason = RevertCutDeadline
		case RevertDeferredRounds:
			rec.Move.Reason = RevertDeferredRounds
		case RevertArchiveRefused:
			rec.Move.Reason = RevertArchiveRefused
		default:
			return fmt.Errorf("provision: revert reason %q has no recorded sentence", reason)
		}
		rec.Move.State = MoveReverting
		return nil
	}
}

// The reasons a landing may record for the rollback it cannot yet start
// (Plan 2 Task 10). As with the revert's, each reaches an operator through
// `carlos move status`, so it is a recorded sentence with a copy row, and
// anything else is refused with nothing written.
const (
	// RollbackConfigChanged: the held render's bundle or overlay etags are
	// not the ones pinned at readiness (§7.6 configured).
	RollbackConfigChanged = "configuration changed during the move"
	// RollbackLandingDeadline: landing_deadline passed before restored.
	RollbackLandingDeadline = "the landing ran past its deadline"
	// RollbackClaimWired: the target runs with CARLOS_HOST_CLAIM on, and
	// the held writes a landing makes have no claim selection yet (G16).
	RollbackClaimWired = "this box cannot land a move while host claims are on"
	// RollbackLandingLayout: the held row is not at the flat database path
	// Phase 1 lands at (G11, D10).
	RollbackLandingLayout = "the landing row is not at the flat path"
	// RollbackInstallMismatch: the built database does not hash to the
	// install manifest's sha256, or does not pass its integrity check.
	RollbackInstallMismatch = "the installed database does not match the build"
	// RollbackFilesOccupied: the files directory holds files, or <host>.env
	// exists, and this landing did not write them (no earlier stay of this
	// installation registered them, so nothing retired them). They are not
	// the landing's to clear (final review, finding 1).
	RollbackFilesOccupied = "the files directory already holds files this move did not write"
	// RollbackManifestOccupied: the prepared files manifest's key under the
	// location already holds a manifest this landing did not write (moves
	// Task 6, C8). The prepared manifest is create-only and the landing's
	// replay reads its own bytes back; anything else there is not the
	// landing's to replace.
	RollbackManifestOccupied = "the destination already holds a files manifest this move did not write"
	// RollbackProbeAttempts: the probe used every attempt it was given
	// without the app answering (§7.7).
	RollbackProbeAttempts = "the probe ran out of attempts"
	// RollbackCommitDeadline: the probe passed, but the commit would have
	// come after the cut or landing deadline (§7.9).
	RollbackCommitDeadline = "the commit came after the deadline"
	// RollbackProbeUnpriced: the admission recorded no probe attempts for
	// the landing (H10: MoveAdmit's CutAttempts is absent), so the probe
	// cannot be bounded.
	RollbackProbeUnpriced = "the cut recorded no probe attempts for this landing"
)

// MoveMarkRollback records on a landing record why the landing stopped
// (§7.6: "any failure from checking to probing, or landing_deadline,
// triggers a rollback"), leaving state and landing_phase as they are:
// Plan 4's rollback transition (§7.8) is what replaces this one call site,
// and until it exists the record carries the reason and the landing pass
// leaves the record alone. The same sentence written again is a no-op; a
// different one over a marked record is a conflict, so the first reason
// stands. reason must be one of the Rollback constants above.
func MoveMarkRollback(g MoveGuard, reason string) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveLanding {
			return fmt.Errorf("%w: a rollback marker is a landing's", ErrMoveConflict)
		}
		if rec.Move.Reason != "" && rec.Move.Reason != reason {
			return fmt.Errorf("%w: the landing already records a reason", ErrMoveConflict)
		}
		switch reason {
		case RollbackConfigChanged:
			rec.Move.Reason = RollbackConfigChanged
		case RollbackLandingDeadline:
			rec.Move.Reason = RollbackLandingDeadline
		case RollbackClaimWired:
			rec.Move.Reason = RollbackClaimWired
		case RollbackLandingLayout:
			rec.Move.Reason = RollbackLandingLayout
		case RollbackInstallMismatch:
			rec.Move.Reason = RollbackInstallMismatch
		case RollbackFilesOccupied:
			rec.Move.Reason = RollbackFilesOccupied
		case RollbackManifestOccupied:
			rec.Move.Reason = RollbackManifestOccupied
		case RollbackProbeAttempts:
			rec.Move.Reason = RollbackProbeAttempts
		case RollbackCommitDeadline:
			rec.Move.Reason = RollbackCommitDeadline
		case RollbackProbeUnpriced:
			rec.Move.Reason = RollbackProbeUnpriced
		default:
			return fmt.Errorf("provision: rollback reason %q has no recorded sentence", reason)
		}
		return nil
	}
}

// MoveProbeAttempt records, on a landing in its probing phase, how many
// probe attempts have begun (§7.7). The count only goes up: the same n again
// is a replay and changes nothing, a smaller one is a late write from an
// earlier attempt and is refused. It is refused too once a rollback reason
// is recorded, because the probe is then over. A record at probing says
// only that the probe is running and how many attempts it has made; whether
// the attempt's first start has been journalled is the claim's to say
// (hostclaim.WakeRetryAttempt), not the record's.
func MoveProbeAttempt(g MoveGuard, n int) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		if m.State != MoveLanding || m.LandingPhase != LandProbing || m.Reason != "" {
			return fmt.Errorf("%w: probe attempts are counted while a landing is probing", ErrMoveConflict)
		}
		have := 0
		if m.Landing != nil {
			have = m.Landing.Attempts
		}
		if n < 1 || n < have {
			return fmt.Errorf("%w: the attempt count does not go down", ErrMoveConflict)
		}
		if m.Landing == nil {
			m.Landing = &MoveLandingJournal{}
		}
		m.Landing.Attempts = n
		return nil
	}
}

// ErrCutDeadline is a commit attempted after the deadline that bounds it
// (§7.4, §7.9): the record CAS is the decision, so the deadline cannot be
// passed between the record write and the claim write.
var ErrCutDeadline = fmt.Errorf("%w: the commit deadline has passed", ErrMoveConflict)

// MoveCommit is the commit (§7.9): landing/probed to committed, in one
// record write. It is refused when a rollback reason is already recorded, so
// a recorded rollback wins over a commit that raced it, and when now is
// after the earlier of cut_deadline and landing_deadline (ruling H8: §7.9
// names the first and §7.8 makes the second a rollback trigger, so the
// earlier bounds the commit). A record with no cut_deadline cannot be shown
// to be in time and is refused. now equal to the deadline is in time.
//
// A replay finds the move already committed and is an ErrMoveConflict; the
// caller re-reads the record and accepts committed for its own move (H22).
func MoveCommit(g MoveGuard, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		if m.State != MoveLanding || m.LandingPhase != LandProbed {
			return fmt.Errorf("%w: commit follows probed", ErrMoveConflict)
		}
		if m.Reason != "" {
			return fmt.Errorf("%w: a rollback is recorded for this landing", ErrMoveConflict)
		}
		if m.CutDeadline == nil || now.After(*m.CutDeadline) || (m.LandingDeadline != nil && now.After(*m.LandingDeadline)) {
			return ErrCutDeadline
		}
		t := now.UTC()
		m.State, m.CommittedAt = MoveCommitted, &t
		return nil
	}
}

// committedStamp stamps one of a committed move's obligations. The first
// stamp stands: a replay changes nothing.
func committedStamp(g MoveGuard, now time.Time, field func(*MoveState) **time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveCommitted {
			return fmt.Errorf("%w: this follows the commit", ErrMoveConflict)
		}
		if p := field(rec.Move); *p == nil {
			t := now.UTC()
			*p = &t
		}
		return nil
	}
}

// MoveOpened stamps opened_at on a committed move: the target is serving
// (§7.9).
func MoveOpened(g MoveGuard, now time.Time) func(*Record) error {
	return committedStamp(g, now, func(m *MoveState) **time.Time { return &m.OpenedAt })
}

// MoveSourceParked stamps source_parked_at on a committed move: the source
// has parked what it kept (§7.9).
func MoveSourceParked(g MoveGuard, now time.Time) func(*Record) error {
	return committedStamp(g, now, func(m *MoveState) **time.Time { return &m.SourceParkedAt })
}

// LandingDeadlineOf is the deadline a landing before restored works to:
// the earlier of the cut's absolute deadline and landing_deadline. Spec
// §7.4 has every pre-commit phase on both participants check the absolute
// deadline; the flip's landing_deadline is a second, separate bound, never
// an extension of it (final review, finding 3). which names the one that
// applies, for the log. The box's landing and rollback decisions (host's
// landingDeadlineOf) and the move's timings (moves.Timings) share it
// (Plan 6 preflight L20).
func LandingDeadlineOf(m *MoveState) (d *time.Time, which string) {
	if m.CutDeadline != nil {
		d, which = m.CutDeadline, "cut_deadline"
	}
	if m.LandingDeadline != nil && (d == nil || m.LandingDeadline.Before(*d)) {
		d, which = m.LandingDeadline, "landing_deadline"
	}
	return d, which
}

// MoveOfflineS is a move's offline time as its close records it (index D11):
// from cut_at to the open that ends the move. That is a committed move's
// opened_at (MoveClose), a rollback's source reopen, rollback.opened_at
// (MoveRolledBackClose), or a revert's reopened_at (MoveRevertClose). ok is
// false while either end is unstamped. The three closes and the move's
// timings (moves.Timings) share it (Plan 6 preflight L20).
func MoveOfflineS(m *MoveState) (seconds float64, ok bool) {
	if m == nil || m.CutAt == nil {
		return 0, false
	}
	var end *time.Time
	switch m.State {
	case MoveCommitted:
		end = m.OpenedAt
	case MoveRollingBack:
		if m.Rollback != nil {
			end = m.Rollback.OpenedAt
		}
	case MoveReverting:
		end = m.ReopenedAt
	}
	if end == nil {
		return 0, false
	}
	return end.Sub(*m.CutAt).Seconds(), true
}

// MoveClose is the close (§7.9 step 5, index D11): a committed move with
// both obligations stamped clears Move and leaves the outcome in LastMove,
// in the same write. The region is already the target's from the flip; a
// record whose region is not is refused rather than closed on a lie. The
// offline time is opened_at minus cut_at (MoveOfflineS), zero when cut_at
// was never stamped. Plan 5 calls it once no obligation under the move
// remains.
func MoveClose(g MoveGuard, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		if m.State != MoveCommitted || m.OpenedAt == nil || m.SourceParkedAt == nil {
			return fmt.Errorf("%w: close follows the open and the source park", ErrMoveConflict)
		}
		if rec.Region != m.To {
			return fmt.Errorf("%w: the record's region is not the target's", ErrMoveConflict)
		}
		out := &MoveOutcome{ID: g.ID, Outcome: "done", At: now.UTC()}
		if s, ok := MoveOfflineS(m); ok {
			out.OfflineS = s
		}
		rec.LastMove, rec.Move = out, nil
		return nil
	}
}

// RollbackTargetFenced is the recorded reason for a verified target fence.
const RollbackTargetFenced = "target fenced"

// RollbackDeclaredLost is the rollback reason the declare-lost door
// records (§7.10): the target box was declared lost before the move
// committed. A recorded sentence, so it names no box; the box is in the
// loss record LossRef names.
const RollbackDeclaredLost = "the target box was declared lost"

// The reasons a move may record when it stops for an operator (§7.8 stuck,
// §7.4 reverting). Each reaches `carlos move status` through
// Move.StuckReason, so each is a recorded sentence with a copy row, and each
// belongs to the phases MoveMarkStuck names for it.
const (
	// StuckBaselineMismatch: the source's files or timers do not hash to
	// what it recorded when it stopped (INV-4). A rollback's clean return
	// at restoring or verified, or a revert's verify.
	StuckBaselineMismatch = "the source no longer matches what it recorded when it stopped"
	// StuckRecoveryProbeAttempts: a recovery probe, a rollback's or a
	// revert's, used every attempt it was given without the app answering
	// (§7.7, nine).
	StuckRecoveryProbeAttempts = "the recovery probe ran out of attempts"
	// StuckRevertProbeBudget: a revert's probes failed for 15 minutes.
	StuckRevertProbeBudget = "the source did not answer within 15 minutes of moving back"
	// StuckReturnFilesOccupied: a dirty return's install (Plan 4 Task 3,
	// ruling J13) found the files directory or <host>.env holding what it
	// did not write after the source's own pre-move files were parked. A
	// rollback whose source is at local.
	StuckReturnFilesOccupied = "the files directory holds files the return did not write"
	// StuckReturnManifestOccupied: a dirty return's install found the
	// prepared files manifest's key under the source's location holding a
	// manifest it did not write (moves Task 6, C8). A rollback whose source
	// is at local.
	StuckReturnManifestOccupied = "the source already holds a files manifest the return did not write"
	// StuckReturnInstallMismatch: the database a dirty return built from its
	// lane does not verify. A rollback whose source is at configured, where
	// the database is installed.
	StuckReturnInstallMismatch = "the returned database does not match the build"
	// StuckReturnUnknownStamp: a dirty return's install found the source's
	// inventory listing paths under a stamp that is neither the source's own
	// S nor one the claim shows superseded on this box (live, unknown, or
	// another holder's). Nothing is moved. A rollback whose source is at
	// local.
	StuckReturnUnknownStamp = "the source holds files of an installation the return cannot account for"
	// The target's dirty return (Plan 4 Task 4, fix round 1, ruling 2):
	// what the target must send back cannot be built from what it holds,
	// so no retry by itself can help. Each is a dirty rollback's whose
	// target is at returning; the target keeps its phase and everything it
	// holds, and a retry runs the return again.
	//
	// StuckReturnTargetDatabase: the stopped database the return must
	// snapshot is missing or not a database.
	StuckReturnTargetDatabase = "the target's stopped database is missing"
	// StuckReturnTargetChanged: the database, its position, the files or
	// the timers are no longer what the target recorded when it stopped.
	StuckReturnTargetChanged = "the target changed after it stopped"
	// StuckReturnTargetFiles: the target's files directory cannot be
	// archived (a special file, over the cap, too many entries, a path too
	// long).
	StuckReturnTargetFiles = "the target's files cannot be archived"
	// StuckReleaseTargetFiles (Plan 5 Task 9a): the target that served
	// stopped, and its files directory cannot be archived when it is
	// digested for its stopped report (a special file, over the cap, too
	// many entries, a path too long). A rollback whose target is at
	// releasing, the source still waiting; the target keeps its phase and
	// everything it holds, and a retry runs the release again.
	StuckReleaseTargetFiles = "the target's files cannot be archived now that it has stopped"
)

// The refusals of this section that reach an operator through a door
// (phase1Conflict passes each through; copy record, Plan 4).
const (
	// RefusedCommitted: MoveRollBack found the record committed. The commit
	// won the race (§7.9); the target is authoritative.
	RefusedCommitted = "the move has committed. it cannot roll back"
	// RefusedNotStuck: MoveRetry on a move that is not stuck.
	RefusedNotStuck = "the move is not stuck. there is nothing to retry"
	// RefusedNothingToDeclare: MoveRollbackTargetLost from a target phase
	// that is past the window a loss can be declared in.
	RefusedNothingToDeclare = "the target holds nothing the move still needs. there is nothing to declare lost"
)

// rollbackReleaseBudget bounds the target's release (§7.8): the console
// pages on ReleaseDeadline.
const rollbackReleaseBudget = 15 * time.Minute

// MoveRollBack is the rollback decision (§7.8): a landing, at any landing
// phase, to rolling_back, in one record write that also moves the region
// back to From. It is the landing's decision against the commit's: a
// committed record refuses it (RefusedCommitted), and once it is written
// the commit is refused, because the state is no longer landing.
//
// rbID is the rollback id the caller minted (moves.NewRollbackID). The
// rollback's reason is the landing's recorded mark when there is one (the
// first reason stands, as in MoveMarkRollback), else reason, which must be
// a recorded Rollback constant (a reason given beside a mark must be one
// too, though the mark wins). The mark stays on Move.Reason: it is copied,
// not moved. LandingPhase and Landing are kept (the target's release reads
// them). A replay finds rolling_back and is an
// ErrMoveConflict; the caller re-reads.
func MoveRollBack(g MoveGuard, rbID, reason string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		if m.State == MoveCommitted {
			return fmt.Errorf("%w: %s", ErrMoveConflict, RefusedCommitted)
		}
		if m.State != MoveLanding {
			return fmt.Errorf("%w: a rollback is a landing's", ErrMoveConflict)
		}
		if !rollbackIDShape.MatchString(rbID) || now.IsZero() {
			return fmt.Errorf("provision: malformed rollback %q", rbID)
		}
		if reason != "" && !recordedRollbackReason(reason) {
			return fmt.Errorf("provision: rollback reason %q has no recorded sentence", reason)
		}
		why := reason
		if m.Reason != "" {
			why = m.Reason
		}
		rb := &MoveRollback{ID: rbID, TargetPhase: RbTargetReleasing, SourcePhase: RbSourceWaiting}
		switch why {
		case RollbackConfigChanged:
			rb.Reason = RollbackConfigChanged
		case RollbackLandingDeadline:
			rb.Reason = RollbackLandingDeadline
		case RollbackClaimWired:
			rb.Reason = RollbackClaimWired
		case RollbackLandingLayout:
			rb.Reason = RollbackLandingLayout
		case RollbackInstallMismatch:
			rb.Reason = RollbackInstallMismatch
		case RollbackFilesOccupied:
			rb.Reason = RollbackFilesOccupied
		case RollbackManifestOccupied:
			rb.Reason = RollbackManifestOccupied
		case RollbackProbeAttempts:
			rb.Reason = RollbackProbeAttempts
		case RollbackCommitDeadline:
			rb.Reason = RollbackCommitDeadline
		case RollbackProbeUnpriced:
			rb.Reason = RollbackProbeUnpriced
		case RollbackTargetFenced:
			rb.Reason = RollbackTargetFenced
		case RollbackDeclaredLost:
			rb.Reason = RollbackDeclaredLost
		default:
			return fmt.Errorf("provision: rollback reason %q has no recorded sentence", why)
		}
		t := now.UTC()
		d := t.Add(rollbackReleaseBudget)
		rb.RequestedAt, rb.ReleaseDeadline = &t, &d
		rec.Region = m.From
		m.State, m.Rollback = MoveRollingBack, rb
		return nil
	}
}

// recordedRollbackReason reports whether r is one of the sentences
// MoveRollBack records.
func recordedRollbackReason(r string) bool {
	switch r {
	case RollbackConfigChanged, RollbackLandingDeadline, RollbackClaimWired, RollbackLandingLayout, RollbackInstallMismatch,
		RollbackFilesOccupied, RollbackManifestOccupied, RollbackProbeAttempts, RollbackCommitDeadline, RollbackDeclaredLost, RollbackTargetFenced, RollbackProbeUnpriced:
		return true
	}
	return false
}

// rollingBack checks the guard and that the record is a rolling_back move
// with its journal.
func rollingBack(g MoveGuard, rec *Record) (*MoveRollback, error) {
	if err := g.check(rec); err != nil {
		return nil, err
	}
	if rec.Move.State != MoveRollingBack || rec.Move.Rollback == nil {
		return nil, fmt.Errorf("%w: the move is %s, not rolling back", ErrMoveConflict, rec.Move.State)
	}
	return rec.Move.Rollback, nil
}

// rbTargetOrder is the target's rollback order (§7.8 target table). A
// clean release skips returning and returned; a dirty one may not.
var rbTargetOrder = []string{RbTargetReleasing, RbTargetReleased, RbTargetReturning, RbTargetReturned, RbTargetCleaning, RbTargetCleaned}

// MoveRollbackTargetPhase advances the target's rollback phase one step
// under the guard (§7.8). Beyond the order: released goes to returning
// only when Dirty and to cleaning only when not, and cleaning waits for the
// source's RecoveredAt (Review Focus 4). It never leaves lost: a declared
// loss has its own closures (MoveRollbackLossCommitted,
// MoveLostTargetParked). stamp runs after the checks.
func MoveRollbackTargetPhase(g MoveGuard, from, to string, stamp func(*MoveRollback)) func(*Record) error {
	return func(rec *Record) error {
		rb, err := rollingBack(g, rec)
		if err != nil {
			return err
		}
		ok := rb.TargetPhase == from
		switch {
		case !ok:
		case from == RbTargetReleased && to == RbTargetCleaning:
			ok = !rb.Dirty
		case from == RbTargetReleased && to == RbTargetReturning:
			ok = rb.Dirty
		default:
			ok = nextIn(rbTargetOrder, from, to)
		}
		if ok && to == RbTargetCleaning && rb.RecoveredAt == nil {
			ok = false
		}
		if !ok {
			return fmt.Errorf("%w: the target's rollback is %s (dirty %t), step expects %s to %s", ErrMoveConflict, rb.TargetPhase, rb.Dirty, from, to)
		}
		rb.TargetPhase = to
		if stamp != nil {
			stamp(rb)
		}
		return nil
	}
}

// The source's rollback orders (§7.8 source table): the clean return
// verifies its own copy, the dirty return installs the target's.
var (
	rbSourceHead  = []string{RbSourceWaiting, RbSourceClaimed, RbSourceRestoring}
	rbSourceClean = []string{RbSourceRestoring, RbSourceVerified, RbSourceProbing}
	rbSourceDirty = []string{RbSourceRestoring, RbSourceParking, RbSourceBuilding, RbSourcePublished, RbSourceLocal, RbSourceConfigured, RbSourceRestored, RbSourceProbing}
	rbSourceTail  = []string{RbSourceProbing, RbSourceProbed, RbSourceRecovered, RbSourceOpened}
)

// MoveRollbackSourcePhase advances the source's rollback phase one step
// under the guard (§7.8). The branch after restoring is the target's
// Dirty: clean verifies, dirty installs. waiting ends only on the target's
// evidence, released clean or returned dirty; a target at lost is never
// taken (§7.10), only the released a loss commit writes. stamp runs after
// the checks.
func MoveRollbackSourcePhase(g MoveGuard, from, to string, stamp func(*MoveRollback)) func(*Record) error {
	return func(rec *Record) error {
		rb, err := rollingBack(g, rec)
		if err != nil {
			return err
		}
		branch := rbSourceClean
		if rb.Dirty {
			branch = rbSourceDirty
		}
		ok := rb.SourcePhase == from && (nextIn(rbSourceHead, from, to) || nextIn(branch, from, to) || nextIn(rbSourceTail, from, to))
		if ok && from == RbSourceWaiting {
			ok = (rb.TargetPhase == RbTargetReleased && !rb.Dirty) || (rb.TargetPhase == RbTargetReturned && rb.Dirty)
		}
		if !ok {
			return fmt.Errorf("%w: the source's rollback is %s (target %s, dirty %t), step expects %s to %s", ErrMoveConflict, rb.SourcePhase, rb.TargetPhase, rb.Dirty, from, to)
		}
		rb.SourcePhase = to
		if stamp != nil {
			stamp(rb)
		}
		return nil
	}
}

// MoveRecordRecoveryInputs records the inputs a rollback's recovery probe
// is about to render (§7.7, J2-R2-5; Plan 4 Task 5, Task 3b concern 3): the
// probe renders the current bundle, overlay and release and records them
// first, so the held render and the launch it then writes are bound to what
// the record says (MovesClaimPolicy). It is legal only on this rollback
// (rbID) while its source is at probing, and writes Rollback.RecoveryInputs
// alone: a dirty return's configured recorded its own, and a restarted
// probe's are the ones it renders now; neither is ever compared with
// move.inputs. The release must be named. in is copied.
func MoveRecordRecoveryInputs(g MoveGuard, rbID string, in MoveInputs) func(*Record) error {
	return func(rec *Record) error {
		if in.Release == "" {
			return fmt.Errorf("provision: recovery inputs name the release")
		}
		rb, err := rollingBack(g, rec)
		if err != nil {
			return err
		}
		if rb.ID != rbID || rb.SourcePhase != RbSourceProbing {
			return fmt.Errorf("%w: recovery inputs are recorded by rollback %s at probing (the record is %s at %s)", ErrMoveConflict, rbID, rb.ID, rb.SourcePhase)
		}
		cp := in
		cp.BundleETags = make(map[string]string, len(in.BundleETags))
		for k, v := range in.BundleETags {
			cp.BundleETags[k] = v
		}
		rb.RecoveryInputs = &cp
		return nil
	}
}

// MoveMarkStuck stops a recovery for an operator (§7.8 stuck, §7.4
// reverting): rolling_back or reverting to stuck, with a recorded reason
// in Move.StuckReason, the one field it writes besides the state and
// StuckAt (Move.Reason and Rollback.Reason keep the cause). Each reason
// belongs to its phases:
//
//   - StuckBaselineMismatch: a rollback whose source is at restoring or
//     verified (a clean return's verification, INV-4), or a revert (its
//     verify);
//   - StuckRecoveryProbeAttempts: a rollback whose source is at probing;
//   - StuckReturnFilesOccupied, StuckReturnManifestOccupied and
//     StuckReturnUnknownStamp: a dirty rollback whose source is at local
//     (the shared installer, Plan 4 Task 3; the prepared manifest, Task 6);
//   - StuckReturnInstallMismatch: a dirty rollback whose source is at
//     configured;
//   - StuckReturnTargetDatabase, StuckReturnTargetChanged and
//     StuckReturnTargetFiles: a dirty rollback whose target is at returning
//     (the source still waiting for it); the target phase is kept, so a
//     retry runs the return again;
//   - StuckReleaseTargetFiles: a rollback whose target is at releasing,
//     the source still waiting (Plan 5 Task 9a); the target phase is kept,
//     so a retry runs the release again;
//   - StuckRevertProbeBudget: a revert;
//   - StuckRecoveryProbeAttempts: a revert too, whose probe has the recovery
//     probe's nine attempts as well as its fifteen minutes (§7.7; Plan 4
//     Task 6 fix round 1).
//
// A rollback records the source phase in StuckAt, which MoveRetry returns
// to; a revert has no Rollback journal and resumes from the claim's
// rev_phase (ruling J20). The same reason again on a stuck record is a
// replay; another is a conflict. Any other text is refused, and nothing is
// written.
func MoveMarkStuck(g MoveGuard, reason string) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		var why string
		switch reason {
		case StuckBaselineMismatch:
			why = StuckBaselineMismatch
		case StuckRecoveryProbeAttempts:
			why = StuckRecoveryProbeAttempts
		case StuckRevertProbeBudget:
			why = StuckRevertProbeBudget
		case StuckReturnFilesOccupied:
			why = StuckReturnFilesOccupied
		case StuckReturnManifestOccupied:
			why = StuckReturnManifestOccupied
		case StuckReturnInstallMismatch:
			why = StuckReturnInstallMismatch
		case StuckReturnUnknownStamp:
			why = StuckReturnUnknownStamp
		case StuckReturnTargetDatabase:
			why = StuckReturnTargetDatabase
		case StuckReturnTargetChanged:
			why = StuckReturnTargetChanged
		case StuckReturnTargetFiles:
			why = StuckReturnTargetFiles
		case StuckReleaseTargetFiles:
			why = StuckReleaseTargetFiles
		default:
			return fmt.Errorf("provision: stuck reason %q has no recorded sentence", reason)
		}
		if m.State == MoveStuck && m.StuckReason == why {
			return nil
		}
		ok := false
		switch m.State {
		case MoveRollingBack:
			if rb := m.Rollback; rb != nil {
				switch why {
				case StuckBaselineMismatch:
					ok = rb.SourcePhase == RbSourceRestoring || rb.SourcePhase == RbSourceVerified
				case StuckRecoveryProbeAttempts:
					ok = rb.SourcePhase == RbSourceProbing
				case StuckReturnFilesOccupied, StuckReturnManifestOccupied, StuckReturnUnknownStamp:
					ok = rb.Dirty && rb.SourcePhase == RbSourceLocal
				case StuckReturnInstallMismatch:
					ok = rb.Dirty && rb.SourcePhase == RbSourceConfigured
				case StuckReturnTargetDatabase, StuckReturnTargetChanged, StuckReturnTargetFiles:
					ok = rb.Dirty && rb.TargetPhase == RbTargetReturning && rb.SourcePhase == RbSourceWaiting
				case StuckReleaseTargetFiles:
					ok = rb.TargetPhase == RbTargetReleasing && rb.SourcePhase == RbSourceWaiting
				}
			}
		case MoveReverting:
			ok = why == StuckBaselineMismatch || why == StuckRevertProbeBudget || why == StuckRecoveryProbeAttempts
		}
		if !ok {
			return fmt.Errorf("%w: the move cannot stop here for that reason", ErrMoveConflict)
		}
		if m.Rollback != nil {
			m.Rollback.StuckAt = m.Rollback.SourcePhase
		}
		m.State = MoveStuck
		m.StuckReason = why
		return nil
	}
}

// MoveRetry is the operator's --retry (§7.8, Task 7): stuck back to the
// recovery it stopped. A rollback returns to rolling_back with the source
// at the phase StuckAt names, so a mismatch verifies again and a probe
// probes again; a revert (no Rollback) returns to reverting. It clears
// Move.StuckReason and StuckAt: the move's cause (Move.Reason) and the
// rollback's (Rollback.Reason) stay. It leaves RecoveryRetryEpoch alone, so
// the legacy (flag-off) retry writes the same bytes it always did; the
// claim-backed retry is MoveRetryRecovery.
func MoveRetry(g MoveGuard) func(*Record) error {
	return moveRetry(g, false)
}

// MoveRetryRecovery is MoveRetry for the claim-backed retry: in the same
// transition it advances RecoveryRetryEpoch, which grants the recovery probe
// a fresh durable start budget.
func MoveRetryRecovery(g MoveGuard) func(*Record) error {
	return moveRetry(g, true)
}

func moveRetry(g MoveGuard, advanceEpoch bool) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		if m.State != MoveStuck {
			return fmt.Errorf("%w: %s", ErrMoveConflict, RefusedNotStuck)
		}
		if advanceEpoch && m.RecoveryRetryEpoch == ^uint64(0) {
			return ErrMoveConflict
		}
		if rb := m.Rollback; rb != nil {
			if rb.StuckAt == "" {
				return fmt.Errorf("%w: the stuck rollback records no phase to return to", ErrMoveConflict)
			}
			m.State, rb.SourcePhase, rb.StuckAt = MoveRollingBack, rb.StuckAt, ""
		} else {
			m.State = MoveReverting
		}
		m.StuckReason = ""
		if advanceEpoch {
			m.RecoveryRetryEpoch++
		}
		return nil
	}
}

// MoveRollbackTargetLost is the declared-loss decision on the record
// (§7.10): the target phase goes to lost, recording the loss record key
// (or the termination fence record key) and when. Legal while the target
// still holds something the source needs: releasing, returning, or
// released dirty and not yet returned. Dirty is left as it was. The same
// lossRef again is a replay; another is a conflict. The source's recovery
// does not take lost: MoveRollbackLossCommitted moves it on once the claim
// carries the declaration.
func MoveRollbackTargetLost(g MoveGuard, lossRef string, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if lossRef == "" || now.IsZero() {
			return fmt.Errorf("provision: malformed loss declaration")
		}
		rb, err := rollingBack(g, rec)
		if err != nil {
			return err
		}
		switch {
		case rb.TargetPhase == RbTargetLost && rb.LossRef == lossRef:
			return nil
		case rb.TargetPhase == RbTargetLost:
			return fmt.Errorf("%w: another loss is recorded for this target", ErrMoveConflict)
		case rb.TargetPhase == RbTargetReleasing, rb.TargetPhase == RbTargetReturning, rb.TargetPhase == RbTargetReleased && rb.Dirty:
		default:
			return fmt.Errorf("%w: %s", ErrMoveConflict, RefusedNothingToDeclare)
		}
		t := now.UTC()
		rb.TargetPhase, rb.LossRef, rb.LostAt = RbTargetLost, lossRef, &t
		return nil
	}
}

// MoveRollbackLossCommitted closes a declared loss on the record (§7.10):
// lost to released, clean, with LossRef as the evidence, so the source's
// waiting takes it as a clean release. The caller writes it only after the
// claim carries the declaration (its loss record committed). A replay
// finds released with that evidence and changes nothing.
func MoveRollbackLossCommitted(g MoveGuard) func(*Record) error {
	return func(rec *Record) error {
		rb, err := rollingBack(g, rec)
		if err != nil {
			return err
		}
		if rb.TargetPhase == RbTargetReleased && rb.LossRef != "" && rb.Evidence == rb.LossRef && !rb.Dirty {
			return nil
		}
		if rb.TargetPhase != RbTargetLost || rb.LossRef == "" {
			return fmt.Errorf("%w: the target's rollback is %s, not lost", ErrMoveConflict, rb.TargetPhase)
		}
		rb.TargetPhase, rb.Dirty, rb.Evidence = RbTargetReleased, false, rb.LossRef
		return nil
	}
}

// MoveRolledBackClose is a rollback's close (§7.8 rolled_back, index D11):
// a rolling_back move whose source has recovered and opened and whose
// target has parked clears Move and leaves the outcome in LastMove, in the
// same write. The region is already From (MoveRollBack); a record whose
// region is not is refused. The offline time is the rollback's OpenedAt
// minus CutAt (MoveOfflineS), zero when CutAt was never stamped.
func MoveRolledBackClose(g MoveGuard, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		rb, err := rollingBack(g, rec)
		if err != nil {
			return err
		}
		m := rec.Move
		if rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil {
			return fmt.Errorf("%w: close follows the recovery, the open and the target park", ErrMoveConflict)
		}
		if rec.Region != m.From {
			return fmt.Errorf("%w: the record's region is not the source's", ErrMoveConflict)
		}
		out := &MoveOutcome{ID: g.ID, Outcome: "rolled_back", At: now.UTC()}
		if s, ok := MoveOfflineS(m); ok {
			out.OfflineS = s
		}
		rec.LastMove, rec.Move = out, nil
		return nil
	}
}

// MoveRevertReopened stamps ReopenedAt on a reverting move: its source
// answered the probe (rev_phase probed, ruling J26). The first stamp
// stands.
func MoveRevertReopened(g MoveGuard, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveReverting {
			return fmt.Errorf("%w: the reopen is a revert's", ErrMoveConflict)
		}
		if rec.Move.ReopenedAt == nil {
			t := now.UTC()
			rec.Move.ReopenedAt = &t
		}
		return nil
	}
}

// MoveRecordRevertInputs records the inputs a revert's probe is about to
// render (Plan 4 Task 6; §7.7, J18): the current bundle, overlay and release,
// written before the render so the held render and the launch are bound to
// what the record says. It is legal on a reverting move that has not
// reopened (a revert has no Rollback journal, so the inputs are the move's
// own RecoveryInputs), and writes nothing else; a later probe's replace an
// earlier one's. They are never compared with Inputs. The release must be
// named. in is copied.
func MoveRecordRevertInputs(g MoveGuard, in MoveInputs) func(*Record) error {
	return func(rec *Record) error {
		if in.Release == "" {
			return fmt.Errorf("provision: recovery inputs name the release")
		}
		if err := g.check(rec); err != nil {
			return err
		}
		if rec.Move.State != MoveReverting || rec.Move.ReopenedAt != nil {
			return fmt.Errorf("%w: a revert's recovery inputs are recorded before it reopens", ErrMoveConflict)
		}
		cp := in
		cp.BundleETags = make(map[string]string, len(in.BundleETags))
		for k, v := range in.BundleETags {
			cp.BundleETags[k] = v
		}
		rec.Move.RecoveryInputs = &cp
		return nil
	}
}

// MoveRevertClose is a revert's close (§7.4 reverting, index D11): the
// caller has seen the claim's rev_phase at opened; the record needs its
// ReopenedAt. Move is cleared and LastMove says rolled_back, the offline
// time counted from CutAt to the reopen (MoveOfflineS). A revert never
// flipped, so the region is From; a record whose region is not is refused.
func MoveRevertClose(g MoveGuard, now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if err := g.check(rec); err != nil {
			return err
		}
		m := rec.Move
		if m.State != MoveReverting || m.ReopenedAt == nil {
			return fmt.Errorf("%w: a revert closes after its source reopens", ErrMoveConflict)
		}
		if rec.Region != m.From {
			return fmt.Errorf("%w: the record's region is not the source's", ErrMoveConflict)
		}
		out := &MoveOutcome{ID: g.ID, Outcome: "rolled_back", At: now.UTC()}
		if s, ok := MoveOfflineS(m); ok {
			out.OfflineS = s
		}
		rec.LastMove, rec.Move = out, nil
		return nil
	}
}

// InstanceStoreEnv is the environment name an instance's private object
// store hangs off: the record's own Environment, or its Slug when it has
// none. Twin of the create door's inline rule (internal/console/
// apiinstances.go, "env, derived := rec.Environment, false"); the door is
// not refactored to call this, so keep the two in step. Shared by the
// console's move door and the host round, which cannot see a console-local
// helper.
func InstanceStoreEnv(rec *Record) string {
	if rec.Environment != "" {
		return rec.Environment
	}
	return rec.Slug
}

// MovePromoteReady is the source box's preparing→ready promotion.
func MovePromoteReady(now time.Time) func(*Record) error {
	return func(rec *Record) error {
		if rec.Move == nil || rec.Move.State != MovePreparing {
			return fmt.Errorf("%w: ready is only reached from preparing", ErrMoveConflict)
		}
		t := now.UTC()
		rec.Move.State = MoveReady
		rec.Move.ReadyAt = &t
		return nil
	}
}

// MoveMarkCutting is the console's cut verb: legal only from ready — the
// tenant freezes writes first, and a cut from preparing would cut over a
// replica that was never caught up.
func MoveMarkCutting() func(*Record) error {
	return func(rec *Record) error {
		if rec.Move == nil || rec.Move.State != MoveReady {
			return fmt.Errorf("%w: cut is only legal from ready", ErrMoveConflict)
		}
		rec.Move.State = MoveCutting
		return nil
	}
}

// MoveAbort marks a move for abort — legal before the point of no return.
// The clear itself is the source box's (MoveAbortComplete), after it has
// deleted the staged copy.
//
// The legal states are listed, not the illegal ones, so a state this
// console has never heard of is refused (G16, instance-moves design §6.1).
// A newer box writes "landing" after the flip, while the target restores
// from staging. Aborting it would send the target down the aborting arm to
// delete that staging before it promoted, and the first wake would seed an
// empty database.
func MoveAbort() func(*Record) error {
	return func(rec *Record) error {
		if rec.Move == nil {
			return fmt.Errorf("%w: %s", ErrMoveConflict, refusedNoMove)
		}
		switch rec.Move.State {
		case MovePreparing, MoveReady:
		case MoveCutting:
			return fmt.Errorf("%w: past the point of no return — the cut completes forward", ErrMoveConflict)
		case MoveAborting:
			return fmt.Errorf("%w: already aborting", ErrMoveConflict)
		default:
			return fmt.Errorf("%w: a move in state %q cannot be aborted", ErrMoveConflict, rec.Move.State)
		}
		rec.Move.State = MoveAborting
		return nil
	}
}

// MoveAbortComplete is the source box's half of a legacy abort: staging is
// gone, the move field clears. It refuses a per-op move (Plan 5 Task 1): that
// completes through MoveGuardedAbortComplete once the target region has
// parked its lane and the claim is released, and its source never deletes
// the target's staging.
func MoveAbortComplete() func(*Record) error {
	return func(rec *Record) error {
		if rec.Move == nil || rec.Move.State != MoveAborting {
			return fmt.Errorf("%w: abort-complete is only legal from aborting", ErrMoveConflict)
		}
		if !rec.Move.Legacy() {
			return fmt.Errorf("%w: a move with an id completes its abort through the guarded transition", ErrMoveConflict)
		}
		rec.Move = nil
		return nil
	}
}

// MoveComplete is the source's terminal act: Region flips to the target and
// Move clears, in ONE write — every reader that knows nothing about moves
// sees the old region right up until it sees the new one.
func MoveComplete() func(*Record) error {
	return func(rec *Record) error {
		if rec.Move == nil || rec.Move.State != MoveCutting {
			return fmt.Errorf("%w: complete is only legal from cutting", ErrMoveConflict)
		}
		rec.Region = rec.Move.To
		rec.Move = nil
		return nil
	}
}

// DeleteRecord retires an instance's route authorization. Object deletion
// only — the parked database and blobs under the regional bucket's prefix
// are deliberately untouched (design §3.2); Store's Delete is idempotent,
// so retiring an already-gone record is not an error.
func DeleteRecord(ctx context.Context, s release.Store, acct, app, host string) error {
	return s.Delete(ctx, RecordKey(acct, app, host))
}

// UpdateRecord rewrites an EXISTING instance record, guarded by the etag the
// caller read it under. Three writes now touch a record and each has its own
// contract, deliberately:
//
//   - CreateRecord is create-only ("" etag): a lost race is a permanent 409,
//     because two signups for one host must never silently merge.
//   - SetRecordRegion is an internal read-modify-write that retries once: it
//     stamps ONE field from a placement decision made elsewhere, so re-reading
//     and re-applying on a lost race changes nothing about what it meant.
//   - This one surfaces the conflict. Its caller is the console's member-facing
//     PATCH door, where the read and the write are one decision — a member sent
//     the upstream set they wanted GIVEN the record they were shown — so a
//     silent retry would re-apply that decision on top of a body nobody read,
//     which is the clobber PutManifestIfMatch's own doc comment records the
//     cost of.
//
// The record is marshaled here, by the same json.Marshal every other writer
// uses: the caller hands over a whole provision.Record (already
// ValidateShape'd) and never hand-builds the object body, so field ORDER and
// omitempty behaviour — which the registry's joined addrs column compares
// verbatim — cannot drift between the create path and this one.
//
// An empty etag is rejected rather than quietly becoming a create-only Put:
// this verb is "rewrite what I read", and a caller that reached here with no
// etag read nothing.
func claimRawUpdateRecord(ctx context.Context, s release.Store, rec Record, etag string) error {
	if etag == "" {
		return fmt.Errorf("provision: update %s/%s/%s: no etag — an update must be guarded by the read it was decided from",
			rec.Account, rec.App, rec.Host)
	}
	prior, currentETag, err := s.Get(ctx, RecordKey(rec.Account, rec.App, rec.Host))
	if err != nil {
		return err
	}
	if currentETag != etag {
		return release.ErrCASConflict
	}
	before, err := DecodeRecord(prior)
	if err != nil {
		return fmt.Errorf("provision: record %s/%s/%s: %w", rec.Account, rec.App, rec.Host, err)
	}
	if !protectedFieldsEqual(before, rec) {
		return ErrMaintenanceProjection
	}
	if !seedEqual(before.Seed, rec.Seed) {
		return ErrSeedProtected
	}
	body, err := json.Marshal(rec)
	if err != nil {
		return err
	}
	_, _, err = WriteRecord(ctx, s, RecordKey(rec.Account, rec.App, rec.Host), prior, body, etag)
	return err
}

// GetRecord reads and validates one instance record. A body whose own
// Host/App/Account disagree with the key path it was read from is refused
// rather than trusted — the same "never guessed at" discipline
// AccountManifest's sqid-in-path-and-body carries.
func GetRecord(ctx context.Context, s release.Store, acct, app, host string) (*Record, error) {
	rec, _, err := GetRecordETag(ctx, s, acct, app, host)
	return rec, err
}

// GetRecordETag is GetRecord carrying the etag its bytes were read under —
// for a caller that must write back GUARDED against what it read
// (UpdateRecord), not against whatever is there at write time. Same seam,
// same reason, as GetManifestETag beside PutManifestIfMatch.
func GetRecordETag(ctx context.Context, s release.Store, acct, app, host string) (*Record, string, error) {
	raw, etag, err := s.Get(ctx, RecordKey(acct, app, host))
	if err != nil {
		return nil, "", err
	}
	rec, err := DecodeRecord(raw)
	if err != nil {
		return nil, "", fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err)
	}
	if rec.Account != acct || rec.App != app || rec.Host != host {
		return nil, "", fmt.Errorf("provision: record %s/%s/%s: record disagrees with its own path", acct, app, host)
	}
	return &rec, etag, nil
}

// recordHosts picks the record hosts out of a listing of RecordsPrefix:
// the keys of the form <prefix><host>.json, in listing order. Anything else
// (a key nested below the level, a name without .json) is not a record at
// this level. Both ListRecords and ListRecordsStrict read through it, so
// the two never disagree about what a record key looks like.
func recordHosts(keys []string, prefix string) []string {
	var hosts []string
	for _, key := range keys {
		rest, ok := strings.CutPrefix(key, prefix)
		if !ok || rest == "" || strings.Contains(rest, "/") {
			continue
		}
		host, ok := strings.CutSuffix(rest, ".json")
		if !ok || host == "" {
			continue
		}
		hosts = append(hosts, host)
	}
	return hosts
}

// ListRecords lists an app's instance records, sorted by host, capped at
// max with truncated reporting whether more existed. The cap is decided on
// the KEY count — how many instances/<host>.json objects exist under the
// prefix — not on how many bodies successfully parse: a handful of
// malformed bodies among the first max keys must not silently pull in
// extra keys past the cap to compensate, and truncated must not depend on
// which records happen to be malformed today. A malformed or
// path-disagreeing body (GetRecord's own checks) is skipped, not fatal.
//
// A negative max is clamped to 0 rather than treated as "no cap" (a caller
// passing an unvalidated int — a console handler's query param, most
// likely — must degrade safely, not panic on the slice below or list
// everything by accident). max=0 lists nothing but still reports
// truncated when any keys exist, the same "cap decided on key count" rule
// applied at its floor.
func ListRecords(ctx context.Context, s release.Store, acct, app string, max int) (recs []Record, truncated bool, err error) {
	recs, truncated, _, err = ListRecordsComplete(ctx, s, acct, app, max)
	return recs, truncated, err
}

// ListRecordsComplete is ListRecords that also names each host whose record
// could not be read (a store error, a malformed body, a body that disagrees
// with its path). A caller that must know whether the list is the whole app
// (the deploy watch's completeness check, Astra plan 6 R3) reads skipped.
// Hosts cut by the cap are not skipped; truncated reports those.
func ListRecordsComplete(ctx context.Context, s release.Store, acct, app string, max int) (recs []Record, truncated bool, skipped []string, err error) {
	if max < 0 {
		max = 0
	}
	prefix := RecordsPrefix(acct, app)
	keys, err := s.List(ctx, prefix)
	if err != nil {
		return nil, false, nil, err
	}
	hosts := recordHosts(keys, prefix)
	sort.Strings(hosts)
	if len(hosts) > max {
		truncated = true
		hosts = hosts[:max]
	}
	for _, host := range hosts {
		rec, err := GetRecord(ctx, s, acct, app, host)
		if err != nil {
			skipped = append(skipped, host)
			continue
		}
		recs = append(recs, *rec)
	}
	return recs, truncated, skipped, nil
}

// ErrRecordsTruncated is ListRecordsStrict's answer when the app holds more
// record keys than the caller's cap.
var ErrRecordsTruncated = errors.New("provision: the app has more records than the cap allows")

// ListRecordsStrict is ListRecords for a caller that decides something from
// the whole record set, or from a record's absence: whether another record
// shares an environment, whether an import is in flight. ListRecords is
// built for display and quietly drops what it cannot show; this one never
// does, and the difference is in three places.
//
//   - A listing that fails, or a record key that will not Get (any error,
//     including not-found: the key was listed a moment ago), is an error and
//     no records come back. ListRecords skips such a record.
//   - More record keys than max is the error ErrRecordsTruncated, not a
//     flag the caller might forget to read. max < 0 is clamped to 0, as in
//     ListRecords, so an app with any record fails at a cap of 0.
//   - A record that reads but does not decode, or whose body disagrees with
//     its own path, is not returned and not skipped silently: its host (the
//     host in the key) is in unreadable. A caller must treat each such host
//     as a record that could be anything, and fail closed on it.
//
// recs is sorted by host. unreadable is non-nil whenever err is nil.
func ListRecordsStrict(ctx context.Context, s release.Store, acct, app string, max int) (recs []Record, unreadable map[string]bool, err error) {
	if max < 0 {
		max = 0
	}
	prefix := RecordsPrefix(acct, app)
	keys, err := s.List(ctx, prefix)
	if err != nil {
		return nil, nil, fmt.Errorf("provision: list records of %s/%s: %w", acct, app, err)
	}
	hosts := recordHosts(keys, prefix)
	if len(hosts) > max {
		return nil, nil, ErrRecordsTruncated
	}
	sort.Strings(hosts)
	unreadable = map[string]bool{}
	for _, host := range hosts {
		raw, _, err := s.Get(ctx, RecordKey(acct, app, host))
		if err != nil {
			return nil, nil, fmt.Errorf("provision: read record %s: %w", RecordKey(acct, app, host), err)
		}
		rec, err := DecodeRecord(raw)
		if err != nil || rec.Account != acct || rec.App != app || rec.Host != host {
			unreadable[host] = true
			continue
		}
		recs = append(recs, rec)
	}
	return recs, unreadable, nil
}

// SocketDir is where a hibernating instance's control socket lives — the
// same tmpfs every other exec-managed route already uses
// (host/bootstrap.go's tmpfiles.d entry).
const SocketDir = "/run/carlos"

// DefaultSocket and DefaultDBPath are the route conventions `carlos add`
// and the reconciler's mint (Task 3) both build on — the "honest shared
// core" design §9 names, so a hand-added route and a provisioned one land
// on identical path shapes.
//
// filepath.Join, not concatenation: dataDir is CARLOS_DATA_DIR, an operator-
// typed env var, and "/data/" is an ordinary way to write a directory.
// Concatenating gave "/data//h.db" — the same file under a second name, and
// two names for one database is how a route and litestream's box-wide config
// come to disagree about what they are replicating. (The literals runAdd
// used before this helper existed were Join'd; this keeps that.)
func DefaultSocket(host string) string          { return filepath.Join(SocketDir, host+".sock") }
func DefaultDBPath(dataDir, host string) string { return filepath.Join(dataDir, host+".db") }

// MaxInstanceID bounds InstanceID. A unix socket address holds MaxUnixPath
// bytes of path and a host may be 253 bytes, so the directories a socket
// lives under are keyed by a bounded identifier, never by the host itself
// (isolated-exec design §3.1, codex R5-3).
const MaxInstanceID = 40

// MaxUnixPath is sockaddr_un's sun_path (108 bytes) minus the NUL Go
// insists on room for.
const MaxUnixPath = 107

// ControlSocketDir and ControlSocketPath are the root instance-control
// channel (design §5). Deliberately NOT under SocketDir: an instance's
// namespace replaces /run/carlos with an empty tmpfs, and nothing an
// instance can reach must resolve to this path.
//
// Defined in internal/controlsock (a leaf package with no imports) and
// re-exported here unchanged: provision already imports internal/release
// (for the store), and internal/unixdial — imported by internal/release's
// unix dial sites — needs this path without pulling provision (and so
// release) back in.
const (
	ControlSocketDir  = controlsock.Dir
	ControlSocketPath = controlsock.Path
)

// InstanceID is the bounded per-host identifier every isolated-exec
// directory, socket, unit instance and slice is keyed by. A host of at
// most MaxInstanceID bytes is its own identifier. A longer host becomes
// "_" + its first 22 bytes + "-" + 16 hex digits of its sha256, also
// MaxInstanceID bytes.
//
// The underscore is the whole reason the two forms can share one
// namespace: no valid host contains one (controlname labels are
// [a-z0-9-]), so a hashed identifier can never equal any short host, and
// two hashed identifiers collide only on a 64-bit hash collision between
// two long hosts with the same 22-byte prefix (codex R6-1). Anything that
// finds two rows with one identifier treats them as a conflict, never as
// the same host.
func InstanceID(host string) string {
	if len(host) <= MaxInstanceID {
		return host
	}
	sum := sha256.Sum256([]byte(host))
	return "_" + host[:22] + "-" + hex.EncodeToString(sum[:8])
}

// InstanceDir is the per-host data directory of the nested layout:
// <dataDir>/instances/<InstanceID(host)>. File names INSIDE it keep the
// full host (a file name may be 255 bytes; a socket path may not).
func InstanceDir(dataDir, host string) string {
	return filepath.Join(dataDir, "instances", InstanceID(host))
}

// NestedDBPath is the database under InstanceDir. It is what the layout
// pass writes into a migrated row's DBPath; DefaultDBPath stays flat in
// this release (design §8.1).
func NestedDBPath(dataDir, host string) string {
	return filepath.Join(InstanceDir(dataDir, host), host+".db")
}

// InstanceRunDir is the per-host socket directory of the nested layout.
func InstanceRunDir(host string) string {
	return filepath.Join(SocketDir, "instances", InstanceID(host))
}

// NestedSocket is the instance's listener under InstanceRunDir; its
// control socket is NestedSocket + ".control", as for every exec route.
func NestedSocket(host string) string {
	return filepath.Join(InstanceRunDir(host), "app.sock")
}

// NestedPaths reports whether a row's database or socket sits anywhere
// under the nested layout's trees: <dataDir>/instances/ or
// <SocketDir>/instances/. It reads the path shape only, whatever the row's
// flags, backing or exact file names say.
//
// That is deliberately looser than activator.Classify, which calls a row
// nested only when both paths equal the nested defaults. The downgrade
// guards (design §8.2, "That order is enforced") use this test instead:
// a row can leave management or be hand-edited and still have its files
// under the instance tree, and a release before the layout, an unflag or
// a re-add would mishandle it all the same. Do not merge the two.
//
// Both paths are cleaned first, so a hand-edited row with a doubled slash
// or a dot segment is judged by the place it names.
func NestedPaths(dataDir string, rt registry.Route) bool {
	return strings.HasPrefix(filepath.Clean(rt.DBPath), filepath.Join(dataDir, "instances")+"/") ||
		strings.HasPrefix(filepath.Clean(rt.Socket), filepath.Join(SocketDir, "instances")+"/")
}

// LogSocket is the agent's per-role log listener for a host (design
// §6.4): outside every tenant mount, keyed by the identifier.
func LogSocket(host, role string) string {
	return filepath.Join(SocketDir, "logs", InstanceID(host)+"."+role+".sock")
}

// LayoutJournalPath is the migration journal for a host in flight
// (isolated-exec design §8.2): under <dataDir>/layout, outside every tenant
// mount. Its presence is what makes a host `fenced` (§8.1); plan 4 writes
// it, and every consumer that must not act on a half-migrated host checks
// for it here.
func LayoutJournalPath(dataDir, host string) string {
	return filepath.Join(dataDir, "layout", host+".json")
}

// LayoutJournalExists reports whether a journal is at LayoutJournalPath.
// It Lstats, so a dangling symlink there counts, and any error other than
// "not exist" counts too: a journal this process cannot even stat is not
// proof there isn't one.
func LayoutJournalExists(dataDir, host string) bool {
	_, err := os.Lstat(LayoutJournalPath(dataDir, host))
	return !errors.Is(err, os.ErrNotExist)
}

// IsControlEndpoint reports whether path is the control channel or
// anything else under its directory. The edge and the agent refuse to
// dial such a path as an upstream (design §5.4): a registry row pointing
// at it is refused before connect(). Delegates to internal/controlsock;
// see the ControlSocketDir/ControlSocketPath comment above for why.
func IsControlEndpoint(path string) bool {
	return controlsock.IsEndpoint(path)
}

// LongestSocketAddress is the longest unix socket path any helper here
// can produce, over the longest valid host, so a test can pin it under
// MaxUnixPath rather than trusting arithmetic in a comment.
func LongestSocketAddress() int {
	host := strings.Repeat("x", 253)
	longest := 0
	for _, p := range []string{
		NestedSocket(host),
		NestedSocket(host) + ".control",
		LogSocket(host, "sidecar"), // the longest role name
	} {
		if len(p) > longest {
			longest = len(p)
		}
	}
	return longest
}
func PutManifest(ctx context.Context, s release.Store, acct, app string, m Manifest) (claimResult0 error) {
	claimResult0 = release.WithAppMutation(ctx, s, acct, app, "provision.PutManifest", func(claimContext context.Context) error {
		claimResult0 = claimRawPutManifest(claimContext, s, acct, app, m)
		return claimResult0
	})
	return
}
func PutManifestIfMatch(ctx context.Context, s release.Store, acct, app string, m Manifest, etag string) (claimResult0 error) {
	claimResult0 = release.WithAppMutation(ctx, s, acct, app, "provision.PutManifestIfMatch", func(claimContext context.Context) error {
		claimResult0 = claimRawPutManifestIfMatch(claimContext, s, acct, app, m, etag)
		return claimResult0
	})
	return
}

// quotedInner is strconv.Quote without the surrounding quotes: the approved
// sentence (MV-N42) carries its own quotes, and the state inside them keeps
// %q's escaping of quotes, backslashes and control characters.
func quotedInner(s string) string {
	q := strconv.Quote(s)
	return q[1 : len(q)-1]
}