package host

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"log"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The moves owners for a landing probe (moves Phase 1 Plan 3 Task 5c; claim
// Plan 9 Group D, D5; rulings H2, H4, H13, H14, H16, H18). The claim's
// dispatcher (ClaimOwnerDispatch) hands this owner set only the move kinds,
// and it dispatches them again by the observed op's kind (ruling Q7): the
// restoring and reverting kinds go to the recovery owners
// (movesRecoveryOwners, movesrecoveryowners.go; Plan 4 Task 9), the moving
// kind is refused, and the landing is served here. Of the landing it serves
// one shape, a landing probe: the claim at landing/probing with this box
// executing as the move's target, the policy a probe, the registry row held
// by the move (and the committed wake, Task 7's, below). Everything else is
// refused with hostbox.ErrUnsupported or hostbox.ErrChanged, before any
// lease, stop or launch.
//
// The claim owns the admission, the restore, the launch, readiness,
// replication and the publication sequence (claimwakeassembly.go). This
// owner set supplies what moves knows:
//
//   - Select (read-only, before admission): the landing is still
//     admissible (its deadline, the record attempt within what T11
//     admitted, the install the record vouches for); the launch-input pin
//     at spawn (§7.7, round-2 finding 14): the live release and the report
//     of the selected launch's own environment are the pinned ones, else
//     ProbeInputsChanged
//     with the recorded sentence "configuration changed during the move",
//     which the landing turns into its rollback (Task 6); then the
//     admission per §7.14. The first probe is a cold admission into the
//     prepared generation, the move id, after PreparedCheck (.files-gen and
//     <dbPath>.gen name it and the holder's incarnation, and the journalled
//     install manifest is the seal's N); its baseline is the build's pin
//     (genselect.SelectPrepared, never an earlier stay's marker, H14); it
//     keeps the flip's stamp and claims sealed coverage when the seal was
//     built from the stopped database. A restarted attempt, once the move
//     id is admitted, is a cold admission into a fresh generation: no kept
//     stamp, no coverage, its baseline the prepared pin while the move id
//     has published nothing and the certified remote (genselect.Select)
//     once it has (H14). The launch identity is the pinned config and
//     release (H4), CARLOS_MOVE_PROBE=1 is in the child's environment (H18)
//     and the readiness path is the app's moves.ready_path (H12).
//   - AdmissionEvidence: the same decisions and the same admissibility,
//     re-derived and compared, pin included, against a baseline selected
//     again, immediately before the CAS.
//   - Materialize, before every start, under the admitted spawn selection,
//     in the order ruled for C1 (fix round 1, finding 8): the held render
//     and the held release, each under its own claim selection, held-config
//     and held-release, through the existing held paths
//     (appconfig.RenderHeldUnder, release.MaterializeHeldUnder) and
//     MovesClaimPolicy's probing rows (H13), the render's etags compared
//     with the pins before anything is selected or written; then the
//     verified prepared marker, published under the admitted writer's
//     fence; then the §5.1 start check (moves.StartCheck, not weakened)
//     against the claim after the admission's CAS; only then the spawn.
//   - Publish: Group D's call before replication finds the prepared marker
//     Materialize published and records it; it writes only under the fence
//     Materialize recorded for the session. The marker names the move id
//     with the build's pin and install.json's digest, superseding the
//     current marker (an earlier stay's, which Plan 2 keeps) by its exact
//     ETag, or created when there is none. PublishFenced: a restarted
//     attempt's fresh generation, after its own first root, as serving
//     publishes it. PublicationEvidence: this process's receipt for the
//     admitted session, read back.
//
// Production: the moves wake source (movesWakeSources, Plan 5 Task 9b)
// opens this owner set for the claim with both switches on, and its
// registration opens the gate that arms the parks.
type movesProbeOwners struct {
	HomeRegion, FleetHomeRegion string
	// Claims is the claim store (the deployment store in production).
	Claims  release.Store
	Records MovesRecordReader
	// ReadyPath is the app manifest's moves.ready_path, or "/".
	ReadyPath           func(ctx context.Context, account, app string) (string, error)
	Data                genselect.Reader
	Bucket, Region, Box string
	// DataDir is the root of the op journals (moves.JournalDir).
	DataDir  string
	FilesDir func(host string) (dir, envPath string)
	Markers  ClaimMarkerStore
	Route    func(host string) (*registry.Route, error)
	// Launch is the held row's launch: the live executable, its digest,
	// arguments, environment and version (the claim's launch snapshot), and
	// the moves env report of that same environment, computed at capture
	// from the one composition that produced the env (claimMoveLaunchSource:
	// ExecBackend.LaunchLayers, moves.ReportLaunch). The report's Digest is
	// the launch's move-portable pin. It is the moves owners' own value:
	// never persisted, and never digested by the claim (fix round 4).
	Launch func(context.Context, registry.Route) (ClaimServingLaunch, moves.EnvReport, error)
	// EnvKey is the env report's HMAC key (moves.ReportKeyFromSecrets), the
	// key Launch's report is made with. The pin is checked against the
	// report of the selected launch's own environment, so no report can be
	// decoupled from what runs (fix round 1, finding 3). Normalisation
	// contract: the report covers the launch's K=V list, last value wins
	// (childEnv's dedupe), with the probe-only variables (probeOnlyVars:
	// CARLOS_MOVE_PROBE) removed, since the source never ran with them and
	// the probe adds CARLOS_MOVE_PROBE=1 after the pin, and its digest is the
	// move-portable one (moves.PortableDigest). The production Launch is this
	// box's own composition (claim Plan 10's launch accessor).
	EnvKey []byte
	// Admitted is the number of probe attempts T11 admitted (H10).
	Admitted func(context.Context, *provision.Record) (int, error)
	// Works selects a held action's claim work at the admitted capture
	// (hostbox.WorkSource.SelectAt).
	Works  func(ctx context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error)
	Render func(ctx context.Context, rt registry.Route, op string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error)
	// Release makes version live for the held row before a start. On a box
	// whose agent cannot write /opt/carlos it is heldReleaseProbeRelease
	// (Task 7b): root publishes, inside root's own held-release selection.
	Release func(ctx context.Context, rt registry.Route, version string, work hostwrite.Work) error
	NewID   func() string
	Now     func() time.Time
	// Page pages an operator for a refused start check, once per op
	// (movespaging.go, §13): PageOperator once claim Plan 10 registers this
	// owner set; nil, nothing pages.
	Page func(ctx context.Context, level, subject, body string) error
	// Files is the moves C8 files manifest provider (claim §6.2a P6-D3;
	// Task 6 builds it). The recovery owners' Materialize establishes the
	// admitted generation's files manifest through it before every start;
	// without it every recovery kind is refused, typed, before anything is
	// read (movesrecoveryowners.go).
	Files ClaimFilesManifest
	// pageWarned is the pages warned unsent for want of a pager.
	pageWarned sync.Map

	mu        sync.Mutex
	published map[string]ClaimPublicationReceipt
	// envPins are the move-portable pins Select captured, by host, each
	// bound to the exact launch environment it was computed for
	// (captureEnvPin): what Materialize checks the same inputs against.
	// Memory only, never persisted, never digested by the claim. Each lives
	// as long as its wake (holdEnvPin) and never past its deadline.
	envPins   map[string]capturedEnvPin
	envPinSeq uint64
	// fences are the admitted spawn selections Materialize ran under, by
	// host and session: the live fence for the session's marker writes.
	fences map[string]hostwrite.Work
	// lossValidated is each host's last LossSources answer, which the
	// recovery's AdmissionEvidence of the same admission consumes
	// (movesloss.go).
	lossValidated map[string]lossValidation
}

var (
	_ ClaimWakeInputs        = (*movesProbeOwners)(nil)
	_ ClaimFencedPublication = (*movesProbeOwners)(nil)
)

// ProbeInputsChanged is a probe refused because a launch input is not the
// one pinned at readiness. Reason is the landing's recorded rollback
// sentence (provision.RollbackConfigChanged); the landing records it.
type ProbeInputsChanged struct{ Reason string }

func (e *ProbeInputsChanged) Error() string {
	return "host: the probe's launch inputs are not the pinned ones: " + e.Reason
}

func probeInputsChanged() error {
	return &ProbeInputsChanged{Reason: provision.RollbackConfigChanged}
}

// owners is the owner set's ClaimKindOwners.
func (m *movesProbeOwners) owners() ClaimKindOwners {
	return ClaimKindOwners{Inputs: m, AdmissionEvidence: m.AdmissionEvidence, PublicationEvidence: m.PublicationEvidence}
}

// MovesOwners is ClaimMovesOwners for the probe: this owner set and ops,
// with the moves loss owner (the resolve owner's signed-intent policy,
// hostops.MoveLossRecoveryOwner) and the loss baseline validator
// (LossSources, movesloss.go) the claim's writer admits a declared or fenced
// return through. ProbeReady stays nil, so the claim binds its own probe
// readiness check (claimProbeSocketReady: the declared path, 2xx only, a
// peer in the started role's cgroup; H12).
func (m *movesProbeOwners) MovesOwners(ops OperationAttempts) ClaimMovesOwners {
	return ClaimMovesOwners{LossRecovery: hostops.MoveLossRecoveryOwner(), LossSources: m.LossSources, ClaimKindOwners: m.owners(), Operations: ops}
}

func (m *movesProbeOwners) valid() bool {
	return m != nil && m.Claims != nil && m.Records != nil && m.ReadyPath != nil && m.Data != nil && m.Bucket != "" && m.Region != "" && m.Box != "" &&
		m.DataDir != "" && m.FilesDir != nil && m.Markers != nil && m.Route != nil && m.Launch != nil && len(m.EnvKey) != 0 && m.Admitted != nil && m.Works != nil &&
		m.Render != nil && m.Release != nil && m.NewID != nil && m.Now != nil
}

// probeLanding is what one landing probe's checks read.
type probeLanding struct {
	homeRegion string
	// committed is a landing wake after the commit (Plan 3 Task 7): the
	// claim at landing/committed, the record committed.
	committed bool
	op        string
	holder    *hostclaim.Holder
	rec       *provision.Record
	seal      moves.Seal
	manifest  moves.InstallManifest
}

// landing checks that o is a landing probe this box may run and reads the
// record, the seal and the journalled install manifest.
func (m *movesProbeOwners) landing(ctx context.Context, o hostclaim.Observed) (probeLanding, error) {
	var v probeLanding
	if !m.valid() {
		return v, hostwrite.ErrUnavailable
	}
	if o.Absent || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Op.Kind != hostclaim.KindLanding {
		return v, hostbox.ErrUnsupported
	}
	c := o.Claim
	var bindingErr error
	v.homeRegion, bindingErr = park.VerifiedHome(m.FleetHomeRegion, m.HomeRegion, m.Region, c.Op.Intent["home_region"])
	if bindingErr != nil {
		return v, bindingErr
	}
	switch c.Op.Phase {
	case hostclaim.PhaseLandingProbing:
	case hostclaim.PhaseLandingCommitted:
		v.committed = true
	default:
		return v, hostbox.ErrChanged
	}
	if c.Host != o.Host || c.Op.ID == "" || c.Op.Exec != m.Box || c.Op.Intent["target"] != m.Box ||
		c.Op.Intent["source"] == m.Box || !park.SameRegion(m.HomeRegion, c.Op.Intent["to"], m.Region) || len(c.Op.Participants) != 2 || c.Op.Participants[1] != m.Box {
		return v, hostbox.ErrChanged
	}
	for _, lost := range c.Lost {
		if lost == m.Box {
			return v, hostbox.ErrChanged
		}
	}
	v.op, v.holder = c.Op.ID, c.Holder
	rec, err := m.Records(ctx, v.holder.Account, v.holder.App, o.Host)
	if err != nil {
		return v, err
	}
	if rec == nil || rec.Host != o.Host || rec.Move == nil {
		return v, hostbox.ErrChanged
	}
	mv := rec.Move
	if park.PlacementRegion(mv.From) == "" || mv.ID != v.op || mv.Incarnation != v.holder.Incarnation || !park.PlacementMatches(m.HomeRegion, mv.To, m.Region) || mv.Inputs == nil || mv.Inputs.Release == "" {
		return v, hostbox.ErrChanged
	}
	if v.committed {
		if mv.State != provision.MoveCommitted {
			return v, hostbox.ErrChanged
		}
	} else if mv.State != provision.MoveLanding || mv.LandingPhase != provision.LandProbing || mv.Reason != "" {
		return v, hostbox.ErrChanged
	}
	v.rec = rec
	if v.seal, err = journalledSeal(c); err != nil {
		return v, err
	}
	j, err := moves.ReadJournal(m.DataDir, o.Host, v.op)
	if err != nil {
		return v, err
	}
	raw, ok := j.Steps[landStepInstall]
	if !ok {
		return v, fmt.Errorf("%w: the landing journalled no install manifest", hostbox.ErrChanged)
	}
	if err := json.Unmarshal([]byte(raw), &v.manifest); err != nil {
		return v, fmt.Errorf("the journalled install manifest: %w", err)
	}
	if v.manifest.Op != v.op || v.manifest.Host != o.Host || v.manifest.N != v.seal.N || v.manifest.N == 0 || v.manifest.SnapshotKey == "" || v.manifest.SnapshotVersion == "" {
		return v, fmt.Errorf("%w: the install manifest is not this move's build at the seal", hostbox.ErrChanged)
	}
	return v, nil
}

// heldRoute is the registry row the probe runs: this host, held by the move.
func heldRoute(rt registry.Route, host, op string) error {
	if rt.Host != host || !registry.IsHold(rt.Status) || rt.HoldOp != op || !filepath.IsAbs(rt.DBPath) {
		return hostbox.ErrChanged
	}
	return nil
}

// firstAdmission reports whether the holder's writer is still the move's
// source: the flip leaves the source's session in place, and every
// admission on this box makes it the writer. A first admission is the
// prepared one; any later one is a restarted attempt into a fresh
// generation. It reads no admission history, so trimming cannot change it.
func firstAdmission(h *hostclaim.Holder, box string) bool {
	return h.WriterBox != box
}

// admissible is the landing's own word that an admission may happen now
// (fix round 1, finding 1): before the landing's deadline (the earlier of
// cut_deadline and landing_deadline, H8), at a record attempt within what
// T11 admitted (H10), over the install the record vouches for.
func (m *movesProbeOwners) admissible(ctx context.Context, v probeLanding) error {
	mv := v.rec.Move
	d, _ := landingDeadlineOf(mv)
	if d == nil || !m.Now().Before(*d) {
		return fmt.Errorf("%w: the landing is past its deadline", hostbox.ErrChanged)
	}
	admitted, err := m.Admitted(ctx, v.rec)
	if err != nil {
		return err
	}
	if mv.Landing == nil || mv.Landing.Attempts < 1 || mv.Landing.Attempts > moves.AttemptsFor(moves.ProbeLanding, admitted) {
		return fmt.Errorf("%w: the record's probe attempt is not one T11 admitted", hostbox.ErrChanged)
	}
	if mv.Landing.InstalledDBSHA256 != "sha256:"+v.manifest.DBSHA256 {
		return fmt.Errorf("%w: the record does not vouch for this install", hostbox.ErrChanged)
	}
	return nil
}

// probeOnlyVars are the variables the probe adds to the launch and the
// readiness pin never saw.
var probeOnlyVars = map[string]bool{"CARLOS_MOVE_PROBE": true}

// capturedEnvPin is a launch environment's move-portable pin as Select
// computed it at capture (Launch's report), bound to that environment by
// its fingerprint. seq names this capture, so the end of an earlier wake
// never drops a later one; until is the wake's deadline, after which the
// entry is gone whether or not the wake ever reached Materialize; held is
// set once its wake's start loop holds it (holdEnvPin).
type capturedEnvPin struct {
	env, pin string
	seq      uint64
	until    time.Time
	held     bool
}

// envPinHold bounds a captured pin whose wake names no deadline (the
// committed wake's policy).
const envPinHold = 15 * time.Minute

// envFingerprint identifies a launch environment exactly: every entry, in
// order.
func envFingerprint(env []string) string {
	sum := sha256.Sum256([]byte(strings.Join(env, "\x00")))
	return hex.EncodeToString(sum[:])
}

// captureEnvPin records the pin Select computed for the inputs it returns
// (env is ClaimLaunchInputs.Env), replacing any earlier one for host. The
// entry lasts until its wake ends (holdEnvPin) and never past deadline (the
// wake's StartPolicy.Deadline, or envPinHold when it names none): a wake
// abandoned before Materialize leaves nothing behind longer than that.
// Every capture first drops the expired entries.
func (m *movesProbeOwners) captureEnvPin(host string, env []string, pin string, deadline time.Time) {
	now := m.now()
	if deadline.IsZero() {
		deadline = now.Add(envPinHold)
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.envPins == nil {
		m.envPins = map[string]capturedEnvPin{}
	}
	m.pruneEnvPins(now)
	m.envPinSeq++
	m.envPins[host] = capturedEnvPin{env: envFingerprint(env), pin: pin, seq: m.envPinSeq, until: deadline}
}

// now is the owners' clock, the wall clock when none is set.
func (m *movesProbeOwners) now() time.Time {
	if m.Now == nil {
		return time.Now()
	}
	return m.Now()
}

// pruneEnvPins drops the entries past their deadline. m.mu is held.
func (m *movesProbeOwners) pruneEnvPins(now time.Time) {
	for host, c := range m.envPins {
		if !now.Before(c.until) {
			delete(m.envPins, host)
		}
	}
}

// capturedPin is the pin Select captured for exactly env on host; false
// when Select captured none for it in this process, or it has expired
// (Materialize then refuses, fail closed).
func (m *movesProbeOwners) capturedPin(host string, env []string) (string, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.pruneEnvPins(m.now())
	c, ok := m.envPins[host]
	if !ok || c.env != envFingerprint(env) {
		return "", false
	}
	return c.pin, true
}

// holdEnvPin ties host's captured pin to its wake: ctx is the context the
// activator's start loop passes Materialize, which it cancels when the
// wake ends, completed, failed or abandoned. Every retry of the same
// launch runs under the same context and finds the entry; the end of the
// wake drops it, and never a later capture for the same host (seq). Held
// once per capture.
func (m *movesProbeOwners) holdEnvPin(ctx context.Context, host string) {
	m.mu.Lock()
	c, ok := m.envPins[host]
	if !ok || c.held {
		m.mu.Unlock()
		return
	}
	c.held = true
	m.envPins[host] = c
	m.mu.Unlock()
	seq := c.seq
	context.AfterFunc(ctx, func() { m.dropEnvPin(host, seq) })
}

// dropEnvPin removes host's captured pin if it is still capture seq.
func (m *movesProbeOwners) dropEnvPin(host string, seq uint64) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if c, ok := m.envPins[host]; ok && c.seq == seq {
		delete(m.envPins, host)
	}
}

// localEnvDigest is the digest of the whole launch environment, every
// variable but the probe-only ones (moves.LocalDigest): the recovery's
// check against inputs this same box recorded just before the wake (fix
// round 2, platform-1e's same-box rule).
func localEnvDigest(key []byte, env []string) string {
	return moves.LocalDigest(key, launchVars(env))
}

// launchVars is a launch's K=V list as a map, last value wins, the
// probe-only variables dropped.
func launchVars(env []string) map[string]string {
	vars := map[string]string{}
	for _, kv := range env {
		k, v, ok := strings.Cut(kv, "=")
		if !ok || probeOnlyVars[k] {
			continue
		}
		vars[k] = v
	}
	return vars
}

// manifestDigest is install.json's digest: the build writes the manifest
// as json.Marshal of the journalled value (moves.BuildGeneration).
func manifestDigest(m moves.InstallManifest) (string, error) {
	raw, err := json.Marshal(m)
	if err != nil {
		return "", err
	}
	sum := sha256.Sum256(raw)
	return hex.EncodeToString(sum[:]), nil
}

func manifestPin(m moves.InstallManifest) genselect.Pin {
	return genselect.Pin{SnapshotKey: m.SnapshotKey, SnapshotVersion: m.SnapshotVersion, SnapshotTXID: m.N, SnapshotCRC64: m.SnapshotCRC64}
}

// landingMarker reports whether the location's current marker already
// names this landing: the move id's verified prepared marker, or a fresh
// generation proved to descend from the move's admission (descendsFromMove).
// Anything else (an earlier stay's, though its cold admission on this box
// and region is still in the history; another incarnation's) is superseded
// by the prepared one, never taken for a restart (fix round 2, finding 8).
func (m *movesProbeOwners) landingMarker(marker *genselect.Marker, host string, v probeLanding, digest string) bool {
	h := v.holder
	if marker == nil || marker.Incarnation != h.Incarnation || marker.Host != host || marker.Box != m.Box {
		return false
	}
	if marker.Op == v.op {
		return marker.Pin == manifestPin(v.manifest) && marker.ManifestSHA256 == digest
	}
	return m.descendsFromMove(h, marker.Op, v.op)
}

// descendsFromMove is positive proof that generation g is a restart of this
// landing: walking back from g, each step a cold admission on this box and
// region into the current generation, its predecessor chain reaches the
// move's prepared generation (the admission into the move id itself, which
// names the op). A chain that leaves this box or region, loops, or is
// trimmed before the move id is no proof.
func (m *movesProbeOwners) descendsFromMove(h *hostclaim.Holder, g, op string) bool {
	seen := map[string]bool{}
	for g != "" && g != op && !seen[g] {
		seen[g] = true
		found := false
		for i := len(h.Admissions) - 1; i >= 0; i-- {
			a := h.Admissions[i]
			if a.Generation != g || a.Mode != "cold" {
				continue
			}
			if a.Box != m.Box || !park.SameRegion(m.HomeRegion, a.Region, m.Region) {
				return false
			}
			g, found = a.Predecessor, true
			break
		}
		if !found {
			return false
		}
	}
	if g != op {
		return false
	}
	for _, a := range h.Admissions {
		if a.Generation == op && a.Box == m.Box && park.SameRegion(m.HomeRegion, a.Region, m.Region) && a.Mode == "cold" {
			return true
		}
	}
	return false
}

// predecessorRegion is AdmitWriter's predecessor region: the last
// admission's, or the move's source region before any.
func predecessorRegion(h *hostclaim.Holder, from string) (string, error) {
	if !provision.ValidRegion(from) {
		return "", hostbox.ErrChanged
	}
	if n := len(h.Admissions); n > 0 {
		return h.Admissions[n-1].Region, nil
	}
	return park.PlacementRegion(from), nil
}

func (m *movesProbeOwners) request(host string, h *hostclaim.Holder) genselect.Request {
	return genselect.Request{Bucket: m.Bucket, Region: m.Region, Host: host, Holder: park.HolderRegionView(*h, m.HomeRegion, m.Region)}
}

func preparedPin(v probeLanding) genselect.PreparedPin {
	// The build's own pin, from the journalled manifest.
	return genselect.PreparedPin{Generation: v.op, SnapshotKey: v.manifest.SnapshotKey, SnapshotVersion: v.manifest.SnapshotVersion, SnapshotTXID: v.manifest.N, SnapshotCRC64: v.manifest.SnapshotCRC64}
}

func markerKey(h *hostclaim.Holder) string {
	return strings.TrimSuffix(h.Location, "/") + "/generation.json"
}

// currentMarker is the location's current generation.json version, or nil.
func (m *movesProbeOwners) currentMarker(ctx context.Context, h *hostclaim.Holder) (*claimMarkerVersion, error) {
	return currentMarker(ctx, m.Data, m.Bucket, h)
}

// currentMarker is the location's current generation.json version, or
// nil when the key has none; a delete marker, or two latest versions, is
// ErrChanged. The moves probe owners and region HA's relocation inputs
// share it (8d fix round 1, M3).
func currentMarker(ctx context.Context, data genselect.Reader, bucket string, h *hostclaim.Holder) (*claimMarkerVersion, error) {
	key := markerKey(h)
	versions, err := data.ListVersions(ctx, bucket, key)
	if err != nil {
		return nil, err
	}
	var current *claimMarkerVersion
	for _, v := range versions {
		if v.Key == key && v.IsLatest {
			if current != nil || v.DeleteMarker {
				return nil, hostbox.ErrChanged
			}
			current = &claimMarkerVersion{ID: v.VersionID, ETag: v.ETag}
		}
	}
	return current, nil
}

// baseline is the admission's baseline (H14): the build's pin for the
// prepared generation; for a restarted attempt, the prepared pin while the
// location's marker is not this landing's (the move id published nothing
// beyond the build; an earlier stay's pin is never a baseline), and the
// certified remote once it is.
func (m *movesProbeOwners) baseline(ctx context.Context, host string, v probeLanding, prepared bool) (*genselect.Plan, error) {
	if prepared {
		return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v))
	}
	digest, err := manifestDigest(v.manifest)
	if err != nil {
		return nil, err
	}
	marker, err := m.readMarker(ctx, v.holder)
	if err != nil {
		return nil, err
	}
	if !m.landingMarker(marker, host, v, digest) {
		return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v))
	}
	return genselect.Select(ctx, m.Data, m.request(host, v.holder))
}

