package activator

import (
	"context"
	"errors"
	"os/exec"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"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"
)

const (
	pwHost = "shop.example.com"
	pwOp   = "mv_probewake"
)

type pwRoutes struct {
	mu sync.Mutex
	rt registry.Route
}

func (r *pwRoutes) Get(string) (*registry.Route, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	rt := r.rt
	return &rt, nil
}

// pwRuntime is a claim runtime that records what the activator asked of it.
type pwRuntime struct {
	mu        sync.Mutex
	observed  hostclaim.Observed
	calls     []string
	policies  []startPolicy
	failedFor []hostclaim.Observed
	stop      error
	stopped   int
}

func (r *pwRuntime) note(call string, p startPolicy) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.calls = append(r.calls, call)
	r.policies = append(r.policies, p)
}

func (r *pwRuntime) admitted() hostclaim.Observed {
	o := r.observed
	c := *o.Claim
	h := *c.Holder
	h.Session, h.Generation, h.WriterBox = h.Session+5, pwOp, "box"
	c.Holder = &h
	o.Claim, o.ETag = &c, "admitted"
	return o
}

func (r *pwRuntime) runtime(t *testing.T) *ClaimRuntime {
	work, _ := (hostwrite.Config{}).Select(t.Context(), pwHost, "spawn")
	return &ClaimRuntime{Box: "box",
		Read: func(_ context.Context, host string) (hostclaim.Observed, error) {
			r.mu.Lock()
			defer r.mu.Unlock()
			if len(r.calls) > 1 {
				return r.admitted(), nil
			}
			return r.observed, nil
		},
		StopForeign: func(context.Context, string) error { r.mu.Lock(); r.stopped++; r.mu.Unlock(); return nil },
		Prerequisites: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) error {
			r.note("prerequisites", p)
			return nil
		},
		Admit: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (hostclaim.Observed, error) {
			r.note("admit", p)
			return r.admitted(), nil
		},
		Prepare: func(_ context.Context, _ registry.Route, got hostclaim.Observed, p startPolicy) (ClaimStart, error) {
			r.note("prepare", p)
			return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} }, Host: pwHost, 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 { r.note("restore", p); return nil },
				Materialize: func(context.Context) error { r.note("materialize", p); return r.stop },
				Build:       func() *exec.Cmd { return nil }, Ready: func(context.Context, *exec.Cmd) error { return nil },
				Reserve:   func(_ context.Context, n int) error { r.note("reserve", p); return nil },
				Failed:    func(context.Context, int) error { r.note("failed", p); return nil },
				Healthy:   func(context.Context, int, *exec.Cmd) error { return nil },
				Replicate: func(context.Context, string) (Rep, error) { return nil, nil },
				Publish:   func(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) { return o, nil },
				Retain:    func(Proc, Rep, error) {}}, nil
		},
		FailedAdmission: func(_ context.Context, o hostclaim.Observed, _ error) error {
			r.mu.Lock()
			r.failedFor = append(r.failedFor, o)
			r.mu.Unlock()
			return nil
		}}
}

func pwFlip() hostclaim.Observed {
	return hostclaim.Observed{Host: pwHost, ETag: "flip", Claim: &hostclaim.Claim{Host: pwHost,
		Holder: &hostclaim.Holder{Incarnation: "inc", Location: "instances/" + pwHost, Stamp: 18, Session: 15, Generation: "src", WriterBox: "ie-box"},
		Op:     hostclaim.Op{Kind: hostclaim.KindLanding, ID: pwOp, Phase: hostclaim.PhaseLandingProbing, Exec: "box"}}}
}

func pwService(t *testing.T, r *pwRuntime) (*Service, *pwRoutes) {
	t.Helper()
	routes := &pwRoutes{rt: registry.Route{Host: pwHost, Kind: "instance", Status: "landing", HoldOp: pwOp, Socket: "/run/x.sock", DBPath: "/data/x.db"}}
	a := New(Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: r.runtime(t)}, &fakeBackend{}, nil)
	a.Routes = routes
	return a, routes
}

func pwConfig() ProbeConfig {
	return ProbeConfig{Host: pwHost, Op: pwOp, HoldOp: pwOp, Deadline: time.Now().Add(time.Minute),
		Prepared: &moves.InstallManifest{Op: pwOp, Host: pwHost, N: 42}, Inputs: &provision.MoveInputs{Release: "v1"}}
}

