package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The landing on the target (moves spec §5.1, §7.5, §7.6), behind
// CARLOS_MOVES_PHASE1. After the flip the record's region is the target's
// and reconcile has minted the held row, so the target's pass visits the
// record (OwnsRoute) and phase1MoveRound dispatches it here (ruling G34).
// One pass drives one landing record as far as it can, as the cut does:
// each phase does its work, then advances landing_phase by one guarded CAS,
// then mirrors the step into the claim's op.phase (§7.5: the record is the
// decision, the claim follows it). Nothing is carried in memory between
// passes but the own-location lease (building through published): every
// phase resumes from the record, the claim, this box's own bucket and the
// op journal (moves.Journal, <DataDir>/.moves/<host>/<op>/). checking is
// the landing's own; parking to restored run in the generation installer
// the dirty return shares (movesinstall.go, Plan 4 Task 3), bound to this
// landing by installSource.
//
// Phase names (ruling G8, the cut's convention). A record at X means every
// phase before X is complete and X's own work may have begun; a replay of X
// redoes X's work from durable state, and each phase's comment says what it
// finds. The one name read the other way is `restored`: it is written by
// the CAS that ends `configured`, after the database is installed and its
// markers written, stamping landing.installed_db_sha256, so Plan 3's probe
// reads `restored` as "the install is complete". Every later phase is the
// probe's (Plan 3).
//
// The claim never leads the record. Every phase starts with syncClaimPhase:
// equal is the common case; a claim behind the record (a crash between the
// record CAS and the claim CAS) is walked forward along LandingOrder with
// one claim CAS per adjacent step; a claim ahead is left alone and logged, and the phase's own
// RequireMoveOp then refuses, so nothing is written over it.
//
// Nothing here ever serves from moves/: the lane is read by exact version
// into the generation the build writes and into the files directory the
// archive is extracted into (M14, M15). A failure that is permanent for this
// landing is a rollback (§7.8), which Plan 4 builds: until then the record
// carries the reason (provision.MoveMarkRollback) and this pass leaves a
// marked record alone. The landing's deadline passing before `restored`
// marks it too: the earlier of the cut's absolute cut_deadline and the
// flip's landing_deadline (spec §7.4; landingDeadlineOf).
//
// Local installation (final review, findings 1 and 2). Every destructive
// local mutation (the files and env install, .files-gen, the database
// rename and <dbPath>.gen) runs inside the host guard at the installation
// stamp captured from the record (landingRun.guarded, the pattern of
// hostbox ColdRestore.Install): the claim and the record are read fresh
// there and must still be this landing at this phase, and the destination
// is checked again there. Retirement (park.Local) takes the guard itself
// and runs outside those blocks. The landing never deletes at the live
// paths: the archive is extracted into the op's own staging and moved into
// place only when nothing it did not write is there.
//
// Arming (ruling G17; Plan 5 Task 9a). armPhase1 wires the hooks that exist
// on main (the guard, park.Local, the own-location lease, RenderHeld, the
// registry setters, the restore-verification record, the marker writer).
// ParkLeases, ParkView, ParkAuthorize and ParkLocalAuthorize are the
// claim's (MoveParks), and ParkRunHeld is park.Engine.RunHeld over the
// claim's leases (M5), all set by armMovesParks only once the wake-source
// gate opens (movesWakeSourceRegistered, Plan 5 Task 9b); nothing stands in
// for them. landingArmed therefore answers false in production and the pass
// logs a standing notice instead of landing; the tests arm everything with
// fakes (movesparkfakes_test.go).
// Under
// CARLOS_HOST_CLAIM the landing's held writes select their Work from the
// claim (Task 7, Q2): the timers as held-timers, the render as held-config
// and the release as held-release, which the configured step now publishes
// (T7 only stages under the claim; Paul, 2026-10-12). hostbox.Policy.Moves,
// MovesClaimPolicy (movespolicy.go), decides them. The landing still refuses
// at checking with a recorded reason (ruling G16, RollbackClaimWired) until
// the wake-source gate opens (movesWakeSourceRegistered), and lands after.

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

// landingRequestLimit is the residue parks' request limit: the longest a
// request against the location runs (hostbox.RequestLimit, the cut's too).
const landingRequestLimit = cutRequestLimit

