package activator

import (
	"context"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"sync"
	"testing"
	"time"

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

// The recovery probe entry (claim Plan 10 J2, moves Plan 4 Task 9), driven
// through the claim runtime the host assembles: K6, Prerequisites, Admit,
// Prepare and the start loop. RoleLaunch.Start still refuses every launch
// (claim Plan 9 Group D's isolated root launcher), so a full run ends in that
// refusal as an ordinary failed start; retention of a healthy start is driven
// through finishRecoveryProbe with the instance a healthy startAdmitted yields.

const probeHost = "x.example"

func probeClaim(kind, phase, exec string) *hostclaim.Claim {
	return &hostclaim.Claim{
		Holder: &hostclaim.Holder{Incarnation: "inc", Stamp: 1, Session: 1, Generation: "one", Location: "instances/x.example/inc", WriterBox: "box"},
		Op:     hostclaim.Op{Kind: kind, ID: "move", Phase: phase, Exec: exec},
	}
}

func probeObserved(c *hostclaim.Claim) hostclaim.Observed {
	return hostclaim.Observed{Host: probeHost, ETag: "one", Claim: c}
}

func probeAdmitted(o hostclaim.Observed) hostclaim.Observed {
	c := *o.Claim
	h := *c.Holder
	h.Session, h.Generation = h.Session+1, "two"
	c.Holder = &h
	return hostclaim.Observed{Host: o.Host, ETag: "two", Claim: &c}
}

func probeInputs() *provision.MoveInputs {
	return &provision.MoveInputs{Release: "release", OverlayETag: "overlay", BundleETags: map[string]string{"main": "etag"}}
}

func probePolicy(in *provision.MoveInputs) StartPolicy {
	return StartPolicy{MaxStarts: 1, Window: 30 * time.Second, Probe: true, Inputs: in}
}

// probeRun records what the claim runtime saw, in order.
type probeRun struct {
	mu                        sync.Mutex
	steps                     []string
	policies                  []StartPolicy
	failedAdmission           *hostclaim.Observed
	reserved, failed, stopped int
}

func (r *probeRun) step(name string, p StartPolicy) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.steps = append(r.steps, name)
	r.policies = append(r.policies, p)
}

func (r *probeRun) seen() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]string(nil), r.steps...)
}

// probeService wires a Service whose claim runtime reads obs, admits it to
// the next session and prepares a start that reaches RoleLaunch.Start.
func probeService(t *testing.T, obs hostclaim.Observed) (*Service, *ClaimRuntime, *probeRun) {
	t.Helper()
	a, _, _, _ := testActivator(t, time.Hour)
	run := &probeRun{}
	admitted := probeAdmitted(obs)
	work, _ := (hostwrite.Config{}).Select(t.Context(), probeHost, "spawn")
	reads := 0
	r := &ClaimRuntime{
		RecoveryAttemptsReady: func() bool { return true },
		Box:                   "box",
		// The wake's first read sees obs; every later one (startAdmitted's
		// fresh read) sees the admission.
		Read: func(_ context.Context, host string) (hostclaim.Observed, error) {
			run.step("read", StartPolicy{})
			run.mu.Lock()
			defer run.mu.Unlock()
			if reads++; reads > 1 {
				return admitted, nil
			}
			return obs, nil
		},
		StopForeign: func(context.Context, string) error { run.step("stop-foreign", StartPolicy{}); return nil },
		Prerequisites: func(_ context.Context, rt registry.Route, o hostclaim.Observed, p StartPolicy) error {
			run.step("prerequisites", p)
			return nil
		},
		Admit: func(_ context.Context, _ registry.Route, o hostclaim.Observed, p StartPolicy) (hostclaim.Observed, error) {
			run.step("admit", p)
			return admitted, nil
		},
		Prepare: func(_ context.Context, _ registry.Route, o hostclaim.Observed, p StartPolicy) (ClaimStart, error) {
			run.step("prepare", p)
			return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} }, Host: probeHost, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"},
				OpenProbe:   func(context.Context, time.Time) (time.Time, error) { return time.Now(), nil },
				Restore:     func(context.Context) error { return nil },
				Materialize: func(context.Context) error { return nil },
				Build:       func() *exec.Cmd { return nil },
				Ready:       func(context.Context, *exec.Cmd) error { return nil },
				Reserve:     func(context.Context, int) error { run.mu.Lock(); run.reserved++; run.mu.Unlock(); return nil },
				Failed:      func(context.Context, int) error { run.mu.Lock(); run.failed++; run.mu.Unlock(); return nil },
				Healthy:     func(context.Context, int, *exec.Cmd) error { return nil },
				Replicate:   func(context.Context, string) (Rep, error) { return &fakeRep{}, nil },
				Publish:     func(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) { return o, nil },
				Retain:      func(Proc, Rep, error) { t.Error("a refused start is not retained") },
			}, nil
		},
		FailedAdmission: func(_ context.Context, o hostclaim.Observed, _ error) error {
			run.mu.Lock()
			run.failedAdmission = &o
			run.mu.Unlock()
			return nil
		},
	}
	a.cfg.HostWrites = hostwrite.Config{Enabled: true}
	a.cfg.Claims = r
	return a, r, run
}