// ProbeWake is the claim path with the probe's policy (Plan 3 Task 5a, H2):
// the claim's own Prerequisites, Admit and admitted start sequence, each
// given the probe policy (five starts, the 30 s window, the deadline, the
// prepared manifest and the pins). It adds no admission or restore of its
// own. A failure after the admission is ErrProbeFailed, the admitted tree is
// settled through FailedAdmission, and the entry is gone.
func TestProbeWakeRunsTheClaimPathWithTheProbePolicy(t *testing.T) {
	stop := errors.New("materialize failed")
	r := &pwRuntime{observed: pwFlip(), stop: stop}
	a, _ := pwService(t, r)
	cfg := pwConfig()
	_, err := a.ProbeWake(t.Context(), cfg)
	if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, stop) || errors.Is(err, ErrProbeRefused) {
		t.Fatalf("err = %v", err)
	}
	if got := strings.Join(r.calls, ","); got != "prerequisites,admit,prepare,restore,reserve,materialize,failed" {
		t.Fatalf("calls: %s", got)
	}
	want := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
	for i, p := range r.policies {
		if p.Gate != nil {
			t.Fatalf("%s ran a probe under a pin gate", r.calls[i])
		}
		p.Gate = nil
		if !reflect.DeepEqual(p, want) {
			t.Fatalf("%s ran under %+v", r.calls[i], p)
		}
	}
	if len(r.failedFor) != 1 || r.failedFor[0].ETag != "admitted" {
		t.Fatalf("FailedAdmission: %+v", r.failedFor)
	}
	if a.Owns(pwHost) {
		t.Fatal("a failed probe left its entry")
	}
}

// A probe is never planned (merge of the zero-downtime scheduler, plan 5):
// with the scheduler armed, ProbeWake still runs the landing's prepared
// build under its own policy. The planner is not asked, nothing is pinned,
// published or fenced for a plan, and no plan gate, selection check or
// fallback reaches the probe's starts.
func TestProbeWakeIsNeverPlanned(t *testing.T) {
	stop := errors.New("materialize failed")
	r := &pwRuntime{observed: pwFlip(), stop: stop}
	a, _ := pwService(t, r)
	planned := 0
	a.Planner = func(context.Context, registry.Route) WakePlan {
		planned++
		return WakePlan{Refuse: "planner consulted for a probe"}
	}
	cfg := pwConfig()
	_, err := a.ProbeWake(t.Context(), cfg)
	if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, stop) {
		t.Fatalf("err = %v", err)
	}
	if planned != 0 {
		t.Fatalf("the planner was consulted %d times for a probe", planned)
	}
	want := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
	for i, p := range r.policies {
		if p.Gate != nil || p.PinCheck != nil || p.Fallback != nil || p.Pin != nil || !p.HealthBy.IsZero() {
			t.Fatalf("%s ran a probe under a plan's policy: %+v", r.calls[i], p)
		}
		if !reflect.DeepEqual(p, want) {
			t.Fatalf("%s ran under %+v", r.calls[i], p)
		}
	}
}