// localMarkers reads .files-gen (nil when absent) and <dbPath>.gen ("" when
// absent).
func (m *movesProbeOwners) localMarkers(host, dbPath string) (*moves.FilesGen, string, error) {
	dir, _ := m.FilesDir(host)
	if !filepath.IsAbs(dir) {
		return nil, "", hostwrite.ErrUnavailable
	}
	var files *moves.FilesGen
	raw, err := os.ReadFile(filepath.Join(dir, archive.MarkerName))
	switch {
	case errors.Is(err, fs.ErrNotExist):
	case err != nil:
		return nil, "", err
	default:
		var fg moves.FilesGen
		dec := json.NewDecoder(bytes.NewReader(raw))
		dec.DisallowUnknownFields()
		if err := dec.Decode(&fg); err != nil {
			return nil, "", fmt.Errorf("%w: %v", moves.ErrStartFiles, err)
		}
		files = &fg
	}
	db, err := hostbox.ReadGenerationMarker(dbPath + ".gen")
	if err != nil {
		return nil, "", err
	}
	return files, db, nil
}

// launchIdentity is the probe's spawn identity (H4): the pinned config and
// release, bound to the record by MovesClaimPolicy.
func launchIdentity(in *provision.MoveInputs) hostwrite.Identity {
	return hostwrite.Identity{Config: HeldConfigIdentity(in.BundleETags, in.OverlayETag).Config, Release: in.Release}
}

