package host

import (
	"context"
	"encoding/json"
	"errors"
	"os/exec"
	"strings"
	"testing"
	"time"

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

const (
	paHost = "shop.bab.oncarlos.com"
	paOp   = "mv_probeattempts"
)

// paClaim is the landing claim after the first probe's admission: uk-box
// executes at probing and is the writer of session 20 in generation paOp,
// admitted cold from the source's session 15.
func paClaim(t *testing.T) *hostclaim.Claim {
	t.Helper()
	at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
	inc, err := hostclaim.Incarnation("bab", "jam", paHost, at)
	if err != nil {
		t.Fatal(err)
	}
	c := &hostclaim.Claim{V: 2, Host: paHost, Gen: 20,
		Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + paHost + ".json", Account: "bab", App: "jam", CreatedAt: at, Incarnation: inc, Location: "instances/" + paHost, Generation: paOp, Stamp: 18, Session: 20, WriterBox: "uk-box",
			Admissions: []hostclaim.AdmissionRecord{
				{Session: 15, Box: "ie-box", Generation: "src", Predecessor: "", Region: "ie", Mode: "cold"},
				{Session: 20, Box: "uk-box", Generation: paOp, Predecessor: "src", Region: "uk", PredecessorRegion: "ie", Mode: "cold", Coverage: "sealed"}}},
		Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: paOp, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}}}
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	return c
}

type paFixture struct {
	t       *testing.T
	claims  release.Store
	rec     *provision.Record
	ops     *movesProbeAttempts
	start   OperationStart
	ids     int
	admit   int
	dataDir string
}

func newPAFixture(t *testing.T) *paFixture {
	t.Helper()
	f := &paFixture{t: t, claims: release.NewDirStore(t.TempDir()), dataDir: t.TempDir(), admit: 2}
	c := paClaim(t)
	f.putClaim(c)
	o := f.read()
	f.rec = &provision.Record{Host: paHost, Account: "bab", App: "jam", Region: "uk",
		Move: &provision.MoveState{ID: paOp, Incarnation: c.Holder.Incarnation, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: mpPins(), Landing: &provision.MoveLandingJournal{Attempts: 1}}}
	f.ops = f.attempts()
	in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Executable: "/opt/carlos/live/" + paHost, ExecutableSHA256: strings.Repeat("a", 64), Args: []string{"--socket", "/s"}, Env: []string{"CARLOS_MOVE_PROBE=1"}, ReadyPath: "/ready"}
	h := o.Claim.Holder
	f.start = OperationStart{Admitted: o, Policy: activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}, Inputs: in,
		Owner: hostguard.FenceTarget{Host: paHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Op: paOp, Phase: hostclaim.PhaseLandingProbing, Box: "uk-box", Region: "uk", Enrollment: "enr-uk", BootID: "boot-1", ControllerID: "ctl-1", ClaimETag: o.ETag}}
	return f
}

// attempts is a fresh controller's provider over the same durable state.
func (f *paFixture) attempts() *movesProbeAttempts {
	return &movesProbeAttempts{Claims: f.claims, DataDir: f.dataDir, Settle: 20 * time.Millisecond, Ready: time.Second,
		Records:  func(context.Context, string, string, string) (*provision.Record, error) { return f.rec, nil },
		Admitted: func(context.Context, *provision.Record) (int, error) { return f.admit, nil },
		NewID:    func() string { f.ids++; return "id-" + strings.Repeat("x", f.ids) },
		Now:      func() time.Time { return time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC) }}
}

func (f *paFixture) putClaim(c *hostclaim.Claim) {
	f.t.Helper()
	body, err := json.Marshal(c)
	if err != nil {
		f.t.Fatal(err)
	}
	key, err := hostclaim.Key(paHost)
	if err != nil {
		f.t.Fatal(err)
	}
	etag := ""
	if _, current, err := f.claims.Get(f.t.Context(), key); err == nil {
		etag = current
	}
	if _, err := f.claims.Put(f.t.Context(), key, body, etag); err != nil {
		f.t.Fatal(err)
	}
}

func (f *paFixture) read() hostclaim.Observed {
	f.t.Helper()
	o, err := hostclaim.Read(f.t.Context(), f.claims, paHost)
	if err != nil {
		f.t.Fatal(err)
	}
	return o
}

