package host

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

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"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 generation installer (moves spec §7.6, §7.8; Plan 4 Task 3), shared by
// the landing on the target and the dirty return on the source. It runs the
// six phases parking, building, published, local, configured and restored
// exactly as Plan 2 Task 10 wrote them for the landing; installSource says
// what differs between the two kinds of install (rulings J13, J14, J20, J24):
//
//   - the claim check: a landing's op is landing at the phase, mirrored into
//     the claim by every record step; a return's op is restoring/claimed with
//     the source executing at its pre-move stamp S throughout, and the claim
//     does not follow the source's install phases;
//   - the record check: landing at the phase with no rollback reason, or
//     rolling_back, dirty, with the source at the phase;
//   - the refusal action: a landing marks its rollback (MoveMarkRollback); a
//     return stops for an operator (MoveMarkStuck) with a recorded reason;
//   - the phase advance: MoveLandingPhase or MoveRollbackSourcePhase, bound
//     by the caller, so neither kind is told apart by a type switch;
//   - the generation: the move id for a landing, the rollback id for a return
//     (J14: a return is cold and never reuses S's generation name). It names
//     the build, the install's own journal, <dbPath>.gen and .files-gen;
//   - configured: a landing renders and requires its pinned inputs; a return
//     renders the current ones and records them in Rollback.RecoveryInputs,
//     never compared with move.inputs (J2-R2-5);
//   - local: a landing parks an earlier stay's artifacts tagged S'; a return
//     parks the source's own pre-move installation before anything is
//     extracted (J13, fix round 1): RetireArtifacts for S and for every
//     stamp the claim shows superseded on this box, a stuck mark for any
//     other stamp, then the moves-owned rename (H5(b), parkLocalArtifacts)
//     of every pre-move artifact still at the flat paths, inside the host
//     guard. placeStaged and restore then find nothing there, and refuse
//     anything that appears;
//   - the authoritative region the claim's view names for its parks (J24,
//     §5.1): the source's for a landing, the target's for a return, each the
//     other side of the move before this side's probe is admitted.
//
// The phase-name convention is Plan 2's (G8): restored is written by the CAS
// that ends configured, after the database is installed and its markers
// written. Every record write goes through a provision transition (the
// record_seq rule). The installer is unarmed in production, as the landing
// is (until the wake-source gate opens, movesWakeSourceRegistered, Plan 5
// Task 9b): only the landing's pass,
// the recovery and the tests call it.

// installStepPreMove is the moves-owned rename of a return's pre-move
// artifacts (parkLocalArtifacts' step: the retirement id).
const installStepPreMove = "pre_move_parked"

// installStepResidueKept is a dirty return's no-op parking (Task 9a fix
// round 3, ruling (a)): the sentence saying the target never admitted the
// move, so the pre-move generation stays the source's live state.
const installStepResidueKept = "residue_kept"

// installStamp is what a phase step records besides the phase: at the CAS
// that ends configured, the installed database's digest (a landing records
// it in landing.installed_db_sha256) and the inputs a return rendered.
type installStamp struct {
	InstalledDBSHA256 string
	RecoveryInputs    *provision.MoveInputs
}

// installRefusal is why an install stops for good: installSource.Refuse maps
// it to its kind's recorded sentence.
type installRefusal int

const (
	refuseFilesOccupied    installRefusal = iota + 1 // local: the files paths hold what the install did not write
	refuseInstallMismatch                            // configured: the build does not verify
	refuseConfigChanged                              // configured: a landing's render is not its pin
	refuseDeadline                                   // configured: a landing ran past its deadline
	refuseUnknownStamp                               // local: a return found paths under a stamp it cannot account for
	refuseManifestOccupied                           // local: the prepared files manifest's key holds a manifest the install did not write
)

// installSource is one kind of install: what the six phases need from the
// caller (Plan 4 Task 3's interface, with the rulings' additions).
type installSource struct {
	OpID, Host string
	// Role is roleTarget for a landing, roleSource for a dirty return.
	Role string
	// Name is the install's word in the log: "landing" or "return".
	Name string
	// Lane is the staging or return lane prefix in this box's own bucket;
	// every object the seal names must be under it.
	Lane string
	// Seal reads the sealed set from the claim the phase works from.
	Seal func(*hostclaim.Claim) (moves.Seal, error)
	// Baseline is what the sealed set restores to: the final TXID, the
	// files and timers digests, and the carried columns.
	Baseline *provision.MoveSource
	// GenerationOp names the generation this install builds and the
	// install's own journal; HoldOp is the op the row is held by.
	GenerationOp, HoldOp string
	// PinInputs are a landing's pinned inputs; nil for a return.
	PinInputs *provision.MoveInputs
	// Phases are the six phase names of this install, in order.
	Phases [6]string
	// Phase is the record transition one step makes, bound to the guard.
	Phase func(from, to string, st installStamp) func(*provision.Record) error
	// Mirror is the claim half of a record step; nil when the claim does
	// not follow this install's phases.
	Mirror func(ctx context.Context, from, to string) error
	// Sync is each phase's first claim read (a landing walks a claim that
	// is behind forward); ClaimCheck is what the phase and the guard require
	// of the claim at phase.
	Sync       func(ctx context.Context) (hostclaim.Observed, error)
	ClaimCheck func(phase string) func(*hostclaim.Claim) error
	// RecordPhase answers the phase a record is at while it is still this
	// install's, and an error otherwise.
	RecordPhase func(*provision.Record) (string, error)
	// Refuse writes the refusal: written is true when the record took it,
	// sentence is the recorded reason.
	Refuse func(ctx context.Context, why installRefusal) (written bool, sentence string, err error)
	// PastDeadline reports a landing's deadline passed; nil for a return.
	PastDeadline func() bool
}

func (src installSource) validate() error {
	if src.OpID == "" || src.Host == "" || (src.Role != roleTarget && src.Role != roleSource) || src.Lane == "" || src.Seal == nil ||
		src.GenerationOp == "" || src.HoldOp == "" || src.Phase == nil || src.Sync == nil || src.ClaimCheck == nil ||
		src.RecordPhase == nil || src.Refuse == nil {
		return errors.New("moves: an install source names its op, role, lane, seal, generation and hooks")
	}
	for _, p := range src.Phases {
		if p == "" {
			return errors.New("moves: an install source names its six phases")
		}
	}
	return nil
}

// kind is the claim op kind this install runs under.
func (src installSource) kind() string {
	if src.Role == roleSource {
		return hostclaim.KindRestoring
	}
	return hostclaim.KindLanding
}

// installRun is one call's state for one install.
type installRun struct {
	w     *movesWorker
	s     release.Store
	rec   provision.Record
	g     provision.MoveGuard
	src   installSource
	guard hostguard.Guard
	local park.Local
	// refused is the sentence a refusal recorded during this call.
	refused string
}

// installGeneration runs the record's current install phase once (one phase
// per call, as the landing's pass and the cut step): parking, building,
// published, local or configured. advanced is true when it wrote the record
// (a phase step or the refusal); refused is the recorded sentence when the
// refusal was written. A record at restored is the probe's and is left as it
// is; one that is no longer this install's is an error.
func installGeneration(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, src installSource) (advanced bool, refused string, err error) {
	if err := src.validate(); err != nil {
		return false, "", err
	}
	phase, err := src.RecordPhase(&rec)
	if err != nil {
		return false, "", err
	}
	if rec.Move.ID != src.OpID || rec.Host != src.Host {
		return false, "", fmt.Errorf("moves: the install of %s is not %s's", src.OpID, rec.Host)
	}
	ir := newInstallRun(s, w, rec, src)
	switch phase {
	case src.Phases[0]:
		advanced, err = ir.parking(ctx)
	case src.Phases[1]:
		advanced, err = ir.building(ctx)
	case src.Phases[2]:
		advanced, err = ir.published(ctx)
	case src.Phases[3]:
		advanced, err = ir.local_(ctx)
	case src.Phases[4]:
		advanced, err = ir.configured(ctx)
	case src.Phases[5]:
		return false, "", nil
	default:
		return false, "", fmt.Errorf("moves: %s is not a phase of the %s's install", phase, src.Name)
	}
	return advanced, ir.refused, err
}

