package hostclaim

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/regionname"
	"github.com/carlosframework/platform/internal/release"
)

// Relocation (claim §4 "Same-region relocation", region HA spec 1 §5.8):
// the reasons, the phases and the bound on target replacements. The
// releasing phase is main's PhaseReleasing (the same "releasing" a
// restoring op writes) and probing is main's PhaseRelocatingProbing
// (sessionstep.go): one constant per phase name.
const (
	ReasonDrain     = "drain"
	ReasonRebalance = "rebalance"
	ReasonFailover  = "failover"

	PhaseStarting    = "starting"
	PhasePinning     = "pinning"
	PhaseInstalling  = "installing"
	PhaseInstalled   = "installed"
	PhaseProbeFailed = "probe-failed"

	// MaxTargetReplacements is spec 1 §5.8's bound: after two
	// replacements of the target in one operation it goes to stuck.
	MaxTargetReplacements = 2

	// MaxProbeAttempts is §5.8's probe bound: at most three admissions
	// per target installation within the probe deadline; running out is
	// probe-failed. Attempts are counted in the claim before they are
	// made, so a crash or a lost response never buys a fourth.
	MaxProbeAttempts = 3

	// FailoverPinDelay is §5.8's failover pin rule: pin no earlier than
	// the fence's observed_at plus 30 seconds, so requests S3 had already
	// accepted from the terminated box have completed (H3's window).
	FailoverPinDelay = 30 * time.Second
)

var (
	ErrRelocation             = errors.New("hostclaim: relocation transition refused")
	ErrReplacementsExceeded   = errors.New("hostclaim: relocation has used its target replacements")
	ErrAlreadyRecorded        = errors.New("hostclaim: relocation journal entry already recorded")
	ErrProbeAttemptsExhausted = errors.New("hostclaim: the target has used its probe attempts")
)

// RelocationStart is the start CAS's intent (spec 1 §5.8, claim §4). Fence
// is set for a failover only and names the source's termination fence.
type RelocationStart struct {
	OpID, By       string
	Now            time.Time
	Region, Reason string
	// HomeRegion is the starting box's verified home binding (the fleet's
	// home roster name when Region is it, hostbox.Actor.HomeRegion), or ""
	// with no alias. It lets the start guard see the home's chores under
	// either of its names (regionname.SameRegion), and is recorded as the
	// op's "home_region" intent, as BeginAdmittedMove records a move's.
	HomeRegion                  string
	SourceBox, SourceEnrollment string
	TargetBox, TargetEnrollment string
	DirectoryRevision           uint64
	AdmissionID                 string
	Fence                       *FenceRef
	Discharge                   *LocalDischarge
}

// Release is the journal entry a live box writes after its release
// sequence (spec 1 §5.8 steps 1 to 4): its stopped database's final TXID,
// its final files manifest sequence and its full stop receipt.
type Release struct {
	ReleasedBy    string `json:"released_by"`
	Enrollment    string `json:"enrollment"`
	Generation    string `json:"generation"`
	Session       uint64 `json:"session"`
	FinalTXID     uint64 `json:"final_txid"`
	ManifestSeq   int64  `json:"manifest_seq"`
	Receipt       string `json:"receipt"`
	ReceiptSHA256 string `json:"receipt_sha256"`
}

// ReleaseAuthority verifies a journaled release's full stop receipt: the
// immutable object at Receipt hashes to ReceiptSHA256 and is complete stop
// evidence (claim §6.1, P6-D1) for this host, operation, box, incarnation
// and session. Plan 6b's receipt format is its only implementation.
type ReleaseAuthority interface {
	VerifyRelease(context.Context, Claim, Release) error
}

// RecoveryPin is the journal's reference to one pinned recovery set
// (§5.8 "The pinned recovery set"): the immutable pin object holding the
// exact versions, its digest, the database position it reaches, the files
// manifest it pins, and when it was taken.
type RecoveryPin struct {
	Key           string    `json:"key"`
	SHA256        string    `json:"sha256"`
	Generation    string    `json:"generation"`
	MaxTXID       uint64    `json:"max_txid"`
	FilesManifest string    `json:"files_manifest"`
	FilesSeq      int64     `json:"files_seq"`
	PinnedAt      time.Time `json:"pinned_at"`
	// Empty is the evidenced empty bootstrap (claim §6.4a "Empty
	// bootstrap", spec 1 §5.8 "A host with nothing replicated"): no LTX
	// file exists because no transaction was ever committed. It is legal
	// only when a journaled release reports final_txid 0, or the holder
	// never admitted a writer (session 0). A listing that merely finds
	// nothing is never evidence (JournalPin refuses it).
	Empty bool `json:"empty,omitempty"`
	// Release is the app release version the relocation installs (R9 on
	// amadan #236). The pinner sets it at pinning from the region
	// controller's authoritative registry route for the host, never from
	// the box, the request or the box's install journal; the installing
	// CAS copies it to op.journal.install_release, which moves' held-stage
	// relay compares with the request's version. Required: a pin without
	// it is refused.
	Release string `json:"release"`
}

// RecoveryPinKey is where the pin selected against claim generation gen
// lives: control/relocations/<host>/<op>/pin-<gen>.json. The key binds the
// object to the exact observation it was selected against, so a pin whose
// journal CAS lost (the claim moved) is abandoned, never reused, and a
// retry selects afresh at the new generation (B2-3's round binding).
func RecoveryPinKey(host, op string, gen uint64) string {
	return "control/relocations/" + host + "/" + op + "/pin-" + strconv.FormatUint(gen, 10) + ".json"
}

// PreparedAuthority is the `installed` CAS's record of the remote build
// (§5.8 "Building the prepared generation"): the install manifest of
// generation P, verified complete by a fresh read.
type PreparedAuthority struct {
	Generation      string `json:"generation"`
	InstallManifest string `json:"install_manifest"`
	SHA256          string `json:"sha256"`
	SnapshotTXID    uint64 `json:"snapshot_txid"`
	// FilesSeq and FilesManifestSHA256 identify P's prepared files
	// manifest (claim rev 6.30 R18-1, moves rev 7.22 §7.6): create-only,
	// at sequence 0, never replaced; FilesManifestSHA256 is the digest of
	// the exact bytes a fresh read verified, the exact version the first
	// admission consumes as its baseline (the session publisher then
	// starts at sequence 1). SourceFilesSeq is provenance, not identity:
	// the pinned source manifest's sequence it was built from, which must
	// be the pin's (rev 15, round 14 finding 1). Required always, an empty
	// bootstrap included: a P without its files manifest is not built.
	FilesSeq            int64  `json:"files_seq"`
	SourceFilesSeq      int64  `json:"source_files_seq"`
	FilesManifestSHA256 string `json:"files_manifest_sha256"`
	// InstallStamp binds the prepared manifest to the installation that
	// built it ({location, P, op id, installation stamp}): ToInstalled
	// sets it from install_stamp in the CAS that journals the authority.
	InstallStamp uint64 `json:"install_stamp,omitempty"`
}