// Op journal steps the landing writes (moves.Journal). Each sub-step of
// `local` is a step, so a replay skips what is durable and finishes the
// rest; `.files-gen` is written last and vouches for them all (§7.6).
const (
	landStepPrevious      = "previous"       // the earlier stay's stamp S' (decimal), noted before Activate
	landStepResidue       = "residue"        // the generations whose residue parks were queued (JSON list)
	landStepInstall       = "install"        // the completed install manifest (JSON)
	landStepPublished     = "published"      // the build is complete and recorded (R10-3: no generation.json)
	landStepRetired       = "retired"        // the earlier stay's artifacts are parked, or there were none
	landStepStaged        = "staged"         // the archive is extracted into <JournalDir>/extract/ (JSON stagedFiles)
	landStepFilesPlaced   = "files_placed"   // the install receipt: the staged tree is at the live paths (JSON stagedFiles)
	landStepExtracted     = "extracted"      // the archive is in the files directory and <host>.env
	landStepInventory     = "inventory"      // the placed files and env are registered under F
	landStepTimers        = "timers"         // the timers table is the sealed snapshot
	landStepColumns       = "columns"        // replication and next_due carried over
	landStepVerify        = "verify"         // the restore-verification entry is gone
	landStepFilesManifest = "files_manifest" // the prepared files manifest's binding (JSON preparedManifest: key, sha256, op, stamp; C8, Task 6)
	landStepFilesGen      = "files_gen"      // .files-gen written
	landStepPreInstall    = "pre_install"    // the database's siblings absent before its install (JSON preInstall)
	landStepInventoryDB   = "inventory_db"   // the installed database and its marker are registered under F (carry (b))
	landStepPreRender     = "pre_render"     // the held render's ownership before its write (JSON preRender, carry (c))
	landStepRendered      = "rendered"       // the held render's file, "sha256:<hex>", journalled even when its ack failed (carry (c))
)

// landingFilesManifestSkipped is what the files_manifest step recorded
// before C8 (the prepared-manifest build, Task 6) existed. Nothing writes
// it now; a journal that holds it counts as not done (filesManifestBuilt),
// so a replay builds the manifest (ruling Q3 item 2).
const landingFilesManifestSkipped = "skipped: the prepared files manifest needs C8 (files.BuildPrepared)"

// landingRun is one pass's state for one landing record.
type landingRun struct {
	w     *movesWorker
	s     release.Store
	rec   provision.Record
	g     provision.MoveGuard
	guard hostguard.Guard
}

// parkHooksArmed reports whether the claim's three park hooks are wired:
// the leases, the view and the authorizer (M17). Every moves arm requires
// all three; nothing stands in for a missing one.
func parkHooksArmed(cfg moveCfg) bool {
	return cfg.ParkLeases != nil && cfg.ParkView != nil && cfg.ParkAuthorize != nil
}

// landingArmed reports whether this box can run the landing: every hook
// the phases call is wired, the claim's three park hooks, RunHeld (M5: the
// residue parks run under the landing's own lease) and the claim's local
// authorizer (M9: the install's park.Local) among them. In production
// armMovesParks sets those only once the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b), so this answers false there.
func landingArmed(cfg moveCfg) bool {
	return cfg.Box != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" && parkHooksArmed(cfg) && cfg.ParkRunHeld != nil && cfg.ParkLocalAuthorize != nil &&
		cfg.OwnLease != nil && cfg.RenderHeld != nil && cfg.ReplaceTimers != nil && cfg.SetReplication != nil &&
		cfg.SetNextDue != nil && cfg.RemoveVerifyEntry != nil && cfg.WriteGen != nil && cfg.Route != nil &&
		cfg.FilesDir != nil && cfg.Versioned != nil && cfg.ReleaseReady != nil && cfg.SecretsPriv != nil
}

