package activator

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"maps"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rolloutwords"
)

// planGenBackend is plan 3's SwapBackend (startPinnedAt's staged spawn) plus
// plan 2's KillGen over the package's fakeBackend. Its generations are
// fakeProcs a test can crash. It also starts a legacy sidecar on a
// generation's staged build (SidecarGenStarter), recording the generation
// it was handed.
type planGenBackend struct {
	*fakeBackend
	gmu         sync.Mutex
	gens        []Gen
	procs       map[uint64]*fakeProc
	killed      []uint64
	killErr     error
	gate        chan struct{}
	sidecarGens []Gen
}

func newPlanGenBackend() *planGenBackend {
	return &planGenBackend{fakeBackend: &fakeBackend{}, procs: map[uint64]*fakeProc{}}
}

func (b *planGenBackend) StartStaged(ctx context.Context, _ registry.Route, g Gen) (Gen, Proc, error) {
	b.gmu.Lock()
	gate := b.gate
	b.gmu.Unlock()
	if gate != nil {
		<-gate
	}
	// The caller's exec gate at the launch point (final review F1), as the
	// real backend asks it after its preparation.
	if err := runExecGate(ctx); err != nil {
		return g, nil, err
	}
	b.gmu.Lock()
	defer b.gmu.Unlock()
	g.Socket = "/run/carlos/g/test." + strconv.FormatUint(g.N, 10) + ".sock"
	g.Control = g.Socket + ".control"
	p := &fakeProc{label: "gen:" + g.Build, rec: b.fakeBackend}
	b.gens = append(b.gens, g)
	b.procs[g.N] = p
	return g, p, nil
}

func (b *planGenBackend) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault }
func (b *planGenBackend) GenReady(registry.Route, Gen, Proc)          {}
func (b *planGenBackend) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error {
	return nil
}
func (b *planGenBackend) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error {
	return nil
}
func (b *planGenBackend) GenSwitched(registry.Route, Gen, Proc) error { return nil }

func (b *planGenBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error {
	b.gmu.Lock()
	defer b.gmu.Unlock()
	b.killed = append(b.killed, n)
	if b.killErr != nil {
		return b.killErr
	}
	if p := b.procs[n]; p != nil {
		p.exit()
	}
	return nil
}

func (b *planGenBackend) StartSidecarGen(_ context.Context, _ registry.Route, g Gen) (Proc, error) {
	b.gmu.Lock()
	b.sidecarGens = append(b.sidecarGens, g)
	b.gmu.Unlock()
	return &fakeProc{label: "sidecar", rec: b.fakeBackend}, nil
}

var (
	_ SwapBackend       = (*planGenBackend)(nil)
	_ genKiller         = (*planGenBackend)(nil)
	_ SidecarGenStarter = (*planGenBackend)(nil)
)

func (b *planGenBackend) started() []Gen {
	b.gmu.Lock()
	defer b.gmu.Unlock()
	return append([]Gen(nil), b.gens...)
}

func (b *planGenBackend) kills() []uint64 {
	b.gmu.Lock()
	defer b.gmu.Unlock()
	return append([]uint64(nil), b.killed...)
}

func (b *planGenBackend) procOf(n uint64) *fakeProc {
	b.gmu.Lock()
	defer b.gmu.Unlock()
	return b.procs[n]
}

func (b *planGenBackend) sidecarsOn() []Gen {
	b.gmu.Lock()
	defer b.gmu.Unlock()
	return append([]Gen(nil), b.sidecarGens...)
}

// stagedID writes a staged build and returns its identity: plan 3's
// startPinnedAt verifies the bytes before anything runs.
func stagedID(t *testing.T, dir, ver string) release.StagedIdentity {
	t.Helper()
	body := []byte("build " + ver)
	path := filepath.Join(dir, ver)
	if err := os.WriteFile(path, body, 0o755); err != nil {
		t.Fatal(err)
	}
	sum := sha256.Sum256(body)
	return release.StagedIdentity{Version: ver, StagedPath: path, Digest: hex.EncodeToString(sum[:]), Demand: "hold-" + ver}
}

type genLog struct {
	mu  sync.Mutex
	evs []GenEvent
}

func (l *genLog) add(ev GenEvent) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.evs = append(l.evs, ev)
}

func (l *genLog) kinds() []GenEventKind {
	l.mu.Lock()
	defer l.mu.Unlock()
	out := make([]GenEventKind, 0, len(l.evs))
	for _, ev := range l.evs {
		out = append(out, ev.Kind)
	}
	return out
}

func (l *genLog) wait(t *testing.T, kind GenEventKind) GenEvent {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		l.mu.Lock()
		for _, ev := range l.evs {
			if ev.Kind == kind {
				l.mu.Unlock()
				return ev
			}
		}
		l.mu.Unlock()
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("no %s event; saw %v", kind, l.kinds())
	return GenEvent{}
}

func planActivator(t *testing.T, plan WakePlan) (*Service, *planGenBackend, *genLog) {
	t.Helper()
	b := newPlanGenBackend()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
	a.gens = true // plan 1's switch, as CARLOS_SWAP=overlap sets it before New
	evs := &genLog{}
	a.OnGen = evs.add
	a.Planner = func(context.Context, registry.Route) WakePlan { return plan }
	a.genHealth = func(context.Context, Gen) error { return nil }
	return a, b, evs
}

func shrinkFirstWake(t *testing.T, budget, reserve, settle time.Duration) {
	t.Helper()
	b, r, s, k := firstWakeBudget, firstWakeKillReserve, firstWakeSettle, firstWakeKillBound
	firstWakeBudget, firstWakeKillReserve, firstWakeSettle = budget, reserve, settle
	t.Cleanup(func() { firstWakeBudget, firstWakeKillReserve, firstWakeSettle, firstWakeKillBound = b, r, s, k })
}

func TestFirstWakeFallsBackToTheGoodBuildWithinTheBudget(t *testing.T) {
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	var mu sync.Mutex
	deadlines := map[string]time.Time{}
	a.genHealth = func(ctx context.Context, g Gen) error {
		dl, _ := ctx.Deadline()
		mu.Lock()
		deadlines[g.Build] = dl
		mu.Unlock()
		if g.Build == "v2" {
			return errors.New("GET / answered 500")
		}
		return nil
	}
	start := time.Now()
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("the waiting request was not served by the good build: %v", err)
	}
	done()
	gens := b.started()
	if len(gens) != 2 || gens[0].Build != "v2" || gens[1].Build != "v1" {
		t.Fatalf("spawned %+v, want v2 then the good v1", gens)
	}
	mu.Lock()
	window := deadlines["v2"].Sub(start)
	mu.Unlock()
	if window < 29*time.Second || window > 31*time.Second {
		t.Fatalf("B's health window ends %v after the wake began, want 30s: the 55s deadline minus 25s", window)
	}
	if k := b.kills(); len(k) != 1 || k[0] != gens[0].N {
		t.Fatalf("killed %v, want B's generation %d alone, by cgroup.kill", k, gens[0].N)
	}
	ev := evs.wait(t, GenFellBack)
	if ev.Failed.Version != "v2" || ev.Gen.Build != "v1" {
		t.Fatalf("fallback event = %+v", ev)
	}
	if g, ok := a.Current("one.example.com"); !ok || g.Build != "v1" {
		t.Fatalf("Current = %+v %v, want the good build serving", g, ok)
	}
}