func TestRecoveryProbeRunsTheClaimRuntimeWithTheProbePolicyAndExactInputs(t *testing.T) {
	t.Parallel()
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			obs := probeObserved(probeClaim(kind, hostclaim.PhaseProbing, "box"))
			a, _, run := probeService(t, obs)
			in := probeInputs()
			want := *in
			err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in))
			// The start reached RoleLaunch.Start, which refuses until Group D:
			// an honest failed start, never a success.
			if !errors.Is(err, ErrClaimIsolationUnavailable) {
				t.Fatalf("err = %v, want the launcher's refusal", err)
			}
			steps := run.seen()
			if len(steps) < 4 || steps[0] != "read" || steps[1] != "prerequisites" || steps[2] != "admit" || steps[len(steps)-1] != "prepare" {
				t.Fatalf("steps = %v", steps)
			}
			run.mu.Lock()
			defer run.mu.Unlock()
			for i, p := range run.policies {
				if run.steps[i] == "read" || run.steps[i] == "stop-foreign" {
					continue
				}
				if !p.Probe || !reflect.DeepEqual(p.Inputs, in) || p.MaxStarts != 1 || p.Window != 30*time.Second || p.Prepared != nil {
					t.Fatalf("%s saw %+v, want the caller's probe policy", run.steps[i], p)
				}
			}
			if in.Release != want.Release || in.OverlayETag != want.OverlayETag || in.BundleETags["main"] != "etag" {
				t.Fatalf("inputs changed: %+v", in)
			}
			if run.reserved != 1 || run.failed != 1 {
				t.Fatalf("reserved=%d failed=%d, want one refused start", run.reserved, run.failed)
			}
			if run.failedAdmission == nil || run.failedAdmission.Claim.Holder.Session != 2 {
				t.Fatalf("the failed admission was not handed back: %+v", run.failedAdmission)
			}
			if a.Owns(probeHost) {
				t.Fatal("a failed probe left an entry")
			}
		})
	}
}

func TestRecoveryProbeK6StopsForeignBeforeAnyAdmission(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "elsewhere"))
	a, _, run := probeService(t, obs)
	err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs()))
	if !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("err = %v", err)
	}
	if got := run.seen(); len(got) != 2 || got[0] != "read" || got[1] != "stop-foreign" {
		t.Fatalf("steps = %v, want read then stop-foreign only", got)
	}
	if a.Owns(probeHost) {
		t.Fatal("entry left")
	}
}

func TestRecoveryProbeRefusesWrongKindOrPhaseBeforePrerequisites(t *testing.T) {
	t.Parallel()
	for _, c := range []*hostclaim.Claim{
		probeClaim(hostclaim.KindRestoring, hostclaim.PhaseClaimed, "box"),
		probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbed, "box"),
		probeClaim(hostclaim.KindReverting, hostclaim.PhaseOpened, "box"),
		probeClaim(hostclaim.KindLanding, "probing", "box"),
		{Holder: &hostclaim.Holder{Incarnation: "inc", Session: 1, Generation: "one", Location: "l", WriterBox: "box"}, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}},
	} {
		t.Run(c.Op.Kind+"/"+c.Op.Phase, func(t *testing.T) {
			a, _, run := probeService(t, probeObserved(c))
			err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs()))
			if !errors.Is(err, hostbox.ErrChanged) {
				t.Fatalf("err = %v", err)
			}
			if got := run.seen(); len(got) != 1 || got[0] != "read" {
				t.Fatalf("steps = %v, want the read alone", got)
			}
			if a.Owns(probeHost) {
				t.Fatal("entry left")
			}
		})
	}
}

