// Package hostrelocate is the regional worker's half of relocation (region
// HA spec 1 §5.8, claim §4 L1): starting a relocation, and replaying every
// worker CAS from the claim alone. Any active enrolled box of the region
// may act (claim P6-D5); correctness never depends on which, because every
// step is a claim CAS conditional on the operation, its phase and the
// claim's ETag. The participants' halves (acknowledgement, release,
// install, probe) are not here: they run on the boxes, under the host
// guard, through Plan 6b's box assembly.
//
// This package only READS termination fences. hostfence.TerminationWorker
// writes them (control/box-fences/<enrollment>.json).
package hostrelocate

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"slices"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfence"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/termfence"
)

const (
	// ProbeDeadline is §5.8's probe deadline: a live target whose probe has
	// not passed this long after entering probing goes to probe-failed.
	ProbeDeadline = 10 * time.Minute
	// ReleaseDeadline is this plan's stopping policy for a release that
	// never completes (§5.8: "a flush that never reaches N goes to stuck";
	// "a probe-failed target that cannot finish its release sequence also
	// goes to stuck"). The spec names no number; this one matches the
	// probe deadline. A live box whose release is not journaled this long
	// after entering releasing or probe-failed leaves the operation stuck
	// for the operator: nothing moves away from a live box without its
	// stop evidence.
	ReleaseDeadline = 10 * time.Minute
)

var (
	ErrUnsupported  = errors.New("hostrelocate: a required provider is missing")
	ErrNotWorker    = errors.New("hostrelocate: this box is not an active, unfenced worker of the region")
	ErrNotYet       = errors.New("hostrelocate: waiting for a participant or a time bound")
	ErrRefused      = errors.New("hostrelocate: relocation refused")
	ErrNoController = errors.New("hostrelocate: no execution controller is open for this participation")
)

// Lease is the replica lease on holder.location (claim C3), as
// lease.Acquire returns it. The worker holds it from selecting a pin until
// the pin is journaled, so a park worker (which takes the same lease to
// retire) cannot retire the versions in between (claim §4: "New pins must
// be established under the required leases and fresh claim authority
// before conflicting retirement can proceed").
type Lease interface {
	Lost() <-chan struct{}
	Release(context.Context) error
}

// Request asks for one relocation. Target is required; Source must be the
// claim's executor. Reason is drain, rebalance or failover.
type Request struct {
	Reason, Source, Target string
}

// Owner is the app-admission registration owner of a box-started
// operation (claim §5.1, "A regional worker starting relocating").
type Owner struct {
	Box, Enrollment, Controller string
}

// Admission is claim §5.1's app-admission registration for a box-started
// relocation: Register binds the saved claim observation and holder
// incarnation while admission is open; Recheck is the final check before
// the single start CAS. Claim Plan 5's app gate implements it.
type Admission interface {
	Register(context.Context, hostclaim.Observed, Owner, string) (string, error)
	Recheck(context.Context, string) error
}

// Pinner selects, writes and returns one pinned recovery set (§5.8).
type Pinner interface {
	Pin(context.Context, hostclaim.Observed, time.Time) (hostclaim.RecoveryPin, error)
}

// Prepared verifies the remote build of the operation's prepared
// generation P (§5.8 "Building the prepared generation"): done is true only
// when a fresh read verifies P's snapshot, level-0 anchor and completed
// install manifest (or, for a host with nothing replicated, P's files
// manifest alone). Moves' shared install routine (L16) writes them.
type Prepared interface {
	Verify(context.Context, hostclaim.Observed) (a hostclaim.PreparedAuthority, done bool, err error)
}