// LocalInstall is one box's local install receipt (§5.8, B1-2, B2-5).
type LocalInstall struct {
	Box                  string `json:"box"`
	Enrollment           string `json:"enrollment"`
	InstallStamp         uint64 `json:"install_stamp"`
	FilesManifestVersion int64  `json:"files_manifest_version"`
	JournalSHA256        string `json:"journal_sha256"`
}

// Probe is the target's journaled probe result for its admitted session.
type Probe struct {
	Box     string `json:"box"`
	Session uint64 `json:"session"`
	// At is when the probe passed (rev 4, ruling 67): the worker marks the
	// op stuck ReleaseDeadline after At if no stop receipt is journaled.
	// Zero in a probe journaled before rev 4: the worker then takes the
	// probe as passed at probing_since plus ProbeDeadline, the latest it
	// can have passed, and counts ReleaseDeadline from there.
	At time.Time `json:"at,omitzero"`
}

// Target names a box for the relocation's target slot.
type Target struct {
	Box, Enrollment string
}

func journalJSON(c *Claim, key string, v any) error {
	raw, err := json.Marshal(v)
	if err != nil {
		return err
	}
	if c.Op.Journal == nil {
		c.Op.Journal = map[string]string{}
	}
	c.Op.Journal[key] = string(raw)
	return nil
}

func readJournal(op Op, key string, v any) bool {
	raw, ok := op.Journal[key]
	return ok && json.Unmarshal([]byte(raw), v) == nil
}

// journalUint is a journaled counter or stamp: absent is 0, its starting
// value; a value that does not parse is an error, never 0 (absence or
// malformed data is never evidence).
func journalUint(op Op, key string) (uint64, error) {
	raw, ok := op.Journal[key]
	if !ok || raw == "" {
		return 0, nil
	}
	n, err := strconv.ParseUint(raw, 10, 64)
	if err != nil {
		return 0, fmt.Errorf("%w: journal %s is malformed: %q", ErrRelocation, key, raw)
	}
	return n, nil
}

// Replacements is the number of target replacements the operation has
// used (§5.8's bound, MaxTargetReplacements). A malformed count is an
// error, never 0.
func Replacements(c *Claim) (uint64, error) {
	if c == nil {
		return 0, ErrMalformed
	}
	return journalUint(c.Op, "replacements")
}

func requireRelocation(c *Claim, opID string, phases ...string) error {
	if err := RequireOp(opID)(c); err != nil {
		return err
	}
	if c.Holder == nil || c.Op.Kind != KindRelocating || !slices.Contains(phases, c.Op.Phase) {
		return fmt.Errorf("%w: %s/%s", ErrWrongState, c.Op.Kind, c.Op.Phase)
	}
	if c.Op.Journal["stuck"] != "" {
		return fmt.Errorf("%w: operation is stuck", ErrRelocation)
	}
	ensureJournal(c)
	return nil
}

// ensureJournal makes the journal writable: an empty one round-trips as nil.
func ensureJournal(c *Claim) {
	if c.Op.Journal == nil {
		c.Op.Journal = map[string]string{}
	}
}

// requireAcks is the D6 barrier: every current participant has recorded
// its execution acknowledgement before the operation leaves its phase.
func requireAcks(c *Claim) error {
	for _, box := range c.Op.Participants {
		if c.Op.Journal["exec_ack."+box] == "" {
			return fmt.Errorf("%w: %s has not acknowledged", ErrRelocation, box)
		}
	}
	return nil
}

func failover(op Op) bool {
	return op.Intent["reason"] == ReasonFailover || op.Journal["reason"] == ReasonFailover
}

// RelocationStartGuard is §4's start guard with the relocation exception
// (L11): a park-generation chore in the region does not block, a
// park-replica in the region does, a park-local on the target does, and a
// park-local on a fenced source does not block a failover. home is the
// verified home binding or "": a park-replica queued under the home's
// other name (park.HomeRegion, as moves queues it) is the same region
// (regionname.SameRegion), as BeginAdmittedMove's start guard reads both.
func RelocationStartGuard(c *Claim, home, region, source, target string, sourceFenced bool) error {
	if c.Holder == nil || c.Op.Kind != KindServing || c.Op.ID != "" {
		return fmt.Errorf("%w: kind %q", ErrWrongState, c.Op.Kind)
	}
	fenced := ""
	if sourceFenced {
		fenced = source
	}
	if err := completionBarrier(c, fenced); err != nil {
		return err
	}
	for _, o := range c.Obligations {
		switch {
		case o.Kind == "park-replica" && regionname.SameRegion(home, o.Region, region):
			return fmt.Errorf("%w: park-replica in %s", ErrPendingChore, region)
		case o.Kind == "park-local" && o.Box == target:
			return fmt.Errorf("%w: park-local on %s", ErrPendingChore, target)
		case o.Kind == "park-local" && o.Box == source && !sourceFenced:
			return fmt.Errorf("%w: park-local on %s", ErrPendingChore, source)
		}
	}
	return nil
}

// OwesLocal reports whether box owes a park-local chore on this claim: a
// local retirement not yet done (spec 1 §2: such a box is never chosen to
// install the host, and §4: a park-local on the target refuses).
func OwesLocal(c *Claim, box string) bool {
	for _, o := range c.Obligations {
		if o.Kind == "park-local" && o.Box == box {
			return true
		}
	}
	return false
}

// LocalChoreBoxes lists every box that owes a park-local chore on the
// claim, for a chooser's exclusions.
func LocalChoreBoxes(c *Claim) []string {
	var out []string
	for _, o := range c.Obligations {
		if o.Kind == "park-local" && o.Box != "" && !slices.Contains(out, o.Box) {
			out = append(out, o.Box)
		}
	}
	return out
}

// fenceBox puts box on lost with its fence reference (L2, L4), passes the
// regional work the box owned (park-generation, park-replica owned by
// box:<box>) to its region, region:<r>, keeping each obligation's id,
// operation, retirement, region, generation and park scope (L3, exactly
// as DeclareLost does for a declared loss, lost.go), and, when the volume
// evidence qualifies (L3), discharges that box's park-local chores for
// this host, all in the same CAS. It never removes anything else. D4: the
// same CAS retains the loss outside the operation (evidence_refs,
// loss_refs) with the box's last writer session, the discharge verdict and
// timer knowledge "unknown", so the console can materialize the loss record
// after the operation is long gone.
func fenceBox(c *Claim, box string, ref FenceRef, observedAt string, d *LocalDischarge) error {
	key := "fence." + box
	raw, err := json.Marshal(ref)
	if err != nil {
		return err
	}
	if old, ok := c.Op.Journal[key]; ok && old != string(raw) {
		return fmt.Errorf("%w: fence reference already bound", ErrFence)
	}
	if c.Op.Journal == nil {
		c.Op.Journal = map[string]string{}
	}
	c.Op.Journal[key] = string(raw)
	c.Op.Journal[key+".observed_at"] = observedAt
	if !slices.Contains(c.Lost, box) {
		c.Lost = append(c.Lost, box)
	}
	var reassigned []ObligationRef
	for i := range c.Obligations {
		o := &c.Obligations[i]
		if (o.Kind == "park-generation" || o.Kind == "park-replica") && o.Owner == "box:"+box {
			if o.Region == "" {
				return fmt.Errorf("%w: regional obligation %q names no region", ErrWrongState, o.ID)
			}
			o.Owner = "region:" + o.Region
			reassigned = append(reassigned, ObligationRef{ID: o.ID, Op: o.Op, Retirement: o.Retirement})
		}
	}
	if d == nil {
		d = &LocalDischarge{Reason: "no volume evidence supplied"}
	}
	if err := journalJSON(c, "discharge."+box, d); err != nil {
		return err
	}
	if d.Qualified {
		// Each chore is completed with CompleteObligation's own semantics,
		// so a discovery operation whose final chore this discharges is
		// marked complete exactly as a retirement would (delete.go).
		for {
			var chore *Obligation
			for i := range c.Obligations {
				if c.Obligations[i].Kind == "park-local" && c.Obligations[i].Box == box {
					chore = &c.Obligations[i]
					break
				}
			}
			if chore == nil {
				break
			}
			if err := CompleteObligation(chore.Op, chore.ID)(c); err != nil {
				return err
			}
		}
	}
	return retainAutomaticLoss(c, box, ref, observedAt, *d, reassigned)
}