// Spec §6.4: once a hold build has executed, nothing falls back.
func TestAHoldFirstWakeNeverFallsBack(t *testing.T) {
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true, Fallback: &good})
	a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
	if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); err == nil {
		t.Fatal("a failed hold first wake was served")
	}
	if gens := b.started(); len(gens) != 1 {
		t.Fatalf("spawned %+v, want B alone", gens)
	}
	if ev := evs.wait(t, GenFailed); ev.Mode != ModeHold || ev.Failed.Version != "v2" {
		t.Fatalf("failure event = %+v", ev)
	}
}

// Spec §7.4, §8.1–8.4 (P5 review finding 3): a planned wake re-checks its
// build at the exec, after the lease and the restore. Here a sibling's
// failure holds v2 between the planning and the exec: the wake is planned
// again on the restored state and boots the good build, and v2 never
// execs. Without the re-check v2 ran after its hold.
func TestAPlannedWakeRechecksItsBuildAtTheExec(t *testing.T) {
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{})
	var held atomic.Bool
	check := func(build string) error {
		if build == "v2" && held.Load() {
			return errors.New("v2 failed on a sibling")
		}
		return nil
	}
	var plans atomic.Int32
	a.Planner = func(context.Context, registry.Route) WakePlan {
		if plans.Add(1) == 1 {
			held.Store(true) // the sibling's failure lands after this plan is made
			return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: check}
		}
		return WakePlan{Boot: good, Mode: ModeOverlap, Check: check}
	}
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("the wake: %v", err)
	}
	done()
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v1" {
		t.Fatalf("spawned %+v, want the good v1 alone", gens)
	}
	if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) {
		t.Fatalf("events %v: a build that never ran was recorded", k)
	}
}

// PF7: the selection gate is asked again AT the exec, inside startPinnedAt's
// gate (after the staged bytes are verified and the tier read), not only
// before the spawn. A failure recorded in that moment holds the build: it
// never execs, the wake is abandoned (errPlanStale), and nothing is
// recorded against the build.
func TestThePlanCheckAtTheExecRefusesWithoutRecording(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{})
	var calls atomic.Int32
	check := func(build string) error {
		if build == "v2" && calls.Add(1) >= 2 {
			return errors.New("v2 failed on a sibling")
		}
		return nil
	}
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: check}
	}
	_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if !errors.Is(err, errPlanStale) {
		t.Fatalf("err = %v, want errPlanStale from the exec's gate", err)
	}
	if got := classifyStep(nil, err); got != stepStale {
		t.Fatalf("classifyStep = %v, want stepStale: a refusal at the exec is abandonment", got)
	}
	if gens := b.started(); len(gens) != 0 {
		t.Fatalf("spawned %+v: a held build execed", gens)
	}
	if calls.Load() < 2 {
		t.Fatalf("Check asked %d times, want the second at the exec", calls.Load())
	}
	time.Sleep(50 * time.Millisecond)
	if k := evs.kinds(); len(k) != 0 {
		t.Fatalf("events %v: a refusal before the exec was recorded", k)
	}
}

// PF7: the wake's own locally held lease is part of every planned exec's
// gate. A renewal that fails after the restore execs nothing: no
// generation starts, nothing is recorded, nothing falls back.
func TestAPlannedWakeWhoseLeaseIsLostExecsNothing(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{})
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good,
			Check: func(string) error {
				// Asked after the lease and the restore: lose the lease here.
				b.lastRep().setErr(errors.New("lease renewal failed"))
				return nil
			}}
	}
	_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if !errors.Is(err, errSwapLeaseLost) {
		t.Fatalf("err = %v, want the lost lease (errSwapLeaseLost)", err)
	}
	if gens := b.started(); len(gens) != 0 {
		t.Fatalf("spawned %+v on a lost lease", gens)
	}
	time.Sleep(50 * time.Millisecond)
	if k := evs.kinds(); len(k) != 0 {
		t.Fatalf("events %v: a refusal before the exec was recorded", k)
	}
}

// Spec §7.4 (P5 review rule a): a wake's fallback passes the ownership
// check at its own exec. Here the host is fenced while B fails its health:
// the good build is not started. B's failure is still reported. Checked
// only before B's exec, the fallback ran on a fenced host.
func TestAFallbackOnAFencedHostIsNotStarted(t *testing.T) {
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	const host = "one.example.com"
	a.genHealth = func(_ context.Context, g Gen) error {
		if g.Build != "v2" {
			return nil
		}
		a.mu.Lock()
		a.fences = map[string]uint64{host: 1} // maintenance fences the host meanwhile
		a.mu.Unlock()
		return errors.New("GET / answered 500")
	}
	if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
		t.Fatal("a fenced host's fallback served")
	}
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v2" {
		t.Fatalf("spawned %+v, want B alone", gens)
	}
	evs.wait(t, GenFailed)
}

func TestAnUnprovenKillStartsNoSecondGeneration(t *testing.T) {
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	a.genHealth = func(_ context.Context, g Gen) error { return errors.New("GET / answered 500") }
	b.killErr = ErrStillRunning
	if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); err == nil {
		t.Fatal("served while B's end was unproven")
	}
	if gens := b.started(); len(gens) != 1 {
		t.Fatalf("spawned %+v: two generations against one database while B may still run", gens)
	}
	evs.wait(t, GenFailed)
}

func TestAnExitInsideTheFirstTenSecondsFailsTheBuild(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 400*time.Millisecond)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	evs.wait(t, GenHealthy)
	if _, ok := a.Current("one.example.com"); ok {
		t.Fatal("Current must hide a first wake under observation, so nothing swaps it meanwhile")
	}
	b.procOf(b.started()[0].N).crash()
	if ev := evs.wait(t, GenFailed); ev.Failed.Version != "v2" {
		t.Fatalf("failure event = %+v", ev)
	}
}