// probeEnv is the launch environment with CARLOS_MOVE_PROBE=1 (H18), the
// one entry for that name.
func probeEnv(env []string) []string {
	out := make([]string, 0, len(env)+1)
	for _, kv := range env {
		if !strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
			out = append(out, kv)
		}
	}
	return append(out, "CARLOS_MOVE_PROBE=1")
}

// Select chooses the probe's launch inputs and admission. It is read-only.
func (m *movesProbeOwners) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
	if recoveryKind(o) {
		return m.recovery().Select(ctx, rt, o, p)
	}
	v, err := m.landing(ctx, o)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if v.committed {
		if p.Probe || p.Prepared != nil || p.Inputs != nil {
			return ClaimLaunchInputs{}, hostbox.ErrUnsupported
		}
		return m.selectCommitted(ctx, rt, o, v, p.Deadline)
	}
	if !p.Probe {
		return ClaimLaunchInputs{}, hostbox.ErrUnsupported
	}
	pins := v.rec.Move.Inputs
	if (p.Inputs != nil && !reflect.DeepEqual(*p.Inputs, *pins)) || (p.Prepared != nil && *p.Prepared != v.manifest) {
		return ClaimLaunchInputs{}, hostbox.ErrChanged
	}
	if err := heldRoute(rt, o.Host, v.op); err != nil {
		return ClaimLaunchInputs{}, err
	}
	if err := m.admissible(ctx, v); err != nil {
		return ClaimLaunchInputs{}, err
	}
	h := v.holder
	prepared := firstAdmission(h, m.Box)
	predecessor, err := predecessorRegion(h, v.rec.Move.From)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	admission := hostclaim.Admission{ObservedOpID: v.op, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: m.Box, Mode: "cold",
		KeepStamp: h.Stamp > h.Session, Region: m.Region, PredecessorRegion: predecessor}
	if prepared {
		files, db, err := m.localMarkers(o.Host, rt.DBPath)
		if err != nil {
			return ClaimLaunchInputs{}, err
		}
		if err := moves.PreparedCheck(v.op, *h, files, db); err != nil {
			return ClaimLaunchInputs{}, err
		}
		admission.NewGeneration = v.op
		if v.seal.FromDB {
			admission.Coverage = "sealed"
		}
	} else {
		admission.NewGeneration = m.NewID()
		if admission.NewGeneration == v.op || !claimFreshGeneration(h, admission.NewGeneration) {
			return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
		}
	}
	plan, err := m.baseline(ctx, o.Host, v, prepared)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	admission.RemoteMaxTXID = plan.MaxTXID
	ready, err := m.ReadyPath(ctx, h.Account, h.App)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	// The launch-input pin at spawn: the authoritative check, after
	// readiness's (Plan 2's configured), since the inputs can change between.
	l, report, err := m.Launch(ctx, rt)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	if l.Version != pins.Release {
		return ClaimLaunchInputs{}, probeInputsChanged()
	}
	// A v1 pin is refused whatever it hashes: it cannot tell an instance's
	// box variable from the box's (moves.EnvReportVersion). The launch's own
	// pin is its report's, computed at capture with its env.
	if !moves.CurrentEnvPin(pins.EnvReport) || report.Digest != pins.EnvReport {
		return ClaimLaunchInputs{}, probeInputsChanged()
	}
	in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(pins), Baseline: plan, Admission: admission,
		Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: probeEnv(l.Env),
		DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit, Prepared: prepared, ReadyPath: ready}
	m.captureEnvPin(rt.Host, in.Env, report.Digest, p.Deadline)
	return in, nil
}