func newInstallRun(s release.Store, w *movesWorker, rec provision.Record, src installSource) *installRun {
	cfg := w.cfg
	ir := &installRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}, src: src}
	ir.guard = cfg.Guard
	if ir.guard.Claims == nil {
		ir.guard.Claims = hostguard.StoreReader{Store: s}
	}
	ir.local = cfg.Local
	ir.local.Guard, ir.local.Claims = ir.guard, s
	if ir.local.Box == "" {
		ir.local.Box = cfg.Box
	}
	if ir.local.Authorize == nil {
		// park.Local's authorizer: the claim's MoveParks local authorizer
		// (ParkLocalAuthorize) behind this install's own check; the claim's
		// engine authorizer refuses a park-local (M9). Without it every local
		// chore refuses (landingArmed requires it).
		ir.local.Authorize = movesLocalAuthorize(cfg.ParkLocalAuthorize, ir.authorizeChore)
	}
	return ir
}

func (ir *installRun) cfg() moveCfg { return ir.w.cfg }

// installAuthority is the install's authority re-check for the prepared
// manifest's bucket writes (fix round 1, finding 1; region HA's
// authorityStore pattern): a fresh read of the claim, the install's own
// claim check at phase (the op at that phase with this box executing) and
// the holder still at the captured stamp. It runs immediately before every
// write attempt, retries included, so a delayed build whose executor,
// phase or stamp moved writes nothing more.
func (ir *installRun) installAuthority(ctx context.Context, captured hostguard.Captured, phase string) func() error {
	check := ir.src.ClaimCheck(phase)
	return func() error {
		if err := ctx.Err(); err != nil {
			return err
		}
		obs, err := hostclaim.Read(ctx, ir.s, ir.rec.Host)
		if err != nil {
			return err
		}
		if obs.Absent || obs.Claim == nil {
			return fmt.Errorf("%w: %s has no claim", errLandingChanged, ir.rec.Host)
		}
		if err := check(obs.Claim); err != nil {
			return err
		}
		if obs.Claim.Holder == nil || obs.Claim.Holder.Stamp != captured.Stamp {
			return fmt.Errorf("%w: the claim's holder is not at the installation stamp %d", errLandingChanged, captured.Stamp)
		}
		return nil
	}
}

// installAuthorityStore is the versioned store the prepared build writes
// through: every write re-checks the install's authority first. Reads are
// the store's own; the build never deletes.
type installAuthorityStore struct {
	lease.VersionedStore
	check func() error
}

func (s installAuthorityStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
	if err := s.check(); err != nil {
		return "", err
	}
	return s.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
}

func (s installAuthorityStore) CopyVersion(ctx context.Context, sb, sk, sv, db, dk string, createOnly bool) (string, string, error) {
	if err := s.check(); err != nil {
		return "", "", err
	}
	return s.VersionedStore.CopyVersion(ctx, sb, sk, sv, db, dk, createOnly)
}

func (installAuthorityStore) DeleteVersion(context.Context, string, string, string) error {
	return fmt.Errorf("%w: the prepared build never deletes", errLandingChanged)
}

// preparedManifestJournalled reports whether the journal records the
// prepared files manifest as built: the step done with this install's
// binding (decodePreparedManifest: P's key, the op and the captured
// stamp), not the "skipped" value journals written before C8 existed,
// which a replay builds (ruling Q3 item 2; the step itself is create-only
// and replay-safe). Any other value is an error (fix round 1, finding 3):
// an empty or malformed step, or a binding for another key, op or stamp,
// is a damaged or foreign journal, and the install neither advances over
// it nor rebuilds over it.
func preparedManifestJournalled(j *moves.Journal, key, op string, stamp uint64) (bool, error) {
	v, ok := j.Steps[landStepFilesManifest]
	if !ok || v == landingFilesManifestSkipped {
		return false, nil
	}
	if _, err := decodePreparedManifest(v, key, op, stamp); err != nil {
		return false, err
	}
	return true, nil
}

// fault is the test seam after a side effect (the landing's points).
func (ir *installRun) fault(point string) error {
	if f := ir.w.cfg.landFault; f != nil {
		return f(point)
	}
	return nil
}

// sync is each phase's first step: the kind's claim read, then its check at
// phase (K5's revalidation).
func (ir *installRun) sync(ctx context.Context, phase string) (hostclaim.Observed, error) {
	obs, err := ir.src.Sync(ctx)
	if err != nil {
		return obs, err
	}
	if err := ir.src.ClaimCheck(phase)(obs.Claim); err != nil {
		return obs, fmt.Errorf("move %s: %w", ir.g.ID, err)
	}
	return obs, nil
}

// seal reads the sealed set from the claim and requires every object it
// names in this install's lane.
func (ir *installRun) seal(c *hostclaim.Claim) (moves.Seal, error) {
	seal, err := ir.src.Seal(c)
	if err != nil {
		return seal, err
	}
	for _, class := range [][]moves.SealedObject{seal.Replica, seal.Files, seal.Timers, seal.State} {
		for _, o := range class {
			if !strings.HasPrefix(o.Key, ir.src.Lane) {
				return seal, fmt.Errorf("%w: %s is not in the %s's lane %s", moves.ErrBadSeal, o.Key, ir.src.Name, ir.src.Lane)
			}
		}
	}
	return seal, nil
}

// transition writes one record transition. A conflict is someone else's
// decision (a rollback, a stuck mark, a console write): logged, and the call
// ends without error.
func (ir *installRun) transition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, ir.s, ir.rec.Account, ir.rec.App, ir.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", ir.rec.Host, what, ir.g.ID, err)
			return false, nil
		}
		return false, fmt.Errorf("%s: %w", what, err)
	}
	ir.w.logTransition(ctx, ir.s, pre, *after)
	return true, nil
}

// advance writes the record's step and then, for a landing, mirrors it into
// the claim. A mirror that fails is logged: the next pass's sync writes it.
func (ir *installRun) advance(ctx context.Context, from, to string, st installStamp) (bool, error) {
	name := ir.src.Name
	ok, err := ir.transition(ctx, "the "+name+" "+from+" to "+to, ir.src.Phase(from, to, st))
	if !ok {
		return ok, err
	}
	log.Printf("moves: %s: %s %s is %s", ir.rec.Host, name, ir.g.ID, to)
	if err := ir.fault("after-record-" + to); err != nil {
		return true, err
	}
	if ir.src.Mirror != nil {
		if err := ir.src.Mirror(ctx, from, to); err != nil {
			log.Printf("moves: %s: the claim half of %s %s waits for the next pass: %v", ir.rec.Host, name, to, err)
		}
	}
	return true, nil
}

// refuse writes the kind's refusal; written is true when the record took it.
func (ir *installRun) refuse(ctx context.Context, why installRefusal) (bool, error) {
	ok, sentence, err := ir.src.Refuse(ctx, why)
	if ok {
		ir.refused = sentence
	}
	return ok, err
}

func (ir *installRun) pastDeadline() bool { return ir.src.PastDeadline != nil && ir.src.PastDeadline() }

// route is the host's registry row on this box.
func (ir *installRun) route() (registry.Route, error) {
	rt, err := ir.cfg().Route(ir.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", ir.rec.Host)
	}
	return *rt, nil
}

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

// journal is the install's own op journal, named by its generation.
func (ir *installRun) journal() (*moves.Journal, error) {
	return moves.ReadJournal(ir.cfg().DataDir, ir.rec.Host, ir.src.GenerationOp)
}