func TestRecoveryProbeRefusesMissingRuntimeOrPolicyBeforeAnyRead(t *testing.T) {
	t.Parallel()
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	in := probeInputs()
	runtimes := map[string]func(*Service, *ClaimRuntime){
		"host writes off":     func(a *Service, _ *ClaimRuntime) { a.cfg.HostWrites.Enabled = false },
		"no claims":           func(a *Service, _ *ClaimRuntime) { a.cfg.Claims = nil },
		"no box":              func(_ *Service, r *ClaimRuntime) { r.Box = "" },
		"no stop foreign":     func(_ *Service, r *ClaimRuntime) { r.StopForeign = nil },
		"no prerequisites":    func(_ *Service, r *ClaimRuntime) { r.Prerequisites = nil },
		"no admit":            func(_ *Service, r *ClaimRuntime) { r.Admit = nil },
		"no prepare":          func(_ *Service, r *ClaimRuntime) { r.Prepare = nil },
		"no failed admission": func(_ *Service, r *ClaimRuntime) { r.FailedAdmission = nil },
	}
	for name, breakIt := range runtimes {
		t.Run(name, func(t *testing.T) {
			a, r, run := probeService(t, obs)
			breakIt(a, r)
			if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in)); !errors.Is(err, hostwrite.ErrUnavailable) {
				t.Fatalf("err = %v", err)
			}
			if got := run.seen(); len(got) != 0 || a.Owns(probeHost) {
				t.Fatalf("steps = %v owns=%v", got, a.Owns(probeHost))
			}
		})
	}
	policies := map[string]StartPolicy{
		"ordinary":       {MaxStarts: 1, Window: 30 * time.Second, Inputs: in},
		"no inputs":      {MaxStarts: 1, Window: 30 * time.Second, Probe: true},
		"prepared":       {MaxStarts: 1, Window: 30 * time.Second, Probe: true, Inputs: in, Prepared: &moves.InstallManifest{Op: "move"}},
		"no starts":      {Window: 30 * time.Second, Probe: true, Inputs: in},
		"too many":       {MaxStarts: 6, Window: 30 * time.Second, Probe: true, Inputs: in},
		"no window":      {MaxStarts: 1, Probe: true, Inputs: in},
		"window too big": {MaxStarts: 1, Window: 31 * time.Second, Probe: true, Inputs: in},
	}
	for name, p := range policies {
		t.Run(name, func(t *testing.T) {
			a, _, run := probeService(t, obs)
			if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, p); !errors.Is(err, hostbox.ErrUnsupported) {
				t.Fatalf("err = %v", err)
			}
			if got := run.seen(); len(got) != 0 || a.Owns(probeHost) {
				t.Fatalf("steps = %v", got)
			}
		})
	}
	a, _, run := probeService(t, obs)
	if err := a.RecoveryProbe(t.Context(), registry.Route{}, probePolicy(in)); !errors.Is(err, hostbox.ErrUnsupported) || len(run.seen()) != 0 {
		t.Fatalf("no host: %v", err)
	}
}

func TestRecoveryProbeRefusesAnOrdinaryEntryAndShutdown(t *testing.T) {
	t.Parallel()
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	for _, state := range []actState{actWaking, actAwake, actStopping} {
		a, _, run := probeService(t, obs)
		a.entries[probeHost] = &actEntry{host: probeHost, state: state, ready: make(chan struct{}), stopped: make(chan struct{})}
		if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs())); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("state %v: %v", state, err)
		}
		if got := run.seen(); len(got) != 0 {
			t.Fatalf("state %v: steps %v", state, got)
		}
	}
	a, _, run := probeService(t, obs)
	a.shuttingDown = true
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs())); !errors.Is(err, ErrFenced) || len(run.seen()) != 0 {
		t.Fatalf("shutdown: %v", err)
	}
}