// TestProbeWakeRefusesUnheldRowOrForeignExec (the activator's half; the
// owners' is TestProbeRefusesUnheldRowOrForeignExec in internal/host): the
// row must be held with hold_op == HoldOp and the claim must name this box
// op.exec under op Op at probing, else ErrProbeRefused before any admission.
func TestProbeWakeRefusesUnheldRowOrForeignExec(t *testing.T) {
	for name, tc := range map[string]struct {
		route func(*registry.Route)
		claim func(*hostclaim.Claim)
		cfg   func(*ProbeConfig)
	}{
		"a ready row":        {route: func(r *registry.Route) { r.Status, r.HoldOp = "ready", "" }},
		"held by another op": {route: func(r *registry.Route) { r.HoldOp = "mv_other" }},
		"a foreign executor": {claim: func(c *hostclaim.Claim) { c.Op.Exec = "other-box" }},
		"another op":         {claim: func(c *hostclaim.Claim) { c.Op.ID = "mv_other" }},
		"not probing":        {claim: func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseLandingCommitted }},
		"a serving claim":    {claim: func(c *hostclaim.Claim) { c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} }},
		"another hold op":    {cfg: func(c *ProbeConfig) { c.HoldOp = "rb_other" }},
	} {
		t.Run(name, func(t *testing.T) {
			r := &pwRuntime{observed: pwFlip()}
			if tc.claim != nil {
				tc.claim(r.observed.Claim)
			}
			a, routes := pwService(t, r)
			if tc.route != nil {
				tc.route(&routes.rt)
			}
			cfg := pwConfig()
			if tc.cfg != nil {
				tc.cfg(&cfg)
			}
			_, err := a.ProbeWake(t.Context(), cfg)
			if !errors.Is(err, ErrProbeRefused) {
				t.Fatalf("err = %v", err)
			}
			if len(r.calls) != 0 {
				t.Fatalf("the claim path ran: %v", r.calls)
			}
			if a.Owns(pwHost) {
				t.Fatal("a refused probe left an entry")
			}
		})
	}
	// A live entry for the host (an earlier probe's process, or a wake) is
	// never replaced by a probe.
	r := &pwRuntime{observed: pwFlip()}
	a, _ := pwService(t, r)
	a.mu.Lock()
	a.entries[pwHost] = &actEntry{host: pwHost, state: actWaking, ready: make(chan struct{})}
	a.mu.Unlock()
	if _, err := a.ProbeWake(t.Context(), pwConfig()); !errors.Is(err, ErrProbeRefused) || len(r.calls) != 0 {
		t.Fatalf("over a live entry: %v %v", err, r.calls)
	}
	// Malformed configurations never reach the claim.
	for _, mutate := range []func(*ProbeConfig){
		func(c *ProbeConfig) { c.Host = "" }, func(c *ProbeConfig) { c.Op = "" }, func(c *ProbeConfig) { c.HoldOp = "" },
		func(c *ProbeConfig) { c.Deadline = time.Time{} },
	} {
		cfg := pwConfig()
		mutate(&cfg)
		if _, err := a.ProbeWake(t.Context(), cfg); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("malformed config %+v: %v", cfg, err)
		}
	}
}

// The cut deadline is checked before the admission, with the window not yet
// open (brief step 4): a deadline already past is ErrAttemptOver and nothing
// reaches the claim. A deadline that passes during the admitted sequence
// ends it with ErrAttemptOver through ErrProbeFailed (startAdmitted's loop,
// claimloop_linux_test.go). The 30 s window itself opens only inside
// RoleLaunch.Start, which main still refuses
// (TestStartAdmittedLoopWindowAndReadinessAtTheBoundary).
func TestProbeWakeEndsAtTheCutDeadline(t *testing.T) {
	t.Parallel()
	r := &pwRuntime{observed: pwFlip()}
	a, _ := pwService(t, r)
	cfg := pwConfig()
	cfg.Deadline = time.Now().Add(-time.Second)
	if _, err := a.ProbeWake(t.Context(), cfg); !errors.Is(err, ErrAttemptOver) || errors.Is(err, ErrProbeFailed) || len(r.calls) != 0 {
		t.Fatalf("err = %v calls = %v", err, r.calls)
	}
	// Past the deadline once admitted: the admitted sequence ends at once.
	cfg.Deadline = time.Now().Add(100 * time.Millisecond)
	r.stop = errors.New("unused")
	slow := r.runtime(t)
	admit := slow.Admit
	slow.Admit = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p startPolicy) (hostclaim.Observed, error) {
		time.Sleep(150 * time.Millisecond)
		return admit(ctx, rt, o, p)
	}
	a.cfg.Claims = slow
	_, err := a.ProbeWake(t.Context(), cfg)
	if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, ErrAttemptOver) {
		t.Fatalf("err = %v", err)
	}
	for _, call := range r.calls {
		if call == "reserve" {
			t.Fatal("a start was reserved past the deadline")
		}
	}
}

// TestProbeAttemptEndsAtTheWindow (brief): blocked on the claim's isolated
// root launcher, as the loop test it would extend is.
func TestProbeAttemptEndsAtTheWindow(t *testing.T) {
	t.Skip("blocked: RoleLaunch.Start and ClaimRoleAssembly.Preflight return ErrClaimIsolationUnavailable on main " +
		"(claim Plan 9 Group D did not replace them), so OpenProbe and Ready never run through startAdmitted. " +
		"When they do: OpenProbe answering a time 31 s ago, readiness answering after it, gives ErrProbeFailed and ErrAttemptOver.")
}

