package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// buildStates is the agent's side of settings spec §4.2: the build-state
// object beside each instance's replica, under its lease, and its registry
// copy root reads.
//
// The object sits beside the lease's lock.json (ruling R-CLAIMPREFIX). With
// host claims off that is the hibernate bucket at HibernatePrefix; with them
// on, the claim actor's bucket at the claim holder's Location (never the
// session's g-<gen>/ prefix). A wake that holds the lease says which on its
// context (activator.WithBuildStateLocation), and that wins.
type buildStates struct {
	hibBucket string                   // CARLOS_HIBERNATE_BUCKET
	prefix    func(host string) string // HibernatePrefix, host claims off
	// claims reads a host's claim; nil means host claims are off. With them
	// on, every exec instance wakes through its claim (doWake ->
	// doClaimWake), so every host is located through it.
	claims      func(context.Context, string) (hostclaim.Observed, error)
	claimBucket string // the claim actor's bucket, where claimed wakes lease
	// staged is the legacy move round's staging prefix for host in the
	// hibernate bucket (activator.MovesPrefix). nil unless host claims and
	// Phase 1 are both off (ruling R-STAGEDREAD).
	staged   func(host string) string
	storeFor func(ctx context.Context, bucket string) (release.Store, error)
	reg      *registry.Registry
	optDir   string
	// bakeDir is the data dir every agent and root pass hands release.BakeDir:
	// root's refusal records (release.ReadRefusal) live under it.
	bakeDir string
	now     func() time.Time
	// budget bounds one refreshAll; 0 means refreshBudget.
	budget time.Duration
	// beforePut, when set (tests only), runs in update between its read and
	// its conditional Put.
	beforePut func(ctx context.Context, store release.Store, key string)

	mu    sync.Mutex
	hosts map[string]*sync.Mutex // per-host: a read or update and its column write are one step
	// resume is the first host the last refreshAll did not reach; the next
	// one starts there (ruling R-REFRESH-ROTATE). Guarded by mu.
	resume string
	// written is each object write last put, by bucket and key, for the
	// next write's conditional first try (writtenCopy). Guarded by mu.
	written map[string]writtenCopy
}

// newBuildStates is nil, nil on a box with no replica bucket: it runs no
// hibernating instances, so it has no build state to keep. cfg is the
// activator's, after configureWakeClaims: cfg.HostWrites.Enabled is the
// CARLOS_HOST_CLAIM switch and cfg.Claims its wake runtime.
func newBuildStates(cfg *activator.Config, reg *registry.Registry, optDir string) (*buildStates, error) {
	if cfg == nil || cfg.Bucket == "" {
		return nil, nil
	}
	regions, err := buildStateRegions(cfg)
	if err != nil {
		return nil, err
	}
	b := &buildStates{hibBucket: cfg.Bucket, prefix: func(h string) string { return activator.HibernatePrefix(cfg.Prefix, h) },
		reg: reg, optDir: optDir, bakeDir: dataDir(), now: time.Now}
	if cfg.HostWrites.Enabled {
		b.claims, b.claimBucket = cfg.Claims.Read, cfg.Claims.Bucket
	} else if !moves.Enabled(os.Getenv) {
		// The legacy move round's staging is promoted at the first wake's
		// restore only with Phase 1 off (ExecBackend.Phase1, set from the
		// same switch); with host claims on the state is at the claim
		// holder's location, which a Phase 1 landing fills.
		b.staged = activator.MovesPrefix
	}
	b.storeFor = cachedStores(regions, lease.ExecutionStore)
	return b, nil
}

// buildStateRegions is the one bucket the box leases in, with the region its
// leases sign for. With host claims on every wake is a claimed wake
// (doWake -> doClaimWake) and leases in the claim actor's bucket, so only
// that bucket is registered: the hibernate bucket is never located there,
// even when it is the same bucket under another region setting.
func buildStateRegions(cfg *activator.Config) (map[string]string, error) {
	if !cfg.HostWrites.Enabled {
		return map[string]string{cfg.Bucket: cfg.Region}, nil
	}
	if cfg.Claims == nil || cfg.Claims.Read == nil || cfg.Claims.Bucket == "" {
		return nil, errors.New("host claims are on but the claim reader or the claim bucket is missing")
	}
	return map[string]string{cfg.Claims.Bucket: cfg.Claims.LeaseRegion}, nil
}