func (ir *installRun) journalDir() string {
	return moves.JournalDir(ir.cfg().DataDir, ir.rec.Host, ir.src.GenerationOp)
}

// buildDir is BuildGeneration's scratch, <JournalDir>/build/ (ruling G34),
// on the data disk so the result renames into DBPath.
func (ir *installRun) buildDir() string { return filepath.Join(ir.journalDir(), "build") }

// versioned is this box's own bucket by exact version.
func (ir *installRun) versioned(ctx context.Context) (lease.VersionedStore, error) {
	cfg := ir.cfg()
	if cfg.Versioned == nil {
		return nil, errors.New("no versioned store on this box")
	}
	return cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
}

// parking (§7.6, D6): whatever an earlier stay of this incarnation left at
// holder.location in this box's own bucket is parked before the build
// writes beside it, one park-generation chore per generation (the origin as
// @root), scoped to the location with the flip's time as the evidence
// horizon. On the target that is an earlier stay's residue; on the source
// of a dirty return it is the pre-move generations (§7.8, D6). Each chore is
// run once per pass through the park engine. The location's lease is the
// install's own, held from here (a landing's from checking) through
// published, and each chore runs under it through park.Engine.RunHeld over
// the claim's leases (ParkRunHeld, M5). ErrPending is the expected answer: the acknowledgement waits for
// the horizon, and is C4's runner's. The phase advances once every chore's
// manifest exists and its deletion set is gone (Engine.Parked).
// Per-generation chores are what let the acknowledgement finish after the
// probe makes the new generation live here.
//
// A replay finds parking and continues from the claim: the residue list is
// journalled so the chores are the same ones, the queue is idempotent per
// generation, and Engine.Run resumes a chore from its manifest.
func (ir *installRun) parking(ctx context.Context) (bool, error) {
	cfg := ir.cfg()
	host, id, gen := ir.rec.Host, ir.g.ID, ir.src.GenerationOp
	phase, next := ir.src.Phases[0], ir.src.Phases[1]
	obs, err := ir.sync(ctx, phase)
	if err != nil {
		return false, err
	}
	holder := *obs.Claim.Holder
	location := strings.TrimSuffix(holder.Location, "/")
	vs, err := ir.versioned(ctx)
	if err != nil {
		return false, err
	}
	j, err := ir.journal()
	if err != nil {
		return false, err
	}
	_, listed := j.Steps[landStepResidue]
	if !listed && ir.src.Role == roleSource && !targetAdmitted(ir.rec.Move, obs.Claim) {
		// A dirty return whose target never admitted this move (ruling
		// (a), fix round 3): holder.generation is still the pre-move one,
		// the source's live state, so there is no residue to park. This
		// is decided from the claim's recorded admissions only (fix round
		// 4: not the landing phase), never from a park view's
		// refusal, which can be transient.
		why := fmt.Sprintf("the target never admitted move %s: the claim holds no admission of its generation; the pre-move generation stays the source's live state", id)
		j.Phase = phase
		j.Mark(landStepResidue, "[]")
		j.Mark(installStepResidueKept, why)
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
		log.Printf("moves: %s: the %s parks nothing: %s", host, ir.src.Name, why)
		return ir.advance(ctx, phase, next, installStamp{})
	}
	var gens []string
	if raw, ok := j.Steps[landStepResidue]; ok {
		if err := json.Unmarshal([]byte(raw), &gens); err != nil {
			return false, fmt.Errorf("the journal's residue list: %w", err)
		}
	} else {
		objs, err := vs.ListVersions(ctx, cfg.OwnBucket, location+"/")
		if err != nil {
			return false, fmt.Errorf("list %s: %w", location, err)
		}
		seen := map[string]bool{}
		for _, o := range objs {
			if o.Key == location+"/lock.json" {
				continue
			}
			if movesHolderBuildState(location, o.Key) {
				// The build state beside the lease is the marker's twin
				// (ruling R-PARK): no generation's, never residue here.
				// The build places the moved state over it.
				continue
			}
			g, ok := landingGenerationOf(location, o.Key)
			if !ok {
				return false, fmt.Errorf("%s is not an object of a generation this %s can classify", o.Key, ir.src.Name)
			}
			if g == gen || g == landingMarkerGeneration {
				// The install's own generation is never residue, and the
				// marker is superseded by the probe, never parked.
				continue
			}
			if g == "" {
				g = hostclaim.RootGeneration
			}
			seen[g] = true
		}
		for g := range seen {
			gens = append(gens, g)
		}
		sort.Strings(gens)
		body, err := json.Marshal(gens)
		if err != nil {
			return false, err
		}
		j.Phase = phase
		j.Mark(landStepResidue, string(body))
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
	}
	if len(gens) == 0 {
		return ir.advance(ctx, phase, next, installStamp{})
	}
	if ir.rec.Move.FlippedAt == nil {
		return false, fmt.Errorf("move %s: the record has no flip time", id)
	}
	// The parks run under the build's own lease on the location, which the
	// install holds from here (a landing from checking) through published:
	// the residue scope's lease set is that lock (M6), so Engine.Run would
	// wait on the install itself. park.Engine.RunHeld (ParkRunHeld) runs
	// each chore under the held lease instead, borrowed through the claim's
	// AggregateLeases (M5).
	held, waiting, err := ir.lease(ctx, holder)
	if waiting || err != nil {
		return false, err
	}
	scope := moves.ResidueScope(holder, *ir.rec.Move.FlippedAt)
	if obs, err = hostclaim.Update(ctx, ir.s, obs, hostclaim.QueueResidueParks(id, cfg.parkRegion(), scope, gens, obligationID)); err != nil {
		return false, fmt.Errorf("queue the residue parks: %w", err)
	}
	if err := ir.fault("after-residue-queued"); err != nil {
		return false, err
	}
	eng := ir.engine(vs)
	pending := 0
	for _, o := range obs.Claim.Obligations {
		if o.Op != id || o.Kind != "park-generation" || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) {
			continue
		}
		w := park.Work{Host: host, ID: o.ID, Op: id}
		if cfg.ParkRunHeld == nil {
			return false, fmt.Errorf("move %s: %w", id, &ClaimDependencyError{Contracts: []string{"the claim's MoveParks RunHeld (armMovesParks)"}})
		}
		if err := cfg.ParkRunHeld(ctx, eng, w, held); err != nil && !errors.Is(err, park.ErrPending) {
			return false, fmt.Errorf("park generation %s: %w", o.Generation, err)
		}
		parked, err := eng.Parked(ctx, w)
		if errors.Is(err, park.ErrChanged) {
			continue // acknowledged already
		}
		if err != nil {
			return false, fmt.Errorf("park generation %s: %w", o.Generation, err)
		}
		if !parked {
			pending++
		}
	}
	if pending > 0 {
		ir.w.notice(host+"#parking", "moves: %s: %d residue park(s) are not complete yet", host, pending)
		return false, nil
	}
	return ir.advance(ctx, phase, next, installStamp{})
}

// targetAdmitted reports whether the claim carries evidence that the
// move's target admitted a writer: the holder's admissions include the
// move's prepared generation (an admission appends its record atomically,
// and ReturnToSource keeps them). The record's landing phase is not
// evidence: it enters probing before the probe runs (fix round 4), so a
// landing that stopped at probing before its admission never admitted.
func targetAdmitted(m *provision.MoveState, c *hostclaim.Claim) bool {
	if m == nil || c == nil || c.Holder == nil {
		return false
	}
	for _, a := range c.Holder.Admissions {
		if a.Generation == m.ID {
			return true
		}
	}
	return false
}