// AdmissionEvidence re-derives Select's admission against the observation
// and a baseline selected again, immediately before the admission's CAS.
func (m *movesProbeOwners) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
	if recoveryKind(o) {
		return m.recovery().AdmissionEvidence(ctx, o, in, baseline)
	}
	v, err := m.landing(ctx, o)
	if err != nil {
		return err
	}
	if v.committed {
		return m.committedEvidence(ctx, o, v, in, baseline)
	}
	if err := m.admissible(ctx, v); err != nil {
		return err
	}
	h := v.holder
	predecessor, err := predecessorRegion(h, v.rec.Move.From)
	if err != nil {
		return err
	}
	if in.ObservedOpID != v.op || in.ObservedPhase != o.Claim.Op.Phase || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != m.Box || !park.SameRegion(m.HomeRegion, in.Region, m.Region) ||
		in.Mode != "cold" || in.PredecessorCertified || in.LossID != "" || len(in.AdditionalWritesLost) != 0 || in.KeepStamp != (h.Stamp > h.Session) ||
		!park.SameRegion(v.homeRegion, in.PredecessorRegion, predecessor) || baseline == nil || in.RemoteMaxTXID != baseline.MaxTXID {
		return hostbox.ErrChanged
	}
	prepared := firstAdmission(h, m.Box)
	switch {
	case prepared:
		coverage := ""
		if v.seal.FromDB {
			coverage = "sealed"
		}
		pin := manifestPin(v.manifest)
		if in.NewGeneration != v.op || in.Coverage != coverage || baseline.Generation != v.op || baseline.Pin == nil || *baseline.Pin != pin ||
			baseline.MaxTXID != v.manifest.N || v.manifest.N != v.seal.N {
			return hostbox.ErrChanged
		}
		rt, err := m.route(o.Host, v.op)
		if err != nil {
			return err
		}
		files, db, err := m.localMarkers(o.Host, rt.DBPath)
		if err != nil {
			return err
		}
		if err := moves.PreparedCheck(v.op, *h, files, db); err != nil {
			return err
		}
	default:
		if in.NewGeneration == v.op || !claimFreshGeneration(h, in.NewGeneration) || in.Coverage != "" {
			return hostbox.ErrChanged
		}
	}
	fresh, err := m.baseline(ctx, o.Host, v, prepared)
	if err != nil {
		return err
	}
	if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) || !reflect.DeepEqual(fresh.Pin, baseline.Pin) {
		return hostbox.ErrChanged
	}
	return nil
}

func (m *movesProbeOwners) route(host, op string) (registry.Route, error) {
	rt, err := m.Route(host)
	if err != nil {
		return registry.Route{}, err
	}
	if rt == nil {
		return registry.Route{}, hostbox.ErrChanged
	}
	if err := heldRoute(*rt, host, op); err != nil {
		return registry.Route{}, err
	}
	return *rt, nil
}

// admittedProbe checks that o is the admitted probe session in's admission
// made: this box writes the admission's new generation.
func admittedProbe(o hostclaim.Observed, in ClaimLaunchInputs, box string) error {
	h := o.Claim.Holder
	if h.WriterBox != box || h.Generation != in.Admission.NewGeneration || h.Generation == "" {
		return hostbox.ErrChanged
	}
	return nil
}

// heldIdentityWork runs its work under the held write's own identity.
type heldIdentityWork struct {
	Work hostwrite.Work
	ID   hostwrite.Identity
}

func (w heldIdentityWork) With(ctx context.Context, _ hostwrite.Identity, fn func() error) error {
	if w.Work == nil {
		return hostwrite.ErrUnavailable
	}
	return w.Work.With(ctx, w.ID, fn)
}