// retainAutomaticLoss is D4's retention, written once per (operation, box,
// fence): a replay finds its record and changes nothing.
func retainAutomaticLoss(c *Claim, box string, ref FenceRef, observedAt string, d LocalDischarge, reassigned []ObligationRef) error {
	id := automaticLossID(c.Op.ID, box, ref)
	for _, l := range c.EvidenceRefs {
		if l.ID == id {
			return nil
		}
	}
	role := "target"
	if box == c.Op.Intent["source_box"] {
		role = "source"
	}
	loss := AutomaticLoss{ID: id, Box: box, Enrollment: ref.Enrollment, Fence: ref, ObservedAt: observedAt,
		Op: c.Op.ID, Kind: c.Op.Kind, Phase: c.Op.Phase, Role: role, Discharge: d, Reassigned: reassigned, Timers: TimersUnknown}
	if h := c.Holder; h != nil {
		loss.Incarnation, loss.Stamp = h.Incarnation, h.Stamp
		// The box's last writer session: the live one when it is still
		// the writer, otherwise its last admission on this holder.
		if h.WriterBox == box {
			loss.Generation, loss.Session = h.Generation, h.Session
		}
		for i := len(h.Admissions) - 1; i >= 0; i-- {
			if h.Admissions[i].Box == box {
				last := h.Admissions[i]
				loss.LastAdmission = &last
				if loss.Session == 0 {
					loss.Generation, loss.Session = last.Generation, last.Session
				}
				break
			}
		}
	}
	c.EvidenceRefs = append(c.EvidenceRefs, loss)
	if !slices.Contains(c.LossRefs, id) {
		c.LossRefs = append(c.LossRefs, id)
	}
	return nil
}

func removeParticipant(c *Claim, box string) {
	c.Op.Participants = slices.DeleteFunc(slices.Clone(c.Op.Participants), func(b string) bool { return b == box })
}

func addParticipant(c *Claim, box string) {
	if !slices.Contains(c.Op.Participants, box) {
		c.Op.Participants = append(c.Op.Participants, box)
	}
}

// installedBoxes is every box this operation handed op.exec for an
// installation; each gets a park-local unless discharged (claim §4:
// "Every abandoned local installation receives an exact park-local").
func noteInstalledBox(c *Claim, box string) error {
	var boxes []string
	readJournal(c.Op, "installed_boxes", &boxes)
	if !slices.Contains(boxes, box) {
		boxes = append(boxes, box)
	}
	return journalJSON(c, "installed_boxes", boxes)
}

// StartRelocation is the start CAS (spec 1 §5.8, claim §4). A drain or
// rebalance starts with op.exec on the source; a failover cites the
// source's termination fence, starts with op.exec "none", keeps the stamp,
// and puts the source on lost. The caller has already checked the roster
// (hostactors.Directory.CheckBoxStart), the target's eligibility and the
// app-admission registration; this CAS records what it checked.
func StartRelocation(ctx context.Context, s release.Store, observed Observed, in RelocationStart, authority FenceAuthority) (Observed, error) {
	if observed.Absent || observed.Claim == nil || observed.ETag == "" {
		return Observed{}, ErrWrongState
	}
	switch in.Reason {
	case ReasonDrain, ReasonRebalance:
		if in.Fence != nil {
			return Observed{}, fmt.Errorf("%w: only a failover cites a fence", ErrRelocation)
		}
	case ReasonFailover:
		if in.Fence == nil {
			return Observed{}, fmt.Errorf("%w: a failover cites the source's fence", ErrRelocation)
		}
	default:
		return Observed{}, fmt.Errorf("%w: reason %q", ErrRelocation, in.Reason)
	}
	if in.HomeRegion != "" && regionname.PlacementRegion(in.HomeRegion) != in.HomeRegion {
		return Observed{}, fmt.Errorf("%w: home region %q is not a placement code", ErrRelocation, in.HomeRegion)
	}
	if in.OpID == "" || in.Region == "" || in.AdmissionID == "" || in.DirectoryRevision == 0 || in.Now.IsZero() ||
		!validBoxReference(in.SourceBox) || !validBoxReference(in.TargetBox) || in.SourceBox == "none" || in.TargetBox == "none" ||
		in.SourceBox == in.TargetBox || in.SourceEnrollment == "" || in.TargetEnrollment == "" {
		return Observed{}, fmt.Errorf("%w: incomplete start", ErrRelocation)
	}
	var observedAt string
	if in.Fence != nil {
		v, err := VerifyFence(ctx, s, authority, *in.Fence)
		if err != nil {
			return Observed{}, err
		}
		if v.record.Box != in.SourceBox || v.record.Enrollment != in.SourceEnrollment || v.binding.Region != in.Region {
			return Observed{}, ErrFence
		}
		observedAt = v.record.ObservedAt
	}
	return Update(ctx, s, observed, func(c *Claim) error {
		if c.Op.Exec != in.SourceBox || slices.Contains(c.Lost, in.TargetBox) {
			return fmt.Errorf("%w: the source is not the executor, or the target is lost", ErrWrongState)
		}
		if err := RelocationStartGuard(c, in.HomeRegion, in.Region, in.SourceBox, in.TargetBox, in.Fence != nil); err != nil {
			return err
		}
		exec, participants := in.SourceBox, []string{in.SourceBox, in.TargetBox}
		if in.Fence != nil {
			exec, participants = "none", []string{in.TargetBox}
		}
		c.Op = Op{Kind: KindRelocating, ID: in.OpID, Phase: PhaseStarting, By: in.By, StartedAt: in.Now.UTC(),
			Exec: exec, Participants: participants,
			Intent: map[string]string{
				"region": in.Region, "reason": in.Reason,
				"source_box": in.SourceBox, "source_enrollment": in.SourceEnrollment,
				"target_box": in.TargetBox, "target_enrollment": in.TargetEnrollment,
				"stamp_from":         strconv.FormatUint(c.Holder.Stamp, 10),
				"generation_from":    c.Holder.Generation,
				"directory_revision": strconv.FormatUint(in.DirectoryRevision, 10),
				"admission_id":       in.AdmissionID,
			},
			Journal: map[string]string{}}
		if in.HomeRegion != "" {
			c.Op.Intent["home_region"] = in.HomeRegion
		}
		if in.Fence != nil {
			return fenceBox(c, in.SourceBox, *in.Fence, observedAt, in.Discharge)
		}
		return nil
	})
}