func TestAnIntentionalStopInsideTheWindowRecordsNothing(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 400*time.Millisecond)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	evs.wait(t, GenHealthy)
	a.BounceFor("one.example.com", CauseConfig)
	time.Sleep(800 * time.Millisecond)
	// Spec §6.4: nothing about the build is recorded. Spec §9.1: the
	// observation is still closed, by GenCancelled, or the scheduler shows
	// the host settling for ever and the deploy watch never completes.
	if k := evs.kinds(); !slices.Equal(k, []GenEventKind{GenHealthy, GenCancelled}) {
		t.Fatalf("events %v: want healthy, then cancelled, and nothing recorded against the build", k)
	}
}

func TestAFirstWakeThatSurvivesSettles(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if ev := evs.wait(t, GenSettled); ev.Gen.Build != "v2" {
		t.Fatalf("settled event = %+v", ev)
	}
	g, ok := a.Current("one.example.com")
	if !ok || g.Build != "v2" {
		t.Fatalf("Current = %+v %v after the settle", g, ok)
	}
	// PF10: the planned generation carries its staged identity, so the
	// scheduler can record it as good (it needs the digest) and the legacy
	// sidecar binds to it.
	if g.StagedPath != next.StagedPath || g.Digest != next.Digest || g.Demand != next.Demand || g.N == 0 {
		t.Fatalf("Current = %+v, want generation of %+v with its staged identity", g, next)
	}
}

func TestFirstWakeWaitersGetTheLongerBudget(t *testing.T) {
	old := wakeWaitBound
	wakeWaitBound = 100 * time.Millisecond
	t.Cleanup(func() { wakeWaitBound = old })
	shrinkFirstWake(t, 600*time.Millisecond, 200*time.Millisecond, time.Millisecond)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")

	first, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	b.gate = make(chan struct{})
	time.AfterFunc(250*time.Millisecond, func() { close(b.gate) })
	done, err := first.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("a first wake's waiter gave up at the ordinary bound: %v", err)
	}
	done()

	plain, b2, _ := planActivator(t, WakePlan{Boot: good, Mode: ModeOverlap})
	b2.gate = make(chan struct{})
	time.AfterFunc(250*time.Millisecond, func() { close(b2.gate) })
	if _, err := plain.beginRequest(context.Background(), testRoute("two.example.com")); !errors.Is(err, errWakeTimeout) {
		t.Fatalf("an ordinary wake's waiter = %v, want errWakeTimeout at wakeWaitBound", err)
	}
}

func TestARefusedPlanTakesNoLease(t *testing.T) {
	a, b, _ := planActivator(t, WakePlan{Refuse: "held down after build v2 failed in hold mode"})
	_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err == nil || !strings.Contains(err.Error(), "refusing to wake") {
		t.Fatalf("err = %v", err)
	}
	if acquires, _, _ := b.counts(); acquires != 0 {
		t.Fatal("a refused wake took the lease")
	}
}

// Claims on (plan 3 Task 10 runs the fallback inside startAdmitted): a
// claimed first wake that fell back reports as one, a failed one names the
// build that failed, and one where nothing of the build ran reports nothing.
func TestAClaimedFirstWakeReportsLikeAnUnclaimedOne(t *testing.T) {
	good := release.StagedIdentity{Version: "v1", StagedPath: "/opt/v1", Digest: "d1"}
	next := release.StagedIdentity{Version: "v2", StagedPath: "/opt/v2", Digest: "d2"}
	a := &Service{}
	e := &actEntry{host: "one.example.com", plan: WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good},
		pinFailed: &Gen{Build: "v2"}, pinFailReason: "GET / answered 500"}
	w := a.claimedPlanResult(e, &wokenInstance{spawnVersion: "v1", gen: Gen{N: 9}})
	if !w.fellBack || w.failed == nil || w.failed.Version != "v2" || w.planned.Build != "v1" || w.planned.N != 9 {
		t.Fatalf("fell back: %+v", w)
	}
	if w := a.claimedPlanResult(e, nil); w == nil || w.failed == nil || w.failed.Version != "v2" {
		t.Fatalf("failed: %+v", w)
	}
	e.pinFailed = nil
	if w := a.claimedPlanResult(e, nil); w != nil {
		t.Fatalf("nothing of the build ran, yet %+v", w)
	}
}

// shrinkClaimStartBackoff sets startAdmitted's backoff base for one test.
func shrinkClaimStartBackoff(t *testing.T, d time.Duration) {
	t.Helper()
	old := claimStartBackoff
	claimStartBackoff = d
	t.Cleanup(func() { claimStartBackoff = old })
}

// PF17: every claims-on launch refuses on this base (RoleLaunch.Start
// answers ErrClaimIsolationUnavailable before its exec hook). A planned
// claimed wake carries the plan into the start policy (the pin, the good
// build, the health cap, the ownership gate and the selection gate), meets
// the closed launch, and records nothing: no failure event, no fallback.
// The refused start is retried on startAdmitted's backoff, whose schedule
// TestStartAdmittedLoopStopsAfterFiveStartsWithDoublingBackoff pins; here
// it is shrunk so the retries do not take 15 s.
func TestAPlannedClaimedWakeRefusesAtTheLaunchRecordingNothing(t *testing.T) {
	shrinkClaimStartBackoff(t, 10*time.Millisecond)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	const host = "x.example"
	obs := claimedObs(host)
	obs.Claim.Op.Exec = "box"
	obs.Claim.Holder.WriterBox = "box"
	admitted := obs
	claim := *obs.Claim
	holder := *claim.Holder
	holder.Session, holder.Generation = 2, "two"
	claim.Holder = &holder
	admitted.Claim = &claim
	work, _ := (hostwrite.Config{}).Select(context.Background(), host, "spawn")
	var (
		pmu      sync.Mutex
		policies []startPolicy
		reads    atomic.Int32
	)
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Hour, HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
		Read: func(context.Context, string) (hostclaim.Observed, error) {
			if reads.Add(1) == 1 {
				return obs, nil
			}
			return admitted, nil
		},
		StopForeign:     func(context.Context, string) error { return nil },
		FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil },
		Prerequisites:   func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil },
		Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
			return admitted, nil
		},
		Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) {
			pmu.Lock()
			policies = append(policies, p)
			pmu.Unlock()
			return claimedStartFor(admitted, work, p.Pin.Build), nil
		}}}, &fakeBackend{}, reg)
	evs := &genLog{}
	a.OnGen = evs.add
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: func(string) error { return nil }}
	}
	_, err = a.beginRequest(context.Background(), registry.Route{Host: host, Kind: "instance", Hibernate: true})
	if !errors.Is(err, ErrClaimIsolationUnavailable) {
		t.Fatalf("err = %v, want the closed launch", err)
	}
	pmu.Lock()
	defer pmu.Unlock()
	if len(policies) != 1 {
		t.Fatalf("prepared %d times, want once", len(policies))
	}
	p := policies[0]
	if p.Pin == nil || p.Pin.Build != "v2" || p.Pin.StagedPath != next.StagedPath || p.Pin.Digest != next.Digest {
		t.Fatalf("policy pin = %+v, want the planned build", p.Pin)
	}
	if p.Fallback == nil || p.Fallback.Build != "v1" || p.HealthBy.IsZero() || p.Gate == nil || p.PinCheck == nil {
		t.Fatalf("policy = %+v: want the good build, the health cap, the ownership and selection gates", p)
	}
	time.Sleep(50 * time.Millisecond)
	if k := evs.kinds(); len(k) != 0 {
		t.Fatalf("events %v: a launch refused before its exec recorded something", k)
	}
}