// Materialize runs before every start of the admitted probe, in the ruled
// order: the held render (its etags compared with the pins before anything
// is selected or written) and the held release under their own selections,
// the launch environment's pin again over the exact env that will run, the
// verified prepared marker under the admitted writer's fence, then the
// start check. The spawn follows only if all of it passed.
func (m *movesProbeOwners) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	if recoveryKind(o) {
		return m.recovery().Materialize(ctx, o, in, work)
	}
	// The captured pin lives as long as this wake: every start attempt
	// materializes under the start loop's context, which ends with the
	// wake, completed, failed or abandoned.
	m.holdEnvPin(ctx, o.Host)
	v, err := m.landing(ctx, o)
	if err != nil {
		return err
	}
	if err := admittedProbe(o, in, m.Box); err != nil {
		return err
	}
	if work == nil {
		return hostwrite.ErrUnavailable
	}
	// Without the C8 provider every landing wake is refused by name before
	// anything is rendered or released (whole-branch fixes, item 1), as the
	// recovery owners refuse theirs.
	if m.Files == nil {
		return claimFilesRefusal()
	}
	if v.committed {
		return m.materializeCommitted(ctx, o, v, in, work)
	}
	pins := v.rec.Move.Inputs
	if in.Identity != launchIdentity(pins) {
		return hostbox.ErrChanged
	}
	fence := writerFence(work, in.Identity)
	// The admitted spawn selection still holds under the launch identity.
	if err := fence(ctx); err != nil {
		return err
	}
	rt, err := m.route(o.Host, v.op)
	if err != nil {
		return err
	}
	if rt.DBPath != in.DBPath {
		return hostbox.ErrChanged
	}
	h := v.holder
	captured := hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}
	hr, err := m.Render(ctx, rt, v.op, func(pre appconfig.HeldRender) (hostwrite.Work, error) {
		// The pin decides before any selection or write (fix round 1,
		// finding 4): a changed configuration is the landing's rollback
		// sentence, never the policy's refusal of the write.
		if !sameBundle(pre.BundleETags, pins.BundleETags) || pre.OverlayETag != pins.OverlayETag {
			return nil, probeInputsChanged()
		}
		w, err := m.Works(ctx, o.Host, movesActionHeldConfig, captured)
		if err != nil {
			return nil, err
		}
		return heldIdentityWork{Work: w, ID: HeldConfigIdentity(pre.BundleETags, pre.OverlayETag)}, nil
	})
	if err != nil {
		return err
	}
	if !sameBundle(hr.BundleETags, pins.BundleETags) || hr.OverlayETag != pins.OverlayETag {
		return probeInputsChanged()
	}
	released, err := m.Works(ctx, o.Host, movesActionHeldRelease, captured)
	if err != nil {
		return err
	}
	if err := m.Release(ctx, rt, pins.Release, heldIdentityWork{Work: released, ID: hostwrite.Identity{Release: pins.Release}}); err != nil {
		return err
	}
	// The launch Materialize prepares is the one Select pinned: its env's
	// pin as captured (captureEnvPin), never recomputed from another read.
	if pin, ok := m.capturedPin(o.Host, in.Env); !ok || !moves.CurrentEnvPin(pins.EnvReport) || pin != pins.EnvReport {
		return probeInputsChanged()
	}
	m.recordFence(o.Host, h.Session, work)
	receipt, err := m.preparedMarker(ctx, o.Host, v, fence)
	if err != nil {
		return err
	}
	if in.Prepared {
		m.record(o.Host, h.Session, receipt)
	}
	if err := m.repairFilesGen(ctx, o, v, work, in.Identity); err != nil {
		return err
	}
	// The admitted generation's files manifest through C8 before the start
	// check (whole-branch fixes, item 1): the probe's takes P's prepared
	// manifest at the journalled version, stages the env file on disk and
	// drops the adoption record; a refusal stops the start.
	if err := m.Files.Establish(ctx, o, in, work); err != nil {
		return err
	}
	return m.startCheck(ctx, o, in)
}

// writerFence is the admitted writer's live fence over its spawn selection:
// the attempt's lease, the session, stamp and generation, and the policy,
// checked afresh on every call (claimLeaseWork over hostbox's selection).
func writerFence(work hostwrite.Work, id hostwrite.Identity) func(context.Context) error {
	return func(ctx context.Context) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		return work.With(ctx, id, func() error { return nil })
	}
}

// preparedMarker establishes the verified prepared marker under fence: the
// install.json the build wrote is the journalled manifest, and the location's
// generation.json names this landing (left as it is) or is superseded by the
// move id's marker with the build's pin and that manifest's digest, by
// If-Match on the current version or If-None-Match when there is none. The
// store runs fence before every request attempt. It returns the receipt for
// the move id's marker, or the zero receipt when a later landing generation's
// marker is current.
func (m *movesProbeOwners) preparedMarker(ctx context.Context, host string, v probeLanding, fence func(context.Context) error) (ClaimPublicationReceipt, error) {
	h := v.holder
	digest, err := manifestDigest(v.manifest)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
	defer cancel()
	if err := m.verifyInstall(bounded, h, v, digest); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	pin := manifestPin(v.manifest)
	root := genselect.File{Key: v.manifest.SnapshotKey, VersionID: v.manifest.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: v.manifest.N}
	current, err := m.currentMarker(bounded, h)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if current != nil {
		marker, err := m.readMarker(bounded, h)
		if err != nil {
			return ClaimPublicationReceipt{}, err
		}
		if m.landingMarker(marker, host, v, digest) {
			if marker.Op != v.op {
				return ClaimPublicationReceipt{}, nil
			}
			return ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}, nil
		}
	}
	body, err := json.Marshal(genselect.Marker{Op: v.op, Incarnation: h.Incarnation, Host: host, Box: m.Box, InstalledAt: m.Now().UTC(), ManifestSHA256: digest, Pin: pin})
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	expected := ""
	if current != nil {
		expected = current.ETag
	}
	if err := fence(bounded); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	version, err := m.Markers.PutMarker(bounded, m.Bucket, markerKey(h), body, expected, fence)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if version == "" {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	return ClaimPublicationReceipt{MarkerVersion: version, Root: root}, nil
}

// verifyInstall requires the generation's current install.json to be the
// journalled manifest, byte for byte (its digest is the marker's).
func (m *movesProbeOwners) verifyInstall(ctx context.Context, h *hostclaim.Holder, v probeLanding, digest string) error {
	key := moves.InstallManifestKey(moves.GenerationPrefix(h.Location, v.op))
	versions, err := m.Data.ListVersions(ctx, m.Bucket, key)
	if err != nil {
		return err
	}
	id := ""
	for _, ver := range versions {
		if ver.Key == key && ver.IsLatest && !ver.DeleteMarker {
			id = ver.VersionID
		}
	}
	if id == "" {
		return fmt.Errorf("%w: the generation has no install.json", hostbox.ErrChanged)
	}
	raw, err := m.Data.GetVersion(ctx, m.Bucket, key, id)
	if err != nil {
		return err
	}
	sum := sha256.Sum256(raw)
	if hex.EncodeToString(sum[:]) != digest {
		return fmt.Errorf("%w: install.json is not the journalled manifest", hostbox.ErrChanged)
	}
	return nil
}

// startCheck is §5.1's check against the claim after the admission's CAS,
// read fresh: generation.json from this region's location, .files-gen and
// <dbPath>.gen from this box.
func (m *movesProbeOwners) startCheck(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	fresh, err := hostclaim.Read(ctx, m.Claims, o.Host)
	if err != nil {
		return err
	}
	if !claimSameWriter(o, fresh, hostclaim.SessionContinues) {
		return hostbox.ErrChanged
	}
	marker, err := m.readMarker(ctx, fresh.Claim.Holder)
	if err != nil {
		return err
	}
	files, db, err := m.localMarkers(o.Host, in.DBPath)
	if err != nil {
		return err
	}
	if err := moves.StartCheck(fresh.Claim, marker, files, db); err != nil {
		// A refused start check pages an operator once per op (§13,
		// movespaging.go), and the agent's log says what became of the page.
		op := fresh.Claim.Op.ID
		pagingLogf("moves: %s: %s refused to start %s for %s (%s): %v", o.Host, m.Box, o.Host, op, m.pageStartRefused(ctx, o.Host, op), err)
		return err
	}
	return nil
}

// readMarker is the location's current generation.json, nil when absent.
func (m *movesProbeOwners) readMarker(ctx context.Context, h *hostclaim.Holder) (*genselect.Marker, error) {
	current, err := m.currentMarker(ctx, h)
	if err != nil || current == nil {
		return nil, err
	}
	raw, err := m.Data.GetVersion(ctx, m.Bucket, markerKey(h), current.ID)
	if err != nil {
		return nil, err
	}
	var marker genselect.Marker
	if err := json.Unmarshal(raw, &marker); err != nil {
		return nil, fmt.Errorf("%w: %v", moves.ErrStartMarker, err)
	}
	return &marker, nil
}

// EmptyAuthority: a move never admits the empty bootstrap, a recovery
// included.
func (m *movesProbeOwners) EmptyAuthority(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	if recoveryKind(o) {
		return m.recovery().EmptyAuthority(ctx, o, in)
	}
	return hostbox.ErrUnsupported
}