// The probe's entry stays held: an edge reload (RetireExcept) never stops a
// probe entry while the registry row is still held by the probe's hold op,
// and does once it is not (H18).
func TestRetireExceptKeepsTheHeldProbeEntry(t *testing.T) {
	t.Parallel()
	a, _, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	held := registry.Route{Host: "one.example.com", Kind: "instance", Socket: "/tmp/nope.sock", DBPath: "/tmp/one.example.com.db", Hibernate: true}
	if err := reg.Upsert(held); err != nil {
		t.Fatal(err)
	}
	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	release()
	if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
		t.Fatal(err)
	}
	a.mu.Lock()
	a.entries["one.example.com"].probe = &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp}
	a.mu.Unlock()

	a.RetireExcept(map[string]bool{})
	time.Sleep(50 * time.Millisecond)
	if !a.Owns("one.example.com") {
		t.Fatal("an edge reload retired the held probe's entry")
	}
	if err := reg.SetHold("one.example.com", "landing", "mv_other"); err != nil {
		t.Fatal(err)
	}
	a.RetireExcept(map[string]bool{})
	waitFor(t, "the entry no longer held for its probe to retire", func() bool { return !a.Owns("one.example.com") })
}

// Finding 6: a probe entry published after RetireExcept's snapshot of the
// held probes, and before its retirement pass, is not retired: only entries
// the snapshot validated are.
func TestRetireExceptKeepsAProbeInsertedBetweenTheLocks(t *testing.T) {
	a, _, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	held := registry.Route{Host: "two.example.com", Kind: "instance", Socket: "/tmp/nope.sock", DBPath: "/tmp/two.example.com.db", Hibernate: true}
	if err := reg.Upsert(held); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetHold("two.example.com", "landing", pwOp); err != nil {
		t.Fatal(err)
	}
	inserted := &actEntry{host: "two.example.com", route: held, state: actAwake, ready: make(chan struct{}), proc: &fakeProc{}, rep: &fakeRep{},
		probe: &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp}}
	close(inserted.ready)
	retireBetweenLocks = func() {
		a.mu.Lock()
		a.entries[inserted.host] = inserted
		a.mu.Unlock()
	}
	t.Cleanup(func() { retireBetweenLocks = nil })
	a.RetireExcept(map[string]bool{})
	retireBetweenLocks = nil
	a.mu.Lock()
	state := inserted.state
	a.mu.Unlock()
	if state != actAwake || !a.Owns(inserted.host) {
		t.Fatal("a held probe inserted between the locks was retired")
	}
	// The next reload validates it like any other.
	a.RetireExcept(map[string]bool{})
	if !a.Owns(inserted.host) {
		t.Fatal("a held probe was retired on the next reload")
	}
}

// Finding 7: a cancel that lands after the probe's entry is published and
// before its wake starts still stops the wake: nothing is admitted.
func TestProbeWakeCancelBeforeTheWakeStarts(t *testing.T) {
	r := &pwRuntime{observed: pwFlip()}
	a, _ := pwService(t, r)
	rt := a.cfg.Claims
	prereq := rt.Prerequisites
	rt.Prerequisites = func(ctx context.Context, route registry.Route, o hostclaim.Observed, p startPolicy) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		return prereq(ctx, route, o, p)
	}
	ctx, cancel := context.WithCancel(t.Context())
	probeWakePublished = cancel
	t.Cleanup(func() { probeWakePublished = nil })
	_, err := a.ProbeWake(ctx, pwConfig())
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("err = %v", err)
	}
	for _, call := range r.calls {
		if call == "admit" || call == "prerequisites" {
			t.Fatalf("the cancelled probe ran %v", r.calls)
		}
	}
	if a.Owns(pwHost) {
		t.Fatal("the cancelled probe left its entry")
	}
}