// cachedStores opens one store per bucket the box leases in, with the
// lease's own S3 configuration (lease.ExecutionStore: the region the lock
// was signed for), and keeps it. Any other bucket is refused.
func cachedStores(regions map[string]string, open func(ctx context.Context, bucket, region string) (release.Store, error)) func(context.Context, string) (release.Store, error) {
	var mu sync.Mutex
	stores := map[string]release.Store{}
	return func(ctx context.Context, bucket string) (release.Store, error) {
		region, ok := regions[bucket]
		if !ok {
			return nil, fmt.Errorf("bucket %q is not one this box leases in", bucket)
		}
		mu.Lock()
		defer mu.Unlock()
		if s := stores[bucket]; s != nil {
			return s, nil
		}
		s, err := open(ctx, bucket, region)
		if err != nil {
			return nil, err
		}
		stores[bucket] = s
		return s, nil
	}
}

// locate is the store and prefix of host's build state: the context's
// location when a wake set one; otherwise, with host claims on, the claim
// holder's Location in the claim bucket (an unreadable claim is an error,
// never the hibernate prefix); otherwise the hibernate bucket and prefix.
func (b *buildStates) locate(ctx context.Context, host string) (release.Store, string, error) {
	_, s, prefix, err := b.locateIn(ctx, host)
	return s, prefix, err
}

// locateIn is locate with the bucket it chose, which keys write's copy of
// the object it last wrote.
func (b *buildStates) locateIn(ctx context.Context, host string) (string, release.Store, string, error) {
	bucket, prefix := b.hibBucket, ""
	if l, ok := activator.BuildStateLocationFrom(ctx); ok {
		bucket, prefix = l.Bucket, l.Prefix
	} else if b.claims != nil {
		obs, err := b.claims(ctx, host)
		if err != nil {
			return "", nil, "", fmt.Errorf("read claim: %w", err)
		}
		if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.Claim.Holder.Location == "" {
			return "", nil, "", fmt.Errorf("claim of %s has no holder location", host)
		}
		bucket, prefix = b.claimBucket, obs.Claim.Holder.Location
	} else {
		prefix = b.prefix(host)
	}
	prefix = strings.TrimSuffix(prefix, "/")
	s, err := b.storeFor(ctx, bucket)
	if err != nil {
		return "", nil, "", err
	}
	return bucket, s, prefix, nil
}

func (b *buildStates) hostLock(host string) *sync.Mutex {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.hosts == nil {
		b.hosts = map[string]*sync.Mutex{}
	}
	m := b.hosts[host]
	if m == nil {
		m = &sync.Mutex{}
		b.hosts[host] = m
	}
	return m
}

// getState reads the object at key: nil and "" when there is none, an error
// when it does not parse or names another host. That error names the key:
// update never overwrites such an object (ruling R-BS-CORRUPT), so every
// spawn on the host is refused until an operator repairs or deletes it.
func getState(ctx context.Context, store release.Store, key, host string) (*release.BuildState, string, error) {
	raw, etag, err := store.Get(ctx, key)
	switch {
	case errors.Is(err, os.ErrNotExist):
		return nil, "", nil
	case err != nil:
		return nil, "", err
	}
	var s release.BuildState
	if err := json.Unmarshal(raw, &s); err != nil || s.Host != host {
		return nil, "", fmt.Errorf("build state %s for %s does not parse or names another host; repair or delete that object", key, host)
	}
	return &s, etag, nil
}

// hasReplica is whether prefix holds any object besides the lease's
// lock.json and build-state.json. Three keys decide it, so a store with a
// bounded listing (release.PageLister) is asked for three; any other lists
// the prefix whole.
func hasReplica(ctx context.Context, store release.Store, prefix string) (bool, error) {
	var keys []string
	var err error
	if pl, ok := store.(release.PageLister); ok {
		keys, err = pl.ListPage(ctx, prefix+"/", "", 3)
	} else {
		keys, err = store.List(ctx, prefix+"/")
	}
	if err != nil {
		return false, err
	}
	for _, k := range keys {
		rel := strings.TrimPrefix(k, prefix+"/")
		if rel != "lock.json" && rel != "build-state.json" {
			return true, nil
		}
	}
	return false, nil
}