// FenceSource is §5.8's source-loss rule (B1-6): a drain or rebalance
// whose source is terminated becomes a failover in one CAS citing the
// source's fence. In starting it stays in starting with the source off the
// barrier; in releasing before a release is journaled it goes to pinning
// with the fence as evidence; after a release (pinning on) the source only
// goes on lost. Every case keeps the stamp and the intent.
func FenceSource(ctx context.Context, s release.Store, observed Observed, opID string, ref FenceRef, d *LocalDischarge, authority FenceAuthority) (Observed, error) {
	if observed.Absent || observed.Claim == nil {
		return Observed{}, ErrWrongState
	}
	before := observed.Claim.Op
	source := before.Intent["source_box"]
	v, err := VerifyFence(ctx, s, authority, ref)
	if err != nil {
		return Observed{}, err
	}
	if v.record.Box != source || v.record.Enrollment != before.Intent["source_enrollment"] || v.binding.Region != before.Intent["region"] {
		return Observed{}, ErrFence
	}
	return Update(ctx, s, observed, func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindRelocating || c.Op.Intent["source_box"] != source {
			return ErrWrongState
		}
		ensureJournal(c)
		// Before its release is journaled the source's writes are not
		// kept by a release, so the op becomes a failover. After it, only
		// the source's liveness changes.
		if c.Op.Phase == PhaseStarting || (c.Op.Phase == PhaseReleasing && c.Op.Journal["release"] == "") {
			c.Op.Journal["reason"] = ReasonFailover
			removeParticipant(c, source)
			c.Op.Exec = "none"
			if c.Op.Phase == PhaseReleasing {
				c.Op.Phase = PhasePinning
			}
		}
		return fenceBox(c, source, ref, v.record.ObservedAt, d)
	})
}

// ToReleasing is a drain's barrier CAS: from starting, once every
// participant has acknowledged. A failover never enters releasing. now
// starts the release deadline (hostrelocate.ReleaseDeadline): a source
// that never journals its release goes to stuck.
func ToReleasing(opID string, now time.Time) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseStarting); err != nil {
			return err
		}
		if failover(c.Op) {
			return fmt.Errorf("%w: a failover skips releasing", ErrRelocation)
		}
		if err := requireAcks(c); err != nil {
			return err
		}
		if now.IsZero() {
			return fmt.Errorf("%w: releasing needs its start time", ErrRelocation)
		}
		c.Op.Phase = PhaseReleasing
		c.Op.Journal["releasing_since"] = now.UTC().Format(time.RFC3339Nano)
		return nil
	}
}

// JournalRelease records a live box's completed release sequence. The
// source records it in releasing; a failed target in probe-failed. The
// release must be of the box's own current writer session.
func JournalRelease(opID string, r Release) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseReleasing, PhaseProbeFailed); err != nil {
			return err
		}
		h := c.Holder
		writerMismatch := h.Session != 0 && h.WriterBox != r.ReleasedBy // a host never woken has no writer
		if r.ReleasedBy == "" || r.ReleasedBy != c.Op.Exec || r.Enrollment == "" || r.Receipt == "" || !isHex64(r.ReceiptSHA256) ||
			r.Generation != h.Generation || r.Session != h.Session || writerMismatch {
			return fmt.Errorf("%w: release is not the executor's current session", ErrRelocation)
		}
		want := c.Op.Intent["source_enrollment"]
		if c.Op.Phase == PhaseProbeFailed {
			want = c.Op.Intent["target_enrollment"]
		}
		if r.Enrollment != want {
			return fmt.Errorf("%w: release enrollment", ErrRelocation)
		}
		var old Release
		if readJournal(c.Op, "release", &old) && c.Op.Journal["release.phase"] == c.Op.Phase {
			if old == r {
				return ErrAlreadyRecorded
			}
			return fmt.Errorf("%w: a different release is journaled", ErrRelocation)
		}
		c.Op.Journal["release.phase"] = c.Op.Phase
		return journalJSON(c, "release", r)
	}
}

// PinAfterRelease leaves releasing (a drain) or probe-failed (a failed
// target, §5.8's nested relocation) for pinning, citing the journaled
// release's verified receipt. op.exec becomes "none" and the stamp is kept
// (claim R6-4, D5's non-installing form). From probe-failed, next names
// the replacement target, which must not be the failed one.
func PinAfterRelease(ctx context.Context, s release.Store, observed Observed, opID string, authority ReleaseAuthority, next *Target) (Observed, error) {
	if observed.Absent || observed.Claim == nil || authority == nil {
		return Observed{}, ErrWrongState
	}
	var r Release
	if !readJournal(observed.Claim.Op, "release", &r) {
		return Observed{}, fmt.Errorf("%w: no journaled release", ErrRelocation)
	}
	if err := authority.VerifyRelease(ctx, *observed.Claim, r); err != nil {
		return Observed{}, err
	}
	return Update(ctx, s, observed, func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseReleasing, PhaseProbeFailed); err != nil {
			return err
		}
		var now Release
		if !readJournal(c.Op, "release", &now) || now != r || c.Op.Journal["release.phase"] != c.Op.Phase || r.ReleasedBy != c.Op.Exec {
			return fmt.Errorf("%w: release changed", ErrRelocation)
		}
		if c.Op.Phase == PhaseReleasing {
			if next != nil {
				return fmt.Errorf("%w: a drain keeps its target", ErrRelocation)
			}
			// §5.8: a target replaced in releasing acknowledges before the
			// operation leaves the phase.
			if err := requireAcks(c); err != nil {
				return err
			}
		} else {
			if next == nil || next.Box == r.ReleasedBy {
				return fmt.Errorf("%w: a nested relocation names another target", ErrRelocation)
			}
			if err := replaceTargetSlot(c, *next); err != nil {
				return err
			}
			c.Op.Journal["released_by_target"] = r.ReleasedBy
		}
		c.Op.Exec, c.Op.Phase = "none", PhasePinning
		delete(c.Op.Journal, "pin")
		return nil
	})
}

// PinFailover leaves starting for pinning on a failover: the source's
// fence is journaled, the target is named (a converted committed landing
// names it in starting, NameTarget) and every remaining participant has
// acknowledged, so the pin is selected only after the target's
// acknowledgement.
func PinFailover(opID string) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseStarting); err != nil {
			return err
		}
		if !failover(c.Op) || c.Op.Journal["fence."+c.Op.Intent["source_box"]] == "" || c.Op.Exec != "none" {
			return fmt.Errorf("%w: pinning without a release needs the source's fence", ErrRelocation)
		}
		if c.Op.Intent["target_box"] == "" {
			return fmt.Errorf("%w: a failover names its target before it pins", ErrRelocation)
		}
		if err := requireAcks(c); err != nil {
			return err
		}
		c.Op.Phase = PhasePinning
		return nil
	}
}