// OpenProbe journals the first start's time once per record attempt and
// returns it to every later start, a restarted controller included; the
// next record attempt opens its own window.
func TestMovesProbeAttemptsOpenProbeIsJournalledOnce(t *testing.T) {
	f := newPAFixture(t)
	ctx := t.Context()
	t1 := time.Date(2026, 10, 9, 12, 0, 0, 0, time.UTC)
	got, err := f.ops.OpenProbe(ctx, f.start, t1)
	if err != nil || !got.Equal(t1) {
		t.Fatalf("open: %v %v", got, err)
	}
	if got, err := f.ops.OpenProbe(ctx, f.start, t1.Add(5*time.Second)); err != nil || !got.Equal(t1) {
		t.Fatalf("a second start: %v %v", got, err)
	}
	if got, err := f.attempts().OpenProbe(ctx, f.start, t1.Add(9*time.Second)); err != nil || !got.Equal(t1) {
		t.Fatalf("after a controller restart: %v %v", got, err)
	}
	// The next record attempt is a new cold admission: the same session
	// never opens its window.
	f.rec.Move.Landing.Attempts = 2
	t2 := t1.Add(time.Minute)
	if _, err := f.ops.OpenProbe(ctx, f.start, t2); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("the next record attempt on the same session: %v", err)
	}
	next := f.readmit(21)
	if got, err := f.ops.OpenProbe(ctx, next, t2); err != nil || !got.Equal(t2) {
		t.Fatalf("the next record attempt's admission: %v %v", got, err)
	}
	j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
	if err != nil || !j.Done("probe_opened_1") || !j.Done("probe_opened_2") || j.Steps["probe_session_20"] != "1" || j.Steps["probe_session_21"] != "2" {
		t.Fatalf("journal: %+v %v", j, err)
	}
}

// readmit stores the claim after a new cold admission at session and
// returns that admission's start.
func (f *paFixture) readmit(session uint64) OperationStart {
	f.t.Helper()
	c := paClaim(f.t)
	c.Holder.Session, c.Gen = session, session
	c.Holder.Generation = "fresh"
	c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: session, Box: "uk-box", Generation: "fresh", Predecessor: paOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
	f.putClaim(c)
	s := f.start
	s.Admitted = f.read()
	s.Owner.Session, s.Owner.ClaimETag = session, s.Admitted.ETag
	return s
}

// A session is bound to the record attempt it first journalled (at
// Reserve(0) or OpenProbe): once the record's index moves past it, every
// method refuses and CurrentAttempt answers nothing, even across a
// controller restart.
func TestMovesProbeAttemptsBindTheSessionToOneRecordAttempt(t *testing.T) {
	f := newPAFixture(t)
	ctx := t.Context()
	if err := f.ops.Reserve(ctx, f.start, 0); err != nil {
		t.Fatal(err)
	}
	if f.ops.CurrentAttempt(f.start).ID == "" {
		t.Fatal("no reservation")
	}
	f.rec.Move.Landing.Attempts = 2
	if a := f.ops.CurrentAttempt(f.start); a.ID != "" {
		t.Fatalf("a reservation of record attempt 1 answered for attempt 2: %+v", a)
	}
	if err := f.ops.Failed(ctx, f.start, 0); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("Failed: %v", err)
	}
	if err := f.ops.Reserve(ctx, f.start, 0); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("Reserve(0): %v", err)
	}
	if _, err := f.attempts().OpenProbe(ctx, f.start, time.Now()); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("OpenProbe after a restart: %v", err)
	}
	if err := f.ops.Healthy(ctx, f.start, 0, exec.Command("/bin/true"), func(context.Context) error { return nil }); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("Healthy: %v", err)
	}
	next := f.readmit(21)
	if err := f.ops.Reserve(ctx, next, 0); err != nil {
		t.Fatalf("the new admission: %v", err)
	}
}