// Worker drives relocation for one region. Every field is required.
type Worker struct {
	Claims    release.Store // the deployment (control) bucket
	Directory hostactors.Directory
	Fences    hostclaim.FenceAuthority   // hostfence.ClaimFenceAuthority
	Releases  hostclaim.ReleaseAuthority // Plan 6b's stop receipt reader
	Admission Admission
	// Census is the box-started start census (claim D7 with L5 and L20):
	// hostactors.Directory.CensusBoxRelocation with the deployment's
	// evidence and floor.
	Census func(ctx context.Context, starter string, participants []string) error
	// Choose names a target in the operation's region, never one in
	// exclude (hostops.ExecutorChooser behind an adapter).
	Choose func(ctx context.Context, obs hostclaim.Observed, exclude []string) (hostclaim.Target, error)
	// Eligible refuses a named target that the complete eligible set
	// (§2, boxchoice.Eligible over the roster and the edges files) does
	// not hold: stale edges, a missing capability or secrets key, a
	// draining box (hostops.ExecutorChooser.Permits).
	Eligible func(ctx context.Context, obs hostclaim.Observed, box string) error
	// Relocatable refuses a host whose authoritative record is not a
	// relocatable shape: a unit- or socket-backed route, a tunnel, a
	// sidecar (the same shapes Plan 6b's wake refuses).
	Relocatable func(ctx context.Context, obs hostclaim.Observed) error
	// Leases acquires the replica lease on the host's location in the
	// region's data bucket (lease.Acquire).
	Leases   func(ctx context.Context, location string) (Lease, error)
	Pins     Pinner
	Prepared Prepared
	// Discharge judges a fenced box's volume records (boxvolumes.Discharge).
	// It never gates anything: nil, or an unqualified verdict, leaves the
	// box's park-local chores open for an operator (H19).
	Discharge func(context.Context, hostactors.Box, termfence.Record) *hostclaim.LocalDischarge
	// Notify bumps the control serial after an executor CAS (§5.3,
	// "Prompt changes"). A failed bump costs cadence, never correctness.
	Notify func(context.Context)
	Self   termfence.WrittenBy
	Region string
	// HomeRegion is this box's verified home binding (hostbox.Actor's, the
	// fleet's home roster name when Region is it), or "" with no alias. It
	// is the one binding under which the home's two names, Region and
	// park.HomeRegion, are one region (park.SameRegion), as main's claim
	// owners read it (park.VerifiedHome against the op's "home_region").
	HomeRegion string
	Controller string
	NewID      func() string
	Now        func() time.Time
}

func (w Worker) ready() error {
	if w.Claims == nil || w.Directory.Store == nil || w.Fences == nil || w.Releases == nil || w.Admission == nil || w.Census == nil ||
		w.Choose == nil || w.Eligible == nil || w.Relocatable == nil || w.Leases == nil || w.Pins == nil || w.Prepared == nil || w.Discharge == nil || w.Notify == nil || w.Self.Box == "" || w.Self.Enrollment == "" ||
		w.Region == "" || w.Controller == "" || w.NewID == nil || w.Now == nil {
		return ErrUnsupported
	}
	return nil
}

// self checks, on every call, that this box is an active, unfenced worker
// enrolled in the region (§5.8 "Each worker first checks its own roster
// entry is active in this region").
func (w Worker) self(ctx context.Context) (hostactors.Observed, error) {
	o, err := w.Directory.Read(ctx)
	if err != nil {
		return hostactors.Observed{}, err
	}
	b, ok := o.Roster.Boxes[w.Self.Box]
	if !ok || b.State != "active" || b.Enrollment != w.Self.Enrollment || b.Region != w.Region {
		return hostactors.Observed{}, ErrNotWorker
	}
	// A home binding is the signed roster's home, and only the home's own
	// boxes hold it (hostbox.IdentityProvider.Verify's rule).
	if w.HomeRegion != "" && (w.HomeRegion != o.Roster.HomeRegion || w.HomeRegion != w.Region) {
		return hostactors.Observed{}, ErrNotWorker
	}
	fenced, err := w.Directory.TerminationFence(ctx, o, w.Self.Box, w.Now())
	if err != nil || fenced {
		return hostactors.Observed{}, errors.Join(ErrNotWorker, err)
	}
	return o, nil
}