// Task 6 fix round 1: LiveProbe answers the awake entry a passing probe left,
// with its op and admitted session; nothing for a waking entry, an ordinary
// wake, or a probe that never passed.
func TestLiveProbeAnswersThePassingProbeEntry(t *testing.T) {
	a, _, _, _ := testActivator(t, time.Hour)
	admitted := hostclaim.Observed{Host: pwHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 20, Stamp: 18, Generation: pwOp}}}
	probe := &probeEntry{op: pwOp, holdOp: pwOp, admitted: &admitted}
	a.mu.Lock()
	a.entries[pwHost] = &actEntry{host: pwHost, state: actAwake, probe: probe}
	a.entries["plain.example.com"] = &actEntry{host: "plain.example.com", state: actAwake}
	a.entries["waking.example.com"] = &actEntry{host: "waking.example.com", state: actWaking, probe: &probeEntry{op: pwOp, admitted: &admitted}}
	a.entries["unpassed.example.com"] = &actEntry{host: "unpassed.example.com", state: actAwake, probe: &probeEntry{op: pwOp}}
	a.mu.Unlock()
	op, res, ok := a.LiveProbe(pwHost)
	if !ok || op != pwOp || res != (ProbeResult{Session: 20, Stamp: 18, Generation: pwOp}) {
		t.Fatalf("the passing probe: %q %+v %v", op, res, ok)
	}
	for _, host := range []string{"plain.example.com", "waking.example.com", "unpassed.example.com", "absent.example.com"} {
		if _, _, ok := a.LiveProbe(host); ok {
			t.Fatalf("%s answered", host)
		}
	}
}

// pwHeldEntry is an awake probe entry for one.example.com, held by pwOp.
func pwHeldEntry(t *testing.T, a *Service) *actEntry {
	t.Helper()
	admitted := hostclaim.Observed{Host: "one.example.com", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 20, Stamp: 18, Generation: pwOp}}}
	e := &actEntry{host: "one.example.com", route: testRoute("one.example.com"), state: actAwake, ready: make(chan struct{}), proc: &fakeProc{}, rep: &fakeRep{},
		probe: &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp, admitted: &admitted}}
	close(e.ready)
	a.mu.Lock()
	a.entries[e.host] = e
	a.mu.Unlock()
	return e
}

// Task 7: the target's open marks the live probe entry for its op open;
// nothing else is marked.
func TestOpenProbeEntryMarksTheLiveProbe(t *testing.T) {
	a, _, _, _ := testActivator(t, time.Hour)
	e := pwHeldEntry(t, a)
	if a.OpenProbeEntry("one.example.com", "mv_other") || a.OpenProbeEntry("absent.example.com", pwOp) {
		t.Fatal("another op's or an absent entry was marked")
	}
	if !a.OpenProbeEntry("one.example.com", pwOp) {
		t.Fatal("the probe's entry was not marked")
	}
	a.mu.Lock()
	phase := e.probe.phase
	a.mu.Unlock()
	if phase != probeOpening {
		t.Fatal("not marked opening")
	}
}

// Task 7, the open race (deterministic): an edge reload that read the routes
// before the hold was released, and retires after it, keeps the probe entry
// the open marked: its row is ready, and the next reload manages it. Once
// the edge manages it, it is an ordinary serving entry.
func TestRetireExceptKeepsTheOpenedProbeAcrossTheOpen(t *testing.T) {
	t.Parallel()
	a, _, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	if err := reg.Upsert(testRoute("one.example.com")); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
		t.Fatal(err)
	}
	e := pwHeldEntry(t, a)
	// The reload's managed set was read while the row was held.
	managed := map[string]bool{}
	if !a.OpenProbeEntry("one.example.com", pwOp) {
		t.Fatal("not marked")
	}
	if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
		t.Fatalf("release: %v %v", ok, err)
	}
	a.RetireExcept(managed)
	a.mu.Lock()
	state := e.state
	a.mu.Unlock()
	if state != actAwake || !a.Owns("one.example.com") {
		t.Fatal("the opened probe's entry was retired across the open")
	}
	// The next reload manages the ready row: the entry is ordinary from here.
	a.RetireExcept(map[string]bool{"one.example.com": true})
	a.mu.Lock()
	taken := e.probe.managed
	a.mu.Unlock()
	if !taken {
		t.Fatal("a managed entry is still a probe")
	}
	// An entry held by the probe but never opened, whose row is ready, is
	// not the probe's: it is retired as before.
	a2, _, _, reg2 := testActivator(t, time.Hour)
	a2.Routes = reg2
	if err := reg2.Upsert(testRoute("one.example.com")); err != nil {
		t.Fatal(err)
	}
	pwHeldEntry(t, a2)
	a2.RetireExcept(map[string]bool{})
	waitFor(t, "the unopened probe on a ready row to retire", func() bool { return !a2.Owns("one.example.com") })
}