// execStagedBackend runs planned wakes through the REAL exec backend's
// StartStaged (real processes, plan 3's real flat and foreign spawn paths)
// over fakeBackend's lease and restore. Its KillGen stops the generation's
// process, because a test fixture is not cgroup v2.
type execStagedBackend struct {
	*fakeBackend
	eb    *ExecBackend
	smu   sync.Mutex
	procs map[uint64]Proc
}

func (b *execStagedBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) {
	got, p, err := b.eb.StartStaged(ctx, rt, g)
	if p != nil {
		b.smu.Lock()
		b.procs[g.N] = p
		b.smu.Unlock()
	}
	return got, p, err
}
func (b *execStagedBackend) GenTier(rt registry.Route, build string) cgroup.Limit {
	return b.eb.GenTier(rt, build)
}
func (b *execStagedBackend) GenReady(rt registry.Route, g Gen, p Proc) { b.eb.GenReady(rt, g, p) }
func (b *execStagedBackend) RaiseGenParent(ctx context.Context, rt registry.Route, inc, succ cgroup.Limit) error {
	return b.eb.RaiseGenParent(ctx, rt, inc, succ)
}
func (b *execStagedBackend) RestoreGenParent(ctx context.Context, rt registry.Route, tier cgroup.Limit) error {
	return b.eb.RestoreGenParent(ctx, rt, tier)
}
func (b *execStagedBackend) GenSwitched(rt registry.Route, g Gen, p Proc) error {
	return b.eb.GenSwitched(rt, g, p)
}
func (b *execStagedBackend) Layout(rt registry.Route) (Layout, string) { return b.eb.Layout(rt) }
func (b *execStagedBackend) KillGen(ctx context.Context, _ registry.Route, n uint64) error {
	b.smu.Lock()
	p := b.procs[n]
	b.smu.Unlock()
	if p == nil {
		return nil
	}
	if err := p.Stop(ctx); err != nil {
		return err
	}
	if !p.Exited() {
		return ErrStillRunning
	}
	return nil
}

// stagedScript writes an executable staged build and returns its identity.
func stagedScript(t *testing.T, dir, ver, body string) release.StagedIdentity {
	t.Helper()
	path := filepath.Join(dir, ver)
	if err := os.WriteFile(path, []byte(body), 0o755); err != nil {
		t.Fatal(err)
	}
	sum := sha256.Sum256([]byte(body))
	return release.StagedIdentity{Version: ver, StagedPath: path, Digest: hex.EncodeToString(sum[:])}
}

// Spec §8.2 on the real exec backend: B never answers, so its failure must
// be seen inside the wake's own budget, recorded, and the good build booted
// in its place. StartStaged hands back the running process at once (plan 3
// Task 1), so startPinnedAt's health deadline (55 s − 25 s, shrunk here) is
// the only one, and the kill reaches a real process. A StartStaged that
// waited out the backend's own 30 s spawnHealthTimeout and returned a nil
// handle made spawnPlanned read "nothing executed": no failure recorded, no
// fallback, and the waiter timed out. The foreign row runs the same way on
// its canonical socket (plan 1's reading): refusing it failed every planned
// wake of a foreign row.
func TestAPlannedFirstWakeOnTheRealExecBackendFallsBackInTime(t *testing.T) {
	for _, layout := range []string{"flat", "foreign"} {
		t.Run(layout, func(t *testing.T) {
			oldBound := wakeWaitBound
			wakeWaitBound = time.Second
			t.Cleanup(func() { wakeWaitBound = oldBound })
			shrinkFirstWake(t, 4*time.Second, 2*time.Second, time.Millisecond) // B's health window: 2 s
			// The floor shrinks with the window, or B (2 s) would never be
			// exec'd under the spec's 20 s floor.
			shrinkVerdictFloor(t, time.Second)
			eb := testExecBackend(t)
			eb.RunDirOverride = t.TempDir()
			host := "real.example.com"
			stageLive(t, eb, host, "v0")
			rt := registry.Route{Host: host, Kind: "instance", Hibernate: true, DBPath: filepath.Join(eb.DataDir, host+".db"),
				Socket: filepath.Join(eb.RunDirOverride, host+".sock")}
			want := LayoutFlat
			if layout == "foreign" {
				rt.Socket, want = filepath.Join(t.TempDir(), "hand-set.sock"), LayoutForeign
			}
			if got, _ := eb.Layout(rt); got != want {
				t.Fatalf("fixture layout %v, want %v", got, want)
			}
			dir := t.TempDir()
			next := stagedScript(t, dir, "v2", "#!/bin/sh\nexec sleep 30\n") // never binds
			good := stagedScript(t, dir, "v1", "#!/bin/sh\nexec sleep 30\n")
			b := &execStagedBackend{fakeBackend: &fakeBackend{}, eb: eb, procs: map[uint64]Proc{}}
			reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
			if err != nil {
				t.Fatal(err)
			}
			a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
			a.gens = true
			evs := &genLog{}
			a.OnGen = evs.add
			a.Planner = func(context.Context, registry.Route) WakePlan {
				return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
			}
			// B gets the real check: GET / on its real socket, which nothing
			// binds, bounded only by the caller's deadline. The good build is
			// let through, because a shell script cannot answer HTTP.
			a.genHealth = func(ctx context.Context, g Gen) error {
				if g.Build == "v1" {
					return nil
				}
				return waitHealthyCtx(ctx, g.Socket, spawnHealthTimeout)
			}
			t.Cleanup(func() { a.Shutdown(context.Background()) })
			start := time.Now()
			done, err := a.beginRequest(context.Background(), rt)
			if err != nil {
				t.Fatalf("the waiter was not served by the good build: %v", err)
			}
			done()
			if took := time.Since(start); took > 4*time.Second {
				t.Fatalf("served after %v, past the first-wake budget", took)
			}
			ev := evs.wait(t, GenFellBack)
			if ev.Failed.Version != "v2" || ev.Gen.Build != "v1" {
				t.Fatalf("fallback event = %+v", ev)
			}
			b.smu.Lock()
			procs := maps.Clone(b.procs)
			b.smu.Unlock()
			if len(procs) != 2 {
				t.Fatalf("%d generations spawned, want B and the good build", len(procs))
			}
			for n, p := range procs {
				if n != ev.Gen.N && !p.Exited() {
					t.Fatalf("B's process (generation %d) outlived its kill", n)
				}
			}
		})
	}
}