// building (§7.6, §5.1 step 2): the generation built from the seal's pinned
// replica versions under the location's lease (moves.BuildGeneration: the
// intent, the snapshot, the anchor, the completed install manifest, every
// one create-only), with the result left at <JournalDir>/build/<host>.db
// for the restore. The manifest goes into the journal.
//
// A replay finds building and builds again: BuildGeneration reuses what it
// finds under the generation's keys and refuses anything it did not make.
func (ir *installRun) building(ctx context.Context) (bool, error) {
	cfg := ir.cfg()
	host, id, gen := ir.rec.Host, ir.g.ID, ir.src.GenerationOp
	phase, next := ir.src.Phases[1], ir.src.Phases[2]
	obs, err := ir.sync(ctx, phase)
	if err != nil {
		return false, err
	}
	holder := *obs.Claim.Holder
	held, waiting, err := ir.lease(ctx, holder)
	if waiting || err != nil {
		return false, err
	}
	seal, err := ir.seal(obs.Claim)
	if err != nil {
		return false, err
	}
	src := ir.src.Baseline
	if src == nil || seal.N != src.FinalTXID {
		return false, fmt.Errorf("move %s: the seal is at TXID %d and the record at %d", id, seal.N, srcTXID(src))
	}
	vs, err := ir.versioned(ctx)
	if err != nil {
		return false, err
	}
	if err := held.Check(ctx); err != nil {
		return false, fmt.Errorf("the location's lease: %w", err)
	}
	// Every generation request runs under live lease authority (fix round
	// 1, item 3): the store checks the lease before each call, as the cut's
	// guarded writes check theirs, the requests run under a context the
	// lease's loss cancels (leaseContext, fix round 2), and the lease is
	// checked again before the build's completion is recorded.
	m, err := moves.BuildGeneration(leaseContext(ctx, held), moves.BuildInputs{Store: leasedStore{VersionedStore: vs, check: held.Check}, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
		GenerationPrefix: moves.GenerationPrefix(holder.Location, gen), Op: gen, Host: host, WorkDir: ir.buildDir(), Now: cfg.Now})
	if err != nil {
		return false, fmt.Errorf("build the generation: %w", err)
	}
	if err := ir.placeBuildState(leaseContext(ctx, held), leasedStore{VersionedStore: vs, check: held.Check}, seal, holder); err != nil {
		return false, err
	}
	if err := held.Check(ctx); err != nil {
		return false, fmt.Errorf("the location's lease after the build: %w", err)
	}
	body, err := json.Marshal(m)
	if err != nil {
		return false, err
	}
	j, err := ir.journal()
	if err != nil {
		return false, err
	}
	j.Phase = phase
	j.Mark(landStepInstall, string(body))
	if err := j.Commit(cfg.DataDir); err != nil {
		return false, err
	}
	if err := ir.fault("after-build"); err != nil {
		return false, err
	}
	log.Printf("moves: %s: generation %s built at TXID %d", host, gen, m.N)
	return ir.advance(ctx, phase, next, installStamp{})
}

// placeBuildState copies the sealed build state (the seal's State class,
// the lane's state/ key) beside the location's lease,
// holder.Location/build-state.json in this box's own bucket (settings spec
// rev 11 §4.3a, ruling R-LANDING). building calls it under the location's
// lease, so it lands before the probe and the first wake: their record
// merges onto the moved state, and a held instance never reads as new
// here. It copies exactly the sealed version (ruling R-SEAL-STATE): a
// request a stale round signed can land a newer version in the lane until
// the seal's bound, and that version is never placed. A seal with no state
// (the source had none, or a seal from before the class existed) places
// nothing, and removes every version of a build state an earlier stay left
// at the location (Task 12a): it is the marker's twin, so the landing's park
// passes it over, and otherwise that stay's state would read as this
// instance's. A replay copies, or removes, again.
func (ir *installRun) placeBuildState(ctx context.Context, vs lease.VersionedStore, seal moves.Seal, holder hostclaim.Holder) error {
	bucket := ir.cfg().OwnBucket
	dst := release.BuildStateKey(strings.TrimSuffix(holder.Location, "/"))
	if len(seal.State) == 0 {
		vers, err := vs.ListVersions(ctx, bucket, dst)
		if err != nil {
			return fmt.Errorf("list a leftover build state at %s: %w", dst, err)
		}
		for _, v := range vers {
			if v.Key != dst {
				continue
			}
			if err := vs.DeleteVersion(ctx, bucket, dst, v.VersionID); err != nil {
				return fmt.Errorf("remove a leftover build state at %s (version %s): %w", dst, v.VersionID, err)
			}
			log.Printf("moves: %s: the %s carries no build state; removed version %s of an earlier stay's at %s", ir.rec.Host, ir.src.Name, v.VersionID, dst)
		}
		return nil
	}
	o := seal.State[0]
	if o.Key != ir.src.Lane+moves.BuildStateSuffix {
		return fmt.Errorf("%w: the sealed build state %s is not the %s's lane key", moves.ErrBadSeal, o.Key, ir.src.Name)
	}
	if _, _, err := vs.CopyVersion(ctx, bucket, o.Key, o.VersionID, bucket, dst, false); err != nil {
		return fmt.Errorf("place the build state at %s: %w", dst, err)
	}
	log.Printf("moves: %s: the %s placed the moved build state at %s", ir.rec.Host, ir.src.Name, dst)
	return nil
}

// published (§7.6): the build is complete and recorded; generation.json is
// not written here (R10-3: the probe's cold admission publishes it). The
// journal records the phase, the record advances to local and the location's
// lease is released: the build was the only writer that needed it.
func (ir *installRun) published(ctx context.Context) (bool, error) {
	phase, next := ir.src.Phases[2], ir.src.Phases[3]
	if _, err := ir.sync(ctx, phase); err != nil {
		return false, err
	}
	j, err := ir.journal()
	if err != nil {
		return false, err
	}
	if !j.Done(landStepInstall) {
		return false, fmt.Errorf("move %s: published without an install manifest in the journal", ir.g.ID)
	}
	j.Phase = phase
	j.Mark(landStepPublished, ir.cfg().Now().UTC().Format(time.RFC3339Nano))
	if err := j.Commit(ir.cfg().DataDir); err != nil {
		return false, err
	}
	ok, err := ir.advance(ctx, phase, next, installStamp{})
	if ok {
		ir.releaseLease()
	}
	return ok, err
}