// JournalPin records the pinned recovery set in pinning. The pin's key
// must be RecoveryPinKey for this claim generation: the object was selected
// against exactly this observation (B2-3). After a release it must reach
// the journaled final TXID and files sequence; on a failover it must be
// taken no earlier than the fence's observed_at plus FailoverPinDelay. An
// Empty pin needs its evidence: a release at final_txid 0, or a holder
// that never admitted a writer.
func JournalPin(opID string, p RecoveryPin) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhasePinning); err != nil {
			return err
		}
		if p.Key != RecoveryPinKey(c.Host, opID, c.Gen) {
			return fmt.Errorf("%w: pin %s was not selected against this observation (gen %d)", ErrRelocation, p.Key, c.Gen)
		}
		if !isHex64(p.SHA256) || p.FilesManifest == "" || p.PinnedAt.IsZero() || (p.Empty != (p.MaxTXID == 0)) || p.Release == "" {
			return fmt.Errorf("%w: incomplete pin", ErrRelocation)
		}
		var r Release
		released := readJournal(c.Op, "release", &r)
		if released {
			if p.MaxTXID < r.FinalTXID || p.FilesSeq < r.ManifestSeq {
				return fmt.Errorf("%w: pin %d/%d is short of the release's %d/%d", ErrRelocation, p.MaxTXID, p.FilesSeq, r.FinalTXID, r.ManifestSeq)
			}
		}
		if p.Empty && !((released && r.FinalTXID == 0) || (!released && c.Holder.Session == 0)) {
			return fmt.Errorf("%w: an empty pin needs a release at txid 0 or a holder that never had a writer", ErrRelocation)
		}
		for key := range c.Op.Journal {
			// A fence journaled without its observation time (a move
			// converted to relocating, which binds fence.<box> alone) owes
			// the hold too: absence is never evidence, so the pin waits
			// until FenceSource has consumed the fence and journaled when
			// it was observed (H3).
			if strings.HasPrefix(key, "fence.") && !strings.HasSuffix(key, ".observed_at") && c.Op.Journal[key+".observed_at"] == "" {
				return fmt.Errorf("%w: %s has no observation time; consume the fence through FenceSource before pinning", ErrRelocation, key)
			}
		}
		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 || p.PinnedAt.Before(t.Add(FailoverPinDelay)) {
					return fmt.Errorf("%w: pinned before the fence's observation plus %s", ErrRelocation, FailoverPinDelay)
				}
			}
		}
		if readJournal(c.Op, "pin", new(RecoveryPin)) {
			return ErrAlreadyRecorded
		}
		return journalJSON(c, "pin", p)
	}
}

// NameTarget names the target of an operation that has none: a move
// converted to relocating before its flip (claim §4, moves §7.10), at
// pinning; or a committed landing converted to a failover
// (ConvertCommittedLanding), in starting, where the named box then
// acknowledges before PinFailover selects any pin, as an initial
// failover's target does. The named box acknowledges (D6) before
// ToInstalling hands it op.exec. The move's own target (move_target_box,
// in the region the move was going to) leaves the barrier in the same CAS:
// readiness binds to the box named here (moves §7.10), and a failover's
// participant is its target alone (§5.8). It never held op.exec on the
// converted operation.
func NameTarget(opID string, t Target) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseStarting, PhasePinning); err != nil {
			return err
		}
		if c.Op.Phase == PhaseStarting && (!failover(c.Op) || c.Op.Exec != "none") {
			return fmt.Errorf("%w: only a failover with no executor names its target in starting", ErrRelocation)
		}
		if c.Op.Intent["target_box"] != "" {
			return fmt.Errorf("%w: the operation already has a target", ErrRelocation)
		}
		if moved := c.Op.Intent["move_target_box"]; moved != "" {
			removeParticipant(c, moved)
		}
		return replaceTargetSlot(c, t)
	}
}

// ToInstalling hands op.exec to the target for a new installation: a new
// stamp (this CAS's gen, R6-4), journaled as install_stamp, a fresh
// prepared generation id, and the target added to touched.boxes before its
// first effect. A converted operation with no target names it here.
func ToInstalling(opID string, t Target, prepared string) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhasePinning); err != nil {
			return err
		}
		if !readJournal(c.Op, "pin", new(RecoveryPin)) || prepared == "" || prepared == c.Holder.Generation {
			return fmt.Errorf("%w: installing needs a pin and a fresh prepared generation", ErrRelocation)
		}
		if t.Box != c.Op.Intent["target_box"] || t.Enrollment != c.Op.Intent["target_enrollment"] || slices.Contains(c.Lost, t.Box) {
			return fmt.Errorf("%w: not the operation's target", ErrRelocation)
		}
		if OwesLocal(c, t.Box) {
			return fmt.Errorf("%w: park-local on %s", ErrPendingChore, t.Box)
		}
		if err := requireAcks(c); err != nil {
			return err
		}
		return install(c, t.Box, prepared, PhaseInstalling)
	}
}

// abandonPrepared journals the operation's current P as abandoned when
// no writer opened it, before a transition replaces or drops it: nothing
// will open it now, and the serving CAS parks it (rev 17, finding F3;
// rev 18: every transition that discards an unopened P calls it).
func abandonPrepared(c *Claim) error {
	old := c.Op.Journal["prepared"]
	if old == "" || preparedOpened(c) {
		return nil
	}
	var abandoned []string
	readJournal(c.Op, "abandoned_prepared", &abandoned)
	if slices.Contains(abandoned, old) {
		return nil
	}
	return journalJSON(c, "abandoned_prepared", append(abandoned, old))
}

// install hands op.exec to box for an installation at a new stamp. The
// same CAS journals install_stamp and install_release, the journaled pin's
// release (R9): a replayed or re-targeted installation copies the same
// pin's value and never reads anything else. The two keys are written
// together and only here, and no transition clears one without the other
// (only the serving CAS, which ends the operation, drops both), so
// install_release always names the release of the installation
// install_stamp names.
func install(c *Claim, box, prepared, phase string) error {
	var pin RecoveryPin
	if !readJournal(c.Op, "pin", &pin) || pin.Release == "" {
		return fmt.Errorf("%w: an installation needs the journaled pin's release", ErrRelocation)
	}
	if c.Op.Journal["prepared"] != prepared {
		if err := abandonPrepared(c); err != nil {
			return err
		}
	}
	stamp := c.Gen + 1 // the resulting gen: Update sets gen = observed+1
	c.Op.Exec, c.Op.Phase = box, phase
	c.Holder.Stamp = stamp
	c.Op.Journal["install_stamp"] = strconv.FormatUint(stamp, 10)
	c.Op.Journal["install_release"] = pin.Release
	c.Op.Journal["prepared"] = prepared
	delete(c.Op.Journal, "local_install")
	if !slices.Contains(c.Touched.Boxes, box) {
		c.Touched.Boxes = append(c.Touched.Boxes, box)
	}
	return noteInstalledBox(c, box)
}