// PF10 (spec §12 "Legacy sidecar binding"): the co-spawned legacy sidecar
// of a planned wake runs the build the wake left serving. After a first
// wake fell back, that is the good build, by its staged path, never B's
// and never the live link.
func TestAFallenBackFirstWakeBindsItsSidecarToTheGoodBuild(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	b.sidecarStaged = true
	a.genHealth = func(_ context.Context, g Gen) error {
		if g.Build == "v2" {
			return errors.New("GET / answered 500")
		}
		return nil
	}
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	ev := evs.wait(t, GenFellBack)
	sc := b.sidecarsOn()
	if len(sc) != 1 || sc[0].StagedPath != good.StagedPath || sc[0].Digest != good.Digest || sc[0].N != ev.Gen.N {
		t.Fatalf("sidecar started on %+v, want the good build's generation %d at %s", sc, ev.Gen.N, good.StagedPath)
	}
	if n := b.sidecarStartCount(); n != 0 {
		t.Fatalf("%d live-link sidecar starts, want none", n)
	}
	if ev.Gen.StagedPath != good.StagedPath || ev.Gen.Digest != good.Digest || ev.Gen.Demand != good.Demand {
		t.Fatalf("fallback event's generation %+v lacks the good build's staged identity", ev.Gen)
	}
}

// Spec §6.5: a socketless planned wake is ready once it is still running
// 10 s after it started, and a clean exit inside that window is a
// one-shot's success: the same rule ReplaceWorker applies. The flag-off
// awaitWorkerSettle fails every early exit, exit 0 included, so a one-shot's
// first wake was recorded as a failed build (and could halt its cohort).
func TestAPlannedOneShotWorkerThatExitsCleanlyIsReady(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	evs := &genLog{}
	a.OnGen = evs.add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	b.workerDeadOnStart = true // the child exits at once, cleanly
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v (a one-shot's clean exit is its success)", err)
	}
	ev := evs.wait(t, GenSettled)
	if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) {
		t.Fatalf("events %v: a clean exit was recorded against the build", k)
	}
	if calls := b.calls(); len(calls) != 1 {
		t.Fatalf("spawns %v: a clean exit fell back to the good build", calls)
	}
	// A worker's generation is compared by its staged artifact (the
	// scheduler's acknowledged), so the event names it, and it is a worker's.
	if !ev.Gen.Worker || ev.Gen.StagedPath != next.StagedPath || ev.Gen.Digest != next.Digest {
		t.Fatalf("settled event's generation = %+v, want the worker's v2 artifact", ev.Gen)
	}
}

// plannedHostKill is plan 4's stagedFake plus plan 3's KillHost, failing
// while bad: a worker host cgroup that still holds a process. (The name
// stagedHostKill is worker_retained_pin_test.go's.)
type plannedHostKill struct {
	*stagedFake
	bad atomic.Bool
}

func (s *plannedHostKill) KillHost(context.Context, registry.Route) error {
	if s.bad.Load() {
		return errors.New("the host cgroup still holds a process")
	}
	return nil
}

// Spec §7.4, §7.9: a failed socketless first wake falls back to the good
// build only once the failed worker's host cgroup is proven empty (plan 3's
// genTerminated on a Worker generation). Its leader's exit alone let the
// good build start beside a descendant of the failed one.
func TestAFailedWorkerFirstWakeFallsBackOnlyOnceItsCgroupIsEmpty(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	hk := &plannedHostKill{stagedFake: b}
	hk.bad.Store(true)
	a.backend = hk
	a.OnGen = (&genLog{}).add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	go func() {
		waitFor(t, "the v2 child", func() bool { return len(b.calls()) == 1 })
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	_ = a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal"))
	if calls := b.calls(); len(calls) != 1 {
		t.Fatalf("spawns %v: the good build started beside the failed worker's unproven host cgroup", calls)
	}
}

// The same, with the host cgroup proven empty: the good build replaces the
// failed worker, through startWorker's gate, and the wake reports the
// fallback with the good build's artifact.
func TestAFailedWorkerFirstWakeFallsBackToTheGoodBuild(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	hk := &plannedHostKill{stagedFake: b}
	a.backend = hk
	evs := &genLog{}
	a.OnGen = evs.add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	go func() {
		waitFor(t, "the v2 child", func() bool { return len(b.calls()) == 1 })
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v, want the good build serving", err)
	}
	calls := b.calls()
	if len(calls) != 2 || calls[0] == nil || calls[0].Version != "v2" || calls[1] == nil || calls[1].Version != "v1" {
		t.Fatalf("spawns %v, want v2 then the good v1, each by name", calls)
	}
	ev := evs.wait(t, GenFellBack)
	if ev.Failed.Version != "v2" || ev.Gen.StagedPath != good.StagedPath || !ev.Gen.Worker {
		t.Fatalf("fallback event = %+v", ev)
	}
	if got := a.entryServing("bot.x.internal"); got.StagedPath != good.StagedPath {
		t.Fatalf("serving %+v, want the good build", got)
	}
}

// breakOnAcquire is a staged backend that makes the generations file
// unwritable as the wake takes its lease: after the wake's own publication
// (doWake's step 0.8) and before startWorker's before-exec publication.
type breakOnAcquire struct {
	*stagedFake
	t    *testing.T
	path string
}

func (b *breakOnAcquire) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
	breakGenerationsFile(b.t, b.path)
	return b.stagedFake.Acquire(ctx, rt)
}

// PF9: a planned worker spawn refused because its build could not be
// published first (ErrBuildNotPublished) ran nothing of the build. It is
// abandonment: no failure recorded, nothing falls back. The wake's own
// publication (step 0.8) succeeds here; the file breaks afterwards, so the
// refusal is startWorker's before-exec publication, with the spawn attempt
// reaching the backend and refused inside it.
func TestAPlannedWorkerRefusedForPublicationRecordsNothing(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	a.backend = &breakOnAcquire{stagedFake: b, t: t, path: a.GenerationsFile}
	evs := &genLog{}
	a.OnGen = evs.add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal"))
	if !errors.Is(err, ErrBuildNotPublished) {
		t.Fatalf("WakeWorker = %v, want ErrBuildNotPublished", err)
	}
	if n := b.workerStarts(); n != 0 {
		t.Fatalf("%d worker children started for an unpublished build", n)
	}
	if calls := b.calls(); len(calls) != 1 {
		t.Fatalf("spawn attempts %v: a refusal before the exec fell back", calls)
	}
	time.Sleep(50 * time.Millisecond)
	if k := evs.kinds(); len(k) != 0 {
		t.Fatalf("events %v: a refusal before the exec was recorded", k)
	}
}