// local (§7.6), each sub-step journalled and skipped on a replay once done,
// in this order (round-2 finding 8): (1) the installation record must be at
// the holder's stamp; (2) what is there before is parked: on the target an
// earlier stay's artifacts tagged S' through park.Local (which refuses a
// path tagged with another stamp), or a park-local left with nothing to
// retire is completed; on the source of a dirty return its own pre-move
// installation (retirePreMove); (3) the sealed archive, read by exact
// version and verified, is extracted into this install's own staging,
// <JournalDir>/extract/, and then, inside the host guard, moved to the files
// directory and <host>.env when nothing this install did not write is there
// (installFiles: the install never deletes at the live paths, final review
// finding 1; a retry clears only its staging); (4) the new paths are
// registered under the holder's stamp; (5) the timers table is the sealed
// snapshot with its registrations, the carried columns are set and the
// restore-verification entry removed; (6) the prepared files manifest (C8,
// Task 6): the placed tree, <host>.env and the wake state as create-only
// objects and one create-only sequence-0 manifest under files/g-<gen>/ in
// the location, each write re-checking the install's authority, the
// binding journalled with the manifest's digest and exact version, and a
// manifest already there that is not this install's refused
// (RollbackManifestOccupied); (7) the files directory is synced
// and .files-gen written last, so its presence vouches for every step
// before it.
//
// A replay finds local and finishes what the journal does not record;
// nothing here is read by the probe before .files-gen exists.
func (ir *installRun) local_(ctx context.Context) (bool, error) {
	cfg := ir.cfg()
	host, id, gen := ir.rec.Host, ir.g.ID, ir.src.GenerationOp
	phase, next := ir.src.Phases[3], ir.src.Phases[4]
	obs, err := ir.sync(ctx, phase)
	if err != nil {
		return false, err
	}
	holder := *obs.Claim.Holder
	src := ir.src.Baseline
	if src == nil || src.Files == nil {
		return false, fmt.Errorf("move %s: the record has no source baselines", id)
	}
	seal, err := ir.seal(obs.Claim)
	if err != nil {
		return false, err
	}
	j, err := ir.journal()
	if err != nil {
		return false, err
	}
	j.Phase = phase
	mark := func(step, value string) error {
		j.Mark(step, value)
		if err := j.Commit(cfg.DataDir); err != nil {
			return err
		}
		return ir.fault("after-" + step)
	}
	captured, err := ir.capture(holder)
	if err != nil {
		return false, err
	}
	if !j.Done(landStepRetired) {
		if ir.src.Role == roleSource {
			outcome, err := ir.retirePreMove(ctx, captured, j, mark)
			switch {
			case err != nil:
				return false, err
			case outcome == restoreMarked:
				return true, nil // the record changed: the refusal
			case outcome == restoreWaiting:
				return false, nil
			}
		} else if raw, ok := j.Steps[landStepPrevious]; ok {
			earlier, err := strconv.ParseUint(raw, 10, 64)
			if err != nil || earlier == 0 {
				return false, fmt.Errorf("the journal's previous stamp %q", raw)
			}
			var chore *hostclaim.Obligation
			for i := range obs.Claim.Obligations {
				if o := &obs.Claim.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
					chore = o
				}
			}
			switch {
			case chore != nil && hostclaim.InventoryHasStamp(obs.Claim, cfg.Box, earlier):
				if err := ir.local.RetireArtifacts(ctx, host, captured, chore.ID, id, earlier); err != nil {
					return false, fmt.Errorf("retire the earlier stay's artifacts: %w", err)
				}
				log.Printf("moves: %s: the earlier stay's artifacts (stamp %d) are parked", host, earlier)
			case chore != nil:
				if _, err := hostclaim.Update(ctx, ir.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, earlier)); err != nil {
					return false, fmt.Errorf("complete the local park: %w", err)
				}
			}
		}
		if err := mark(landStepRetired, ""); err != nil {
			return false, err
		}
	}
	dir, envPath := cfg.FilesDir(host)
	if !j.Done(landStepExtracted) {
		outcome, err := ir.installFiles(ctx, j, mark, captured, seal, *src.Files, dir, envPath)
		switch {
		case err != nil:
			return false, err
		case outcome == restoreMarked:
			return true, nil // the record changed: the refusal
		case outcome == restoreWaiting:
			return false, nil
		}
	}
	if !j.Done(landStepInventory) {
		// Only what this install has placed (carry (b)): the files directory
		// and <host>.env, which installFiles moved into place when nothing
		// it did not write was there. The database and its marker are
		// registered by configured, once the install proves the database at
		// DBPath is this build (registerDatabase): registered here, before
		// the install, a rollback's retirement of F would park whatever
		// database stands at DBPath, a foreign one included.
		paths := []string{host, host + ".env"}
		if err := ir.local.RecordInventory(ctx, host, captured, "flat", paths, ir.src.ClaimCheck(phase)); err != nil {
			return false, fmt.Errorf("record the inventory: %w", err)
		}
		if err := mark(landStepInventory, ""); err != nil {
			return false, err
		}
	}
	if !j.Done(landStepTimers) {
		rows, err := ir.sealedTimers(ctx, seal, src.TimersSHA256)
		if err != nil {
			return false, err
		}
		if err := cfg.ReplaceTimers(ctx, host, rows); err != nil {
			return false, fmt.Errorf("replace the timers: %w", err)
		}
		if err := mark(landStepTimers, strconv.Itoa(len(rows))); err != nil {
			return false, err
		}
	}
	if !j.Done(landStepColumns) {
		if err := cfg.SetReplication(host, src.Replication); err != nil {
			return false, fmt.Errorf("set replication: %w", err)
		}
		if err := cfg.SetNextDue(host, src.NextDue); err != nil {
			return false, fmt.Errorf("set next_due: %w", err)
		}
		if err := mark(landStepColumns, ""); err != nil {
			return false, err
		}
	}
	if !j.Done(landStepVerify) {
		if err := cfg.RemoveVerifyEntry(ctx, ir.s, host); err != nil {
			return false, fmt.Errorf("remove the restore-verification entry: %w", err)
		}
		if err := mark(landStepVerify, ""); err != nil {
			return false, err
		}
	}
	built, err := preparedManifestJournalled(j, hostfiles.ManifestKey(movesFilesPrefix(holder.Location, gen)), id, captured.Stamp)
	if err != nil {
		return false, err
	}
	if !built {
		// Spec rev 7.22 §7.6 (L17, R19-1; C8, Task 6): the prepared files
		// manifest under files/g-<generation>/ in the location, bound to
		// {location, g-<generation>, op, stamp}, where the generation is the
		// installation's prepared destination generation (the move id, or
		// the rollback id of a dirty return), never holder.generation. It
		// carries the placed files, <host>.env and the wake state the
		// install put in place above, as create-only objects and one
		// create-only sequence-0 manifest through the box's own versioned
		// store, after the claim sync and stamp capture above (op.exec's
		// installation authority). The binding is journalled with the
		// manifest's digest; a replay reuses the objects by digest and reads
		// the manifest back by key. A journal that records the step as
		// skipped (written before C8 existed) counts as not done (ruling Q3
		// item 2). Another manifest at the key is not this install's to
		// replace: the refusal (RollbackManifestOccupied).
		// The image is the install's own placement, by the install receipt
		// (files_placed: the tree's identity, the env file's identity or
		// none): a vanished tree or env file refuses, and the manifest
		// carries an env item exactly when the archive carried one (fix
		// round 1, finding 4).
		placed, err := journalledStaged(j, landStepFilesPlaced)
		if err != nil {
			return false, err
		}
		if ok, err := sameIdentity(dir, placed.Files); err != nil || !ok {
			return false, fmt.Errorf("%w: %s is not the files directory this %s placed (%v)", errLandingChanged, dir, ir.src.Name, err)
		}
		wantEnv := placed.Env != ""
		if wantEnv {
			if ok, err := sameIdentity(envPath, placed.Env); err != nil || !ok {
				return false, fmt.Errorf("%w: %s is not the env file this %s placed (%v)", errLandingChanged, envPath, ir.src.Name, err)
			}
		}
		rows, err := ir.sealedTimers(ctx, seal, src.TimersSHA256)
		if err != nil {
			return false, err
		}
		vs, err := ir.versioned(ctx)
		if err != nil {
			return false, err
		}
		// Every write re-checks the authority first (installAuthority): a
		// stale build never wins the create-only manifest.
		authority := installAuthorityStore{VersionedStore: vs, check: ir.installAuthority(ctx, captured, phase)}
		pm, err := buildPreparedManifest(ctx, authority, cfg.OwnBucket, holder.Location, gen, id, captured.Stamp, dir, envPath, wantEnv, hostfiles.WakeStateOf(host, rows, src.NextDue, src.Replication))
		if errors.Is(err, hostfiles.ErrManifestExists) {
			log.Printf("moves: %s: the prepared files manifest of %s is not this install's: %v", host, id, err)
			return ir.refuse(ctx, refuseManifestOccupied)
		}
		if err != nil {
			return false, fmt.Errorf("build the prepared files manifest: %w", err)
		}
		raw, err := json.Marshal(pm)
		if err != nil {
			return false, err
		}
		if err := mark(landStepFilesManifest, string(raw)); err != nil {
			return false, err
		}
		log.Printf("moves: %s: the prepared files manifest %s is written (sha256 %s)", host, pm.Key, pm.SHA256)
	}
	if err := ir.fault("before-files-gen"); err != nil {
		return false, err
	}
	marker, err := json.Marshal(filesGenMarker{Op: gen, Incarnation: holder.Incarnation})
	if err != nil {
		return false, err
	}
	placed, err := journalledStaged(j, landStepFilesPlaced)
	if err != nil {
		return false, err
	}
	// The marker goes into the files directory this install placed, and
	// only while the installation is still this install's (final review,
	// finding 2): inside the guard, the directory is the receipt's.
	if err := ir.guarded(ctx, captured, phase, func() error {
		if ok, err := sameIdentity(dir, placed.Files); err != nil || !ok {
			return fmt.Errorf("%w: %s is not the files directory this %s placed (%v)", errLandingChanged, dir, ir.src.Name, err)
		}
		if err := syncPath(dir); err != nil {
			return err
		}
		return writeFileAtomic(filepath.Join(dir, archive.MarkerName), marker, 0o644)
	}); err != nil {
		return false, fmt.Errorf("write %s: %w", archive.MarkerName, err)
	}
	if err := mark(landStepFilesGen, ""); err != nil {
		return false, err
	}
	log.Printf("moves: %s: files, environment, timers and columns installed; %s written", host, archive.MarkerName)
	return ir.advance(ctx, phase, next, installStamp{})
}