// read returns host's build state (nil when there is none) and whether its
// replica has any object besides the lease's lock.json and build-state.json.
//
// A legacy move's staging, when there is one, is read in place of the own
// prefix (settings spec rev 11 §4.3a): between the cut and the first wake
// the replica and the build state are only there, and the wake's restore
// promotes the staging over the own prefix whenever it holds anything
// (promoteStagedReplica: staging wins, and its mirror deletes what the
// staging lacks). So the staging is what the instance's state will be. A
// held instance must not read as new in that window, or root installs the
// head. update still writes only the own prefix: the wake promotes first.
//
// When both hold a state, an own state with a strictly later At wins (a
// tie is the staging's, as at restore): a wake that promoted the
// staging and then failed to drop it (dropStagedReplica is best-effort)
// leaves a pre-move staged state behind its own newer record, and that
// record is the instance's state.
func (b *buildStates) read(ctx context.Context, host string) (*release.BuildState, bool, error) {
	staged, stagedReplica, ok, err := b.readStaged(ctx, host)
	if err != nil {
		return nil, false, err
	}
	if ok && staged == nil {
		return nil, stagedReplica, nil
	}
	st, replica, err := b.readOwn(ctx, host)
	if err != nil {
		return nil, false, err
	}
	if ok && (st == nil || !st.At.After(staged.At)) {
		return staged, stagedReplica || replica, nil
	}
	return st, replica || (ok && stagedReplica), nil
}

// readOwn reads host's build state and replica at its own location.
func (b *buildStates) readOwn(ctx context.Context, host string) (*release.BuildState, bool, error) {
	store, prefix, err := b.locate(ctx, host)
	if err != nil {
		return nil, false, err
	}
	st, _, err := getState(ctx, store, release.BuildStateKey(prefix), host)
	if err != nil {
		return nil, false, err
	}
	replica, err := hasReplica(ctx, store, prefix)
	if err != nil {
		return nil, false, err
	}
	return st, replica, nil
}

// readStaged reads a legacy move's staging for host (ruling R-STAGEDREAD):
// ok is true when moves/<host>/ in the hibernate bucket holds any key of
// the legacy round's, the same "anything staged" promoteStagedReplica
// promotes on. Phase 1 op lanes (moves/<host>/<mv_ or rb_ id>/...) are not
// the legacy round's and are skipped. The staged build-state.json is the
// state, and any other staged object but lock.json is a replica. Off (ok
// false) unless staged is set: host claims and Phase 1 both off.
func (b *buildStates) readStaged(ctx context.Context, host string) (*release.BuildState, bool, bool, error) {
	if b.staged == nil {
		return nil, false, false, nil
	}
	prefix := strings.TrimSuffix(b.staged(host), "/")
	store, err := b.storeFor(ctx, b.hibBucket)
	if err != nil {
		return nil, false, false, err
	}
	keys, err := store.List(ctx, prefix+"/")
	if err != nil {
		return nil, false, false, fmt.Errorf("list the staged move %s: %w", prefix, err)
	}
	var staged, state, replica bool
	for _, k := range keys {
		rel := strings.TrimPrefix(k, prefix+"/")
		if first, _, ok := strings.Cut(rel, "/"); ok && moves.ValidOpID(first) {
			continue
		}
		staged = true
		switch rel {
		case "build-state.json":
			state = true
		case "lock.json":
		default:
			replica = true
		}
	}
	if !staged {
		return nil, false, false, nil
	}
	var st *release.BuildState
	if state {
		if st, _, err = getState(ctx, store, release.BuildStateKey(prefix), host); err != nil {
			return nil, false, false, err
		}
	}
	return st, replica, true, nil
}