// landingPass is the MoveLanding arm of phase1MoveRound.
func (w *movesWorker) landingPass(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	// The rollback decision (movesrollback.go, J10, J11): a recorded mark
	// or the landing's deadline, at any phase, armed or not. A landing the
	// decision took is abandoned, and with it the location's lease a build
	// keeps across passes (building through published): it is released
	// here, before the pass ends, or the target's cleaning could never take
	// the same lease (the final review's finding 2).
	if done, err := w.rollBackLanding(ctx, s, rec); done || err != nil {
		if done {
			releaseOwnLease(w, rec.Host)
		}
		return err
	}
	if !landingArmed(cfg) {
		w.notice(rec.Host+"#landing", "moves: %s: the landing of %s waits: this box's landing is not armed (the claim's park hooks are set from MoveParks only once this box's moves wake source is registered, Plan 5 Task 9b)", rec.Host, rec.Move.ID)
		return nil
	}
	l := &landingRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}}
	l.guard = cfg.Guard
	if l.guard.Claims == nil {
		l.guard.Claims = hostguard.StoreReader{Store: s}
	}
	defer l.settleLease()
	for i := 0; i < landingStepsPerPass; i++ {
		advanced, err := l.step(ctx)
		if err != nil || !advanced {
			return err
		}
		fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
		if err != nil {
			return fmt.Errorf("re-read the record: %w", err)
		}
		if fresh.Move == nil || fresh.Move.ID != l.g.ID || fresh.Move.Incarnation != l.g.Incarnation || fresh.Move.State != provision.MoveLanding {
			return nil
		}
		l.rec = *fresh
	}
	return nil
}

func (l *landingRun) cfg() moveCfg { return l.w.cfg }

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

// step runs the record's current phase once. advanced is true when it
// wrote the record (a phase or the rollback marker).
func (l *landingRun) step(ctx context.Context) (bool, error) {
	m := l.rec.Move
	if m.State != provision.MoveLanding {
		return false, nil
	}
	if m.Reason != "" {
		// Marked: the next pass's rollback decision (rollBackLanding) rolls
		// it back; markRollback nudged the serial for it.
		return false, nil
	}
	// Every phase checks the landing's deadline at entry: the install's
	// phases (P2-deadline), and restored, probing and probed (H8: the
	// probe works to the same deadline).
	if l.pastDeadline() {
		return l.markRollback(ctx, provision.RollbackLandingDeadline)
	}
	switch m.LandingPhase {
	case provision.LandChecking:
		return l.checking(ctx)
	case provision.LandParking, provision.LandBuilding, provision.LandPublished, provision.LandLocal, provision.LandConfigured:
		// parking to restored are the shared installer's (movesinstall.go).
		advanced, _, err := installGeneration(ctx, l.s, l.w, l.rec, l.installSource())
		return advanced, err
	case provision.LandRestored:
		// The install is complete. The claim half of the last CAS first: a
		// crash after the record's restored, or a mirror that failed, leaves
		// the claim behind (fix round 1, item 1). Then the probe begins,
		// when this box has a probe wake (unarmed in production).
		if _, err := l.syncClaimPhase(ctx); err != nil {
			return false, err
		}
		if l.cfg().Probe == nil {
			l.w.notice(l.rec.Host+"#probe", "moves: %s: landing %s waits at restored: this box has no probe wake (claim Plan 11's launcher and Plan 5 Task 9's arming are needed)", l.rec.Host, l.g.ID)
			return false, nil
		}
		return l.advance(ctx, provision.LandRestored, provision.LandProbing, nil)
	case provision.LandProbing:
		return l.probing(ctx)
	case provision.LandProbed:
		// The probe passed: the commit (moveslandcommit.go).
		return l.probed(ctx)
	}
	return false, nil
}

// landingDeadlineOf is the deadline a landing before restored works to:
// provision.LandingDeadlineOf, the earlier of the cut's absolute deadline
// and landing_deadline, which the move's timings share (Plan 6 preflight
// L20). which names the one that applies, for the log.
func landingDeadlineOf(m *provision.MoveState) (d *time.Time, which string) {
	return provision.LandingDeadlineOf(m)
}

// pastDeadline reports whether the landing's deadline (landingDeadlineOf)
// has passed now. It is checked at every phase's entry (step), and in
// configured again before the install and before the transition that
// names it complete.
func (l *landingRun) pastDeadline() bool {
	d, which := landingDeadlineOf(l.rec.Move)
	if d == nil || !l.cfg().Now().After(*d) {
		return false
	}
	log.Printf("moves: %s: landing %s is past its %s (%s)", l.rec.Host, l.g.ID, which, d.UTC().Format(time.RFC3339))
	return true
}