// installArtifactPaths are an installation's artifacts at the flat paths
// (G11), relative to the data directory: the database with its marker,
// journal, quarantine and litestream shadow, the files directory and
// <host>.env. A dirty return parks them before it installs; the source's
// finalization parks them and the rendered config beside them.
func installArtifactPaths(host string) []string {
	db := host + ".db"
	return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", host, host + ".env"}
}

// retirePreMove is a dirty return's local sub-step (2) (ruling J13; fix
// round 1): the source's own pre-move installation, and anything an earlier
// installation of this holder left on this box, leaves the flat paths before
// anything of the return's is placed there, so placeStaged and restore find
// them empty and never install over them, and the return's registration
// never meets a path tagged with another stamp.
//
// First the claim's inventory for this box, read fresh each round, is
// accounted for stamp by stamp, before anything moves:
//   - S, the source's installation stamp (the holder's after
//     ReturnToSource): retired through park.Local under the host guard
//     against the park-local the source's claimed queued
//     (RetireArtifacts completes it); paths of S listed with no park-local
//     owed is an error, since only the claimed CAS queues it;
//   - a stamp the claim shows superseded on this box (supersededOnBox: a
//     cold admission here issued it, and it is neither S nor the holder's
//     live session; spec §7.8, rev 7.11): retired the same way, against a
//     park-local queued here for it, so its inventory entries clear;
//   - any other stamp (live, unknown, another holder's): nothing is moved,
//     and the return stops for an operator (StuckReturnUnknownStamp); a
//     retry after the operator has accounted for it runs this again.
//
// A park-local owed with nothing listed (a replay after a retirement) is
// completed. Then every pre-move artifact still at the flat paths that no
// inventory entry claims is parked by the moves-owned journalled rename
// (H5(b), parkLocalArtifacts), its plan and its renames inside the host
// guard with the claim and the record read fresh there (fix round 1, ruling
// 1), so a call that resumes after authority moved on renames nothing.
func (ir *installRun) retirePreMove(ctx context.Context, captured hostguard.Captured, j *moves.Journal, mark func(step, value string) error) (restoreOutcome, error) {
	cfg := ir.cfg()
	host, id, phase := ir.rec.Host, ir.g.ID, ir.src.Phases[3]
	stamp := captured.Stamp
	for round := 0; ; round++ {
		if round > 64 {
			return restoreWaiting, fmt.Errorf("move %s: the source's inventory did not clear", id)
		}
		obs, err := ir.sync(ctx, phase)
		if err != nil {
			return restoreWaiting, err
		}
		c := obs.Claim
		var retire []uint64
		for _, x := range inventoryStamps(c, cfg.Box) {
			switch {
			case x == stamp:
				retire = append([]uint64{x}, retire...)
			case supersededOnBox(c, cfg.Box, x):
				retire = append(retire, x)
			default:
				log.Printf("moves: %s: this box's inventory lists paths under stamp %d, which the return of %s cannot account for; nothing is moved", host, x, id)
				ok, err := ir.refuse(ctx, refuseUnknownStamp)
				if err != nil || !ok {
					return restoreWaiting, err
				}
				return restoreMarked, nil
			}
		}
		var chore *hostclaim.Obligation
		for i := range c.Obligations {
			if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
				chore = o
			}
		}
		if len(retire) == 0 {
			if chore != nil {
				if _, err := hostclaim.Update(ctx, ir.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, stamp)); err != nil {
					return restoreWaiting, fmt.Errorf("complete the source's local park: %w", err)
				}
			}
			break
		}
		x := retire[0]
		if chore == nil {
			if x == stamp {
				return restoreWaiting, fmt.Errorf("move %s: this box's inventory lists the source's pre-move paths at stamp %d, and the claim owes no park-local for them", id, stamp)
			}
			if obs, err = hostclaim.Update(ctx, ir.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil {
				return restoreWaiting, fmt.Errorf("queue the local park of stamp %d: %w", x, err)
			}
			for i := range obs.Claim.Obligations {
				if o := &obs.Claim.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
					chore = o
				}
			}
			if chore == nil {
				return restoreWaiting, fmt.Errorf("move %s: the local park of stamp %d was not queued", id, x)
			}
		}
		if err := ir.local.RetireArtifacts(ctx, host, captured, chore.ID, id, x); err != nil {
			return restoreWaiting, fmt.Errorf("retire the source's artifacts of stamp %d: %w", x, err)
		}
		log.Printf("moves: %s: the source's artifacts of stamp %d are retired", host, x)
	}
	if err := ir.fault("before-" + installStepPreMove); err != nil {
		return restoreWaiting, err
	}
	paths := installArtifactPaths(host)
	unclaimed := func(c *hostclaim.Claim) error {
		for _, p := range paths {
			if _, ok := c.Inventory[cfg.Box].Paths[p]; ok {
				return fmt.Errorf("%w: the inventory claims %s", errLandingChanged, p)
			}
		}
		return nil
	}
	err := ir.guardedWith(ctx, captured, phase, unclaimed, func() error {
		_, err := parkLocalArtifacts(ctx, j, mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: captured.Incarnation, Host: host,
			Op: ir.src.GenerationOp, Step: installStepPreMove, Paths: paths, NewID: obligationID, Fault: ir.fault})
		return err
	})
	if err != nil {
		return restoreWaiting, fmt.Errorf("park the source's pre-move artifacts: %w", err)
	}
	return restoreInstalled, nil
}

// inventoryStamps are the distinct stamps box's inventory tags paths with,
// in order.
func inventoryStamps(c *hostclaim.Claim, box string) []uint64 {
	seen := map[uint64]bool{}
	var out []uint64
	for _, s := range c.Inventory[box].Paths {
		if !seen[s] {
			seen[s] = true
			out = append(out, s)
		}
	}
	sort.Slice(out, func(i, k int) bool { return out[i] < out[k] })
	return out
}

// supersededOnBox reports whether the claim shows stamp as an installation
// stamp this holder issued on box and has since moved past: a cold
// admission's stamp is its record's session (hostclaim.AdmitWriter), so a
// cold admission record on box with that session names it, and it is
// neither the holder's stamp nor its live writer session. A stamp the claim
// has no such record of is not this box's to retire.
func supersededOnBox(c *hostclaim.Claim, box string, stamp uint64) bool {
	h := c.Holder
	if h == nil || stamp == 0 || stamp == h.Stamp || stamp == h.Session {
		return false
	}
	for _, a := range h.Admissions {
		if a.Mode == "cold" && a.Box == box && a.Session == stamp {
			return true
		}
	}
	return false
}