// refresh stores what read found. A failed read keeps the last known state
// for root (a known copy is not wiped by a blip; an unread one stays
// unknown) and records the failure in ReadErr, so the box's report says this
// host's build state is incomplete and the housekeeper prunes nothing for its
// app (Astra I1 #6). A good read clears ReadErr.
func (b *buildStates) refresh(ctx context.Context, host string) error {
	m := b.hostLock(host)
	m.Lock()
	defer m.Unlock()
	st, replica, err := b.read(ctx, host)
	if err != nil {
		rt, gerr := b.reg.Get(host)
		prev := release.BuildStateCopy{}
		if gerr == nil && rt != nil {
			prev, _ = release.ParseBuildStateCopy(rt.BuildState)
		}
		prev.ReadErr = err.Error()
		if serr := b.reg.SetBuildState(host, prev.Column()); serr != nil {
			return errors.Join(err, serr)
		}
		return err
	}
	c := release.BuildStateCopy{Known: true, Replica: replica, State: st, ReadAt: b.now().UTC().Truncate(time.Second)}
	return b.reg.SetBuildState(host, c.Column())
}

// keepsBuildState is a row whose spawns run under a lease in the replica
// bucket, so the store keeps a build state beside it (ruling R-WORKER-ROWS):
// an activator-managed exec instance (activator.ActivatorManaged: Kind
// instance, Hibernate, a socket, no Backing, no Tunnel), or a standalone
// worker (Kind "sidecar"; every worker wake takes the lease at
// HibernatePrefix(its host), activator.go doWake step 1, whatever its
// Hibernate flag), exec-backed and not tunnel-served. Never a named unit or
// the platform's own row. Settings apply only to Kind instance; that is
// root's concern, not this store's.
//
// With host claims on (claims) a worker never leases: the claim runtime
// refuses sidecar rows before any lease (claimwakeassembly.go
// Prerequisites), so it has no build state and no claim to locate one by.
func keepsBuildState(rt registry.Route, claims bool) bool {
	if rt.Unit != "" || rt.Platform {
		return false
	}
	if rt.IsSidecar() {
		return !claims && rt.Backing == "" && rt.Tunnel == ""
	}
	return activator.ActivatorManaged(rt)
}

// refreshBudget bounds one refreshAll: it runs inline in the reconcile legs,
// before the adopt pass. A host it does not reach keeps its last copy; a
// host whose read the deadline cuts records the failure like any other.
const refreshBudget = 20 * time.Second

// refreshAll runs after every reconcile, for every row keepsBuildState
// names, within the budget. Each host is located on its own: a location a
// caller's context carries names one wake's lease, so it is masked here. It
// returns how many hosts it did not reach.
//
// The hosts are taken in host order starting at the first one the last
// call did not reach, wrapping round (ruling R-REFRESH-ROTATE): a box whose
// budget runs out every pass reaches every host in turn, rather than
// leaving the same tail unread (and its app unprunable) forever.
func (b *buildStates) refreshAll(ctx context.Context) int { return b.refreshKept(ctx, nil) }

// refreshKept is refreshAll over the rows keep accepts (nil: every row): the
// scoped pass refreshes only the hosts of the apps that moved (S3
// change-index design §4.5). A filtered call neither reads nor moves the
// rotation point, which stays the full refresh's.
func (b *buildStates) refreshKept(ctx context.Context, keep func(registry.Route) bool) int {
	budget := b.budget
	if budget <= 0 {
		budget = refreshBudget
	}
	ctx, cancel := context.WithTimeout(activator.WithBuildStateLocation(ctx, activator.BuildStateLocation{}), budget)
	defer cancel()
	routes, err := b.reg.List()
	if err != nil {
		log.Printf("agent: build state refresh: %v", err)
		return 0
	}
	var hosts []string
	for _, rt := range routes {
		if keepsBuildState(rt, b.claims != nil) && (keep == nil || keep(rt)) {
			hosts = append(hosts, rt.Host)
		}
	}
	sort.Strings(hosts)
	start := 0
	if keep == nil {
		b.mu.Lock()
		start = sort.SearchStrings(hosts, b.resume)
		b.mu.Unlock()
	}
	if start >= len(hosts) {
		start = 0
	}
	hosts = append(hosts[start:len(hosts):len(hosts)], hosts[:start]...)
	for i, host := range hosts {
		if ctx.Err() != nil {
			if keep == nil {
				b.mu.Lock()
				b.resume = host
				b.mu.Unlock()
			}
			skipped := len(hosts) - i
			log.Printf("agent: build state refresh: the %s budget ran out; %d of %d hosts not reached keep their last copy", budget, skipped, len(hosts))
			return skipped
		}
		if err := b.refresh(ctx, host); err != nil {
			log.Printf("agent: build state of %s: %v", host, err)
		}
	}
	return 0
}