// The wake's own publication (Task 12, doWake step 0.8) refuses a planned
// worker before the lease when the file cannot be written at all: nothing is
// taken, no spawn is attempted, let alone a fallback, and nothing is
// recorded.
func TestAPlannedWorkerRefusedAtItsPublicationAttemptsNoSpawn(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	blocker := filepath.Join(t.TempDir(), "not-a-dir")
	if err := os.WriteFile(blocker, nil, 0o644); err != nil {
		t.Fatal(err)
	}
	a.GenerationsFile = filepath.Join(blocker, "generations.json") // unwritable
	evs := &genLog{}
	a.OnGen = evs.add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	// Step 0.8 is plan 5's staged publication, flag on only, so it carries
	// that path's sentinel (final review F9); the worker's own exec-time
	// publication keeps ErrBuildNotPublished (the test above).
	err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal"))
	if !errors.Is(err, errStagedBuildNotPublished) {
		t.Fatalf("WakeWorker = %v, want errStagedBuildNotPublished", err)
	}
	if n := b.workerStarts(); n != 0 {
		t.Fatalf("%d worker children started for an unpublished build", n)
	}
	// The plan is published before the lease (Task 12), so the refusal
	// comes before any spawn is attempted, let alone a fallback.
	if calls := b.calls(); len(calls) != 0 {
		t.Fatalf("spawn attempts %v: an unpublished plan reached a spawn", calls)
	}
	time.Sleep(50 * time.Millisecond)
	if k := evs.kinds(); len(k) != 0 {
		t.Fatalf("events %v: a refusal before the exec was recorded", k)
	}
}

// PF9: the one publication sentinel is abandonment wherever it surfaces.
func TestErrBuildNotPublishedIsAbandonment(t *testing.T) {
	err := fmt.Errorf("start bot.x.internal: %w", fmt.Errorf("%w: disk full", ErrBuildNotPublished))
	if got := classifyStep(nil, err); got != stepStale {
		t.Fatalf("classifyStep = %v, want stepStale", got)
	}
	if got := verdictOf(err); got != gateProceed {
		t.Fatalf("verdictOf = %v: main's flag-off respawn reads verdictOf, which must stay as it was", got)
	}
}

// Final review F9: the worker path keeps main's sentinel text (origin/main
// a3bf7d6c). Zero-downtime plan 6 (Task 14b) gives a worker wake whose
// publication fails a member-safe event of its own, whatever the flag: the
// draft below (zdd.wake.not-published), never the wrapped detail. The staged
// instance path has a sentinel of its own: abandonment too, and its member
// event reads as the maintenance fence's existing wording, never the raw text.
func TestTheWorkerPublicationSentinelKeepsMainsText(t *testing.T) {
	if got, want := ErrBuildNotPublished.Error(), "worker build not published before its exec"; got != want {
		t.Fatalf("ErrBuildNotPublished = %q, want main's %q", got, want)
	}
	worker := fmt.Errorf("start bot.x.internal: %w", fmt.Errorf("%w: disk full", ErrBuildNotPublished))
	const draft = "wake failed: the box could not record which build this instance runs, so it did not start it. the next request tries again"
	if got := memberSafeWakeReason(worker); got != draft {
		t.Fatalf("worker member text = %q, want the draft %q", got, draft)
	}
	if got := memberSafeWakeReason(fmt.Errorf("%w: disk full", ErrBuildNotPublished)); got != draft || strings.Contains(got, "disk full") {
		t.Fatalf("wrapped member text = %q, want the draft alone", got)
	}
	inst := fmt.Errorf("one.example.com: refusing to wake: %w", errStagedBuildNotPublished)
	if errors.Is(inst, ErrBuildNotPublished) || errors.Is(worker, errStagedBuildNotPublished) {
		t.Fatal("the worker and instance sentinels must be distinct")
	}
	if got := classifyStep(nil, inst); got != stepStale {
		t.Fatalf("classifyStep(instance) = %v, want stepStale", got)
	}
	if got := verdictOf(inst); got != gateProceed {
		t.Fatalf("verdictOf(instance) = %v, want unchanged (gateProceed)", got)
	}
	if got, want := memberSafeWakeReason(inst), memberSafeWakeReason(ErrFenced); got != want {
		t.Fatalf("instance member text = %q, want the fence's reviewed wording %q", got, want)
	}
}

// Spec §6.4, §7.4: a wake cancelled by its lifecycle (here Shutdown, as
// maintenance or a stop would) is no failure of the build: no failure
// event, no fallback, so no halted cohort and no hold-down. spawnPlanned
// that marked every error with a process as a failure recorded one.
func TestACancelledFirstWakeRecordsNothing(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	entered := make(chan struct{})
	var once sync.Once
	a.genHealth = func(ctx context.Context, _ Gen) error {
		once.Do(func() { close(entered) })
		<-ctx.Done()
		return ctx.Err()
	}
	go a.beginRequest(context.Background(), testRoute("one.example.com")) //nolint:errcheck
	<-entered
	a.Shutdown(context.Background())
	time.Sleep(100 * time.Millisecond)
	if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) {
		t.Fatalf("events %v: a cancelled wake was recorded against the build", k)
	}
	if gens := b.started(); len(gens) != 1 {
		t.Fatalf("spawned %+v: a cancelled wake fell back", gens)
	}
}

// Spec §6.4, §8.1: a first wake's observation is registered in the awake
// transition, before the generation is reachable. Here the build crashes,
// and the sweep starts stopping it, in the moment between the wake becoming
// ready and its observer running. Registered late (in afterWake), Current
// exposed the build as stable meanwhile, and the observer, finding the
// entry stopping, reported GenCancelled: a real failure of the build lost.
//
// Only the first healthy generation crashes. The fixed planner plans the
// same first wake every time, so a hook that crashed every generation made
// a livelock: the parked request got out only if it took a.mu in the gap
// between ready closing and the hook's crash, which the Go scheduler rarely
// gives it (thousands of recycles at about 0.4 ms each, past the 30 s wake
// bound in about one run in ten, on main too). Now the request either wins
// that gap or rides the recycle to a second wake that stays up.
func TestAFirstWakeIsObservedFromTheMomentItIsReady(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	var visible atomic.Bool
	var crashOnce sync.Once
	a.OnGen = func(ev GenEvent) {
		if ev.Kind == GenHealthy {
			if _, ok := a.Current("one.example.com"); ok {
				visible.Store(true)
			}
			crashOnce.Do(func() {
				b.procOf(ev.Gen.N).crash()
				a.sweepOnce() // the sweep sees the exit before the observer runs
			})
		}
		evs.add(ev)
	}
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if ev := evs.wait(t, GenFailed); ev.Failed.Version != "v2" {
		t.Fatalf("failure event = %+v", ev)
	}
	if slices.Contains(evs.kinds(), GenCancelled) {
		t.Fatalf("events %v: a crash was reported as a cancelled observation", evs.kinds())
	}
	if visible.Load() {
		t.Fatal("Current exposed the first wake as stable before its observation was registered")
	}
}