// releaseLive is the configured step's release (§9.2 as amended
// 2026-10-12): ReleaseLive, else, with CARLOS_HOST_CLAIM off, ReleaseReady,
// which publishes then. Under the claim T7's hook only stages, so it is
// never the fallback: nil, and configured waits.
func (c moveCfg) releaseLive() func(context.Context, release.Store, provision.Record, string) moves.Check {
	switch {
	case c.ReleaseLive != nil:
		return c.ReleaseLive
	case hostwrite.Enabled(os.Getenv):
		return nil
	}
	return c.ReleaseReady
}

// configured (§7.6, §9.1, §9.2): the held row rendered (RenderHeld, the
// move-bound ack). A landing requires the render's bundle and overlay etags
// to be the ones pinned at readiness (else the refusal, "configuration
// changed during the move") and the pinned release live and proven for the
// held row through releaseLive (ruling G4: "waiting for the root stage
// pass" waits, anything else waits until the landing's deadline refuses).
// Under the claim (Paul's T7 ruling, 2026-10-12) the render is selected as
// held-config and the release is published here, under held-release: T7
// only staged it. A
// return renders the current inputs instead, never compared with
// move.inputs, gates on the release live on this box, and records what it
// rendered as Rollback.RecoveryInputs in the CAS that ends the phase
// (J2-R2-5). Then the install the brief lists under `restored`: the built
// database's sha256 against the manifest, lease.CheckDB, the rename into
// DBPath (G34), <dbPath>.gen through hostbox's marker writer naming the
// generation (ruling G2), and the CAS to restored.
//
// A replay finds configured and does it all again: the render and the
// release check are idempotent reads and writes of the same bytes; a
// database already at DBPath with the manifest's sha256 is this install's
// own, so only the marker is written again.
func (ir *installRun) configured(ctx context.Context) (bool, error) {
	cfg := ir.cfg()
	host, id := ir.rec.Host, ir.g.ID
	phase, next := ir.src.Phases[4], ir.src.Phases[5]
	obs, err := ir.sync(ctx, phase)
	if err != nil {
		return false, err
	}
	if obs.Claim.Op.Kind == hostclaim.KindLanding && cfg.registerLanding != nil {
		if err := cfg.registerLanding(ctx, obs); err != nil {
			return false, err
		}
	}
	var pin, recovery *provision.MoveInputs
	if ir.src.Role == roleTarget {
		if pin = ir.src.PinInputs; pin == nil {
			return false, fmt.Errorf("move %s: the record has no pinned inputs", id)
		}
	}
	rt, err := ir.route()
	if err != nil {
		return false, err
	}
	// The render's ownership is journalled before its write (carry (c),
	// fix round 1, I1): the hook runs once the rendered bytes are known and
	// before they are placed, and journals their sha256 with the file's
	// state before the render (pre_render). The rendered sha256 is journalled
	// again as soon as the render answers, even when its config ack failed.
	// A rollback's cleaning parks <host>.config.env only when its content is
	// one this op rendered (rollbackOwnership).
	//
	// Under the claim the render is selected as held-config, and the pins
	// decide first (Task 7, Q2.2): a merge that is not them is the
	// configuration-changed refusal, with nothing selected or written.
	hr, err := cfg.RenderHeld(ctx, ir.s, rt, ir.src.HoldOp, pin, ir.notePreRender)
	if hr.SHA256 != "" {
		if jerr := ir.noteRendered(hr.SHA256); jerr != nil {
			return false, errors.Join(err, jerr)
		}
	}
	if pin != nil && errors.Is(err, errHeldRenderUnpinned) {
		log.Printf("moves: %s: the held render's inputs are not the pinned ones (%v, overlay %q): %v", host, pin.BundleETags, pin.OverlayETag, err)
		return ir.refuse(ctx, refuseConfigChanged)
	}
	if err != nil {
		return false, fmt.Errorf("render the held row: %w", err)
	}
	want := ""
	if pin != nil {
		if !sameBundle(hr.BundleETags, pin.BundleETags) || hr.OverlayETag != pin.OverlayETag {
			log.Printf("moves: %s: the render's etags (%v, overlay %q) are not the pinned ones (%v, overlay %q)", host, hr.BundleETags, hr.OverlayETag, pin.BundleETags, pin.OverlayETag)
			return ir.refuse(ctx, refuseConfigChanged)
		}
		want = pin.Release
	} else {
		live, err := release.LiveVersion(cfg.OptDir, host)
		if err != nil {
			return false, fmt.Errorf("read the live release: %w", err)
		}
		bundle := make(map[string]string, len(hr.BundleETags))
		for k, v := range hr.BundleETags {
			bundle[k] = v
		}
		recovery = &provision.MoveInputs{BundleETags: bundle, OverlayETag: hr.OverlayETag, Release: live}
		want = live
	}
	live := cfg.releaseLive()
	if live == nil {
		ir.w.notice(host+"#release", "moves: %s: release %s cannot be published for the held row: no held-release publication is wired under CARLOS_HOST_CLAIM; retrying", host, want)
		return false, nil
	}
	if c := live(ctx, ir.s, ir.rec, want); !c.OK {
		ir.w.notice(host+"#release", "moves: %s: release %s is not live and proven for the held row yet (%s); retrying", host, want, c.Reason)
		return false, nil
	}
	if err := ir.fault("after-release-ready"); err != nil {
		return false, err
	}
	// The render, the release gate and the install can outlast a landing's
	// deadline (fix round 1, item 4): it is checked again before the install
	// and right before the transition that names it complete.
	if ir.pastDeadline() {
		return ir.refuse(ctx, refuseDeadline)
	}
	sha, outcome, err := ir.restore(ctx, obs)
	if err != nil {
		return false, err
	}
	switch outcome {
	case restoreWaiting:
		return false, nil
	case restoreMarked:
		return true, nil // the record changed: the refusal, not an install
	}
	if err := ir.registerDatabase(ctx, obs, phase); err != nil {
		return false, err
	}
	if ir.pastDeadline() {
		return ir.refuse(ctx, refuseDeadline)
	}
	return ir.advance(ctx, phase, next, installStamp{InstalledDBSHA256: sha, RecoveryInputs: recovery})
}

// registerDatabase registers the installed database and its marker under F
// (carry (b)), once restore has proved the database at DBPath is this
// install's build: renamed into place inside the guard, or found there
// with the build's sha256. Journalled, so a replay registers once.
func (ir *installRun) registerDatabase(ctx context.Context, obs hostclaim.Observed, phase string) error {
	j, err := ir.journal()
	if err != nil {
		return err
	}
	if j.Done(landStepInventoryDB) {
		return nil
	}
	captured, err := ir.capture(*obs.Claim.Holder)
	if err != nil {
		return err
	}
	host := ir.rec.Host
	if err := ir.local.RecordInventory(ctx, host, captured, "flat", []string{host + ".db", host + ".db.gen"}, ir.src.ClaimCheck(phase)); err != nil {
		return fmt.Errorf("register the installed database: %w", err)
	}
	j.Phase = phase
	j.Mark(landStepInventoryDB, "")
	if err := j.Commit(ir.cfg().DataDir); err != nil {
		return err
	}
	return ir.fault("after-" + landStepInventoryDB)
}

// configEnvPath is the held render's file, <DataDir>/<host>.config.env.
func (ir *installRun) configEnvPath() string {
	return filepath.Join(ir.cfg().DataDir, ir.rec.Host+".config.env")
}

// preRender is the landStepPreRender record: <host>.config.env's state
// before the first held render ("absent", "sha256:<hex>" of a regular file,
// or "other"), and the intended set: the sha256 of each content a render of
// this op may have placed, journalled before its write. The set is pruned
// to the file on disk (fix round 4): before each attempt it keeps the hash
// of the content now on disk, only when it is already one of the set's, and
// adds the new content's. So it never holds more than two hashes, and every
// content this op placed that can still be on disk stays named.
type preRender struct {
	Before   string   `json:"before"`
	Intended []string `json:"intended"`
}