// updateTries bounds update's compare-and-swap loop.
const updateTries = 3

// update is the only writer of the object (ruling R-BS-UPDATE). Callers hold
// the host's lease. Under the host's lock it reads the object (absent is a
// nil cur and a create-only Put), merges, and Puts against the etag it read,
// so a write that landed in between is never overwritten: a CAS conflict
// re-reads and merges again, up to updateTries times. The registry copy is
// set under the same lock, so a concurrent refresh cannot put older state
// over it. A failed update writes nothing the caller may run on.
func (b *buildStates) update(ctx context.Context, host string, merge func(cur *release.BuildState) release.BuildState) (release.BuildState, error) {
	return b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) { return merge(cur), true, nil }, nil)
}

// write is update with a merge that may refuse (an error: nothing is
// written) or find nothing to write (false: the object and the registry copy
// are left as they are, and the state as read is returned). replica says
// whether the instance has a replica, for the registry copy; nil is yes (a
// process is about to run against the database).
func (b *buildStates) write(ctx context.Context, host string, merge func(cur *release.BuildState) (release.BuildState, bool, error),
	replica func(ctx context.Context, store release.Store, prefix string) (bool, error)) (release.BuildState, error) {
	m := b.hostLock(host)
	m.Lock()
	defer m.Unlock()
	bucket, store, prefix, err := b.locateIn(ctx, host)
	if err != nil {
		return release.BuildState{}, err
	}
	key := release.BuildStateKey(prefix)
	// The first try goes on the copy this agent last wrote, when it has
	// one: a wake's record is then one conditional write, not a read and a
	// write. The condition is that copy's etag, so the write lands only on
	// the very object the merge saw, and the merge decides exactly what it
	// would on a fresh read. Anything else (the object moved on, the merge
	// refused or had nothing to write on a copy that may be stale) is
	// decided again on a fresh read, as before.
	cached, hasCopy := b.lastWritten(bucket, key)
	tries := updateTries
	if hasCopy {
		tries++ // the copy's try is extra: the fresh reads keep all theirs
	}
	for try := 0; try < tries; try++ {
		var (
			cur  *release.BuildState
			etag string
		)
		fromCopy := hasCopy && try == 0
		if fromCopy {
			cur, etag = cached.state(), cached.etag
		} else if cur, etag, err = getState(ctx, store, key, host); err != nil {
			return release.BuildState{}, err
		}
		next, put, err := merge(cur)
		if fromCopy && (err != nil || !put) {
			continue
		}
		if err != nil {
			return release.BuildState{}, err
		}
		if !put {
			if cur == nil {
				return release.BuildState{}, nil
			}
			return *cur, nil
		}
		next.Schema, next.Host = 1, host
		next.At = b.now().UTC().Truncate(time.Second)
		raw, err := json.MarshalIndent(next, "", "  ")
		if err != nil {
			return release.BuildState{}, err
		}
		if b.beforePut != nil {
			b.beforePut(ctx, store, key)
		}
		newEtag, err := store.Put(ctx, key, raw, etag)
		if errors.Is(err, release.ErrCASConflict) {
			b.forget(bucket, key)
			continue
		} else if err != nil {
			b.forget(bucket, key)
			return release.BuildState{}, err
		}
		b.remember(bucket, key, raw, newEtag)
		// A process about to run against the database has (or is about to
		// have) a replica; any other writer says.
		hasRep := true
		if replica != nil {
			if hasRep, err = replica(ctx, store, prefix); err != nil {
				return next, fmt.Errorf("build state of %s written, its replica not listed: %w", host, err)
			}
		}
		c := release.BuildStateCopy{Known: true, Replica: hasRep, State: &next, ReadAt: b.now().UTC().Truncate(time.Second)}
		if err := b.reg.SetBuildState(host, c.Column()); err != nil {
			return next, fmt.Errorf("build state of %s written, registry copy not: %w", host, err)
		}
		return next, nil
	}
	return release.BuildState{}, fmt.Errorf("build state of %s: %d tries lost the race: %w", host, updateTries, release.ErrCASConflict)
}

// writtenCopy is the object write last put at a key, and the etag that
// write returned: the state the next write's conditional first try goes on.
type writtenCopy struct {
	raw  []byte
	etag string
}