// One probe per host: an identical second call joins the one in flight and
// shares its outcome; a different policy is refused; a request never joins.
func TestRecoveryProbeCoalescesTheSamePolicyAndRefusesOthers(t *testing.T) {
	t.Parallel()
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, r, run := probeService(t, obs)
	entered, release := make(chan struct{}), make(chan struct{})
	var once sync.Once
	refuse := errors.New("prerequisites refused")
	r.Prerequisites = func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
		run.step("prerequisites", p)
		once.Do(func() { close(entered) })
		<-release
		return refuse
	}
	in := probeInputs()
	first := make(chan error, 1)
	go func() {
		first <- a.RecoveryProbe(context.Background(), registry.Route{Host: probeHost}, probePolicy(in))
	}()
	<-entered

	other := probeInputs()
	other.Release = "other"
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(other)); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("different inputs: %v", err)
	}
	longer := probePolicy(in)
	longer.MaxStarts = 2
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, longer); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("different starts: %v", err)
	}
	if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
		t.Fatalf("a request joined the probe: %v", err)
	}
	if err := a.BeginSidecarRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
		t.Fatalf("a sidecar request joined the probe: %v", err)
	}
	if a.Bounce(probeHost) {
		t.Fatal("bounce stopped the probe")
	}

	same := probeInputs()
	second := make(chan error, 1)
	go func() {
		second <- a.RecoveryProbe(context.Background(), registry.Route{Host: probeHost}, probePolicy(same))
	}()
	time.Sleep(100 * time.Millisecond) // let the second call reach the entry
	close(release)
	if err := <-first; !errors.Is(err, refuse) {
		t.Fatalf("first: %v", err)
	}
	if err := <-second; !errors.Is(err, refuse) {
		t.Fatalf("joined: %v", err)
	}
	n := 0
	for _, s := range run.seen() {
		if s == "prerequisites" {
			n++
		}
	}
	if n != 1 || a.Owns(probeHost) {
		t.Fatalf("prerequisites ran %d times, owns=%v", n, a.Owns(probeHost))
	}
}

// A maintenance fence installed while the probe runs cancels it, exactly
// as it cancels an ordinary wake.
func TestRecoveryProbeIsCancelledByAFenceInstalledMidProbe(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, r, run := probeService(t, obs)
	entered := make(chan struct{})
	r.Prerequisites = func(ctx context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
		run.step("prerequisites", p)
		close(entered)
		<-ctx.Done()
		return ctx.Err()
	}
	done := make(chan error, 1)
	go func() {
		done <- a.RecoveryProbe(context.Background(), registry.Route{Host: probeHost}, probePolicy(probeInputs()))
	}()
	<-entered
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if _, err := a.FenceAndStop(ctx, probeHost, 7); err != nil && !errors.Is(err, ErrStopUnproven) {
		t.Fatal(err)
	}
	if err := <-done; !errors.Is(err, context.Canceled) {
		t.Fatalf("err = %v", err)
	}
	if a.Owns(probeHost) {
		t.Fatal("entry left")
	}
}

// heldProbe installs a probe entry as RecoveryProbe does and finishes it with
// the instance a healthy startAdmitted yields.
func heldProbe(t *testing.T, a *Service, fence uint64, admitted hostclaim.Observed, in *provision.MoveInputs) (*actEntry, *fakeProc, *fakeRep, error) {
	t.Helper()
	p := probePolicy(in)
	p.RecoveryFence = fence
	e := &actEntry{host: probeHost, route: registry.Route{Host: probeHost}, state: actWaking, ready: make(chan struct{}), recoveryPolicy: &p, probeHeld: true}
	a.mu.Lock()
	a.entries[probeHost] = e
	a.mu.Unlock()
	e.probeAdmitted = admitted
	proc, rep := &fakeProc{}, &fakeRep{}
	err := a.finishRecoveryProbe(e, fence, &wokenInstance{proc: proc, rep: rep, spawnVersion: "v", stopOutcome: &stopObservation{}}, nil)
	return e, proc, rep, err
}