// Publish is Group D's call for a prepared admission, before replication.
// Materialize has already published the verified prepared marker; this
// finds it and records the receipt. It writes only under the fence
// Materialize recorded for the admitted session, and refuses without one.
func (m *movesProbeOwners) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	if recoveryKind(o) {
		return m.recovery().Publish(ctx, o, in)
	}
	v, err := m.landing(ctx, o)
	if err != nil {
		return err
	}
	if err := admittedProbe(o, in, m.Box); err != nil {
		return err
	}
	h := v.holder
	if !in.Prepared || in.Admission.NewGeneration != v.op {
		return hostbox.ErrChanged
	}
	work := m.fence(o.Host, h.Session)
	if work == nil {
		return fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
	}
	receipt, err := m.preparedMarker(ctx, o.Host, v, writerFence(work, in.Identity))
	if err != nil {
		return err
	}
	if receipt.MarkerVersion == "" {
		return hostbox.ErrChanged
	}
	m.record(o.Host, h.Session, receipt)
	return nil
}

// PublishFenced writes a restarted attempt's fresh generation marker after
// the session's own first root, under the publisher's fence, as the claim's
// serving publisher does (claimServingOwners.PublishFenced).
func (m *movesProbeOwners) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
	if recoveryKind(o) {
		return m.recovery().PublishFenced(ctx, o, in, pub)
	}
	v, err := m.landing(ctx, o)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if err := admittedProbe(o, in, m.Box); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if pub.Sync == nil || pub.Check == nil {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	h := v.holder
	if in.Prepared || h.Generation == v.op {
		return ClaimPublicationReceipt{}, hostbox.ErrChanged
	}
	// The admitted writer's spawn selection, recorded by Materialize: the
	// fence the local files marker is refreshed under (fix round 1).
	work := m.fence(o.Host, h.Session)
	if work == nil {
		return ClaimPublicationReceipt{}, fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
	}
	// The files on disk must be the landing's before anything is published,
	// and their content is recorded as this generation's install record
	// (fix round 3): the digest the refresh, and any later repair, compare.
	var digest string
	if err := work.With(ctx, in.Identity, func() error {
		if err := m.verifyFiles(o.Host, v); err != nil {
			return err
		}
		var err error
		digest, err = m.filesDigest(o.Host)
		return err
	}); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if err := m.recordFilesDigest(o.Host, v.op, h.Generation, digest); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	return m.publishFresh(ctx, o.Host, h, pub, func() error { return m.refreshFilesGen(ctx, o.Host, v, work, in.Identity, digest) })
}

// publishFresh is the fenced publication of the holder's fresh generation,
// shared by the landing's restarted attempt and the recovery owners: the
// session's replicator synced, the generation's own first root selected,
// then generation.json naming the generation superseding the current marker
// by its exact ETag (created when there is none), pub.Check immediately
// before the write and before every attempt. A replay that finds this
// generation's marker from this box records it. refresh runs after the
// marker, before the receipt is recorded (the local files marker).
func (m *movesProbeOwners) publishFresh(ctx context.Context, host string, h *hostclaim.Holder, pub ClaimPublication, refresh func() error) (ClaimPublicationReceipt, error) {
	if err := pub.Sync(ctx); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
	defer cancel()
	plan, err := genselect.Select(bounded, m.Data, m.request(host, h))
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	// Only this session's bound replica client writes under the fresh
	// generation, so its root is the session's own acknowledged upload.
	if plan.Generation != h.Generation || plan.Root == nil || plan.Root.MinTXID != 1 || plan.Root.VersionID == "" {
		return ClaimPublicationReceipt{}, hostbox.ErrChanged
	}
	root := *plan.Root
	current, err := m.currentMarker(bounded, h)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if current != nil && plan.Marker != nil && plan.MarkerVersion == current.ID && plan.Marker.Op == h.Generation {
		if plan.Marker.Box != m.Box || plan.Marker.Incarnation != h.Incarnation {
			return ClaimPublicationReceipt{}, hostbox.ErrChanged
		}
		receipt := ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}
		if err := refresh(); err != nil {
			return ClaimPublicationReceipt{}, err
		}
		m.record(host, h.Session, receipt)
		return receipt, nil
	}
	body, err := json.Marshal(genselect.Marker{Op: h.Generation, Incarnation: h.Incarnation, Host: host, Box: m.Box, InstalledAt: m.Now().UTC()})
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	expected := ""
	if current != nil {
		expected = current.ETag
	}
	if err := pub.Check(bounded); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	version, err := m.Markers.PutMarker(bounded, m.Bucket, markerKey(h), body, expected, pub.Check)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if version == "" {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	receipt := ClaimPublicationReceipt{MarkerVersion: version, Root: root}
	if err := refresh(); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	m.record(host, h.Session, receipt)
	return receipt, nil
}

// recordFence keeps the admitted spawn selection Materialize ran under, by
// host and session: the writer's fence for that session's marker writes and
// its files marker refresh.
func (m *movesProbeOwners) recordFence(host string, session uint64, work hostwrite.Work) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.fences == nil {
		m.fences = map[string]hostwrite.Work{}
	}
	m.fences[claimPublishedKey(host, session)] = work
}

// fence is the admitted spawn selection Materialize recorded for the
// session, or nil.
func (m *movesProbeOwners) fence(host string, session uint64) hostwrite.Work {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.fences[claimPublishedKey(host, session)]
}

// recovery is the recovery owners over this owner set's sources.
func (m *movesProbeOwners) recovery() movesRecoveryOwners { return movesRecoveryOwners{m: m} }

// verifyFiles requires the files directory and <host>.env on disk to be the
// ones the landing placed: the install receipt's identities
// (landStepFilesPlaced). Anything else fails the publication.
func (m *movesProbeOwners) verifyFiles(host string, v probeLanding) error {
	j, err := moves.ReadJournal(m.DataDir, host, v.op)
	if err != nil {
		return err
	}
	placed, err := journalledStaged(j, landStepFilesPlaced)
	if err != nil {
		return fmt.Errorf("%w: %v", hostbox.ErrChanged, err)
	}
	dir, env := m.FilesDir(host)
	if ok, err := sameIdentity(dir, placed.Files); err != nil || !ok {
		return fmt.Errorf("%w: %s is not the files directory the landing placed (%v)", hostbox.ErrChanged, dir, err)
	}
	if placed.Env != "" {
		if ok, err := sameIdentity(env, placed.Env); err != nil || !ok {
			return fmt.Errorf("%w: %s is not the environment file the landing placed (%v)", hostbox.ErrChanged, env, err)
		}
	}
	return nil
}

// refreshFilesGen points .files-gen at the published generation (fix round
// 1, finding 1), inside the admitted writer's fence, after the files on disk
// are proved the landing's again: the files are the same bytes the move
// installed, and the generation that now owns them is the one just
// published. Task 4's start check is unchanged: the next admission's cold
// predecessor is this generation. <dbPath>.gen needs no refresh: every cold
// restore rewrites it to the admitted generation (H11).
func (m *movesProbeOwners) refreshFilesGen(ctx context.Context, host string, v probeLanding, work hostwrite.Work, id hostwrite.Identity, digest string) error {
	h := v.holder
	marker, err := json.Marshal(moves.FilesGen{Op: h.Generation, Incarnation: h.Incarnation})
	if err != nil {
		return err
	}
	dir, _ := m.FilesDir(host)
	return work.With(ctx, id, func() error {
		if err := m.verifyFilesContent(host, v, digest); err != nil {
			return err
		}
		if err := syncPath(dir); err != nil {
			return err
		}
		return writeFileAtomic(filepath.Join(dir, archive.MarkerName), marker, 0o644)
	})
}

// filesPlainStep is the op journal step a published generation's files
// digest is recorded under: its install record (fix round 3).
const filesPlainStep = "files_plain_"

// filesDigest is the files tree's plain digest, the landing's own measure
// (archive.PlainDigest over the files directory and <host>.env; .files-gen
// is not part of it). The tree is capped at 256 MiB.
func (m *movesProbeOwners) filesDigest(host string) (string, error) {
	dir, env := m.FilesDir(host)
	plain, _, _, err := archivePlainDigest(dir, env)
	if err != nil {
		return "", err
	}
	if plain == "" {
		return "", fmt.Errorf("%w: the files tree has no digest", hostbox.ErrChanged)
	}
	return plain, nil
}

// recordFilesDigest journals generation g's files digest, the install record
// a later repair compares. A different digest already recorded for g is a
// conflict.
func (m *movesProbeOwners) recordFilesDigest(host, op, g, digest string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	// Under the journal's one lock (moves.UpdateJournal): the landing's pass
	// and the paging rules write this journal too.
	err := moves.UpdateJournal(m.DataDir, host, op, func(j *moves.Journal) error {
		if have, ok := j.Steps[filesPlainStep+g]; ok {
			if have != digest {
				return fmt.Errorf("%w: generation %s recorded other files", hostbox.ErrChanged, g)
			}
			return errJournalUnchanged
		}
		j.Mark(filesPlainStep+g, digest)
		return nil
	})
	if errors.Is(err, errJournalUnchanged) {
		return nil
	}
	return err
}