// landingBefore reports whether phase comes before other in LandingOrder.
func landingBefore(phase, other string) bool {
	order := provision.LandingOrder()
	return landingIndex(order, phase) >= 0 && landingIndex(order, phase) < landingIndex(order, other)
}

func landingIndex(order []string, phase string) int {
	for i, p := range order {
		if p == phase {
			return i
		}
	}
	return -1
}

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

// advance writes the record's step and then mirrors it into the claim. A
// mirror that fails is logged: the next pass's syncClaimPhase writes it.
func (l *landingRun) advance(ctx context.Context, from, to string, stamp func(*provision.MoveState)) (bool, error) {
	ok, err := l.transition(ctx, "the landing "+from+" to "+to, provision.MoveLandingPhase(l.g, from, to, stamp))
	if !ok {
		return ok, err
	}
	log.Printf("moves: %s: landing %s is %s", l.rec.Host, l.g.ID, to)
	if err := l.fault("after-record-" + to); err != nil {
		return true, err
	}
	if err := l.mirrorPhase(ctx, from, to); err != nil {
		log.Printf("moves: %s: the claim half of landing %s waits for the next pass: %v", l.rec.Host, to, err)
	}
	return true, nil
}

// mirrorPhase is the claim CAS that follows a record step (§7.5).
func (l *landingRun) mirrorPhase(ctx context.Context, from, to string) error {
	obs, err := l.claim(ctx)
	if err != nil {
		return err
	}
	step := hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, from, to)
	if hostclaim.SessionPhase(hostclaim.KindLanding, from) && hostclaim.SessionPhase(hostclaim.KindLanding, to) {
		// probing to probed (and Task 7's committed): the admitted probe
		// session is the serving session, so the CAS must keep it (H3).
		step = hostclaim.KeepWriterSession(l.g.ID, step)
	}
	_, err = hostclaim.Update(ctx, l.s, obs, step)
	return err
}

// markRollback records why the landing stopped (§7.6; Plan 4's rollback
// transition replaces this one call site).
func (l *landingRun) markRollback(ctx context.Context, reason string) (bool, error) {
	ok, err := l.transition(ctx, "the rollback marker", provision.MoveMarkRollback(l.g, reason))
	if ok {
		log.Printf("moves: %s: landing %s stops for a rollback: %s", l.rec.Host, l.g.ID, reason)
		nudgeSerial(ctx, l.s)
	}
	return ok, err
}

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

// syncClaimPhase is every phase's first step (§7.5): the claim's op.phase
// follows the record's landing_phase. Behind, it is walked forward one step
// at a time, each in its own CAS; ahead, it is left alone and logged (the phase's own
// RequireMoveOp then refuses); equal is the common case. The observation
// returned is the one the phase works from, and names this box as the
// executor.
func (l *landingRun) syncClaimPhase(ctx context.Context) (hostclaim.Observed, error) {
	cfg := l.cfg()
	obs, err := l.claim(ctx)
	if err != nil {
		return obs, err
	}
	op := obs.Claim.Op
	if op.Kind != hostclaim.KindLanding {
		return obs, fmt.Errorf("move %s: the claim is %s/%q, not landing", l.g.ID, op.Kind, op.Phase)
	}
	if cfg.Box == "" || op.Exec != cfg.Box {
		return obs, fmt.Errorf("move %s: the claim's executor is %q, not this box", l.g.ID, op.Exec)
	}
	order := provision.LandingOrder()
	ci, ri := landingIndex(order, op.Phase), landingIndex(order, l.rec.Move.LandingPhase)
	switch {
	case ci < 0 || ri < 0:
		return obs, fmt.Errorf("move %s: landing phases %q (claim) and %q (record) are not both known", l.g.ID, op.Phase, l.rec.Move.LandingPhase)
	case ci == ri:
		return obs, nil
	case ci > ri:
		log.Printf("moves: %s: the claim's landing phase %q is ahead of the record's %q; left alone", l.rec.Host, op.Phase, l.rec.Move.LandingPhase)
		return obs, nil
	}
	for i := ci; i < ri; i++ {
		step := hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, order[i], order[i+1])
		if hostclaim.SessionPhase(hostclaim.KindLanding, order[i]) && hostclaim.SessionPhase(hostclaim.KindLanding, order[i+1]) {
			step = hostclaim.KeepWriterSession(l.g.ID, step)
		}
		obs, err = hostclaim.Update(ctx, l.s, obs, step)
		if err != nil {
			return obs, fmt.Errorf("catch the claim up to %s: %w", l.rec.Move.LandingPhase, err)
		}
	}
	log.Printf("moves: %s: the claim's landing phase caught up from %q to %q", l.rec.Host, op.Phase, l.rec.Move.LandingPhase)
	return obs, nil
}