func TestRecoveryProbeRetainsTheHealthyProcessHeldUntilOpened(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	admitted := probeAdmitted(obs)
	a, r, _ := probeService(t, obs)
	// The rollback's fence stays installed through the recovery.
	a.fences = map[string]uint64{probeHost: 9}
	in := probeInputs()
	e, proc, _, err := heldProbe(t, a, 9, admitted, in)
	if err != nil {
		t.Fatal(err)
	}
	a.mu.Lock()
	state, held, kept := e.state, e.probeHeld, e.proc == Proc(proc)
	a.mu.Unlock()
	if state != actAwake || !held || !kept || proc.isStopped() {
		t.Fatalf("state=%v held=%v kept=%v stopped=%v", state, held, kept, proc.isStopped())
	}
	if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
		t.Fatalf("a request reached the held probe: %v", err)
	}
	if a.Bounce(probeHost) || proc.isStopped() {
		t.Fatal("bounce stopped the held probe")
	}

	// A replayed call for the same session answers from the retained process.
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); err != nil {
		t.Fatalf("replay: %v", err)
	}
	moved := probeAdmitted(admitted)
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return moved, nil }
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("replay on another session: %v", err)
	}
	proc.exit()
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("replay on an exited process: %v", err)
	}
}

func TestRecoveryProbeStopsAHealthyStartWhenTheFenceMoved(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	for name, now := range map[string]map[string]uint64{"installed": {probeHost: 3}, "replaced": {probeHost: 12}} {
		t.Run(name, func(t *testing.T) {
			a, _, _ := probeService(t, obs)
			a.fences = now
			start := uint64(0)
			if name != "installed" {
				start = 9
			}
			e, proc, rep, err := heldProbe(t, a, start, probeAdmitted(obs), probeInputs())
			if !errors.Is(err, ErrFenced) {
				t.Fatalf("err = %v", err)
			}
			select {
			case <-e.stopped:
			case <-time.After(5 * time.Second):
				t.Fatal("not stopped")
			}
			if !proc.isStopped() || !rep.isClosed() || a.Owns(probeHost) {
				t.Fatalf("stopped=%v closed=%v owns=%v", proc.isStopped(), rep.isClosed(), a.Owns(probeHost))
			}
		})
	}
}

func TestRecoveryHoldVerdict(t *testing.T) {
	admitted := probeAdmitted(probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")))
	at := func(mutate func(*hostclaim.Claim)) hostclaim.Observed {
		c := *admitted.Claim
		h := *c.Holder
		c.Holder = &h
		mutate(&c)
		return hostclaim.Observed{Host: probeHost, ETag: "later", Claim: &c}
	}
	cases := []struct {
		name string
		obs  hostclaim.Observed
		open bool
		err  bool
	}{
		{"probing", at(func(*hostclaim.Claim) {}), false, false},
		{"probed", at(func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseProbed }), false, false},
		{"opened", at(func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseOpened }), true, false},
		{"released to serving", at(func(c *hostclaim.Claim) { c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} }), true, false},
		{"claimed", at(func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseClaimed }), false, true},
		{"new session", at(func(c *hostclaim.Claim) { c.Holder.Session++ }), false, true},
		{"new generation", at(func(c *hostclaim.Claim) { c.Holder.Generation = "three" }), false, true},
		{"other op", at(func(c *hostclaim.Claim) { c.Op.ID = "other" }), false, true},
		{"other exec", at(func(c *hostclaim.Claim) { c.Op.Exec = "elsewhere" }), false, true},
		{"lost", at(func(c *hostclaim.Claim) { c.Lost = []string{"box"} }), false, true},
		{"serving elsewhere", at(func(c *hostclaim.Claim) {
			c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "elsewhere"}
			c.Holder.WriterBox = "elsewhere"
		}), false, true},
		{"absent", hostclaim.Observed{Host: probeHost, Absent: true}, false, true},
	}
	for _, c := range cases {
		open, err := recoveryHoldVerdict(admitted, c.obs, "box")
		if open != c.open || (err != nil) != c.err || (err != nil && !errors.Is(err, hostbox.ErrChanged)) {
			t.Errorf("%s: open=%v err=%v", c.name, open, err)
		}
	}
}