// fence reads box's termination fence, if any, as the exact bytes a claim
// CAS will cite. Absence is never evidence: no fence is (nil, nil); an
// unreadable or invalid one is an error, never "not fenced".
func (w Worker) fence(ctx context.Context, o hostactors.Observed, box string) (*hostclaim.FenceRef, termfence.Record, error) {
	b, ok := o.Roster.Boxes[box]
	if !ok {
		return nil, termfence.Record{}, fmt.Errorf("%w: %s is not in the roster", ErrRefused, box)
	}
	key := hostfence.FenceKey(b.Enrollment)
	raw, _, err := w.Directory.Store.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		return nil, termfence.Record{}, nil
	}
	if err != nil {
		return nil, termfence.Record{}, err
	}
	var f termfence.Record
	if err := json.Unmarshal(raw, &f); err != nil {
		return nil, termfence.Record{}, err
	}
	if err := hostactors.ValidateTerminationFence(o, box, f, w.Now()); err != nil {
		return nil, termfence.Record{}, err
	}
	sum := sha256.Sum256(raw)
	return &hostclaim.FenceRef{Enrollment: b.Enrollment, Key: key, SHA256: hex.EncodeToString(sum[:])}, f, nil
}

// ConvertLanding is J3 for one landing claim, under the same worker
// authorization as Step (Astra round 6 finding 4): this box must be an
// active, unfenced worker whose current roster entry is the enrollment it
// runs as, in the region the landing lands in. A landing whose op.exec
// box (the move's target) has a termination fence becomes a failover
// relocation when its move committed (guard); an uncommitted one is
// moves' case (ErrWrongState). A landing into another region, one with no
// executor, or one whose target has no fence is left alone with nothing
// written; a caller that is not such a worker is ErrNotWorker with nothing
// written.
func (w Worker) ConvertLanding(ctx context.Context, obs hostclaim.Observed, guard hostclaim.CommittedLandingGuard) (hostclaim.Observed, error) {
	if w.Claims == nil || w.Directory.Store == nil || w.Fences == nil || w.Discharge == nil || w.NewID == nil || w.Now == nil || w.Self.Box == "" || w.Self.Enrollment == "" || w.Region == "" || guard == nil {
		return obs, ErrUnsupported
	}
	o, err := w.self(ctx)
	if err != nil {
		return obs, err
	}
	if obs.Absent || obs.Claim == nil {
		return obs, nil
	}
	c := obs.Claim
	box := c.Op.Exec
	if c.Op.Kind != hostclaim.KindLanding || box == "" || box == "none" {
		return obs, nil
	}
	// The landing's "to" is an engine region (park.HomeRegion for the
	// home); this worker's Region is its roster name. They are one region
	// only under a home binding this box and the move agree on.
	home, err := park.VerifiedHome(o.Roster.HomeRegion, w.HomeRegion, w.Region, c.Op.Intent["home_region"])
	if err != nil {
		return obs, err
	}
	if !park.SameRegion(home, c.Op.Intent["to"], w.Region) {
		return obs, nil
	}
	ref, record, err := w.fence(ctx, o, box)
	if err != nil || ref == nil {
		return obs, err
	}
	d := w.Discharge(ctx, o.Roster.Boxes[box], record)
	return hostclaim.ConvertCommittedLanding(ctx, w.Claims, obs, c.Op.ID, box, *ref, w.Fences, guard, d, w.NewID)
}

// newestFenceObservation is the latest fence.<box>.observed_at the op
// journals. An observation that does not parse, or none at all, is an
// error, never "no bound" (absence is never evidence): the conversion
// journals its target's fence observation in the same CAS that leaves the
// op waiting, so a wait with nothing to bound it is a defect, refused as
// an unreadable installed_at is.
func newestFenceObservation(c *hostclaim.Claim) (time.Time, error) {
	var newest time.Time
	for k, v := range c.Op.Journal {
		if !strings.HasPrefix(k, "fence.") || !strings.HasSuffix(k, ".observed_at") {
			continue
		}
		at, err := time.Parse(time.RFC3339Nano, v)
		if err != nil {
			return time.Time{}, fmt.Errorf("%w: %s of %s: %v", ErrRefused, k, c.Op.ID, err)
		}
		if at.After(newest) {
			newest = at
		}
	}
	if newest.IsZero() {
		return time.Time{}, fmt.Errorf("%w: %s journals no fence observation to bound its wait for a target", ErrRefused, c.Op.ID)
	}
	return newest, nil
}