// state decodes a fresh copy each time, so a merge can never change the
// cached bytes.
func (c writtenCopy) state() *release.BuildState {
	var s release.BuildState
	if json.Unmarshal(c.raw, &s) != nil {
		return nil
	}
	return &s
}

func (b *buildStates) lastWritten(bucket, key string) (writtenCopy, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	c, ok := b.written[bucket+"\x00"+key]
	return c, ok && c.etag != "" && c.state() != nil
}

func (b *buildStates) remember(bucket, key string, raw []byte, etag string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if etag == "" {
		delete(b.written, bucket+"\x00"+key)
		return
	}
	if b.written == nil {
		b.written = map[string]writtenCopy{}
	}
	b.written[bucket+"\x00"+key] = writtenCopy{raw: append([]byte(nil), raw...), etag: etag}
}

func (b *buildStates) forget(bucket, key string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	delete(b.written, bucket+"\x00"+key)
}

// beforeExec is Service.BeforeExec (settings spec §4.2): before build runs
// against host's database, under its lease, the object names build as
// last_executed, carries root's resolution from the consumed file, and
// completes the request if build is its target (release.NextBuildState).
// That write is what completes a request: at first run, not at install.
//
// It fails closed. A consumed file the agent cannot read refuses the exec:
// a write without it could miss a resolution and leave R uncompleted after
// its T ran. update sets the registry copy itself, and refuses an object it
// cannot parse rather than overwrite it (ruling R-BS-CORRUPT). For a row
// settings apply to, a state that now names another held target than build
// (release.HeldExecStale) refuses it too: the wake decided on a stale copy.
// The one exception is the zero-downtime planner's checked selection of
// build in place of that target, which the activator carries on ctx for this
// exec only (activator.PlannedSelection; ruling R-PLANNED-RECOVERY): it is
// admitted, and still recorded. A consumed file another tenant of the
// hostname left (Consumed.ForeignTo) is read as none (ruling R26).
func (b *buildStates) beforeExec(ctx context.Context, host, build string) error {
	if build == "" {
		return fmt.Errorf("no build to record for %s", host)
	}
	consumed, err := release.ReadConsumed(b.optDir, host)
	if err != nil {
		return fmt.Errorf("read root's consumed file for %s: %w", host, err)
	}
	// The held target is revalidated against the state read under the lease
	// (Astra round 1 finding 1): a wake that decided on a stale copy is
	// refused here, before anything is written or run. A route that does
	// not read refuses the exec before any write (Astra r4 finding 3): the
	// tenant check below needs it, and an unread tenant must not merge a
	// hostname-keyed file into the state. A route that is absent (nil, no
	// error) is read as before.
	rt, err := b.reg.Get(host)
	if err != nil {
		return fmt.Errorf("read the route of %s: %w", host, err)
	}
	// A file root wrote for another tenant of this hostname carries nothing
	// into this tenant's state (ruling R26). An untagged one is read as the
	// route's own, as before.
	if rt != nil && consumed.ForeignTo(*rt) {
		consumed = nil
	}
	held := rt != nil && settingsApply(*rt)
	inPlaceOf := ""
	if sel, ok := activator.PlannedSelectionFrom(ctx); ok && sel.Build == build {
		inPlaceOf = sel.InPlaceOf
	}
	_, err = b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) {
		if held {
			if err := release.HeldExecStale(release.DecideHeldWake(b.heldWakeInput(*rt, cur)), cur, build, inPlaceOf); err != nil {
				return release.BuildState{}, false, err
			}
		}
		return release.NextBuildState(cur, consumed, build), true, nil
	}, nil)
	return err
}