func TestRecoveryHoldReviewOpensOrStops(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	admitted := probeAdmitted(obs)
	opened := probeAdmitted(obs)
	oc := *opened.Claim
	oc.Op.Phase = hostclaim.PhaseOpened
	opened.Claim = &oc

	a, r, _ := probeService(t, obs)
	e, proc, _, err := heldProbe(t, a, 0, admitted, probeInputs())
	if err != nil {
		t.Fatal(err)
	}
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
	a.reviewRecoveryHold(e)
	a.mu.Lock()
	held := e.probeHeld
	a.mu.Unlock()
	if !held {
		t.Fatal("released at probing")
	}
	r.Read = func(context.Context, string) (hostclaim.Observed, error) {
		return hostclaim.Observed{}, errors.New("offline")
	}
	a.reviewRecoveryHold(e)
	a.mu.Lock()
	held, state := e.probeHeld, e.state
	a.mu.Unlock()
	if !held || state != actAwake {
		t.Fatal("an unreadable claim changed the hold")
	}
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return opened, nil }
	a.reviewRecoveryHold(e)
	a.mu.Lock()
	held, state = e.probeHeld, e.state
	a.mu.Unlock()
	if held || state != actAwake || proc.isStopped() {
		t.Fatalf("opened: held=%v state=%v", held, state)
	}
	release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost})
	if err != nil {
		t.Fatalf("an opened probe does not serve: %v", err)
	}
	release()

	b, rb, _ := probeService(t, obs)
	e, proc, _, err = heldProbe(t, b, 0, admitted, probeInputs())
	if err != nil {
		t.Fatal(err)
	}
	moved := probeAdmitted(admitted)
	rb.Read = func(context.Context, string) (hostclaim.Observed, error) { return moved, nil }
	b.reviewRecoveryHold(e)
	select {
	case <-e.stopped:
	case <-time.After(5 * time.Second):
		t.Fatal("a changed session kept its process")
	}
	if !proc.isStopped() || b.Owns(probeHost) {
		t.Fatal("not stopped")
	}
}

// With the probe fields unset, the claim wake is main's: the ordinary
// policy, never Probe, never Inputs.
func TestOrdinaryClaimWakeKeepsTheServingPolicy(t *testing.T) {
	c := &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc", Session: 1, Generation: "one", Location: "l", WriterBox: "box"}, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}
	a, r, run := probeService(t, probeObserved(c))
	stop := errors.New("stop")
	r.Prerequisites = func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
		run.step("prerequisites", p)
		return stop
	}
	if _, err := a.doClaimWake(t.Context(), &actEntry{host: probeHost, route: registry.Route{Host: probeHost}}); !errors.Is(err, stop) {
		t.Fatal(err)
	}
	run.mu.Lock()
	defer run.mu.Unlock()
	p := run.policies[len(run.policies)-1]
	if !reflect.DeepEqual(p, StartPolicy{MaxStarts: 5, Window: 30 * time.Second}) {
		t.Fatalf("ordinary policy %+v", p)
	}
}

func TestRecoveryProbeOwnsItsInputs(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, r, _ := probeService(t, obs)
	in := probeInputs()
	called := false
	r.Prerequisites = func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p StartPolicy) error {
		called = true
		in.Release = "changed"
		in.BundleETags["main"] = "changed"
		if p.Inputs.Release != "release" || p.Inputs.BundleETags["main"] != "etag" {
			t.Error("caller mutated captured probe inputs")
		}
		return hostbox.ErrChanged
	}
	_ = a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in))
	if !called {
		t.Fatal("did not reach prerequisites")
	}
}

func TestRecoveryProbeChangedInputsRetireHeldEntry(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, _, _ := probeService(t, obs)
	e, proc, rep, err := heldProbe(t, a, 0, probeAdmitted(obs), probeInputs())
	if err != nil {
		t.Fatal(err)
	}
	in := probeInputs()
	in.Release = "new-release"
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in)); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatal(err)
	}
	a.mu.Lock()
	stopped := e.stopped
	state := e.state
	a.mu.Unlock()
	if state != actStopping && a.Owns(probeHost) {
		t.Fatal("changed inputs stranded old held probe")
	}
	select {
	case <-stopped:
	case <-time.After(5 * time.Second):
		t.Fatal("old probe did not stop")
	}
	if !proc.isStopped() || !rep.isClosed() || a.Owns(probeHost) {
		t.Fatal("old probe retained resources")
	}
}