// Reserve, CurrentAttempt and Failed are bound to the record's attempt
// index and the admitted session: the reservation carries the owner's
// tuple and the launch digest, is stable between reservations, and a next
// start is reserved only after the last one was settled.
func TestMovesProbeAttemptsReserveBindsTheRecordsAttempt(t *testing.T) {
	f := newPAFixture(t)
	ctx := t.Context()
	if a := f.ops.CurrentAttempt(f.start); a.ID != "" {
		t.Fatalf("an attempt before any reservation: %+v", a)
	}
	if err := f.ops.Reserve(ctx, f.start, 0); err != nil {
		t.Fatal(err)
	}
	first := f.ops.CurrentAttempt(f.start)
	digest, _ := ClaimLaunchDigest(f.start.Inputs)
	o := f.start.Owner
	if first.ID == "" || first.InvocationID == "" || first.ID == first.InvocationID || first.Session != o.Session || first.Box != o.Box || first.Enrollment != o.Enrollment || first.BootID != o.BootID || first.ControllerID != o.ControllerID || first.LaunchInputs != digest {
		t.Fatalf("reservation: %+v", first)
	}
	if again := f.ops.CurrentAttempt(f.start); again != first {
		t.Fatalf("CurrentAttempt moved: %+v", again)
	}
	if err := f.ops.Reserve(ctx, f.start, 0); err != nil || f.ops.CurrentAttempt(f.start) != first {
		t.Fatalf("Reserve(0) again is a validation: %v", err)
	}
	if err := f.ops.Reserve(ctx, f.start, 1); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("start 1 before start 0 settled: %v", err)
	}
	if err := f.ops.Failed(ctx, f.start, 1); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("settling a start that was not reserved: %v", err)
	}
	if err := f.ops.Failed(ctx, f.start, 0); err != nil {
		t.Fatal(err)
	}
	if err := f.ops.Reserve(ctx, f.start, 1); err != nil {
		t.Fatal(err)
	}
	second := f.ops.CurrentAttempt(f.start)
	if second.ID == first.ID || second.InvocationID == first.InvocationID || second.Session != o.Session || second.LaunchInputs != digest {
		t.Fatalf("start 1: %+v", second)
	}
	changed := f.start
	changed.Inputs.Env = []string{"CARLOS_MOVE_PROBE=1", "OTHER=1"}
	if err := f.ops.Failed(ctx, f.start, 1); err != nil {
		t.Fatal(err)
	}
	if err := f.ops.Reserve(ctx, changed, 2); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("other launch inputs: %v", err)
	}
}

// Every method refuses once the record or the claim no longer names the
// admitted probe: the record's attempt index outside what T11 admitted, a
// rollback reason, the record past probing, another session.
func TestMovesProbeAttemptsRefuseWhenTheProbeMovedOn(t *testing.T) {
	for name, mutate := range map[string]func(*paFixture){
		"no attempt counted":   func(f *paFixture) { f.rec.Move.Landing = nil },
		"beyond the admitted":  func(f *paFixture) { f.rec.Move.Landing.Attempts = 3 },
		"none admitted":        func(f *paFixture) { f.admit = 0 },
		"a rollback is marked": func(f *paFixture) { f.rec.Move.Reason = provision.RollbackProbeAttempts },
		"the record is probed": func(f *paFixture) { f.rec.Move.LandingPhase = provision.LandProbed },
		"another move":         func(f *paFixture) { f.rec.Move.ID = "mv_other" },
		"another session": func(f *paFixture) {
			c := paClaim(f.t)
			c.Holder.Session, c.Gen = 21, 21
			c.Holder.Admissions[1].Session = 21
			f.putClaim(c)
		},
		"the claim is probed": func(f *paFixture) {
			c := paClaim(f.t)
			c.Op.Phase = hostclaim.PhaseLandingProbed
			f.putClaim(c)
		},
		"a committed wake": func(f *paFixture) { f.start.Policy.Probe = false },
	} {
		t.Run(name, func(t *testing.T) {
			f := newPAFixture(t)
			mutate(f)
			ctx := t.Context()
			if _, err := f.ops.OpenProbe(ctx, f.start, time.Now()); err == nil {
				t.Fatal("OpenProbe")
			}
			if err := f.ops.Reserve(ctx, f.start, 0); err == nil {
				t.Fatal("Reserve")
			}
			if err := f.ops.Healthy(ctx, f.start, 0, exec.Command("/bin/true"), func(context.Context) error { return nil }); err == nil {
				t.Fatal("Healthy")
			}
		})
	}
}