// Spec §6.5, §8.6: a socketless first wake is settled once its readiness
// window is over. A crash after it is ordinary crash recovery, not a failed
// release. A second, HTTP-style observation after the readiness window
// recorded a worker that crashed at 10 s plus a moment as a failed build.
func TestAPlannedWorkerFirstWakeSettlesAtItsReadiness(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	evs := &genLog{}
	a.OnGen = evs.add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatal(err)
	}
	evs.wait(t, GenSettled)
	b.lastWorker().crash() // after its readiness window
	time.Sleep(400 * time.Millisecond)
	if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) {
		t.Fatalf("events %v: a crash after readiness was recorded against the build", k)
	}
}

// The flag off: no Planner, today's wake, no events.
func TestWithoutAPlannerTheWakeIsTodays(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	var fired bool
	a.OnGen = func(GenEvent) { fired = true }
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if b.startCount() != 1 || fired {
		t.Fatalf("starts %d, events %v: want today's Start and nothing else", b.startCount(), fired)
	}
}

func TestDriftFeedsTheSchedulerInsteadOfStopping(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	a.liveVer = func(string) (string, error) { return "v2", nil }
	b.execVerFn = func() string { return "v1" }
	var drifted []string
	var mu sync.Mutex
	a.OnDrift = func(h string) { mu.Lock(); drifted = append(drifted, h); mu.Unlock() }
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Second)
	a.sweepOnce()
	time.Sleep(50 * time.Millisecond)
	if b.lastProc().isStopped() {
		t.Fatal("the sweep stopped a drifted instance; with the flag on the scheduler owns the move (spec §10)")
	}
	mu.Lock()
	defer mu.Unlock()
	if len(drifted) == 0 || drifted[0] != "one.example.com" {
		t.Fatalf("OnDrift saw %v", drifted)
	}
}

type causeSink struct {
	mu   sync.Mutex
	msgs []string
}

func (s *causeSink) App(string, string, string, string) {}
func (s *causeSink) Event(_, _, _, _, msg string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.msgs = append(s.msgs, msg)
}

// Spec §9.4: the bounce event names the attempt's cause.
func TestBounceForNamesTheCause(t *testing.T) {
	a := newTestServiceAwake(t, "one.example.com")
	sink := &causeSink{}
	a.Logs = sink
	if !a.BounceFor("one.example.com", CauseConfig) {
		t.Fatal("BounceFor stopped nothing")
	}
	sink.mu.Lock()
	defer sink.mu.Unlock()
	want := fmt.Sprintf(rolloutwords.EventBounceInstance, rolloutwords.Cause(string(CauseConfig)))
	if len(sink.msgs) != 1 || sink.msgs[0] != want {
		t.Fatalf("events = %v, want %q", sink.msgs, want)
	}
}

// bounceText words both nouns through rolloutwords, and restart's cause
// phrase leaves the flag-off literals in bounceLocked unchanged.
func TestBounceTextIsTheApprovedWording(t *testing.T) {
	for _, cause := range []Cause{CauseRelease, CauseConfig, CauseRestart} {
		if got, want := bounceText(false, cause), fmt.Sprintf(rolloutwords.EventBounceInstance, rolloutwords.Cause(string(cause))); got != want {
			t.Errorf("instance, %s: %q, want %q", cause, got, want)
		}
		if got, want := bounceText(true, cause), fmt.Sprintf(rolloutwords.EventBounceSidecar, rolloutwords.Cause(string(cause))); got != want {
			t.Errorf("sidecar, %s: %q, want %q", cause, got, want)
		}
	}
	if got := bounceText(false, CauseRestart); got != "instance restarting (carlos restart)" {
		t.Errorf("restart bounce %q drifted from the flag-off literal", got)
	}
	if got := bounceText(true, CauseRestart); got != "sidecar restarting (carlos restart)" {
		t.Errorf("restart sidecar bounce %q drifted from the flag-off literal", got)
	}
}

// shrinkVerdictFloor sets the least health window a first wake's B is
// exec'd with, beside shrinkFirstWake's budget, so a test drives the floor
// in milliseconds rather than the spec's seconds.
func shrinkVerdictFloor(t *testing.T, floor time.Duration) {
	t.Helper()
	old := firstWakeMinVerdict
	firstWakeMinVerdict = floor
	t.Cleanup(func() { firstWakeMinVerdict = old })
}

// slowFirstCheck is a plan's selection gate whose first call, checkPlan's
// before the exec, takes d: it stands in for a slow lease, restore or
// seeding, all of which run before B's exec on the wake's own clock.
func slowFirstCheck(d time.Duration) func(string) error {
	var once sync.Once
	return func(string) error {
		once.Do(func() { time.Sleep(d) })
		return nil
	}
}

// Review fix round 1 (risk c): B's health window is fixed from the wake's
// start, so a slow before-exec phase leaves B too little time for a verdict.
// Below firstWakeMinVerdict B is never exec'd: the good build boots through
// the same gate, nothing is recorded against B, and the wake reports as a
// healthy wake of the good build that is not a first wake.
func TestAFirstWakeWithNoTimeForAVerdictBootsTheGoodBuild(t *testing.T) {
	// B's window ends 2 s after the wake began; the floor is 1 s; the work
	// before the exec takes 1.5 s, leaving B about 0.5 s.
	shrinkFirstWake(t, 3*time.Second, time.Second, firstWakeSettle)
	shrinkVerdictFloor(t, time.Second)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: slowFirstCheck(1500 * time.Millisecond)})
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("the waiting request was not served by the good build: %v", err)
	}
	done()
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v1" {
		t.Fatalf("spawned %+v, want the good v1 alone: B must not exec without time for a verdict", gens)
	}
	if k := b.kills(); len(k) != 0 {
		t.Fatalf("killed %v, want nothing", k)
	}
	ev := evs.wait(t, GenHealthy)
	if ev.FirstWake || ev.Gen.Build != "v1" || ev.Gen.StagedPath != good.StagedPath {
		t.Fatalf("healthy event = %+v, want the good build, not a first wake", ev)
	}
	time.Sleep(50 * time.Millisecond) // anything afterWake would emit after GenHealthy
	if k := evs.kinds(); len(k) != 1 {
		t.Fatalf("events %v, want GenHealthy alone: nothing recorded against B, no observation", k)
	}
	if g, ok := a.Current("one.example.com"); !ok || g.Build != "v1" {
		t.Fatalf("Current = %+v %v, want the good build serving and not under observation", g, ok)
	}
}