// requirePhase is the phase's K5 revalidation on the synced observation.
func (l *landingRun) requirePhase(obs hostclaim.Observed, phase string) error {
	if err := hostclaim.RequireMoveOp(l.g.ID, hostclaim.KindLanding, phase)(obs.Claim); err != nil {
		return fmt.Errorf("move %s: %w", l.g.ID, err)
	}
	return nil
}

// landingCheck is the claim check the guard runs before a local write of
// phase: this landing's op, at that phase, with this box executing.
func (l *landingRun) landingCheck(phase string) func(*hostclaim.Claim) error {
	return func(c *hostclaim.Claim) error {
		if err := hostclaim.RequireMoveOp(l.g.ID, hostclaim.KindLanding, phase)(c); err != nil {
			return err
		}
		if c.Op.Exec != l.cfg().Box {
			return fmt.Errorf("%w: the executor is %q, not this box", hostclaim.ErrWrongState, c.Op.Exec)
		}
		return nil
	}
}

// route is the host's registry row on this box.
func (l *landingRun) route() (registry.Route, error) {
	rt, err := l.cfg().Route(l.rec.Host)
	if err != nil {
		return registry.Route{}, fmt.Errorf("read the registry row: %w", err)
	}
	if rt == nil {
		return registry.Route{}, fmt.Errorf("%s has no registry row on this box", l.rec.Host)
	}
	return *rt, nil
}

func (l *landingRun) dbPath() string { return provision.DefaultDBPath(l.cfg().DataDir, l.rec.Host) }

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

// ensureLocalDirs creates the two directories the host guard and park.Local
// require under the data directory (hostguard/ for the installation record,
// .parked/ for retired artifacts). On a box with claims on they exist
// already; a Phase 1 landing on a box without them creates them.
func (l *landingRun) ensureLocalDirs() error {
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(l.cfg().DataDir, d), 0o770); err != nil {
			return err
		}
	}
	return nil
}