// Healthy is the 20 s settle with the process alive, then the second
// readiness request (moves §7.7); the start is journalled healthy only
// after both.
func TestMovesProbeAttemptsHealthySettlesAndAsksAgain(t *testing.T) {
	f := newPAFixture(t)
	ctx := t.Context()
	if err := f.ops.Reserve(ctx, f.start, 0); err != nil {
		t.Fatal(err)
	}
	cmd := exec.Command("sleep", "30")
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { cmd.Process.Kill(); cmd.Wait() })
	began := time.Now()
	asked := 0
	if err := f.ops.Healthy(ctx, f.start, 0, cmd, func(context.Context) error {
		asked++
		if time.Since(began) < f.ops.Settle {
			t.Error("asked before the settle")
		}
		return nil
	}); err != nil || asked != 1 {
		t.Fatalf("healthy: %v asked=%d", err, asked)
	}
	j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
	if err != nil || !strings.Contains(j.Steps["probe_start_1"], `"state":"healthy"`) {
		t.Fatalf("journal: %+v %v", j, err)
	}

	// A second request that fails, or a process that died in the settle,
	// is not healthy.
	g := newPAFixture(t)
	if err := g.ops.Reserve(ctx, g.start, 0); err != nil {
		t.Fatal(err)
	}
	refused := errors.New("503")
	if err := g.ops.Healthy(ctx, g.start, 0, cmd, func(context.Context) error { return refused }); !errors.Is(err, refused) {
		t.Fatalf("a failing second request: %v", err)
	}
	dead := exec.Command("/bin/true")
	if err := dead.Run(); err != nil {
		t.Fatal(err)
	}
	if err := g.ops.Healthy(ctx, g.start, 0, dead, func(context.Context) error { return nil }); err == nil {
		t.Fatal("a dead process was healthy")
	}
	if j, _ := moves.ReadJournal(g.dataDir, paHost, paOp); strings.Contains(j.Steps["probe_start_1"], "healthy") {
		t.Fatalf("journalled healthy: %v", j.Steps)
	}
}

// Task 7: the committed wake's starts are journalled per admitted session,
// with no record attempt (the probe's are over); Healthy accepts at once.
func TestMovesProbeAttemptsCommittedWake(t *testing.T) {
	f := newPAFixture(t)
	ctx := t.Context()
	c := paClaim(t)
	c.Op.Phase = hostclaim.PhaseLandingCommitted
	f.putClaim(c)
	f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, ""
	s := f.start
	s.Admitted, s.Policy.Probe = f.read(), false
	s.Owner.ClaimETag = s.Admitted.ETag
	if err := f.ops.Reserve(ctx, s, 0); err != nil {
		t.Fatal(err)
	}
	first := f.ops.CurrentAttempt(s)
	if first.ID == "" || first.Session != s.Owner.Session {
		t.Fatalf("reservation: %+v", first)
	}
	if err := f.ops.Failed(ctx, s, 0); err != nil {
		t.Fatal(err)
	}
	if err := f.ops.Reserve(ctx, s, 1); err != nil || f.ops.CurrentAttempt(s).ID == first.ID {
		t.Fatalf("start 1: %v", err)
	}
	asked := 0
	if err := f.ops.Healthy(ctx, s, 1, nil, func(context.Context) error { asked++; return nil }); err != nil || asked != 0 {
		t.Fatalf("healthy: %v asked=%d", err, asked)
	}
	f.rec.Move.State = provision.MoveLanding
	if err := f.ops.Reserve(ctx, s, 0); err == nil {
		t.Fatal("a committed wake's start over a landing record")
	}
}

func recoveryPAFixture(t *testing.T, kind string) *paFixture {
	t.Helper()
	f := newPAFixture(t)
	c := f.read().Claim
	c.Op.Kind, c.Op.Phase = kind, hostclaim.PhaseProbing
	c.Op.Exec, c.Holder.WriterBox = "ie-box", "ie-box"
	c.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
	f.rec.Move.State = provision.MoveReverting
	if kind == hostclaim.KindRestoring {
		c.Op.Intent["rollback"] = plan10J2Rollback
		f.rec.Move.State = provision.MoveRollingBack
		f.rec.Move.Rollback = &provision.MoveRollback{ID: plan10J2Rollback, SourcePhase: provision.RbSourceProbing}
	}
	f.putClaim(c)
	f.start.Admitted = f.read()
	f.start.Owner.Box = "ie-box"
	f.start.Owner.Phase = hostclaim.PhaseProbing
	return f
}