// Start writes the start CAS for req (§5.8 "The start CAS", §5.10). It
// refuses unless the claim serves on req.Source; the roster shows the
// starter, the target and (for a drain) the source active, unfenced and
// with no executable update reserved (H14); the census passes; and the
// app admits the registration. A failover requires the source's fence.
func (w Worker) Start(ctx context.Context, host string, req Request) (hostclaim.Observed, error) {
	if err := w.ready(); err != nil {
		return hostclaim.Observed{}, err
	}
	o, err := w.self(ctx)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	obs, err := hostclaim.Read(ctx, w.Claims, host)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindServing || obs.Claim.Op.Exec != req.Source || req.Target == req.Source {
		return hostclaim.Observed{}, fmt.Errorf("%w: %s does not serve on %s", ErrRefused, host, req.Source)
	}
	if obs.Claim.Holder.Family != "instance" {
		return hostclaim.Observed{}, fmt.Errorf("%w: a %s host is not relocatable", ErrRefused, obs.Claim.Holder.Family)
	}
	if err := w.Relocatable(ctx, obs); err != nil {
		return hostclaim.Observed{}, err
	}
	if err := w.Eligible(ctx, obs, req.Target); err != nil {
		return hostclaim.Observed{}, err
	}
	source, sok := o.Roster.Boxes[req.Source]
	target, tok := o.Roster.Boxes[req.Target]
	if !sok || !tok || source.Region != w.Region || target.Region != w.Region {
		return hostclaim.Observed{}, fmt.Errorf("%w: source and target must be enrolled in %s", ErrRefused, w.Region)
	}
	ref, record, err := w.fence(ctx, o, req.Source)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	refs := []string{w.Self.Box, req.Target}
	switch req.Reason {
	case hostclaim.ReasonFailover:
		if ref == nil {
			return hostclaim.Observed{}, fmt.Errorf("%w: a failover needs the source's termination fence", ErrRefused)
		}
	case hostclaim.ReasonDrain, hostclaim.ReasonRebalance:
		if ref != nil {
			return hostclaim.Observed{}, fmt.Errorf("%w: the source is fenced; start a failover", ErrRefused)
		}
		if obs.Claim.Holder.Session == 0 {
			// Ruling 11 (rev 2): a never-woken source has no writer session
			// to stop, and root refuses stop evidence for session 0 until
			// claim Plan 5's no-process witness lands. Until then only a
			// failover relocates such a host.
			return hostclaim.Observed{}, fmt.Errorf("%w: %s never admitted a writer on %s; a drain has nothing to release", ErrRefused, host, req.Source)
		}
		refs = append(refs, req.Source)
	default:
		return hostclaim.Observed{}, fmt.Errorf("%w: reason %q", ErrRefused, req.Reason)
	}
	refs = slices.Compact(slices.Sorted(slices.Values(refs)))
	revision, err := w.Directory.CheckBoxStart(ctx, refs...)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	if err := w.Census(ctx, w.Self.Box, refs); err != nil {
		return hostclaim.Observed{}, err
	}
	id := w.NewID()
	admission, err := w.Admission.Register(ctx, obs, Owner{Box: w.Self.Box, Enrollment: w.Self.Enrollment, Controller: w.Controller}, id)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	if err := w.Admission.Recheck(ctx, admission); err != nil {
		return hostclaim.Observed{}, err
	}
	in := hostclaim.RelocationStart{OpID: id, By: "box:" + w.Self.Box, Now: w.Now(), Region: w.Region, HomeRegion: w.HomeRegion, Reason: req.Reason,
		SourceBox: req.Source, SourceEnrollment: source.Enrollment, TargetBox: req.Target, TargetEnrollment: target.Enrollment,
		DirectoryRevision: revision, AdmissionID: admission, Fence: ref}
	if ref != nil {
		in.Discharge = w.Discharge(ctx, source, record)
	}
	next, err := hostclaim.StartRelocation(ctx, w.Claims, obs, in, w.Fences)
	if err == nil && ref != nil {
		w.Notify(ctx)
	}
	return next, err
}