// carryRequest is moveCfg.CarryRequest (ruling R-SPEC-43A-2; Astra round 1
// finding 3): before a cut seals host's build state at bucket/prefix (the
// lease's location: the hibernate prefix for a legacy cut, the claim
// holder's location for a Phase 1 cut or a return), root's consumed record
// on this box for the route's tenant (not one Consumed.ForeignTo it, ruling
// R26) is merged into it with release.CarryRequest: a resolution and
// a failed completion, never a successful completion, never last_executed.
// The object is written only when that changes its request state, under the
// host's lock and against the etag it read (write), so it never overwrites a
// newer record. A consumed file that does not read refuses the carry, and so
// the cut: a request this box closed would reopen on the destination.
//
// The carry runs inside the carry fence (release.FenceCarry; ruling
// R-CARRY-FENCE): root's publication for host and this carry serialize on
// one lock, and root publishes only for a row no move holds. So the row's
// move hold must already be effective here (every lane sets it before its
// carry: the legacy cut's "moving", the Phase 1 cut's held "moving", the
// return lane's "releasing"); a row in no hold refuses the carry, as root
// could still publish after it. A row gone from the registry gets no
// publication either. Then the fence, held across the read and the write: a
// root record in flight lands first and is carried.
func (b *buildStates) carryRequest(ctx context.Context, host, bucket, prefix string) error {
	rt, err := b.reg.Get(host)
	if err != nil {
		return fmt.Errorf("read the route of %s: %w", host, err)
	}
	if rt != nil && !registry.IsHold(rt.Status) {
		return fmt.Errorf("%s is in no move hold (status %q), so root could still publish a request record after the carry", host, rt.Status)
	}
	unlock, err := release.FenceCarry(ctx, b.optDir, host)
	if err != nil {
		return fmt.Errorf("the carry fence for %s: %w", host, err)
	}
	defer unlock()
	consumed, err := release.ReadConsumed(b.optDir, host)
	if err != nil {
		return fmt.Errorf("read root's consumed file for %s: %w", host, err)
	}
	// Another tenant's file is no request record of this route's (ruling
	// R26): nothing to carry. An untagged one is carried as before.
	if consumed == nil || (rt != nil && consumed.ForeignTo(*rt)) {
		return nil
	}
	ctx = activator.WithBuildStateLocation(ctx, activator.BuildStateLocation{Bucket: bucket, Prefix: prefix})
	_, err = b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) {
		next := release.CarryRequest(cur, consumed)
		return next, !release.SameRequestState(cur, &next), nil
	}, hasReplica)
	return err
}

// moveCarry is moveCfg.CarryRequest for this box: the build-state store's
// carry, a refusal when the store did not start (a cut must not seal a
// state it could not bring up to date), and nil on a box with no replica
// bucket (no build state to carry).
func moveCarry(bs *buildStates, initErr error) func(ctx context.Context, host, bucket, prefix string) error {
	if initErr != nil {
		return func(context.Context, string, string, string) error {
			return fmt.Errorf("the build state store did not start on this box (%v)", initErr)
		}
	}
	if bs == nil {
		return nil
	}
	return bs.carryRequest
}

// buildStateHook is Service.BeforeExec for this box (settings spec §4.2,
// Astra I3 #1). Only a box with no replica bucket configured (bs nil, no
// error) keeps the nil hook: it runs no hibernating instance, so it has no
// build state. A box whose bucket is configured but whose build-state store
// failed to start refuses every exec: the write before execution is
// mandatory, and "could not record" never means "run unrecorded".
func buildStateHook(bs *buildStates, initErr error) func(ctx context.Context, host, build string) error {
	if initErr != nil {
		return func(context.Context, string, string) error {
			return fmt.Errorf("the build state store did not start on this box (%v); restart the agent once it is fixed", initErr)
		}
	}
	if bs == nil {
		return nil
	}
	return bs.beforeExec
}

// heldTarget is the held wake check's answer (activator.Service.HeldTarget;
// settings spec §4.3 rule 3): release.DecideHeldWake (ruling R-HELD-PURE)
// on what root reads and what root wrote: the route's settings column, its
// build-state copy, root's consumed file and refusal record, and the build
// the live link names. The decision is root's rule order (Target): a pin,
// then a pending request's T, then rule 3's target (last_executed, else the
// installed build, each failure-checked). Root's consumed reason counts only
// for the settings column it was decided for: with none, a wake that could
// be held holds and asks root, never "not held" (Astra round 1 finding 4).
// A copy whose last refresh failed is refused as unread (finding 1).
//
// A held, decided target that root does not report failed is offered with
// this box's verified retained copy of it, if any (Staged).
//
// Settings hold only the rows root applies them to (settingsApply; ruling
// R-HELD-SCOPE): every other row, a named unit or an exec worker under an
// app-policy setting, is not held.
func (b *buildStates) heldTarget(host string) activator.HeldWake {
	rt, err := b.reg.Get(host)
	if err != nil || rt == nil || !settingsApply(*rt) {
		return activator.HeldWake{}
	}
	d := release.DecideHeldWake(b.heldWakeInput(*rt, nil))
	h := activator.HeldWake{Target: d.Target, Held: d.Held, Unknown: d.Unknown, Failed: d.Failed, AskRoot: d.AskRoot, Unread: d.Unread}
	if h.Held && h.Target != "" && !h.Failed && !h.AskRoot && !h.Unknown && !h.Unread {
		h.Staged = b.retained(*rt, h.Target)
	}
	return h
}