// checking (§7.6): the refusals this box can know (G16, G11), the flip's
// claim half if the source crashed before it (G12), then the claim's word:
// landing/checking for this move with this box executing and a root-layout
// holder; the source's complete stop evidence; this box in touched.boxes
// (K12, before any local write); move.json's incarnation; the earlier
// stay's stamp S' noted in the journal before the installation record is
// activated at F (holder.stamp, the flip's); and a park-local queued only
// when the inventory has paths tagged S' (ruling G5). No lease is taken
// here: the residue parks take the location's lease themselves.
//
// A replay finds checking and runs it all again: every write is
// idempotent (the flip, Touch and the park queue under the same op, the
// journal step written once, Activate at the same stamp), and Capture after
// Activate reads F, which is never noted as a previous stay.
func (l *landingRun) checking(ctx context.Context) (bool, error) {
	cfg := l.cfg()
	host, id := l.rec.Host, l.g.ID
	// Ruling G16, as Task 7 settles it (Q2.3): with host claims on, the
	// landing's held writes run under the claim's selections (heldRenderer,
	// heldReleaseHooks, heldTimersReplacer) and MovesClaimPolicy decides
	// them, but they are granted only once this box's moves source is
	// registered with the claim. Until the wake-source gate opens
	// (movesWakeSourceRegistered, Q1's one gate) the landing keeps refusing
	// here, as before; once it opens it lands. RollbackClaimWired stays
	// (and stays a legal reason in provision), because old records carry it.
	if hostwrite.Enabled(os.Getenv) {
		if err := movesWakeSourceRegistered(); err != nil {
			log.Printf("moves: %s: landing %s under CARLOS_HOST_CLAIM before the wake source is registered: %v", host, id, err)
			return l.markRollback(ctx, provision.RollbackClaimWired)
		}
	}
	obs, err := l.claim(ctx)
	if err != nil {
		return false, err
	}
	if op := obs.Claim.Op; op.Kind == hostclaim.KindMoving && op.Phase == hostclaim.PhaseSealed && op.Exec == "" && op.Journal[hostclaim.JournalEvidenceSource] != "" {
		// The record flipped and the source died before the claim half
		// (ruling G12): the target writes it, with the same preconditions
		// the source would have met.
		h := obs.Claim.Holder
		if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.Flip(id, cfg.Box, h.Stamp, h.Generation)); err != nil {
			return false, fmt.Errorf("the claim half of the flip: %w", err)
		}
		log.Printf("moves: %s: the claim half of the flip was written by the target", host)
		if err := l.fault("after-target-flip"); err != nil {
			return false, err
		}
	}
	if err := l.requirePhase(obs, provision.LandChecking); err != nil {
		return false, err
	}
	if cfg.Box == "" || obs.Claim.Op.Exec != cfg.Box {
		return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, obs.Claim.Op.Exec)
	}
	if cfg.registerLanding != nil {
		if err := cfg.registerLanding(ctx, obs); err != nil {
			return false, err
		}
	}
	rt, err := l.route()
	if err != nil {
		return false, err
	}
	if rt.DBPath != l.dbPath() {
		log.Printf("moves: %s: the held row's database is at %s, not the flat %s", host, rt.DBPath, l.dbPath())
		return l.markRollback(ctx, provision.RollbackLandingLayout)
	}
	if !registry.IsHold(rt.Status) || rt.HoldOp != id {
		return false, fmt.Errorf("the registry row is %q held by %q, not held by landing %s", rt.Status, rt.HoldOp, id)
	}
	holder := *obs.Claim.Holder
	if _, err := writerLease(cfg.OwnPrefix, host, holder); err != nil {
		return false, fmt.Errorf("move %s: %w", id, err)
	}
	src := l.rec.Move.Source
	if src == nil || src.Evidence == "" {
		return false, fmt.Errorf("move %s: the record has no source baselines", id)
	}
	if obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource] != src.Evidence {
		return false, fmt.Errorf("move %s: the claim cites %q, the record %q", id, obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource], src.Evidence)
	}
	report, err := readStoppedReport(ctx, l.s, src.Evidence)
	if err != nil {
		return false, err
	}
	if report.Host != host || report.OpID != id || report.Role != "source" || report.FinalTXID != src.FinalTXID {
		return false, fmt.Errorf("move %s: %s is %s %s %s at TXID %d, not this move's source at %d", id, src.Evidence, report.Host, report.OpID, report.Role, report.FinalTXID, src.FinalTXID)
	}
	if !containsString(obs.Claim.Touched.Boxes, cfg.Box) {
		if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.Touch(id, nil, []string{cfg.Box})); err != nil {
			return false, fmt.Errorf("touch the claim: %w", err)
		}
	}
	own, err := l.w.mirror(ctx, cfg.OwnBucketRegion)
	if err != nil {
		return false, err
	}
	body, err := own.Get(ctx, cfg.OwnBucket, moves.MoveJSONKey(host, id))
	if err != nil {
		return false, fmt.Errorf("read move.json: %w", err)
	}
	var mj moves.MoveJSON
	if err := json.Unmarshal(body, &mj); err != nil {
		return false, fmt.Errorf("move.json: %w", err)
	}
	if mj.Incarnation != l.rec.Move.Incarnation || holder.Incarnation != l.rec.Move.Incarnation {
		return false, fmt.Errorf("move %s: move.json names incarnation %q, the record %q", id, mj.Incarnation, l.rec.Move.Incarnation)
	}
	if err := l.ensureLocalDirs(); err != nil {
		return false, err
	}
	j, err := l.journal()
	if err != nil {
		return false, err
	}
	captured, err := hostguard.Capture(cfg.DataDir, host)
	switch {
	case errors.Is(err, hostguard.ErrNoRecord):
		// No earlier stay on this box.
	case err != nil:
		return false, fmt.Errorf("read the installation record: %w", err)
	case !j.Done(landStepPrevious) && captured.Stamp != 0 && captured.Stamp != holder.Stamp && hostclaim.InventoryHasStamp(obs.Claim, cfg.Box, captured.Stamp):
		j.Phase = provision.LandChecking
		j.Mark(landStepPrevious, strconv.FormatUint(captured.Stamp, 10))
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
		log.Printf("moves: %s: an earlier stay at stamp %d will be retired by local", host, captured.Stamp)
	}
	if err := l.fault("after-previous"); err != nil {
		return false, err
	}
	if err := l.guard.Activate(ctx, host, cfg.Box, nil); err != nil {
		return false, fmt.Errorf("activate the installation record: %w", err)
	}
	if err := l.fault("after-activate"); err != nil {
		return false, err
	}
	if j.Done(landStepPrevious) {
		if obs, err = l.claim(ctx); err != nil {
			return false, err
		}
		if _, err := hostclaim.Update(ctx, l.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil {
			return false, fmt.Errorf("queue the local park: %w", err)
		}
	}
	// Spec §7.6 checking: the lease on the location in this box's own
	// bucket, held through published (M5). The residue parks run under it
	// (ParkRunHeld), so the landing never waits on its own lock.
	if _, waiting, err := acquireOwnLease(ctx, l.w, host, holder); waiting || err != nil {
		return false, err
	}
	return l.advance(ctx, provision.LandChecking, provision.LandParking, nil)
}