// Step makes at most one worker CAS for host's relocation, the one its
// phase and evidence call for, and returns the claim it left. ErrNotYet
// means a participant or a time bound is awaited; nothing was written.
// A host that is not relocating, or whose operation is stuck, is left
// alone.
func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, error) {
	if err := w.ready(); err != nil {
		return hostclaim.Observed{}, err
	}
	o, err := w.self(ctx)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	obs, err := hostclaim.Read(ctx, w.Claims, host)
	if err != nil || obs.Absent {
		return obs, err
	}
	c := obs.Claim
	if c.Op.Kind != hostclaim.KindRelocating || c.Op.Intent["region"] != w.Region || c.Op.Journal["stuck"] != "" {
		return obs, nil
	}
	// An op bound to a home names it in "home_region"; this box steps it
	// only under the same binding (park.VerifiedHome, main's rule).
	if _, err := park.VerifiedHome(o.Roster.HomeRegion, w.HomeRegion, w.Region, c.Op.Intent["home_region"]); err != nil {
		return obs, err
	}
	id := c.Op.ID
	// Evidence first: a fenced participant changes what every phase may do.
	// A pass that refuses here checks the release deadline first (refused).
	if src := c.Op.Intent["source_box"]; !slices.Contains(c.Lost, src) {
		ref, record, err := w.fence(ctx, o, src)
		if err != nil {
			return w.refused(ctx, obs, err)
		}
		if ref == nil && c.Op.Journal["fence."+src] != "" {
			// A converted move bound the source's fence before FenceSource
			// consumed it. A bound fence that is no longer there is not
			// "not fenced" (absence is never evidence): refuse the pass.
			return w.refused(ctx, obs, fmt.Errorf("%w: %s's fence is bound in the journal but cannot be read", ErrRefused, src))
		}
		if ref != nil {
			next, err := hostclaim.FenceSource(ctx, w.Claims, obs, id, *ref, w.Discharge(ctx, o.Roster.Boxes[src], record), w.Fences)
			if err == nil {
				w.Notify(ctx)
			}
			return next, err
		}
	}
	if dst := c.Op.Intent["target_box"]; dst != "" && !slices.Contains(c.Lost, dst) {
		ref, record, err := w.fence(ctx, o, dst)
		if err != nil {
			return w.refused(ctx, obs, err)
		}
		if ref != nil {
			return w.replace(ctx, obs, *ref, w.Discharge(ctx, o.Roster.Boxes[dst], record), record.ObservedAt)
		}
	}
	switch c.Op.Phase {
	case hostclaim.PhaseStarting:
		if c.Op.Intent["target_box"] == "" && c.Op.Exec == "none" {
			// A committed landing converted to a failover
			// (hostclaim.ConvertCommittedLanding) starts with no target: name
			// one here, so it acknowledges before PinFailover selects a pin,
			// as an initial failover's target does.
			next, err := w.next(ctx, obs)
			if err != nil {
				// J3-3: bounded. A converted landing with no eligible
				// target is stuck ReleaseDeadline after its target's
				// fence was observed; before that the chooser's refusal
				// is returned and the pass alerts once.
				at, ferr := newestFenceObservation(c)
				if ferr != nil {
					return obs, errors.Join(err, ferr)
				}
				if w.Now().Sub(at) > ReleaseDeadline {
					return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("no eligible box for the converted landing after %s: %v", ReleaseDeadline, err)))
				}
				return obs, err
			}
			return w.update(ctx, obs, hostclaim.NameTarget(id, next))
		}
		if !acked(c) {
			return obs, ErrNotYet
		}
		if c.Op.Exec == "none" {
			return w.update(ctx, obs, hostclaim.PinFailover(id))
		}
		return w.update(ctx, obs, hostclaim.ToReleasing(id, w.Now()))
	case hostclaim.PhaseReleasing:
		if c.Op.Journal["release"] == "" {
			return w.releaseOverdue(ctx, obs, "releasing_since")
		}
		return hostclaim.PinAfterRelease(ctx, w.Claims, obs, id, w.Releases, nil)
	case hostclaim.PhaseProbeFailed:
		if c.Op.Journal["release"] == "" {
			return w.releaseOverdue(ctx, obs, "probe_failed_since")
		}
		if done, err := exhausted(c); err != nil {
			return obs, err
		} else if done {
			return w.stuckOn(ctx, obs, hostclaim.ErrReplacementsExceeded)
		}
		next, err := w.next(ctx, obs)
		if err != nil {
			// No eligible box yet; a later pass may find one, but not for
			// ever: in a two-box region the source is excluded and there
			// may be none (rev 17, finding F1). ReleaseDeadline after the
			// target failed, the operation is stuck for an operator.
			since, perr := time.Parse(time.RFC3339Nano, c.Op.Journal["probe_failed_since"])
			if perr == nil && w.Now().Sub(since) > ReleaseDeadline {
				return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("no eligible box for the nested relocation after %s: %v", ReleaseDeadline, err)))
			}
			return obs, err
		}
		out, err := hostclaim.PinAfterRelease(ctx, w.Claims, obs, id, w.Releases, &next)
		if err != nil {
			// A live target's replacement counts too: the third is stuck.
			return w.stuckOn(ctx, obs, err)
		}
		return out, nil
	case hostclaim.PhasePinning:
		return w.pinning(ctx, obs)
	case hostclaim.PhaseRelocatingProbing:
		since, err := time.Parse(time.RFC3339Nano, c.Op.Journal["probing_since"])
		if err != nil {
			return obs, err
		}
		if c.Op.Journal["probe"] != "" && c.Op.Journal["probe_stop"] != "" {
			next, err := w.update(ctx, obs, hostclaim.CompleteRelocation(id, w.NewID))
			if err == nil {
				w.Notify(ctx)
			}
			return next, err
		}
		if c.Op.Journal["probe"] != "" {
			// The target stops its probe session first (ruling 49). Ruling
			// 67: bounded. If root's stop receipt cannot be had (the
			// execution lease lost after RecordProbe, so the participant
			// never passes Ready; a stop root keeps refusing), the op is
			// stuck ReleaseDeadline after the probe passed (Probe.At; for an
			// untimed probe journaled before rev 4, probing_since plus
			// ProbeDeadline, the latest it can have passed), for an
			// operator. Nothing is released: the participation's denial,
			// the row's hold and the session stay as they are.
			var p hostclaim.Probe
			passed := since.Add(ProbeDeadline)
			if json.Unmarshal([]byte(c.Op.Journal["probe"]), &p) == nil && !p.At.IsZero() {
				passed = p.At
			}
			if w.Now().Sub(passed) > ReleaseDeadline {
				return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("%s passed its probe but no root stop receipt for session %d within %s: an operator must obtain the stop", c.Op.Exec, p.Session, ReleaseDeadline)))
			}
			return obs, ErrNotYet
		}
		if w.Now().Sub(since) > ProbeDeadline {
			return w.update(ctx, obs, hostclaim.ToProbeFailed(id, w.Now()))
		}
		return obs, ErrNotYet
	case hostclaim.PhaseInstalling:
		a, done, err := w.Prepared.Verify(ctx, obs)
		if err != nil || !done || !acked(c) {
			return obs, errors.Join(ErrNotYet, err)
		}
		return w.update(ctx, obs, hostclaim.ToInstalledAt(id, a, w.Now()))
	case hostclaim.PhaseInstalled:
		// The target's local install and its entry into probing (Plan 6b).
		// Ruling 60: bounded. A target that cannot enter probing within
		// ProbeDeadline (a retired stamp, a participation main denies) is
		// stuck for an operator: no automatic replacement of a live target
		// is made here.
		// A missing or unreadable installed_at is an error, never "no
		// bound" (absence is never evidence), as probing_since is above.
		at, err := time.Parse(time.RFC3339Nano, c.Op.Journal["installed_at"])
		if err != nil {
			return obs, fmt.Errorf("installed_at of %s: %w", id, err)
		}
		if w.Now().Sub(at) > ProbeDeadline {
			return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("%s did not enter probing within %s of installed", c.Op.Exec, ProbeDeadline)))
		}
		return obs, ErrNotYet
	}
	return obs, ErrNotYet
}