// errJournalUnchanged ends a moves.UpdateJournal that has nothing to write;
// its caller treats it as success.
var errJournalUnchanged = errors.New("moves journal: nothing to write")

// recordedFilesDigest is generation g's recorded files digest: the move id's
// is the source's stopped baseline the landing installed (Move.Source.Files);
// a later generation's is the one its publication journalled. false when
// none is recorded.
func (m *movesProbeOwners) recordedFilesDigest(host string, v probeLanding, g string) (string, bool) {
	if g == v.op {
		src := v.rec.Move.Source
		if src == nil || src.Files == nil || src.Files.Plain == "" {
			return "", false
		}
		return src.Files.Plain, true
	}
	j, err := moves.ReadJournal(m.DataDir, host, v.op)
	if err != nil {
		return "", false
	}
	d, ok := j.Steps[filesPlainStep+g]
	return d, ok && d != ""
}

// verifyFilesContent requires both the landing's placement (identity) and
// the recorded content (the plain digest): never identity alone.
func (m *movesProbeOwners) verifyFilesContent(host string, v probeLanding, want string) error {
	if want == "" {
		return fmt.Errorf("%w: no recorded files digest", hostbox.ErrChanged)
	}
	if err := m.verifyFiles(host, v); err != nil {
		return err
	}
	got, err := m.filesDigest(host)
	if err != nil {
		return err
	}
	if got != want {
		return fmt.Errorf("%w: the files on disk are not the recorded ones", hostbox.ErrChanged)
	}
	return nil
}

func (m *movesProbeOwners) record(host string, session uint64, receipt ClaimPublicationReceipt) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.published == nil {
		m.published = map[string]ClaimPublicationReceipt{}
	}
	m.published[claimPublishedKey(host, session)] = receipt
}

// PublicationEvidence requires this process's receipt for the admitted
// session: the current marker is exactly the version it wrote, naming the
// holder's generation, incarnation, host and this box (and, for the move
// id, the build's pin), and the root it named still exists, current.
func (m *movesProbeOwners) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error {
	if recoveryKind(o) {
		return m.recovery().PublicationEvidence(ctx, o)
	}
	v, err := m.landing(ctx, o)
	if err != nil {
		return err
	}
	return m.publicationEvidence(ctx, o.Host, v.holder, v.op, v.manifest)
}

// publicationEvidence is PublicationEvidence's check of the admitted
// session's receipt, shared with the recovery owners. prepared names the
// prepared generation whose marker also carries manifest's pin and digest
// ("" for none).
func (m *movesProbeOwners) publicationEvidence(ctx context.Context, host string, h *hostclaim.Holder, prepared string, manifest moves.InstallManifest) error {
	if h.WriterBox != m.Box {
		return hostbox.ErrChanged
	}
	m.mu.Lock()
	receipt, ok := m.published[claimPublishedKey(host, h.Session)]
	m.mu.Unlock()
	prefix, err := hostbox.GenerationPrefix(*h)
	if !ok || err != nil || receipt.MarkerVersion == "" || receipt.Root.VersionID == "" || !strings.HasPrefix(receipt.Root.Key, prefix) {
		return hostbox.ErrChanged
	}
	plan, err := genselect.Select(ctx, m.Data, m.request(host, h))
	if err != nil {
		return err
	}
	if plan.Generation != h.Generation || plan.Marker == nil || plan.MarkerVersion != receipt.MarkerVersion || plan.Marker.Op != h.Generation || plan.Marker.Box != m.Box || plan.Marker.Incarnation != h.Incarnation || plan.Marker.Host != host {
		return hostbox.ErrChanged
	}
	if prepared != "" && h.Generation == prepared {
		digest, err := manifestDigest(manifest)
		if err != nil {
			return err
		}
		if plan.Marker.Pin != manifestPin(manifest) || plan.Marker.ManifestSHA256 != digest {
			return hostbox.ErrChanged
		}
	}
	versions, err := m.Data.ListVersions(ctx, m.Bucket, receipt.Root.Key)
	if err != nil {
		return err
	}
	exact, current := false, false
	for _, ver := range versions {
		if ver.Key != receipt.Root.Key || ver.DeleteMarker {
			continue
		}
		exact = exact || ver.VersionID == receipt.Root.VersionID
		current = current || ver.IsLatest
	}
	if !exact || !current {
		return hostbox.ErrChanged
	}
	return nil
}

// The committed wake (Plan 3 Task 7; Task 5's concern). After the commit the
// claim stays landing/committed until the source's finalization closes the
// move, and Group D admits a landing wake there (K6's committed row). A wake
// happens only when no process serves the host: the probe's own entry, while
// it lives, is the serving process and is never restarted. The wake is a
// cold admission into a fresh generation from the certified remote (the
// landing's published marker, genselect.Select), continuing the admitted
// session's generation: no kept stamp, no coverage. It launches the row's
// ordinary launch (the claim's launch snapshot, its own identity, which
// MovesClaimPolicyWithInputs's committed check accepts), never the probe
// variable, and the held inputs are not rendered again: after the open the
// row is config sync's.

// committedRoute is the row a committed wake runs: this host's, held by the
// move until the open, then ready.
func committedRoute(rt registry.Route, host, op string) error {
	if rt.Host != host || !filepath.IsAbs(rt.DBPath) {
		return hostbox.ErrChanged
	}
	if (registry.IsHold(rt.Status) && rt.HoldOp == op) || (rt.Status == "ready" && rt.HoldOp == "") {
		return nil
	}
	return hostbox.ErrChanged
}

// withoutProbeVars is a launch environment with the probe-only variables
// removed.
func withoutProbeVars(env []string) []string {
	out := make([]string, 0, len(env))
	for _, kv := range env {
		if k, _, _ := strings.Cut(kv, "="); !probeOnlyVars[k] {
			out = append(out, kv)
		}
	}
	return out
}

func (m *movesProbeOwners) committedAdmission(o hostclaim.Observed, v probeLanding, gen string, plan *genselect.Plan) (hostclaim.Admission, error) {
	h := v.holder
	predecessor, err := predecessorRegion(h, v.rec.Move.From)
	if err != nil {
		return hostclaim.Admission{}, err
	}
	return hostclaim.Admission{ObservedOpID: v.op, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: m.Box, Mode: "cold",
		NewGeneration: gen, KeepStamp: h.Stamp > h.Session, Region: m.Region, PredecessorRegion: predecessor, RemoteMaxTXID: plan.MaxTXID}, nil
}

func (m *movesProbeOwners) selectCommitted(ctx context.Context, rt registry.Route, o hostclaim.Observed, v probeLanding, deadline time.Time) (ClaimLaunchInputs, error) {
	if err := committedRoute(rt, o.Host, v.op); err != nil {
		return ClaimLaunchInputs{}, err
	}
	h := v.holder
	if h.WriterBox != m.Box {
		return ClaimLaunchInputs{}, hostbox.ErrChanged
	}
	gen := m.NewID()
	if gen == v.op || !claimFreshGeneration(h, gen) {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	plan, err := genselect.Select(ctx, m.Data, m.request(o.Host, h))
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	l, report, err := m.Launch(ctx, rt)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	// The committed wake runs the move's pinned inputs (InputPolicy), judged
	// on the launch itself: its environment's report and its release.
	pins := v.rec.Move.Inputs
	id := committedLaunchIdentity(report.Digest, l.Version)
	if !moves.CurrentEnvPin(pins.EnvReport) || id != pinnedLaunchIdentity(pins) {
		return ClaimLaunchInputs{}, fmt.Errorf("%w: the launch is not the move's pinned inputs", hostbox.ErrChanged)
	}
	admission, err := m.committedAdmission(o, v, gen, plan)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: id, Baseline: plan, Admission: admission,
		Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: withoutProbeVars(l.Env),
		DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit}
	m.captureEnvPin(rt.Host, in.Env, report.Digest, deadline)
	return in, nil
}

func (m *movesProbeOwners) committedEvidence(ctx context.Context, o hostclaim.Observed, v probeLanding, in hostclaim.Admission, baseline *genselect.Plan) error {
	h := v.holder
	if baseline == nil || h.WriterBox != m.Box || in.NewGeneration == v.op || !claimFreshGeneration(h, in.NewGeneration) || in.PredecessorCertified || in.LossID != "" || len(in.AdditionalWritesLost) != 0 {
		return hostbox.ErrChanged
	}
	want, err := m.committedAdmission(o, v, in.NewGeneration, baseline)
	if err != nil {
		return err
	}
	if !reflect.DeepEqual(in, want) {
		return hostbox.ErrChanged
	}
	fresh, err := genselect.Select(ctx, m.Data, m.request(o.Host, h))
	if err != nil {
		return err
	}
	if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) || !reflect.DeepEqual(fresh.Pin, baseline.Pin) {
		return hostbox.ErrChanged
	}
	return nil
}