func srcTXID(src *provision.MoveSource) uint64 {
	if src == nil {
		return 0
	}
	return src.FinalTXID
}

// installSource binds the shared installer (movesinstall.go) to this
// landing: the claim follows each record step (syncClaimPhase walks it,
// mirrorPhase writes it), the claim check is the landing op at the phase
// with this box executing, the record check a landing at the phase with no
// rollback reason, the refusal the rollback marker (Plan 4's rollback
// transition replaces markRollback), the phase advance MoveLandingPhase
// stamping landing.installed_db_sha256 at the end of configured, and the
// generation, the hold and the archive all the move id's.
func (l *landingRun) installSource() installSource {
	m := l.rec.Move
	return installSource{
		OpID: l.g.ID, Host: l.rec.Host, Role: roleTarget, Name: "landing",
		Lane: moves.StagingPrefix(l.rec.Host, l.g.ID), Seal: journalledSeal, Baseline: m.Source,
		GenerationOp: l.g.ID, HoldOp: l.g.ID, PinInputs: m.Inputs,
		Phases: [6]string{provision.LandParking, provision.LandBuilding, provision.LandPublished,
			provision.LandLocal, provision.LandConfigured, provision.LandRestored},
		Phase: func(from, to string, st installStamp) func(*provision.Record) error {
			var stamp func(*provision.MoveState)
			if sha := st.InstalledDBSHA256; sha != "" {
				stamp = func(m *provision.MoveState) {
					if m.Landing == nil {
						m.Landing = &provision.MoveLandingJournal{}
					}
					m.Landing.InstalledDBSHA256 = sha
				}
			}
			return provision.MoveLandingPhase(l.g, from, to, stamp)
		},
		Mirror:     l.mirrorPhase,
		Sync:       l.syncClaimPhase,
		ClaimCheck: l.landingCheck,
		RecordPhase: func(rec *provision.Record) (string, error) {
			m := rec.Move
			if m == nil || m.ID != l.g.ID || m.Incarnation != l.g.Incarnation || m.State != provision.MoveLanding || m.Reason != "" {
				return "", fmt.Errorf("%w: the record is no longer landing %s", errLandingChanged, l.g.ID)
			}
			return m.LandingPhase, nil
		},
		Refuse: func(ctx context.Context, why installRefusal) (bool, string, error) {
			var reason string
			switch why {
			case refuseFilesOccupied:
				reason = provision.RollbackFilesOccupied
			case refuseManifestOccupied:
				reason = provision.RollbackManifestOccupied
			case refuseInstallMismatch:
				reason = provision.RollbackInstallMismatch
			case refuseConfigChanged:
				reason = provision.RollbackConfigChanged
			case refuseDeadline:
				reason = provision.RollbackLandingDeadline
			default:
				return false, "", fmt.Errorf("move %s: a landing has no refusal %d", l.g.ID, why)
			}
			ok, err := l.markRollback(ctx, reason)
			if !ok {
				return false, "", err
			}
			return true, reason, err
		},
		PastDeadline: l.pastDeadline,
	}
}