// Task 7, the idle stop: a held probe entry gets no traffic, so the idle
// sweep must not stop it before the open; once opened it idles as any entry.
func TestIdleSweepSparesAHeldProbe(t *testing.T) {
	t.Parallel()
	a, _, clk, _ := testActivator(t, time.Minute)
	e := pwHeldEntry(t, a)
	a.mu.Lock()
	e.lastActive = clk.now().Add(-time.Hour)
	a.mu.Unlock()
	a.sweepOnce()
	a.mu.Lock()
	state := e.state
	a.mu.Unlock()
	if state != actAwake {
		t.Fatal("the idle sweep stopped a held probe")
	}
	if !a.OpenProbeEntry("one.example.com", pwOp) || !a.ConfirmProbeOpened("one.example.com", pwOp) {
		t.Fatal("not opened")
	}
	a.sweepOnce()
	waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
}

// Task 7 fix round 1, finding 2: the open lands between RetireExcept's probe
// snapshot (not opened) and its registry read (ready, unheld). The
// registry's answer is older than the entry's open, so the entry is kept,
// not retired as a probe that never opened.
func TestRetireExceptRevalidatesTheOpenUnderTheFinalLock(t *testing.T) {
	a, _, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	if err := reg.Upsert(testRoute("one.example.com")); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
		t.Fatal(err)
	}
	e := pwHeldEntry(t, a)
	retireBeforeRegistryRead = func() {
		if !a.OpenProbeEntry("one.example.com", pwOp) {
			t.Error("not marked")
		}
		if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
			t.Errorf("release: %v %v", ok, err)
		}
	}
	t.Cleanup(func() { retireBeforeRegistryRead = nil })
	a.RetireExcept(map[string]bool{})
	retireBeforeRegistryRead = nil
	a.mu.Lock()
	state := e.state
	a.mu.Unlock()
	if state != actAwake || !a.Owns("one.example.com") {
		t.Fatal("the probe opened during the retirement was retired")
	}
}

// Final review, ruling 2: two reloads overlap. The older pass snapshots the
// held probe, the open lands, and a newer pass, whose managed set includes
// the host, takes the entry over as ordinary serving, all before the older
// pass reads the registry. Neither pass panics, and the older one does not
// retire the serving entry.
func TestRetireExceptOverlappingReloadsAcrossTheOpen(t *testing.T) {
	a, _, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	if err := reg.Upsert(testRoute("one.example.com")); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
		t.Fatal(err)
	}
	e := pwHeldEntry(t, a)
	retireBeforeRegistryRead = func() {
		retireBeforeRegistryRead = nil // the newer pass runs without the pause
		if !a.OpenProbeEntry("one.example.com", pwOp) {
			t.Error("not marked")
		}
		if ok, err := reg.ReleaseHold("one.example.com", pwOp); err != nil || !ok {
			t.Errorf("release: %v %v", ok, err)
		}
		a.ConfirmProbeOpened("one.example.com", pwOp)
		a.RetireExcept(map[string]bool{"one.example.com": true})
	}
	t.Cleanup(func() { retireBeforeRegistryRead = nil })
	a.RetireExcept(map[string]bool{})
	a.mu.Lock()
	state := e.state
	a.mu.Unlock()
	if state != actAwake || !a.Owns("one.example.com") {
		t.Fatal("the older reload retired the serving probe")
	}
}

// Final review, ruling 3: the mark is "opening"; idle protection ends only
// once the release is confirmed. A failed release leaves the entry
// protected through an idle sweep; a retried open that succeeds ends it.
func TestIdleProtectionHoldsUntilTheReleaseIsConfirmed(t *testing.T) {
	t.Parallel()
	a, _, clk, _ := testActivator(t, time.Minute)
	e := pwHeldEntry(t, a)
	a.mu.Lock()
	e.lastActive = clk.now().Add(-time.Hour)
	a.mu.Unlock()
	if !a.OpenProbeEntry("one.example.com", pwOp) {
		t.Fatal("not marked")
	}
	// The release failed: no confirmation.
	a.sweepOnce()
	a.mu.Lock()
	state := e.state
	a.mu.Unlock()
	if state != actAwake {
		t.Fatal("an opening probe whose release failed was idled")
	}
	// The retried open: marked again, released, confirmed.
	if !a.OpenProbeEntry("one.example.com", pwOp) || !a.ConfirmProbeOpened("one.example.com", pwOp) {
		t.Fatal("the retried open")
	}
	a.sweepOnce()
	waitFor(t, "the opened probe to idle out", func() bool { return !a.Owns("one.example.com") })
}