func (m *movesProbeOwners) materializeCommitted(ctx context.Context, o hostclaim.Observed, v probeLanding, in ClaimLaunchInputs, work hostwrite.Work) error {
	// The identity is the launch that will run (in.Env, in.Version), and it
	// is the pins'.
	pin, ok := m.capturedPin(o.Host, in.Env)
	if !ok || in.Identity != committedLaunchIdentity(pin, in.Version) || in.Identity != pinnedLaunchIdentity(v.rec.Move.Inputs) {
		return hostbox.ErrChanged
	}
	if err := writerFence(work, in.Identity)(ctx); err != nil {
		return err
	}
	m.recordFence(o.Host, v.holder.Session, work)
	rt, err := m.Route(o.Host)
	if err != nil {
		return err
	}
	if rt == nil || rt.DBPath != in.DBPath {
		return hostbox.ErrChanged
	}
	if err := committedRoute(*rt, o.Host, v.op); err != nil {
		return err
	}
	if err := m.repairFilesGen(ctx, o, v, work, in.Identity); err != nil {
		return err
	}
	// The fresh generation's files manifest through C8, copied from the
	// admitted predecessor's, before the start check (item 1).
	if err := m.Files.Establish(ctx, o, in, work); err != nil {
		return err
	}
	return m.startCheck(ctx, o, in)
}

// InputPolicy is the moves owner's input policy: the entry point claim Plan
// 10 rev 5's ClaimOwnerDispatch routes a move kind's input check to (the
// shape of ClaimOwnerBindings.InputPolicy; a move kind never falls back to
// serving's inputs). It answers the committed wake's spawn (Plan 3 Task 7):
// at landing/committed the launch identity, computed by the owner from the
// launch that will run (committedLaunchIdentity: its environment's report
// digest and its release), must be the move's pins' (Move.Inputs.EnvReport
// and Move.Inputs.Release), over a committed record of this move. The
// other move-kind inputs are MovesClaimPolicy's own rows; anything else is
// refused here.
func (m *movesProbeOwners) InputPolicy(ctx context.Context, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
	if m == nil || m.Records == nil || ctx == nil {
		return hostwrite.ErrUnavailable
	}
	if c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseLandingCommitted ||
		s.Phase != hostclaim.PhaseLandingCommitted || s.Action != movesActionSpawn || s.Op != c.Op.ID || s.Host != c.Host {
		return hostbox.ErrChanged
	}
	rec, err := m.Records(ctx, c.Holder.Account, c.Holder.App, c.Host)
	if err != nil {
		return err
	}
	if rec == nil || rec.Host != c.Host || rec.Move == nil || rec.Move.ID != c.Op.ID || rec.Move.Incarnation != c.Holder.Incarnation ||
		rec.Move.State != provision.MoveCommitted || rec.Move.Inputs == nil || rec.Move.Inputs.Release == "" {
		return hostbox.ErrChanged
	}
	if !moves.CurrentEnvPin(rec.Move.Inputs.EnvReport) || id != pinnedLaunchIdentity(rec.Move.Inputs) {
		return fmt.Errorf("%w: the committed launch is not the move's pinned inputs", hostbox.ErrChanged)
	}
	return nil
}

// committedEnvPrefix marks a committed launch's config identity: the report
// digest of its environment.
const committedEnvPrefix = "launch-env:"

// committedLaunchIdentity is the committed wake's identity, computed from
// the launch that will run (fix round 2, item 2): its environment's
// move-portable pin as captured with it (Launch's report, CARLOS_MOVE_PROBE
// excluded) and its release. The input policy compares it with the pins.
func committedLaunchIdentity(pin, version string) hostwrite.Identity {
	return hostwrite.Identity{Config: committedEnvPrefix + pin, Release: version}
}

// pinnedLaunchIdentity is the identity a launch matching the move's pins
// carries.
func pinnedLaunchIdentity(in *provision.MoveInputs) hostwrite.Identity {
	return hostwrite.Identity{Config: committedEnvPrefix + in.EnvReport, Release: in.Release}
}

// repairFilesGen brings .files-gen to the admitted generation's cold
// predecessor T before the start check, inside the admitted writer's fence,
// in two cases, each verified by content (the files tree's plain digest
// against a recorded one, with the landing's placement), never by identity
// alone:
//
//   - an interrupted refresh (Task 7 fix round 2): the published marker
//     names T itself, and .files-gen names an older generation of T's
//     ancestry; the files must match T's recorded digest.
//   - unpublished cold admissions (final review, ruling 1): the published
//     marker names an older generation P, and T reaches P through cold
//     admissions on this box and region, none of which published (a
//     publication would have superseded P's marker); .files-gen names P or
//     one of those admissions. The files must match P's recorded digest:
//     every admission since P restored from P's data and its files are P's.
//
// Anything else is left alone, for the unchanged start check to decide.
func (m *movesProbeOwners) repairFilesGen(ctx context.Context, o hostclaim.Observed, v probeLanding, work hostwrite.Work, id hostwrite.Identity) error {
	h := o.Claim.Holder
	marker, err := m.readMarker(ctx, h)
	if err != nil || marker == nil {
		return err
	}
	published := marker.Op
	if published == "" || marker.Incarnation != h.Incarnation || marker.Host != o.Host {
		return nil
	}
	t, ok := coldPredecessor(h, h.Generation)
	if !ok || t == "" {
		return nil
	}
	dir, _ := m.FilesDir(o.Host)
	files, _, err := m.localMarkers(o.Host, filepath.Join(dir, ".none"))
	if err != nil || files == nil || files.Op == t || files.Incarnation != h.Incarnation {
		return nil
	}
	var vouch string // the published generation whose recorded digest the files must match
	switch {
	case published == t:
		if !ancestorOf(h, t, files.Op) {
			return nil
		}
		vouch = t
	default:
		path, ok := m.unpublishedPath(h, t, published)
		if !ok || !containsString(path[1:], files.Op) {
			return nil
		}
		vouch = published
	}
	want, ok := m.recordedFilesDigest(o.Host, v, vouch)
	if !ok {
		return nil // nothing recorded: the start check decides
	}
	body, err := json.Marshal(moves.FilesGen{Op: t, Incarnation: h.Incarnation})
	if err != nil {
		return err
	}
	return work.With(ctx, id, func() error {
		if m.verifyFilesContent(o.Host, v, want) != nil {
			return nil // not the vouching generation's files: the start check decides
		}
		if err := syncPath(dir); err != nil {
			return err
		}
		log.Printf("moves: %s: .files-gen repaired from %s to %s (vouched by the published %s)", o.Host, files.Op, t, vouch)
		return writeFileAtomic(filepath.Join(dir, archive.MarkerName), body, 0o644)
	})
}

// unpublishedPath walks from back to the published generation to through
// cold admissions, each into the generation walked and on this box and
// region. It returns the generations from from to to, inclusive, or false
// when the chain leaves this box or region, breaks, loops, or never
// reaches to.
func (m *movesProbeOwners) unpublishedPath(h *hostclaim.Holder, from, to string) ([]string, bool) {
	path := []string{from}
	for g := from; g != to; {
		if len(path) > len(h.Admissions)+1 {
			return nil, false
		}
		var found *hostclaim.AdmissionRecord
		for i := len(h.Admissions) - 1; i >= 0; i-- {
			if a := h.Admissions[i]; a.Generation == g && a.Mode == "cold" {
				found = &h.Admissions[i]
				break
			}
		}
		if found == nil || found.Box != m.Box || !park.SameRegion(m.HomeRegion, found.Region, m.Region) || found.Predecessor == "" {
			return nil, false
		}
		g = found.Predecessor
		path = append(path, g)
	}
	return path, true
}

// coldPredecessor is the predecessor of the newest cold admission into g.
func coldPredecessor(h *hostclaim.Holder, g string) (string, bool) {
	for i := len(h.Admissions) - 1; i >= 0; i-- {
		if a := h.Admissions[i]; a.Generation == g && a.Mode == "cold" {
			return a.Predecessor, true
		}
	}
	return "", false
}

// ancestorOf reports whether old is reached by walking g's cold
// predecessors in the holder's admissions.
func ancestorOf(h *hostclaim.Holder, g, old string) bool {
	for steps := 0; steps <= len(h.Admissions); steps++ {
		pred, ok := coldPredecessor(h, g)
		if !ok || pred == "" {
			return false
		}
		if pred == old {
			return true
		}
		g = pred
	}
	return false
}