func (w Worker) pinning(ctx context.Context, obs hostclaim.Observed) (hostclaim.Observed, error) {
	c := obs.Claim
	id := c.Op.ID
	if c.Op.Intent["target_box"] == "" {
		next, err := w.next(ctx, obs)
		if err != nil {
			return obs, err
		}
		return w.update(ctx, obs, hostclaim.NameTarget(id, next))
	}
	if c.Op.Journal["pin"] == "" {
		for key, at := range c.Op.Journal {
			if len(key) > len(".observed_at") && key[len(key)-len(".observed_at"):] == ".observed_at" {
				t, err := time.Parse(time.RFC3339Nano, at)
				if err != nil {
					return obs, err
				}
				if w.Now().Before(t.Add(hostclaim.FailoverPinDelay)) {
					return obs, ErrNotYet
				}
			}
		}
		return w.pinUnderLease(ctx, obs)
	}
	t := hostclaim.Target{Box: c.Op.Intent["target_box"], Enrollment: c.Op.Intent["target_enrollment"]}
	if !acked(c) {
		return obs, ErrNotYet
	}
	next, err := w.update(ctx, obs, hostclaim.ToInstalling(id, t, w.NewID()))
	if err == nil {
		w.Notify(ctx)
	}
	return next, err
}

// pinUnderLease selects and journals the pin while holding the replica
// lease on holder.location, so no park can retire the pinned versions
// between the selection and the journal CAS (claim §4). A lease lost
// before the CAS refuses: the next pass selects afresh.
func (w Worker) pinUnderLease(ctx context.Context, obs hostclaim.Observed) (hostclaim.Observed, error) {
	held, err := w.Leases(ctx, obs.Claim.Holder.Location)
	if err != nil {
		return obs, err
	}
	defer func() { _ = held.Release(ctx) }()
	pin, err := w.Pins.Pin(ctx, obs, w.Now())
	if errors.Is(err, ErrFilesUnevidenced) {
		// H6: instance.key, report.token and the env are continuous; a pin
		// that cannot show them never existed waits for an operator.
		return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, err.Error()))
	}
	if err != nil {
		return obs, err
	}
	select {
	case <-held.Lost():
		return obs, fmt.Errorf("%w: the location lease was lost before the pin was journaled", ErrNotYet)
	default:
	}
	return w.update(ctx, obs, hostclaim.JournalPin(obs.Claim.Op.ID, pin))
}