func TestRecoveryProbeChangedInputsCancelWakingEntry(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, _, _ := probeService(t, obs)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	p := probePolicy(probeInputs())
	e := &actEntry{host: probeHost, recoveryPolicy: &p, probeHeld: true, state: actWaking, wakeCancel: cancel, ready: make(chan struct{})}
	a.entries[probeHost] = e
	in := probeInputs()
	in.Release = "new"
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(in)); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatal(err)
	}
	if ctx.Err() == nil {
		t.Fatal("stale in-flight probe was not cancelled")
	}
	delete(a.entries, probeHost)
}

func TestRecoveryHoldSurvivesEdgeRetireExcept(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, _, _ := probeService(t, obs)
	e, _, _, err := heldProbe(t, a, 0, probeAdmitted(obs), probeInputs())
	if err != nil {
		t.Fatal(err)
	}
	a.RetireExcept(map[string]bool{})
	a.mu.Lock()
	state := e.state
	a.mu.Unlock()
	if state != actAwake {
		t.Fatal("edge reload retired operation-owned recovery probe")
	}
}

func TestRecoveryProbeLateHealthyAfterInputChangeStops(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, _, _ := probeService(t, obs)
	p := probePolicy(probeInputs())
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	e := &actEntry{host: probeHost, recoveryPolicy: &p, probeHeld: true, state: actWaking, wakeCancel: cancel, ready: make(chan struct{}), probeAdmitted: probeAdmitted(obs)}
	a.entries[probeHost] = e
	in := probeInputs()
	in.Release = "new"
	_ = a.joinRecoveryProbe(t.Context(), e, probePolicy(in))
	if ctx.Err() == nil {
		t.Fatal("not cancelled")
	}
	proc, rep := &fakeProc{}, &fakeRep{}
	err := a.finishRecoveryProbe(e, 0, &wokenInstance{proc: proc, rep: rep, stopOutcome: &stopObservation{}}, nil)
	if !errors.Is(err, hostbox.ErrChanged) {
		t.Fatal("late success kept obsolete inputs", err)
	}
	select {
	case <-e.stopped:
	case <-time.After(5 * time.Second):
		t.Fatal("not stopped")
	}
	if !proc.isStopped() || !rep.isClosed() {
		t.Fatal("obsolete process remains")
	}
}

// Recovery must retain the handle returned by the launcher, and only publish
// it as current after the exact admitted session opens.
func TestRecoveryProbeGenerationRoutingAfterOpening(t *testing.T) {
	for _, generated := range []bool{false, true} {
		for _, phase := range []string{hostclaim.PhaseOpened, hostclaim.KindServing} {
			t.Run(fmt.Sprint(generated)+"/"+phase, func(t *testing.T) {
				obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
				admitted := probeAdmitted(obs)
				a, r, _ := probeService(t, obs)
				a.gens = generated
				rt := registry.Route{Host: probeHost, Socket: filepath.Join(t.TempDir(), "app.sock")}
				p := probePolicy(probeInputs())
				e := &actEntry{host: probeHost, route: rt, state: actWaking, ready: make(chan struct{}), recoveryPolicy: &p, probeHeld: true, probeAdmitted: admitted}
				a.entries[probeHost] = e
				proc, rep := &fakeProc{}, &fakeRep{}
				w := &wokenInstance{proc: proc, rep: rep, spawnVersion: "recovered", stopOutcome: &stopObservation{}}
				if generated {
					w.gen = Gen{N: nextGen(), Socket: filepath.Join(filepath.Dir(rt.Socket), "generation.sock"), Build: "recovered"}
				}
				if err := a.finishRecoveryProbe(e, 0, w, nil); err != nil {
					t.Fatal(err)
				}
				if e.gen.N == 0 {
					t.Fatal("recovery lost its generation")
				}
				want := w.gen
				if !generated {
					want = canonicalGen(rt, e.gen.N, w.spawnVersion)
				}
				if !reflect.DeepEqual(e.gen, want) {
					t.Fatalf("retained %+v, want %+v", e.gen, want)
				}
				if _, ok := a.CacheGen(probeHost); ok {
					t.Fatal("held generation became current")
				}
				if generated {
					if _, err := os.Lstat(rt.Socket); !errors.Is(err, os.ErrNotExist) {
						t.Fatalf("held generation linked: %v", err)
					}
				}
				opened := admitted
				c := *admitted.Claim
				c.Op.Phase = phase
				if phase == hostclaim.KindServing {
					c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
				}
				opened.Claim = &c
				r.Read = func(context.Context, string) (hostclaim.Observed, error) { return opened, nil }
				a.reviewRecoveryHold(e)
				if e.probeHeld || e.proc != Proc(proc) || proc.isStopped() {
					t.Fatal("opening replaced the recovered process")
				}
				got, ok := a.ServingGen(probeHost)
				if !ok || got.N != want.N || got.Socket != want.Socket || got.Build != want.Build {
					t.Fatalf("ServingGen = %+v, %v", got, ok)
				}
				if socket, ok := a.AppSocket(probeHost); !ok || socket != want.Socket {
					t.Fatalf("AppSocket = %q, %v", socket, ok)
				}
				if n, ok := a.CacheGen(probeHost); !ok || n != want.N {
					t.Fatalf("CacheGen = %d, %v", n, ok)
				}
				if generated {
					if target, err := os.Readlink(rt.Socket); err != nil || target != want.Socket {
						t.Fatalf("canonical link = %q, %v", target, err)
					}
				}
				release, err := a.beginRequest(t.Context(), rt)
				if err != nil {
					t.Fatal(err)
				}
				release()
			})
		}
	}
}