// notePreRender is the held render's before-write hook. It hashes the file
// now on disk (Lstat'ed, never followed) and journals the intended set as
// {that hash, if the set already names it} plus {the new content's hash},
// before the write. A file whose content the set does not name (the earlier
// state, or foreign content) contributes nothing, and the earlier-state
// record is kept as first seen. The renderer replaces the file by an atomic
// rename (appconfig's renderExecConfigBefore), so the file on disk is always
// one whole content: a hash pruned here is never on disk again unless this
// op renders it again, when it is journalled again first.
func (ir *installRun) notePreRender(sha string) error {
	if sha == "" {
		return errors.New("moves: the held render names no content")
	}
	j, err := ir.journal()
	if err != nil {
		return err
	}
	path := ir.configEnvPath()
	disk := ""
	fi, err := os.Lstat(path)
	switch {
	case errors.Is(err, os.ErrNotExist):
	case err != nil:
		return err
	case fi.Mode().IsRegular():
		if disk, err = fileSHA256(path); err != nil {
			return err
		}
	}
	var pre preRender
	if raw, ok := j.Steps[landStepPreRender]; ok {
		if err := json.Unmarshal([]byte(raw), &pre); err != nil {
			return fmt.Errorf("the journal's %s step: %w", landStepPreRender, err)
		}
	} else {
		pre.Before = "other"
		switch {
		case errors.Is(err, os.ErrNotExist) || fi == nil:
			pre.Before = "absent"
		case disk != "":
			pre.Before = disk
		}
	}
	want := "sha256:" + sha
	var next []string
	if disk != "" && containsString(pre.Intended, disk) {
		next = append(next, disk)
	}
	if want != disk || len(next) == 0 {
		next = append(next, want)
	}
	if strings.Join(next, " ") == strings.Join(pre.Intended, " ") {
		return nil
	}
	pre.Intended = next
	body, err := json.Marshal(pre)
	if err != nil {
		return err
	}
	j.Mark(landStepPreRender, string(body))
	if err := j.Commit(ir.cfg().DataDir); err != nil {
		return err
	}
	return ir.fault("after-" + landStepPreRender)
}

// noteRendered journals the sha256 (hex) of the file the held render wrote,
// as "sha256:<hex>": the latest render's, which is what the file holds.
func (ir *installRun) noteRendered(sha string) error {
	j, err := ir.journal()
	if err != nil {
		return err
	}
	want := "sha256:" + sha
	if j.Steps[landStepRendered] == want {
		return nil
	}
	j.Mark(landStepRendered, want)
	if err := j.Commit(ir.cfg().DataDir); err != nil {
		return err
	}
	return ir.fault("after-" + landStepRendered)
}

// recoveryInstallSource binds the installer to a dirty return on the source
// (§7.8 source table, the dirty branch; rulings J13, J14, J20): the record
// rolling_back and dirty with the source at an install phase; the claim
// restoring/claimed for this rollback with this box, the move's source,
// executing at the flip's journalled source stamp S; the generation and the
// hold named by the rollback id; the return lane moves/<host>/<rb id>/ in
// this box's own bucket; the refusal MoveMarkStuck; the phase advance
// MoveRollbackSourcePhase, stamping RecoveryInputs at the end of configured.
// seal reads the return seal from the claim the phase works from, and
// baseline is what it restores to (the return's evidence): both are the
// caller's (Plan 4 Tasks 4 and 5). Production does not call it yet.
func recoveryInstallSource(w *movesWorker, s release.Store, rec provision.Record, seal func(*hostclaim.Claim) (moves.Seal, error), baseline *provision.MoveSource) (installSource, error) {
	cfg := w.cfg
	m := rec.Move
	if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || !m.Rollback.Dirty || !moves.ValidOpID(m.Rollback.ID) ||
		!strings.HasPrefix(m.Rollback.ID, "rb_") {
		return installSource{}, errors.New("moves: only a dirty rollback's source installs a return")
	}
	if cfg.Box == "" || seal == nil || baseline == nil {
		return installSource{}, errors.New("moves: a return's install names this box, its seal and its baseline")
	}
	g := provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}
	host, rb, box := rec.Host, m.Rollback.ID, cfg.Box
	claimCheck := func(string) func(*hostclaim.Claim) error {
		return func(c *hostclaim.Claim) error {
			if err := hostclaim.RequireMoveOp(g.ID, hostclaim.KindRestoring, hostclaim.PhaseClaimed)(c); err != nil {
				return err
			}
			if c.Op.Exec != box || len(c.Op.Participants) == 0 || c.Op.Participants[0] != box || c.Op.Intent["source"] != box {
				return fmt.Errorf("%w: the executor is %q, not this box as the move's source", hostclaim.ErrWrongState, c.Op.Exec)
			}
			if c.Op.Intent["rollback"] != rb {
				return fmt.Errorf("%w: the op is rollback %q, not %s", hostclaim.ErrWrongState, c.Op.Intent["rollback"], rb)
			}
			stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
			if err != nil || stamp == 0 || c.Holder == nil || c.Holder.Stamp != stamp {
				return fmt.Errorf("%w: the holder is not at the source's pre-move stamp", hostclaim.ErrWrongState)
			}
			return nil
		}
	}
	return installSource{
		OpID: g.ID, Host: host, Role: roleSource, Name: "return",
		Lane: moves.StagingPrefix(host, rb), Seal: seal, Baseline: baseline,
		GenerationOp: rb, HoldOp: rb,
		Phases: [6]string{provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished,
			provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored},
		Phase: func(from, to string, st installStamp) func(*provision.Record) error {
			var stamp func(*provision.MoveRollback)
			if st.RecoveryInputs != nil {
				in := st.RecoveryInputs
				stamp = func(r *provision.MoveRollback) { r.RecoveryInputs = in }
			}
			return provision.MoveRollbackSourcePhase(g, from, to, stamp)
		},
		Sync: func(ctx context.Context) (hostclaim.Observed, error) {
			obs, err := hostclaim.Read(ctx, s, 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", g.ID)
			}
			if err := hostclaim.RequireOp(g.ID)(obs.Claim); err != nil {
				return obs, fmt.Errorf("move %s: %w", g.ID, err)
			}
			if h := obs.Claim.Holder; h == nil || h.Incarnation != g.Incarnation {
				return obs, fmt.Errorf("move %s: the host claim names another incarnation", g.ID)
			}
			return obs, nil
		},
		ClaimCheck: claimCheck,
		RecordPhase: func(r *provision.Record) (string, error) {
			m := r.Move
			if m == nil || m.ID != g.ID || m.Incarnation != g.Incarnation || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID != rb || !m.Rollback.Dirty {
				return "", fmt.Errorf("%w: the record is no longer the dirty return of %s", errLandingChanged, rb)
			}
			return m.Rollback.SourcePhase, nil
		},
		Refuse: func(ctx context.Context, why installRefusal) (bool, string, error) {
			var reason string
			switch why {
			case refuseFilesOccupied:
				reason = provision.StuckReturnFilesOccupied
			case refuseManifestOccupied:
				reason = provision.StuckReturnManifestOccupied
			case refuseInstallMismatch:
				reason = provision.StuckReturnInstallMismatch
			case refuseUnknownStamp:
				reason = provision.StuckReturnUnknownStamp
			default:
				return false, "", fmt.Errorf("move %s: a return's install has no refusal %d", g.ID, why)
			}
			var pre provision.Record
			after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, observedMove(&pre, provision.MoveMarkStuck(g, reason)))
			if err != nil {
				if errors.Is(err, provision.ErrMoveConflict) {
					log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, rb, err)
					return false, "", nil
				}
				return false, "", fmt.Errorf("mark the return stuck: %w", err)
			}
			w.logTransition(ctx, s, pre, *after)
			log.Printf("moves: %s: the return %s stops for an operator: %s", host, rb, reason)
			nudgeSerial(ctx, s)
			return true, reason, nil
		},
	}, nil
}