// refused is a pass that refuses with cause. The release deadline is
// checked before the refusal stands: in releasing or probe-failed,
// ReleaseDeadline after the phase began (ruling 14), the operation is
// stuck instead, whether or not a release was journaled. So a refusal
// that never clears, a source removed from the roster before its pin
// (Task 7's ruling), leaves the operation stuck for an operator, never
// refusing for ever. Only refusals (ErrRefused) count: any other error is
// returned as it is, for a later pass to retry.
func (w Worker) refused(ctx context.Context, obs hostclaim.Observed, cause error) (hostclaim.Observed, error) {
	sinceKey := map[string]string{hostclaim.PhaseReleasing: "releasing_since", hostclaim.PhaseProbeFailed: "probe_failed_since"}[obs.Claim.Op.Phase]
	if sinceKey == "" || !errors.Is(cause, ErrRefused) {
		return obs, cause
	}
	since, err := time.Parse(time.RFC3339Nano, obs.Claim.Op.Journal[sinceKey])
	if err != nil {
		return obs, errors.Join(cause, err)
	}
	if w.Now().Sub(since) <= ReleaseDeadline {
		return obs, cause
	}
	return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, fmt.Sprintf("%s refused past the release deadline (%s): %v", obs.Claim.Op.Phase, ReleaseDeadline, cause)))
}

// releaseOverdue marks the operation stuck when a live box has not
// journaled its release ReleaseDeadline after entering the phase whose
// start key names; before that it waits.
func (w Worker) releaseOverdue(ctx context.Context, obs hostclaim.Observed, sinceKey string) (hostclaim.Observed, error) {
	since, err := time.Parse(time.RFC3339Nano, obs.Claim.Op.Journal[sinceKey])
	if err != nil {
		return obs, err
	}
	if w.Now().Sub(since) <= ReleaseDeadline {
		return obs, ErrNotYet
	}
	return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, fmt.Sprintf("%s has not released %s after %s", obs.Claim.Op.Exec, ReleaseDeadline, obs.Claim.Op.Phase)))
}