// heldWakeInput reads DecideHeldWake's input for rt. state, when set, is
// the build state just read under the lease (beforeExec's revalidation), in
// place of the registry copy. A consumed file or refusal record that does
// not read is no word from root (nil), and so is a consumed file tagged for
// another tenant (Consumed.ForeignTo); a copy that does not parse is unread.
func (b *buildStates) heldWakeInput(rt registry.Route, state *release.BuildState) release.HeldWakeInput {
	in := release.HeldWakeInput{Settings: rt.UpdateSettings}
	// Another tenant's file is no word from root for this route (ruling
	// R26), so a wake that could be held holds and asks root. An untagged
	// file, from before ruling R22, is read as before, as every other read
	// of it is (ruling R28): ignoring its failed decision would let a wake
	// whose target is already live run a build root reports failed, as
	// awaitRootDecision does not wait when the live link names the target.
	if c, _ := release.ReadConsumed(b.optDir, rt.Host); !c.ForeignTo(rt) {
		in.Consumed = c
	}
	in.Refusal, _ = release.ReadRefusal(b.bakeDir, rt.Host)
	in.Installed, _ = release.LiveVersion(b.optDir, rt.Host)
	if state != nil {
		in.Build = release.BuildStateCopy{Known: true, Replica: true, State: state}
	} else {
		in.Build, _ = release.ParseBuildStateCopy(rt.BuildState)
	}
	return in
}

// settingsApply reports a row root applies update settings to: Kind
// instance with no Unit, release's adopt condition (ruling R-HELD-SCOPE).
// Root and the wake must agree on it, or a wake waits for an install root
// never makes.
func settingsApply(rt registry.Route) bool { return rt.Kind == "instance" && rt.Unit == "" }

// retained is v's staged executable on this box when its bytes match its
// staged manifest's sha256 for this platform (release.RetainedStaged, the
// check stageAuthorized's recovery uses), else nil (Astra I3 #5).
func (b *buildStates) retained(rt registry.Route, v string) *release.StagedIdentity {
	id, _ := release.RetainedStaged(b.optDir, release.RouteAccount(rt), rt.App, v)
	return id
}

// wakeHeldBudget bounds wakeTarget's refresh: the wake waits on it.
const wakeHeldBudget = 5 * time.Second

// wakeTarget is heldTarget with the copy refreshed first (§4.2: "refreshed
// at every reconcile and before every wake"), for a route whose settings
// could hold it. A route with default settings, or one root applies no
// settings to (settingsApply), is today's wake: nothing is read for it. A
// refresh that fails refuses the wake as unread (Astra round 1 finding 1):
// the last copy may name a build the data has since moved past.
func (b *buildStates) wakeTarget(ctx context.Context, host string) activator.HeldWake {
	rt, err := b.reg.Get(host)
	if err != nil || rt == nil || !settingsApply(*rt) {
		return activator.HeldWake{}
	}
	if u, err := release.ParseUpdateSettingsLenient(rt.UpdateSettings); err == nil && u.Default() {
		return activator.HeldWake{}
	}
	if keepsBuildState(*rt, b.claims != nil) {
		ctx, cancel := context.WithTimeout(ctx, wakeHeldBudget)
		err := b.refresh(ctx, host)
		cancel()
		if err != nil {
			// Refused, never decided from the last copy (Astra round 1
			// finding 1): the durable state may name another build by now.
			// refresh recorded the failure in the copy, which DecideHeldWake
			// refuses too, unless the registry write itself failed.
			log.Printf("agent: %s not woken: its build state could not be read before the wake: %v", host, err)
			return activator.HeldWake{Held: true, Unread: true}
		}
	}
	return b.heldTarget(host)
}