func TestRecoveryHoldSurvivesAdoptionAtProbed(t *testing.T) {
	t.Parallel()
	for _, kind := range []string{hostclaim.KindReverting, hostclaim.KindRestoring} {
		for _, fenced := range []bool{false, true} {
			for _, api := range []string{"Bounce", "BounceAdopted", "ConvergeAdopted"} {
				t.Run(kind+"/"+fmt.Sprint(fenced)+"/"+api, func(t *testing.T) {
					obs := probeObserved(probeClaim(kind, hostclaim.PhaseProbing, "box"))
					admitted := probeAdmitted(obs)
					a, r, _ := probeService(t, obs)
					a.fences = map[string]uint64{probeHost: 9}
					e, proc, rep, err := heldProbe(t, a, 9, admitted, probeInputs())
					if err != nil {
						t.Fatal(err)
					}
					probed := admitted
					c := *admitted.Claim
					c.Op.Phase = hostclaim.PhaseProbed
					probed.Claim = &c
					r.Read = func(context.Context, string) (hostclaim.Observed, error) { return probed, nil }
					if !fenced {
						delete(a.fences, probeHost)
					}
					a.reviewRecoveryHold(e)
					a.liveVer = func(string) (string, error) { return "adopted", nil }
					var bounced bool
					switch api {
					case "Bounce":
						bounced = a.Bounce(probeHost)
					case "BounceAdopted":
						bounced = a.BounceAdopted(probeHost)
					case "ConvergeAdopted":
						bounced = a.ConvergeAdopted(probeHost, "adopted")
					}
					a.mu.Lock()
					held, state, same := e.probeHeld, e.state, a.entries[probeHost] == e
					a.mu.Unlock()
					if bounced || !held || state != actAwake || !same || proc.isStopped() || rep.isClosed() {
						t.Fatalf("adoption disturbed probed session: bounced=%v held=%v state=%v same=%v", bounced, held, state, same)
					}
					if err := a.RecoveryProbe(t.Context(), e.route, *e.recoveryPolicy); err != nil {
						t.Fatalf("cannot rejoin held session: %v", err)
					}
					if _, err := a.beginRequest(t.Context(), e.route); !errors.Is(err, ErrFenced) {
						t.Fatalf("probed session served: %v", err)
					}
				})
			}
		}
	}
}

func TestRecoveryProbeUnboundAttemptsAreNotReady(t *testing.T) {
	a, r, _ := probeService(t, probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")))
	r.RecoveryAttemptsReady = nil
	if a.RecoveryProbeReady() {
		t.Fatal("recovery probe ready without recovery attempts provider")
	}
}