// ToInstalled records the verified remote build of P (§5.8, the
// no-reinstall boundary): the install manifest at the pin's N (none for an
// empty pin) and, always, P's files manifest at the pinned sequence, each
// verified by a fresh read. Nobody builds P again after this CAS.
func ToInstalled(opID string, a PreparedAuthority) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseInstalling); err != nil {
			return err
		}
		var pin RecoveryPin
		readJournal(c.Op, "pin", &pin)
		if a.Generation == "" || a.Generation != c.Op.Journal["prepared"] || a.SnapshotTXID != pin.MaxTXID ||
			(pin.MaxTXID > 0 && (a.InstallManifest == "" || !isHex64(a.SHA256))) ||
			(pin.MaxTXID == 0 && (a.InstallManifest != "" || a.SHA256 != "")) {
			return fmt.Errorf("%w: prepared authority is not this pin's build of P", ErrRelocation)
		}
		if a.FilesSeq != 0 || a.SourceFilesSeq != pin.FilesSeq || !isHex64(a.FilesManifestSHA256) {
			return fmt.Errorf("%w: prepared authority lacks P's prepared files manifest (sequence 0, built from the pinned sequence)", ErrRelocation)
		}
		if err := requireAcks(c); err != nil {
			return err
		}
		stamp, err := journalUint(c.Op, "install_stamp")
		if err != nil {
			return err
		}
		c.Op.Phase = PhaseInstalled
		a.InstallStamp = stamp
		return journalJSON(c, "prepared_authority", a)
	}
}

// PreparedOpened reports whether P has been opened: journaled as
// prepared_opened by the admission that opened it (rev 12: AdmitWriter
// trims holder.admissions to eight, so the admission list alone could
// forget it), or named by a retained admission.
func PreparedOpened(c *Claim) bool { return preparedOpened(c) }

func preparedOpened(c *Claim) bool {
	p := c.Op.Journal["prepared"]
	if p == "" || c.Holder == nil {
		return false
	}
	if c.Op.Journal["prepared_opened"] == p {
		return true
	}
	for _, a := range c.Holder.Admissions {
		if a.Generation == p {
			return true
		}
	}
	return false
}

// RecordLocalInstall journals the executor's local install receipt in
// installed (B1-2). It is bound to op.exec, the target's enrollment and the
// journaled install_stamp, and refused once any admission has named P
// (B2-5): after that the local install is never remade.
func RecordLocalInstall(opID string, l LocalInstall) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseInstalled); err != nil {
			return err
		}
		if err := requireReceiptOfInstallation(c, l); err != nil {
			return err
		}
		var old LocalInstall
		if readJournal(c.Op, "local_install", &old) {
			if old == l {
				return ErrAlreadyRecorded
			}
			return fmt.Errorf("%w: a different receipt is journaled", ErrRelocation)
		}
		return journalJSON(c, "local_install", l)
	}
}

// ReplaceLocalInstall replaces the journaled receipt old with l after a
// repair of the local install under the same installation (stage B final
// review A3): the repair re-renders the held configuration, which may
// have changed since old was journaled, so the box's durable journal and
// its digest are new while the box, enrollment and install stamp are the
// same. Only before P is admitted (an opened P is never reinstalled), only
// over exactly old, and only for the same installation: the three readers
// of the receipt (the pre-admission verification, the post-admission
// check and the receipt check after P opens) then compare one digest.
func ReplaceLocalInstall(opID string, old, l LocalInstall) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseInstalled, PhaseRelocatingProbing); err != nil {
			return err
		}
		if err := requireReceiptOfInstallation(c, l); err != nil {
			return err
		}
		if l.Box != old.Box || l.Enrollment != old.Enrollment || l.InstallStamp != old.InstallStamp {
			return fmt.Errorf("%w: the repaired receipt is not the journaled installation's", ErrRelocation)
		}
		var cur LocalInstall
		if !readJournal(c.Op, "local_install", &cur) || cur != old {
			return fmt.Errorf("%w: the journaled receipt is not the one being replaced", ErrRelocation)
		}
		if cur == l {
			return ErrAlreadyRecorded
		}
		return journalJSON(c, "local_install", l)
	}
}

// requireReceiptOfInstallation is the receipt's binding to the operation's
// installation (RecordLocalInstall, ReplaceLocalInstall): the executor,
// its enrollment, the journaled install stamp, a well-formed digest, and
// P not yet admitted.
func requireReceiptOfInstallation(c *Claim, l LocalInstall) error {
	stamp, err := journalUint(c.Op, "install_stamp")
	if err != nil {
		return err
	}
	if l.Box != c.Op.Exec || l.Enrollment != c.Op.Intent["target_enrollment"] ||
		l.InstallStamp == 0 || l.InstallStamp != stamp || !isHex64(l.JournalSHA256) {
		return fmt.Errorf("%w: receipt is not this installation's", ErrRelocation)
	}
	if preparedOpened(c) {
		return fmt.Errorf("%w: P was admitted; the local install is never remade", ErrRelocation)
	}
	return nil
}

// ToProbing enters probing, which requires a local install receipt for
// exactly op.exec, its enrollment and the journaled install_stamp. now
// starts the probe deadline (§5.8: 10 minutes).
func ToProbing(opID string, now time.Time) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseInstalled); err != nil {
			return err
		}
		stamp, err := journalUint(c.Op, "install_stamp")
		if err != nil {
			return err
		}
		var l LocalInstall
		if !readJournal(c.Op, "local_install", &l) || l.Box != c.Op.Exec || l.Enrollment != c.Op.Intent["target_enrollment"] ||
			l.InstallStamp != stamp {
			return fmt.Errorf("%w: no local install receipt for the executor", ErrRelocation)
		}
		if err := requireAcks(c); err != nil {
			return err
		}
		if now.IsZero() {
			return fmt.Errorf("%w: probing needs its start time", ErrRelocation)
		}
		c.Op.Phase = PhaseRelocatingProbing
		c.Op.Journal["probing_since"] = now.UTC().Format(time.RFC3339Nano)
		return nil
	}
}

// BeginProbeAttempt counts one probe attempt for the executor before it
// is made (§5.8 "limits"): the fourth is refused with
// ErrProbeAttemptsExhausted, and the participant then enters
// probe-failed. The count is per box and per operation, so a replacement
// target starts at zero and a fenced one is never counted again.
func BeginProbeAttempt(opID, box string) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseRelocatingProbing); err != nil {
			return err
		}
		if box == "" || box != c.Op.Exec {
			return fmt.Errorf("%w: only the executor probes", ErrRelocation)
		}
		key := "probe_attempts." + box
		n, err := journalUint(c.Op, key)
		if err != nil {
			return err
		}
		if n >= MaxProbeAttempts {
			return ErrProbeAttemptsExhausted
		}
		c.Op.Journal[key] = strconv.FormatUint(n+1, 10)
		return nil
	}
}