// next names the operation's next target: never the source, a lost box,
// a box this operation already replaced, or a box that owes a park-local
// chore for the host (spec 1 §2; ReplaceTarget refuses one in its CAS).
func (w Worker) next(ctx context.Context, obs hostclaim.Observed) (hostclaim.Target, error) {
	c := obs.Claim
	exclude := append([]string{c.Op.Intent["source_box"]}, c.Lost...)
	exclude = append(exclude, hostclaim.LocalChoreBoxes(c)...)
	if t := c.Op.Intent["target_box"]; t != "" {
		exclude = append(exclude, t)
	}
	var replaced []string
	if raw := c.Op.Journal["excluded_targets"]; raw != "" {
		if err := json.Unmarshal([]byte(raw), &replaced); err != nil {
			return hostclaim.Target{}, err
		}
	}
	return w.Choose(ctx, obs, append(exclude, replaced...))
}

func (w Worker) replace(ctx context.Context, obs hostclaim.Observed, ref hostclaim.FenceRef, d *hostclaim.LocalDischarge, observedAt string) (hostclaim.Observed, error) {
	// Exhaustion is checked before a box is chosen (§5.8 "The bound"):
	// with the replacements used, no spare is looked for; the dead
	// target's fence is consumed into a stuck pinning (the loss retained,
	// its installation's authority revoked, no probe to wait for).
	done, err := exhausted(obs.Claim)
	if err != nil {
		return obs, err
	}
	if done {
		out, err := hostclaim.FenceExhaustedTarget(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, w.Fences)
		if err == nil {
			w.Notify(ctx)
		}
		return out, err
	}
	next, err := w.next(ctx, obs)
	if err != nil {
		// No eligible box to replace the dead target. A later pass may
		// find one, but not for ever (rev 18, round 17 finding 2):
		// ReleaseDeadline after the fence's observation the fence is
		// consumed and the operation is stuck for an operator.
		at, perr := time.Parse(time.RFC3339Nano, observedAt)
		if perr == nil && w.Now().Sub(at) > ReleaseDeadline {
			out, ferr := hostclaim.FenceTargetWithoutSpare(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, w.Fences,
				fmt.Sprintf("%s terminated and no eligible box replaced it after %s: %v", obs.Claim.Op.Intent["target_box"], ReleaseDeadline, err))
			if ferr == nil {
				w.Notify(ctx)
			}
			return out, ferr
		}
		return obs, err
	}
	out, err := hostclaim.ReplaceTarget(ctx, w.Claims, obs, obs.Claim.Op.ID, ref, d, next, w.NewID(), w.Fences)
	if errors.Is(err, hostclaim.ErrReplacementsExceeded) {
		return w.stuckOn(ctx, obs, err)
	}
	if err == nil {
		w.Notify(ctx)
	}
	return out, err
}

// stuckOn marks the operation stuck when it has used its replacements;
// any other error is returned for a later pass to retry.
func (w Worker) stuckOn(ctx context.Context, obs hostclaim.Observed, cause error) (hostclaim.Observed, error) {
	if !errors.Is(cause, hostclaim.ErrReplacementsExceeded) {
		return obs, cause
	}
	return w.update(ctx, obs, hostclaim.MarkStuck(obs.Claim.Op.ID, cause.Error()))
}

func (w Worker) update(ctx context.Context, obs hostclaim.Observed, tr func(*hostclaim.Claim) error) (hostclaim.Observed, error) {
	return hostclaim.Update(ctx, w.Claims, obs, tr)
}

// exhausted reports whether the operation has used its target
// replacements. A malformed count is an error, never 0.
func exhausted(c *hostclaim.Claim) (bool, error) {
	n, err := hostclaim.Replacements(c)
	if err != nil {
		return false, err
	}
	return n >= hostclaim.MaxTargetReplacements, nil
}

// acked is the D6 barrier as the worker reads it before a CAS that needs it.
func acked(c *hostclaim.Claim) bool {
	for _, box := range c.Op.Participants {
		if c.Op.Journal["exec_ack."+box] == "" {
			return false
		}
	}
	return true
}