// The floor needs a good build to boot in B's place. With none, B is exec'd
// as a wake does today: spawnHealthTimeout from the exec, with no cap.
func TestAFirstWakeWithNoGoodBuildExecsBWhateverTimeIsLeft(t *testing.T) {
	shrinkFirstWake(t, 3*time.Second, time.Second, 50*time.Millisecond)
	shrinkVerdictFloor(t, time.Second)
	dir := t.TempDir()
	next := stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Check: slowFirstCheck(1500 * time.Millisecond)})
	var capped atomic.Bool
	a.genHealth = func(ctx context.Context, g Gen) error {
		if dl, ok := ctx.Deadline(); ok && time.Until(dl) < time.Second {
			capped.Store(true)
		}
		return ctx.Err()
	}
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("the wake: %v", err)
	}
	done()
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v2" {
		t.Fatalf("spawned %+v, want B: there is no good build to boot instead", gens)
	}
	if capped.Load() {
		t.Fatal("B's health check was capped: with no good build it keeps today's full window")
	}
	if ev := evs.wait(t, GenHealthy); !ev.FirstWake || ev.Gen.Build != "v2" {
		t.Fatalf("healthy event = %+v, want B's first wake", ev)
	}
}

// With at least the floor left, B is exec'd as before, and a failing B is
// still a failure that falls back (the real-budget case is
// TestFirstWakeFallsBackToTheGoodBuildWithinTheBudget).
func TestAFirstWakeWithTimeForAVerdictStillExecsB(t *testing.T) {
	shrinkFirstWake(t, 3*time.Second, time.Second, firstWakeSettle)
	shrinkVerdictFloor(t, time.Second)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: slowFirstCheck(0)})
	a.genHealth = func(_ context.Context, g Gen) error {
		if g.Build == "v2" {
			return errors.New("GET / answered 500")
		}
		return nil
	}
	done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("the waiting request was not served by the good build: %v", err)
	}
	done()
	if gens := b.started(); len(gens) != 2 || gens[0].Build != "v2" || gens[1].Build != "v1" {
		t.Fatalf("spawned %+v, want v2 then the good v1", gens)
	}
	if ev := evs.wait(t, GenFellBack); ev.Failed.Version != "v2" || ev.Gen.Build != "v1" {
		t.Fatalf("fallback event = %+v", ev)
	}
}

// Astra R3 ruling 206 and plan review finding 9 (zdd-readiness Task 6): a
// planned first wake whose nested start root accepted, with an uncertain
// outcome, may have run its build. In hold mode that is always a failure of
// the build (GenFailed, under the outcome fence), never a cancellation the
// scheduler could boot the build again after. In overlap only a unit that
// exited before it was observed (errStartedThenExited) is a proven crash at
// the start, so it fails the build, and the good build boots in its place
// only when the backend's kill proved the generation gone; unproven, the
// wake serves nothing. Any other uncertain overlap start keeps its
// GenCancelled with Executed and boots nothing. Each wake sends its token
// exactly one terminal event. Before the fix both hold cases sent
// GenCancelled and the proven overlap crash served nothing.
func TestAPlannedWakeWhoseNestedStartIsUncertain(t *testing.T) {
	t.Parallel()
	unitFailed := errors.New("the unit failed to start")
	// B (v2) is answered as the case says; the good build (v1) starts.
	bAnswers := func(inv string, err error) func(instancectl.StagedExe) (string, error) {
		return func(x instancectl.StagedExe) (string, error) {
			if x.Version == "v2" {
				return inv, err
			}
			return "inv-7", nil
		}
	}
	unproven := errors.New("stop timed out")
	for _, tc := range []struct {
		name      string
		mode      SwapMode
		answer    func(instancectl.StagedExe) (string, error)
		stopErr   error
		want      GenEventKind
		executed  bool // GenCancelled only
		goodAsked bool // root was asked to start the good build, which serves
	}{
		{"hold, the unit exited before it was observed", ModeHold, bAnswers("", nil), nil, GenFailed, false, false},
		{"hold, root's start failed", ModeHold, bAnswers("", unitFailed), nil, GenFailed, false, false},
		{"overlap, the unit exited before it was observed, kill proven", ModeOverlap, bAnswers("", nil), nil, GenFellBack, false, true},
		{"overlap, the unit exited before it was observed, kill unproven", ModeOverlap, bAnswers("", nil), unproven, GenFailed, false, false},
		{"overlap, root's start failed, kill proven", ModeOverlap, bAnswers("", unitFailed), nil, GenCancelled, true, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			b := newLaunchBackend(t, "nested", &fakeSwapBackend{fakeBackend: &fakeBackend{}})
			b.ctl.exeAnswer, b.ctl.stopGenErr = tc.answer, tc.stopErr
			dir := t.TempDir()
			next := markerScript(t, dir, "v2", filepath.Join(t.TempDir(), "v2.ran"))
			good := markerScript(t, dir, "v1", filepath.Join(t.TempDir(), "v1.ran"))
			plan := WakePlan{Boot: next, Mode: tc.mode, FirstWake: true, Token: 7}
			if tc.mode == ModeOverlap {
				plan.Fallback = &good
			}
			a, evs := launchActivator(t, b, plan)
			done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
			if tc.goodAsked {
				if err != nil {
					t.Fatalf("wake = %v, want the good build serving", err)
				}
				done()
			} else if err == nil {
				done()
				t.Fatal("the wake served, want it failed with nothing serving")
			}
			ev := evs.wait(t, tc.want)
			if ev.Token != 7 {
				t.Fatalf("%s event = %+v, want the plan's token", tc.want, ev)
			}
			switch tc.want {
			case GenFailed:
				if ev.Failed.Version != "v2" {
					t.Fatalf("failure event = %+v, want v2 failed", ev)
				}
			case GenFellBack:
				if ev.Failed.Version != "v2" || ev.Gen.Build != "v1" {
					t.Fatalf("fallback event = %+v, want v2 failed and v1 serving", ev)
				}
			case GenCancelled:
				if ev.Executed != tc.executed {
					t.Fatalf("cancellation = %+v, want Executed %v", ev, tc.executed)
				}
			}
			time.Sleep(50 * time.Millisecond)
			if k := evs.kinds(); len(k) != 1 {
				t.Fatalf("events %v, want exactly one terminal event for the token", k)
			}
			b.ctl.mu.Lock()
			var asked []string
			for _, x := range b.ctl.got {
				asked = append(asked, x.Version)
			}
			b.ctl.mu.Unlock()
			if want := map[bool]int{true: 2, false: 1}[tc.goodAsked]; len(asked) != want || asked[0] != "v2" {
				t.Fatalf("root was asked to start %v, want v2 first and %d start(s) in all", asked, want)
			}
		})
	}
}