func TestRecoveryProbeAttemptsDurableBudget(t *testing.T) {
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			f := recoveryPAFixture(t, kind)
			for n := 0; n < 9; n++ {
				// Three starts per wake, then a new admitted session and provider.
				local := n % 3
				if local == 0 && n > 0 {
					c := f.read().Claim
					c.Gen++
					c.Holder.Session = c.Gen
					f.putClaim(c)
					f.start.Admitted = f.read()
					f.start.Owner.Session = c.Gen
					f.ops = f.attempts()
				}
				if _, err := f.ops.OpenProbe(t.Context(), f.start, time.Now()); err != nil {
					t.Fatal(err)
				}
				if err := f.ops.Reserve(t.Context(), f.start, local); err != nil {
					t.Fatalf("start %d: %v", n+1, err)
				}
				if f.ops.CurrentAttempt(f.start).ID == "" {
					t.Fatal("no launch reservation")
				}
				if local == 0 {
					if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
						t.Fatal("idempotent reservation", err)
					}
				}
				err := f.ops.Failed(t.Context(), f.start, local)
				if n == 8 {
					if !errors.Is(err, ErrRecoveryProbeExhausted) {
						t.Fatalf("ninth failure: %v", err)
					}
				} else if err != nil {
					t.Fatal(err)
				}
			}
			f.ops = f.attempts()
			if err := f.ops.Reserve(t.Context(), f.start, 3); !errors.Is(err, ErrRecoveryProbeExhausted) {
				t.Fatalf("restart lost count: %v", err)
			}
		})
	}
}

func TestRecoveryProbeAttemptsOperatorRetryBudget(t *testing.T) {
	t.Parallel()
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			ctx := t.Context()
			f := recoveryPAFixture(t, kind)
			f.rec.Schema, f.rec.Slug, f.rec.Channel = 1, "shop", "stable"
			f.rec.CreatedAt = f.start.Admitted.Claim.Holder.CreatedAt
			if err := provision.CreateRecord(ctx, f.claims, *f.rec); err != nil {
				t.Fatal(err)
			}
			restart := func() {
				f.ops = f.attempts()
				f.ops.Records = func(ctx context.Context, account, app, host string) (*provision.Record, error) {
					return provision.GetRecord(ctx, f.claims, account, app, host)
				}
			}
			restart()
			g := provision.MoveGuard{ID: f.rec.Move.ID, Incarnation: f.rec.Move.Incarnation}
			transition := func(mut func(*provision.Record) error) error {
				_, err := provision.SetRecordMove(ctx, f.claims, f.rec.Account, f.rec.App, paHost, mut)
				return err
			}
			newSession := func() {
				c := f.read().Claim
				c.Gen++
				c.Holder.Session = c.Gen
				f.putClaim(c)
				f.start.Admitted = f.read()
				f.start.Owner.Session = c.Gen
				restart()
			}
			var previous hostclaim.WakeRetryAttempt
			for budget := 0; budget < 3; budget++ {
				if budget > 0 {
					if err := transition(provision.MoveMarkStuck(g, provision.StuckRecoveryProbeAttempts)); err != nil {
						t.Fatal(err)
					}
					if err := transition(provision.MoveRetryRecovery(g)); err != nil {
						t.Fatal(err)
					}
					if got := f.ops.CurrentAttempt(f.start); got.ID != "" {
						t.Fatal("old reservation survived retry", got)
					}
					if err := f.ops.Failed(ctx, f.start, 2); !errors.Is(err, hostbox.ErrChanged) {
						t.Fatalf("old failure settled after retry: %v", err)
					}
				}
				rec, err := provision.GetRecord(ctx, f.claims, f.rec.Account, f.rec.App, paHost)
				if err != nil || rec.Move.RecoveryRetryEpoch != uint64(budget) {
					t.Fatalf("retry epoch not persisted: record=%+v err=%v", rec, err)
				}
				for n := 0; n < 9; n++ {
					local := n % 3
					if local == 0 && n > 0 {
						newSession()
					}
					if err := f.ops.Reserve(ctx, f.start, local); err != nil {
						t.Fatalf("budget %d start %d: %v", budget, n+1, err)
					}
					attempt := f.ops.CurrentAttempt(f.start)
					if attempt.ID == "" || attempt.ID == previous.ID {
						t.Fatalf("start has no fresh reservation: %+v", attempt)
					}
					previous = attempt
					if local == 0 {
						restart()
						if err := f.ops.Reserve(ctx, f.start, 0); err != nil || f.ops.CurrentAttempt(f.start) != attempt {
							t.Fatalf("reservation replay after restart: %v", err)
						}
					}
					if budget > 0 {
						if err := transition(provision.MoveRetryRecovery(g)); !errors.Is(err, provision.ErrMoveConflict) {
							t.Fatalf("retry replay: %v", err)
						}
					}
					err := f.ops.Failed(ctx, f.start, local)
					if n == 8 {
						if !errors.Is(err, ErrRecoveryProbeExhausted) {
							t.Fatalf("budget %d ninth failure: %v", budget, err)
						}
					} else if err != nil {
						t.Fatal(err)
					}
				}
				previousProvider := f.ops
				restart()
				if err := f.ops.Reserve(ctx, f.start, 3); !errors.Is(err, ErrRecoveryProbeExhausted) {
					t.Fatalf("budget %d restart lost exhaustion: %v", budget, err)
				}
				f.ops = previousProvider
			}
			newSession()
			if err := transition(provision.MoveRetryRecovery(g)); !errors.Is(err, provision.ErrMoveConflict) {
				t.Fatalf("exhausted retry replay: %v", err)
			}
			if err := f.ops.Reserve(ctx, f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) {
				t.Fatalf("retry replay or fresh session granted extra starts: %v", err)
			}
		})
	}
}