// RecordProbe journals the executor's passing probe of its admitted session.
func RecordProbe(opID string, p Probe) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseRelocatingProbing); err != nil {
			return err
		}
		stamp, err := journalUint(c.Op, "install_stamp")
		if err != nil {
			return err
		}
		if p.Box != c.Op.Exec || p.Box != c.Holder.WriterBox || p.Session == 0 || p.Session != c.Holder.Session ||
			p.Session < stamp {
			return fmt.Errorf("%w: probe is not of the executor's admitted session", ErrRelocation)
		}
		return journalJSON(c, "probe", p)
	}
}

// ToProbeFailed records that the live target's probe failed or missed its
// deadline. The target then runs its release sequence (JournalRelease);
// now starts its release deadline.
func ToProbeFailed(opID string, now time.Time) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseRelocatingProbing); err != nil {
			return err
		}
		if now.IsZero() {
			return fmt.Errorf("%w: probe-failed needs its start time", ErrRelocation)
		}
		c.Op.Phase = PhaseProbeFailed
		c.Op.Journal["probe_failed_since"] = now.UTC().Format(time.RFC3339Nano)
		delete(c.Op.Journal, "release")
		delete(c.Op.Journal, "release.phase")
		return nil
	}
}

// CompleteRelocation is the serving CAS: the target's probe passed for its
// admitted session, so the host serves on it alone. Every other box this
// operation installed on, and a live source, owes a park-local unless a
// fence discharged it.
func CompleteRelocation(opID string, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseRelocatingProbing); err != nil {
			return err
		}
		var p Probe
		if !readJournal(c.Op, "probe", &p) || p.Box != c.Op.Exec || p.Session != c.Holder.Session || c.Holder.WriterBox != p.Box {
			return fmt.Errorf("%w: no passing probe of the current session", ErrRelocation)
		}
		if !probeStopped(c, p) {
			// Ruling 49: the kind change ends the probe's session, so the
			// session is stopped, with root's receipt, before it.
			return fmt.Errorf("%w: the probe's session is not stopped", ErrRelocation)
		}
		target := c.Op.Exec
		var owed []string
		readJournal(c.Op, "installed_boxes", &owed)
		owed = append([]string{c.Op.Intent["source_box"]}, owed...)
		for _, box := range owed {
			if box == target || box == "" {
				continue
			}
			var d LocalDischarge
			if readJournal(c.Op, "discharge."+box, &d) && d.Qualified {
				continue
			}
			exists := false
			for _, o := range c.Obligations {
				if o.Kind == "park-local" && o.Box == box {
					exists = true
				}
			}
			if !exists {
				c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-local", Box: box, Retirement: newID(), Owner: "box:" + box})
			}
		}
		// Every prepared generation this operation built and no writer
		// opened is parked here, in the region, by the claim's own
		// park-generation chore (rev 17, finding F3); the at-rest sweep
		// stays the backstop the spec names.
		var abandoned []string
		readJournal(c.Op, "abandoned_prepared", &abandoned)
		region := c.Op.Intent["region"]
		for _, g := range abandoned {
			if g == "" || g == c.Holder.Generation || region == "" {
				continue
			}
			queued := false
			for _, o := range c.Obligations {
				if o.Kind == "park-generation" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, region) && o.Generation == g {
					queued = true
				}
			}
			if !queued {
				c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-generation", Region: region, Generation: g, Retirement: newID(), Owner: "region:" + region})
			}
		}
		var stop ProbeStop
		readJournal(c.Op, "probe_stop", &stop)
		done, err := json.Marshal(CompletedRelocation{Op: opID, Stop: stop, Stamp: c.Holder.Stamp, Phase: c.Op.Phase, Generation: c.Holder.Generation})
		if err != nil {
			return err
		}
		// Ruling 56: the serving op keeps the stopped probe session's
		// receipt reference for the next cold wake's handover.
		next := Op{Kind: KindServing, Exec: target, Participants: []string{target}, Journal: map[string]string{JournalCompleted: string(done)}}
		if c.Op.Journal["conversion"] == "committed-landing" {
			// M1 on amadan #236: the move this relocation finished is named
			// on the serving op, so moves' close can find it.
			next.Journal[journalReleasedMove] = opID
		}
		c.Op = next
		return nil
	}
}

// replaceTargetSlot names t as the target, counting a replacement when it
// replaces another box. A replaced box is never named again for this host.
func replaceTargetSlot(c *Claim, t Target) error {
	if !validBoxReference(t.Box) || t.Box == "none" || t.Enrollment == "" || t.Box == c.Op.Intent["source_box"] || slices.Contains(c.Lost, t.Box) {
		return fmt.Errorf("%w: invalid target", ErrRelocation)
	}
	var excluded []string
	readJournal(c.Op, "excluded_targets", &excluded)
	if slices.Contains(excluded, t.Box) {
		return fmt.Errorf("%w: %s was replaced in this operation", ErrRelocation, t.Box)
	}
	if OwesLocal(c, t.Box) {
		// Spec 1 §2: a box owing a local retirement for this host is never
		// handed an installation, in any phase a replacement resumes at
		// (install is called below without ToInstalling's check).
		return fmt.Errorf("%w: park-local on %s", ErrPendingChore, t.Box)
	}
	old := c.Op.Intent["target_box"]
	if old != "" && old != t.Box {
		n, err := journalUint(c.Op, "replacements")
		if err != nil {
			return err
		}
		if n >= MaxTargetReplacements {
			return ErrReplacementsExceeded
		}
		c.Op.Journal["replacements"] = strconv.FormatUint(n+1, 10)
		excluded = append(excluded, old)
		if err := journalJSON(c, "excluded_targets", excluded); err != nil {
			return err
		}
		removeParticipant(c, old)
	}
	c.Op.Intent["target_box"], c.Op.Intent["target_enrollment"] = t.Box, t.Enrollment
	addParticipant(c, t.Box)
	return nil
}