func TestRecoveryProbeAttemptsLegacyBudget(t *testing.T) {
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			f := recoveryPAFixture(t, kind)
			rollback := paOp
			if kind == hostclaim.KindRestoring {
				rollback = plan10J2Rollback
			}
			raw, err := json.Marshal([]string{paOp, rollback, f.rec.Move.Incarnation})
			if err != nil {
				t.Fatal(err)
			}
			j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
			if err != nil {
				t.Fatal(err)
			}
			j.Mark("recovery_attempts_"+string(raw), "9")
			if err := j.Commit(f.dataDir); err != nil {
				t.Fatal(err)
			}
			if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, ErrRecoveryProbeExhausted) {
				t.Fatalf("legacy exhaustion lost: %v", err)
			}
		})
	}
}

func TestRecoveryProbeAttemptsReadinessAndBinding(t *testing.T) {
	f := recoveryPAFixture(t, hostclaim.KindRestoring)
	w := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }}
	if w.recoveryAttemptsReady() {
		t.Fatal("landing-only provider admitted recovery")
	}
	w.Operations = f.ops
	if !w.recoveryAttemptsReady() {
		t.Fatal("recovery provider not ready")
	}
	w.Operations = claimOperationRouter{hostclaim.KindRestoring: {Attempts: f.ops, ProbeReady: w.ProbeReady}}
	if w.recoveryAttemptsReady() {
		t.Fatal("missing revert provider admitted")
	}
	if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
		t.Fatal(err)
	}
	oldKey := recoveryAttemptKey(f.start, 0)
	c := f.read().Claim
	c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"
	f.putClaim(c)
	f.start.Admitted = f.read()
	f.rec.Move.Rollback.ID = c.Op.Intent["rollback"]
	if recoveryAttemptKey(f.start, 0) == oldKey {
		t.Fatal("rollback not bound")
	}
	if got := f.ops.CurrentAttempt(f.start); got.ID != "" {
		t.Fatal("reservation escaped its rollback")
	}
	if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil {
		t.Fatal(err)
	}
	j, err := moves.ReadJournal(f.dataDir, paHost, paOp)
	if err != nil {
		t.Fatal(err)
	}
	if j.Steps[oldKey] != "1" || j.Steps[recoveryAttemptKey(f.start, 0)] != "1" {
		t.Fatal("rollback budgets combined")
	}
	for _, mutate := range []func(*OperationStart){
		func(s *OperationStart) { s.Admitted.Claim.Holder.Incarnation = "other" },
		func(s *OperationStart) { s.Admitted.Claim.Op.ID = "other" },
	} {
		copied := f.start
		raw, _ := json.Marshal(f.start.Admitted.Claim)
		var claim hostclaim.Claim
		_ = json.Unmarshal(raw, &claim)
		copied.Admitted.Claim = &claim
		mutate(&copied)
		if recoveryAttemptKey(copied, 0) == recoveryAttemptKey(f.start, 0) {
			t.Fatal("budget key omitted identity")
		}
	}
}

func TestRecoveryProbeAttemptsRefuseChangedRollback(t *testing.T) {
	f := recoveryPAFixture(t, hostclaim.KindRestoring)
	c := f.read().Claim
	c.Op.Intent["rollback"] = "rb_zzzzzzzzzzzzzzzzzzzzzzzzzz"
	f.putClaim(c)
	if err := f.ops.Reserve(t.Context(), f.start, 0); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("changed rollback admitted: %v", err)
	}
}