// ReplaceTarget is §5.8's replacement table for a target terminated by its
// fence. It records the fence (the old target goes on lost), names next,
// and resumes at the phase the table gives:
//
//	starting, releasing, pinning      -> the same phase
//	installing                        -> installing: next is op.exec, new stamp, fresh P'
//	installed, or probing before P    -> installed: next is op.exec, new stamp, same P, no receipt
//	probing after P, or probe-failed  -> pinning: a nested failover from holder.generation
//
// prepared is P' for installing and ignored otherwise. A third
// replacement is refused with ErrReplacementsExceeded; the worker then
// marks the operation stuck.
func ReplaceTarget(ctx context.Context, s release.Store, observed Observed, opID string, ref FenceRef, d *LocalDischarge, next Target, prepared string, authority FenceAuthority) (Observed, error) {
	if observed.Absent || observed.Claim == nil {
		return Observed{}, ErrWrongState
	}
	old := observed.Claim.Op.Intent["target_box"]
	v, err := VerifyFence(ctx, s, authority, ref)
	if err != nil {
		return Observed{}, err
	}
	if old == "" || v.record.Box != old || v.record.Enrollment != observed.Claim.Op.Intent["target_enrollment"] || v.binding.Region != observed.Claim.Op.Intent["region"] {
		return Observed{}, ErrFence
	}
	return Update(ctx, s, observed, func(c *Claim) error {
		if err := requireRelocation(c, opID, PhaseStarting, PhaseReleasing, PhasePinning, PhaseInstalling, PhaseInstalled, PhaseRelocatingProbing, PhaseProbeFailed); err != nil {
			return err
		}
		if c.Op.Intent["target_box"] != old {
			return ErrStaleOp
		}
		phase := c.Op.Phase
		if err := fenceBox(c, old, ref, v.record.ObservedAt, d); err != nil {
			return err
		}
		if err := replaceTargetSlot(c, next); err != nil {
			return err
		}
		switch {
		case phase == PhaseStarting || phase == PhaseReleasing || phase == PhasePinning:
			return nil
		case phase == PhaseInstalling:
			if prepared == "" || prepared == c.Op.Journal["prepared"] || prepared == c.Holder.Generation {
				return fmt.Errorf("%w: a replacement in installing builds a fresh P'", ErrRelocation)
			}
			return install(c, next.Box, prepared, PhaseInstalling)
		case (phase == PhaseInstalled || phase == PhaseRelocatingProbing) && !preparedOpened(c):
			return install(c, next.Box, c.Op.Journal["prepared"], PhaseInstalled)
		default:
			// After P's first admission, or from probe-failed: nested
			// failover from pinning, pinning from holder.generation as
			// observed. A P no writer opened (a probe deadline before any
			// admission) is recorded before it is dropped.
			if err := abandonPrepared(c); err != nil {
				return err
			}
			c.Op.Exec, c.Op.Phase = "none", PhasePinning
			for _, k := range []string{"pin", "prepared", "prepared_authority", "local_install", "probe", "release", "release.phase"} {
				delete(c.Op.Journal, k)
			}
			return nil
		}
	})
}

// FenceExhaustedTarget is the terminal transition for a target terminated
// after the operation has used its replacements (§5.8 "The bound"; round 3
// finding 5): it consumes the fence (the loss is retained, the box goes on
// lost) and records stuck. While another box is op.exec (the live source
// in starting or releasing; round 8 finding 1) nothing else changes: the
// target's fence revokes nothing of the source's. Otherwise (op.exec is
// the dead target, or none) it revokes the dead installation's authority (op.exec "none", the
// target off the barrier, its receipt dropped), leaves the operation at
// pinning with its pin and, when P was never opened, its remote build kept
// for the operator's replacement, or a nested pinning when P was opened,
// and records stuck in the same CAS. No third target is chosen. MarkStuck's
// "never between installed and a probe" is for a live target; a dead one
// has no probe to make.
func FenceExhaustedTarget(ctx context.Context, s release.Store, observed Observed, opID string, ref FenceRef, d *LocalDischarge, authority FenceAuthority) (Observed, error) {
	return fenceTargetStuck(ctx, s, observed, opID, ref, d, authority, "")
}

// FenceTargetWithoutSpare is FenceExhaustedTarget's transition for a
// terminated target with replacements left but no eligible box to name
// (rev 18, round 17 finding 2): the worker calls it once its bound has
// passed. The fence is consumed in the same CAS (loss retained, the box
// on lost, a dead executor's authority revoked), and the operation is
// stuck for an operator with reason, never left waiting unbounded.
func FenceTargetWithoutSpare(ctx context.Context, s release.Store, observed Observed, opID string, ref FenceRef, d *LocalDischarge, authority FenceAuthority, reason string) (Observed, error) {
	if reason == "" {
		return Observed{}, fmt.Errorf("%w: a stuck operation names its reason", ErrRelocation)
	}
	return fenceTargetStuck(ctx, s, observed, opID, ref, d, authority, reason)
}

// fenceTargetStuck is both: noSpare "" is the exhausted bound.
func fenceTargetStuck(ctx context.Context, s release.Store, observed Observed, opID string, ref FenceRef, d *LocalDischarge, authority FenceAuthority, noSpare string) (Observed, error) {
	if observed.Absent || observed.Claim == nil {
		return Observed{}, ErrWrongState
	}
	old := observed.Claim.Op.Intent["target_box"]
	v, err := VerifyFence(ctx, s, authority, ref)
	if err != nil {
		return Observed{}, err
	}
	if old == "" || v.record.Box != old || v.record.Enrollment != observed.Claim.Op.Intent["target_enrollment"] || v.binding.Region != observed.Claim.Op.Intent["region"] {
		return Observed{}, ErrFence
	}
	return Update(ctx, s, observed, func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindRelocating || c.Op.Intent["target_box"] != old {
			return ErrStaleOp
		}
		ensureJournal(c)
		if c.Op.Journal["stuck"] != "" {
			return ErrAlreadyRecorded
		}
		reason := noSpare
		if reason == "" {
			n, err := journalUint(c.Op, "replacements")
			if err != nil {
				return err
			}
			if n < MaxTargetReplacements {
				return fmt.Errorf("%w: replacements remain; replace the target instead", ErrRelocation)
			}
			reason = fmt.Sprintf("%s terminated after %d target replacements", old, MaxTargetReplacements)
		}
		if err := fenceBox(c, old, ref, v.record.ObservedAt, d); err != nil {
			return err
		}
		removeParticipant(c, old)
		if c.Op.Exec != old && c.Op.Exec != "none" {
			// The executor is another box (the live source in starting or
			// releasing): the target's fence is no evidence against it.
			// Its execution identity, phase and release records stay, so
			// root still accepts its own stop (claimstop.go requires the
			// requester to be op.exec); only stuck is new.
			c.Op.Journal["stuck"] = reason
			return nil
		}
		opened := preparedOpened(c)
		c.Op.Exec, c.Op.Phase = "none", PhasePinning
		keys := []string{"local_install", "probe", "release", "release.phase", "probing_since", "probe_failed_since"}
		if opened {
			// Nested: pinning again from holder.generation, as a replacement
			// after P's first admission would.
			keys = append(keys, "pin", "prepared", "prepared_authority")
		}
		// An unopened P is kept for the operator's replacement; it is on
		// the abandoned list only if a later transition drops it.
		for _, k := range keys {
			delete(c.Op.Journal, k)
		}
		if c.Op.Journal["stuck"] == "" {
			c.Op.Journal["stuck"] = reason
		}
		return nil
	})
}

// MarkStuck stops the operation at its current phase for an operator
// (`carlos ops host resolve`). It never resets the replacement count and
// never invents evidence; every later relocation transition refuses.
func MarkStuck(opID, reason string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindRelocating || reason == "" {
			return ErrWrongState
		}
		ensureJournal(c)
		if c.Op.Phase == PhaseInstalled && c.Op.Journal["local_install"] != "" && !preparedOpened(c) {
			return fmt.Errorf("%w: never stop between installed and a probe", ErrRelocation)
		}
		if c.Op.Journal["stuck"] != "" {
			return ErrAlreadyRecorded
		}
		c.Op.Journal["stuck"] = reason
		return nil
	}
}