package activator

import (
	"context"
	"errors"
	"fmt"
	"os"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

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

// A hold is stop-then-start through the barrier (spec §7.8). A request that
// arrives during the stop is held, not refused, and is served by the new
// generation, which execs the attempt's staged build while the live link
// names something else. This is the witness for mutation M3.
func TestSwapWitnessHoldExecsTheStagedBuild(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	f.b.execVerFn = func() string { return "v3" }
	gate := make(chan struct{})
	f.b.lastProc().setStopGate(gate)
	results := make(chan SwapResult, 1)
	go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) }()
	waitFor(t, "the hold's stop", func() bool {
		f.a.mu.Lock()
		defer f.a.mu.Unlock()
		e := f.a.entries[f.host]
		return e != nil && e.state == actStopping
	})
	stopSeen := time.Now() // the old generation's stop began before this
	served := make(chan error, 1)
	waiting := &doneAskedCtx{Context: context.Background(), asked: make(chan struct{})}
	go func() {
		release, err := f.a.beginRequest(waiting, testRoute(f.host))
		if err == nil {
			release()
		}
		served <- err
	}()
	// The request has reached its wait on the stopping entry: begin asks
	// its context's Done only there (the first select it reaches).
	<-waiting.asked
	select {
	case err := <-served:
		t.Fatalf("the request was answered (%v) while the old generation was still stopping", err)
	default:
	}
	gateOpen := time.Now()
	close(gate)
	var res SwapResult
	select {
	case res = <-results:
	case <-time.After(10 * time.Second):
		t.Fatal("the hold never finished")
	}
	if res.Outcome != SwapDone || res.Gen.Build != "v2" || res.Gen.StagedPath != f.staged {
		t.Fatalf("Swap = %+v", res)
	}
	if held := gateOpen.Sub(stopSeen); res.Drain < held {
		t.Fatalf("Drain = %v, want at least the old generation's gated stop, %v (spec §9.4)", res.Drain, held)
	}
	if err := <-served; err != nil {
		t.Fatalf("the held request was refused: %v", err)
	}
	// #231: one live-link Start (the original wake). The hold's wake went
	// through StartGen alone.
	if got := f.b.startCount(); got != 1 {
		t.Fatalf("Start (live link) calls = %d, want 1", got)
	}
	if gens, _, _ := f.b.calls(); gens != 1 {
		t.Fatalf("StartGen calls = %d, want 1", gens)
	}
	if got := f.appSocket(t); got != res.Gen.Socket {
		t.Fatalf("AppSocket = %q, want the new generation's %q", got, res.Gen.Socket)
	}
}

// doneAskedCtx is a context that signals the first time its Done channel
// is asked for: begin() asks only when it reaches a wait, so the signal is
// the request held on the entry.
type doneAskedCtx struct {
	context.Context
	once  sync.Once
	asked chan struct{}
}

func (c *doneAskedCtx) Done() <-chan struct{} {
	c.once.Do(func() { close(c.asked) })
	return c.Context.Done()
}

// A hold canary is not done when it is healthy: it must survive the 10 s
// settle (spec §6.2, §8.5). Without the settle, swapHold answers SwapDone
// the moment the pinned wake is ready, and a crash a second later neither
// fails the attempt nor halts the rollout.
func TestSwapHoldCanarySettles(t *testing.T) {
	old := swapSettle
	t.Cleanup(func() { swapSettle = old })
	start := func(t *testing.T, f *swapFixture) (chan SwapResult, Gen, *fakeProc) {
		t.Helper()
		at := f.attempt(ModeHold)
		at.Canary = true
		results := make(chan SwapResult, 1)
		go func() { results <- f.a.Swap(context.Background(), at, f.newest) }()
		var g Gen
		var p *fakeProc
		waitFor(t, "the pinned generation awake", func() bool {
			g, p = f.b.lastGen()
			f.a.mu.Lock()
			defer f.a.mu.Unlock()
			e := f.a.entries[f.host]
			return p != nil && e != nil && e.state == actAwake && e.gen.N == g.N
		})
		return results, g, p
	}
	t.Run("an exit in the window is down, with no fallback", func(t *testing.T) {
		swapSettle = 5 * time.Second
		f := newSwapFixture(t)
		results, _, p := start(t, f)
		select {
		case res := <-results:
			t.Fatalf("Swap answered %+v before the settle ended", res)
		case <-time.After(50 * time.Millisecond):
		}
		p.exit()
		res := <-results
		if res.Outcome != SwapDown || !strings.Contains(res.Reason, errSettleExit.Error()) {
			t.Fatalf("Swap = %+v, want down: %v", res, errSettleExit)
		}
		if gens, _, _ := f.b.calls(); gens != 1 || f.b.startCount() != 1 {
			t.Fatalf("a failed hold canary fell back: StartGen %d, Start %d", gens, f.b.startCount())
		}
	})
	t.Run("surviving the window is done", func(t *testing.T) {
		swapSettle = 100 * time.Millisecond
		f := newSwapFixture(t)
		began := time.Now()
		results, g, _ := start(t, f)
		res := <-results
		if res.Outcome != SwapDone || res.Gen.N != g.N {
			t.Fatalf("Swap = %+v", res)
		}
		if took := time.Since(began); took < swapSettle {
			t.Fatalf("done after %s, inside the %s settle", took, swapSettle)
		}
	})
	t.Run("an intentional stop in the window is not a failure", func(t *testing.T) {
		swapSettle = 5 * time.Second
		f := newSwapFixture(t)
		results, _, _ := start(t, f)
		f.a.Bounce(f.host)
		res := <-results
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapStopped.Error()) {
			t.Fatalf("Swap = %+v, want abandoned: %v", res, errSwapStopped)
		}
	})
}

// A hold build that fails its health check leaves nothing serving and does
// not fall back (spec §8.3).
func TestSwapHoldFailureIsDown(t *testing.T) {
	f := newSwapFixture(t)
	f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
	res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
	if res.Outcome != SwapDown {
		t.Fatalf("Swap = %+v, want down", res)
	}
	if f.a.entryCount() != 0 {
		t.Fatal("a failed hold left an entry behind")
	}
	if _, p := f.b.lastGen(); p == nil || !p.isStopped() {
		t.Fatal("the failed hold generation was left running")
	}
	if got := f.b.startCount(); got != 1 {
		t.Fatalf("Start (live link) calls = %d: a failed hold fell back", got)
	}
}

// A pinned wake that fails before it reaches the exec ran nothing of the
// build: the hold is abandoned, never down (index P5/P6, spec §6.4: only a
// hold build that executed is held down). Reported down, plan 5 records a
// failure of a build that never ran and halts its cohort.
func TestSwapHoldRefusedBeforeTheExecIsAbandoned(t *testing.T) {
	t.Parallel()
	for _, tc := range []struct {
		name    string
		arrange func(f *swapFixture)
		want    string
	}{
		{"the lease is held elsewhere after the stop", func(f *swapFixture) {
			f.b.mu.Lock()
			f.b.acquireErr = &errLeaseHeld{holder: "box-b", retryAfter: time.Minute}
			f.b.mu.Unlock()
		}, "instance lease held"},
		{"the restore fails", func(f *swapFixture) {
			f.b.mu.Lock()
			f.b.restoreErr = errors.New("replica unreadable")
			f.b.mu.Unlock()
		}, "replica unreadable"},
		{"the staged build changed after the stop", func(f *swapFixture) {
			f.b.mu.Lock()
			f.b.restoreErr = nil
			f.b.mu.Unlock()
			f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() {
				if err := os.WriteFile(f.staged, []byte("not build v2"), 0o755); err != nil {
					panic(err)
				}
			}}
		}, "staging"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSwapFixture(t)
			tc.arrange(f)
			res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want) {
				t.Fatalf("Swap = %+v, want abandoned: %s", res, tc.want)
			}
			if gens, _, _ := f.b.calls(); gens != 0 {
				t.Fatalf("StartGen calls = %d, want 0", gens)
			}
		})
	}
}

// lockedSink is an EventSink safe for the wake goroutine and the test.
type lockedSink struct {
	mu     sync.Mutex
	events []eventRec
}

func (s *lockedSink) App(string, string, string, string) {}
func (s *lockedSink) Event(acct, app, host, level, msg string) {
	s.mu.Lock()
	s.events = append(s.events, eventRec{acct, app, host, level, msg})
	s.mu.Unlock()
}

func (s *lockedSink) errors() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	var out []string
	for _, e := range s.events {
		if e.level == "error" {
			out = append(out, e.msg)
		}
	}
	return out
}

// A pinned wake's refusal is the attempt's business, not a member's: an
// abandoned hold emits no "wake failed" event, and no swap text reaches the
// member's timeline (plan 5's EvSwapDown carries a failed hold's line). A
// pinned build that ran and failed is reported in the wording any failed
// wake already uses.
func TestSwapHoldWakeEventsCarryNoSwapText(t *testing.T) {
	t.Parallel()
	t.Run("superseded before the exec emits nothing", func(t *testing.T) {
		f := newSwapFixture(t)
		sink := &lockedSink{}
		f.a.Logs = sink
		var once sync.Once
		f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }}
		if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapAbandoned {
			t.Fatalf("Swap = %+v, want abandoned", res)
		}
		if got := sink.errors(); len(got) != 0 {
			t.Fatalf("member error events %q, want none for an abandoned hold", got)
		}
	})
	t.Run("a mismatched spawn's text stays in the log", func(t *testing.T) {
		f := newSwapFixture(t)
		sink := &lockedSink{}
		f.a.Logs = sink
		f.a.backend = &mismatchSwapBackend{fakeSwapBackend: f.b}
		if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
			t.Fatalf("Swap = %+v, want down: the mismatched process ran", res)
		}
		for _, msg := range sink.errors() {
			if strings.Contains(msg, "the hold chose") || strings.Contains(msg, "swap:") {
				t.Fatalf("member event %q carries swap text", msg)
			}
		}
	})
	t.Run("a pinned build that failed its health uses the existing wording", func(t *testing.T) {
		f := newSwapFixture(t)
		sink := &lockedSink{}
		f.a.Logs = sink
		f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
		if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
			t.Fatalf("Swap = %+v, want down", res)
		}
		got := sink.errors()
		if len(got) != 1 || !strings.HasPrefix(got[0], "wake failed: ") || strings.Contains(got[0], "swap:") {
			t.Fatalf("member error events %q, want one existing \"wake failed: …\" line", got)
		}
	})
}

// mismatchSwapBackend runs a different build from the one it was asked for.
type mismatchSwapBackend struct {
	*fakeSwapBackend
}

func (b *mismatchSwapBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) {
	ran, p, err := b.fakeSwapBackend.StartStaged(ctx, rt, g)
	ran.Build = "v9"
	return ran, p, err
}

// acquireHookBackend runs hook inside Acquire: the window between a hold's
// re-check at its stop and its pinned wake's exec.
type acquireHookBackend struct {
	*fakeSwapBackend
	hook func()
}

func (b *acquireHookBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
	b.hook()
	return b.fakeSwapBackend.Acquire(ctx, rt)
}

// Spec §6.3, §7.4, §8.1: a hold's attempt checks run again at the pinned
// wake's exec and once more before its result counts. Checked only when
// the stop finished, a newer attempt (or a halted cohort, which bumps the
// revision) landing during the wake still exec'd the superseded build and
// answered SwapDone, which records it good.
func TestSwapHoldReChecksAtTheExecAndBeforeTheResult(t *testing.T) {
	t.Run("superseded before the exec", func(t *testing.T) {
		f := newSwapFixture(t)
		var once sync.Once
		f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }}
		res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
			t.Fatalf("Swap = %+v, want abandoned: superseded", res)
		}
		if gens, _, _ := f.b.calls(); gens != 0 {
			t.Fatalf("the superseded hold build was exec'd (%d)", gens)
		}
	})
	t.Run("superseded before the result", func(t *testing.T) {
		f := newSwapFixture(t)
		f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil }
		res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
			t.Fatalf("Swap = %+v, want abandoned: a superseded result is never done", res)
		}
	})
}

// killingSwapBackend is fakeSwapBackend plus plan 2's KillGen. fail decides,
// per generation, whether the child is still occupied.
type killingSwapBackend struct {
	*fakeSwapBackend
	kmu  sync.Mutex
	fail func(n uint64) bool
}

func (k *killingSwapBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error {
	k.kmu.Lock()
	defer k.kmu.Unlock()
	if k.fail != nil && k.fail(n) {
		return errors.New("the child still holds a process")
	}
	return nil
}

func (k *killingSwapBackend) setFail(f func(uint64) bool) {
	k.kmu.Lock()
	k.fail = f
	k.kmu.Unlock()
}

// A failed wake gives its lease back only once its generation's child is
// proven empty (spec §7.4). Until then the host takes no new wake, and the
// sweep retries the proof. cleanupFailedWake that accepts the leader's exit
// releases the lease over a surviving descendant and lets a new wake start
// beside it.
func TestSwapHoldFailedWakeKeepsTheLeaseUntilItsChildIsEmpty(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	kb := &killingSwapBackend{fakeSwapBackend: f.b}
	kb.setFail(func(n uint64) bool { return n != f.incumbent.N }) // the hold's new generation only
	f.a.backend = kb
	f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
	before := f.b.releaseCount()
	if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
		t.Fatalf("Swap = %+v, want down", res)
	}
	if got := f.b.releaseCount(); got != before+1 {
		t.Fatalf("lease releases %d -> %d: want the old generation's only, the failed wake's kept", before, got)
	}
	if _, err := f.a.beginRequest(context.Background(), testRoute(f.host)); err == nil || !strings.Contains(err.Error(), errWakeUnproven.Error()) {
		t.Fatalf("a new wake = %v, want refused while the failed generation is unproven", err)
	}
	kb.setFail(nil)
	f.a.sweepOnce()
	waitFor(t, "the failed wake's lease given back", func() bool { return f.b.releaseCount() == before+2 })
}

// The same when the failed wake's leader itself never stops: an unproven
// stop (ErrStillRunning) is no termination, so the lease is kept until the
// sweep's retried stop and proof hold. Retaining only when the leader had
// exited released the lease over a live process.
func TestSwapHoldFailedWakeWhoseLeaderNeverStopsKeepsTheLease(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	f.b.gmu.Lock()
	f.b.genStuck = true
	f.b.gmu.Unlock()
	f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") }
	before := f.b.releaseCount()
	if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown {
		t.Fatalf("Swap = %+v, want down", res)
	}
	if got := f.b.releaseCount(); got != before+1 {
		t.Fatalf("lease releases %d -> %d: the failed wake's lease was given back over a live leader", before, got)
	}
	_, p := f.b.lastGen()
	p.exit()
	f.a.sweepOnce()
	waitFor(t, "the failed wake's lease given back", func() bool { return f.b.releaseCount() == before+2 })
}

// The re-check before the spawn happens when the stop finishes. A fence, a
// newer attempt or the agent's shutdown abandons the hold, and nothing is
// spawned.
func TestSwapHoldRefusedAtTheStop(t *testing.T) {
	t.Parallel()
	for _, tc := range []struct {
		name string
		act  func(t *testing.T, f *swapFixture)
		want error
	}{
		{"fenced", func(_ *testing.T, f *swapFixture) {
			f.a.mu.Lock()
			if f.a.fences == nil {
				f.a.fences = map[string]uint64{}
			}
			f.a.fences[f.host] = 5
			f.a.mu.Unlock()
		}, errSwapFenced},
		{"superseded", func(_ *testing.T, f *swapFixture) { f.rev.Add(1) }, errSwapSuperseded},
		{"shutdown", func(t *testing.T, f *swapFixture) {
			go f.a.Shutdown(context.Background())
			// The shutdown is announced before the old generation's stop is
			// let go, so the outcome does not hang on goroutine scheduling.
			waitFor(t, "the shutdown announced", func() bool {
				f.a.mu.Lock()
				defer f.a.mu.Unlock()
				return f.a.shuttingDown
			})
		}, errSwapShutdown},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSwapFixture(t)
			gate := make(chan struct{})
			f.b.lastProc().setStopGate(gate)
			results := make(chan SwapResult, 1)
			go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) }()
			waitFor(t, "the hold's stop", func() bool {
				f.a.mu.Lock()
				defer f.a.mu.Unlock()
				e := f.a.entries[f.host]
				return e != nil && e.state == actStopping
			})
			tc.act(t, f)
			close(gate)
			res := <-results
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
				t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
			}
			if gens, _, _ := f.b.calls(); gens != 0 {
				t.Fatal("a refused hold spawned a generation")
			}
		})
	}
}

// Claims on: a hold needs a claim runtime that can start a pinned
// generation of its own. Without one it is refused BEFORE anything stops.
func TestSwapHoldClaimsOnNeedsPins(t *testing.T) {
	f := newSwapFixture(t)
	f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}}
	res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
	if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errClaimsNoGenerations.Error()) {
		t.Fatalf("Swap = %+v", res)
	}
	if f.b.lastProc().isStopped() {
		t.Fatal("a refused hold stopped the incumbent")
	}
}

// A row that cannot run a staged generation is refused BEFORE anything
// stops (spec §7.8 is stop-then-start: a start that cannot happen must not
// cost the instance its running process). Refused after the stop, the
// pinned wake failed and the hold answered SwapDown for a build that never
// ran. Swaps off refuses a hold as it refuses an overlap.
func TestSwapHoldRefusesARowThatCannotRunAStagedBuild(t *testing.T) {
	t.Parallel()
	for _, tc := range []struct {
		name    string
		arrange func(f *swapFixture)
		want    error
	}{
		{"swap off", func(f *swapFixture) { f.a.gens = false }, errSwapOff},
		{"unit-backed", func(f *swapFixture) {
			f.a.mu.Lock()
			f.a.entries[f.host].route.Backing = registry.BackingUnit
			f.a.mu.Unlock()
		}, errSwapKind},
		{"fenced layout", func(f *swapFixture) {
			f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFenced, "a migration journal" }
		}, errSwapLayout},
		{"worker", func(f *swapFixture) {
			f.a.mu.Lock()
			f.a.entries[f.host].worker = true
			f.a.mu.Unlock()
		}, errSwapKind},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSwapFixture(t)
			tc.arrange(f)
			res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
				t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
			}
			if f.b.lastProc().isStopped() {
				t.Fatal("a refused hold stopped the incumbent")
			}
			if gens, _, _ := f.b.calls(); gens != 0 || f.b.startCount() != 1 {
				t.Fatalf("a refused hold spawned: StartGen %d, Start %d", gens, f.b.startCount())
			}
			f.a.mu.Lock()
			e := f.a.entries[f.host]
			held := e != nil && e.holdNext != nil
			f.a.mu.Unlock()
			if held {
				t.Fatal("a refused hold left its spec on the entry")
			}
		})
	}
}

// The caller is released when the old generation's stop outlasts
// wakeWaitBound, the bound §7.8 holds requests for, or when the caller's own
// context ends. The hold is withdrawn under the entry lock, so the stop's
// eventual finish installs no pinned wake and the next request wakes the
// ordinary way. Waiting on the stop alone held the caller for as long as an
// unproven stop stayed pending.
func TestSwapHoldReleasesItsCallerWhileTheStopIsPending(t *testing.T) {
	for _, tc := range []struct {
		name string
		run  func(f *swapFixture) SwapResult
		want error
	}{
		{"the stop outlasts wakeWaitBound", func(f *swapFixture) SwapResult {
			return f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
		}, errHoldStopSlow},
		{"the caller's context ends", func(f *swapFixture) SwapResult {
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			go func() {
				// Cancel once the hold's stop has begun (bounded: the test's
				// own assertion reports a hold that never got there).
				for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) {
					f.a.mu.Lock()
					e := f.a.entries[f.host]
					stopping := e != nil && e.state == actStopping
					f.a.mu.Unlock()
					if stopping {
						break
					}
				}
				cancel()
			}()
			return f.a.Swap(ctx, f.attempt(ModeHold), f.newest)
		}, errSwapStopped},
	} {
		t.Run(tc.name, func(t *testing.T) {
			old := wakeWaitBound
			t.Cleanup(func() { wakeWaitBound = old })
			f := newSwapFixture(t)
			if errors.Is(tc.want, errHoldStopSlow) {
				wakeWaitBound = 100 * time.Millisecond
			}
			gate := make(chan struct{})
			incumbent := f.b.lastProc()
			incumbent.setStopGate(gate)
			res := tc.run(f)
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, tc.want.Error()) {
				close(gate)
				t.Fatalf("Swap = %+v, want abandoned: %v", res, tc.want)
			}
			f.a.mu.Lock()
			prev := f.a.entries[f.host]
			if prev == nil || prev.state != actStopping || prev.holdNext != nil {
				f.a.mu.Unlock()
				close(gate)
				t.Fatal("want the old entry still stopping, with its hold withdrawn")
			}
			stopped := prev.stopped
			f.a.mu.Unlock()
			close(gate)
			<-stopped
			// finishStop would install the pinned wake in the critical section
			// that closes stopped, so its absence now is final.
			f.a.mu.Lock()
			next := f.a.entries[f.host]
			f.a.mu.Unlock()
			if next != nil {
				t.Fatalf("a withdrawn hold installed an entry (state %v)", next.state)
			}
			if gens, _, _ := f.b.calls(); gens != 0 {
				t.Fatalf("a withdrawn hold exec'd its build (%d)", gens)
			}
		})
	}
}

// A claims-on pinned wake runs the hold's attempt gate before anything is
// admitted: a superseded hold reaches neither the prerequisites nor the
// admission.
func TestDoClaimWakeRunsThePinGateFirst(t *testing.T) {
	obs := hostclaim.Observed{Host: "x.example", ETag: "one", Claim: &hostclaim.Claim{
		Holder: &hostclaim.Holder{Location: "instances/x.example/inc", Session: 1, Generation: "one", WriterBox: "box"},
		Op:     hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"},
	}}
	admitted := false
	a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Pins: true,
		Read:            func(context.Context, string) (hostclaim.Observed, error) { return obs, 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 {
			admitted = true
			return nil
		},
		Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
			admitted = true
			return obs, nil
		},
		Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
			admitted = true
			return ClaimStart{}, nil
		}}}}
	e := &actEntry{host: obs.Host, route: registry.Route{Host: obs.Host, Kind: "instance"}, pin: &Gen{Build: "v2"},
		pinCheck: func() error { return errSwapSuperseded }}
	if _, err := a.doClaimWake(context.Background(), e); !errors.Is(err, errSwapSuperseded) || admitted {
		t.Fatalf("doClaimWake = %v (admitted %v), want the pin gate's refusal before admission", err, admitted)
	}
}

// A claim runtime that prepared a different build from the pinned one is
// refused before Restore touches the database.
func TestStartAdmittedRefusesAPinMismatch(t *testing.T) {
	obs := hostclaim.Observed{Host: "x.example", ETag: "one", Claim: &hostclaim.Claim{
		Holder: &hostclaim.Holder{Location: "instances/x.example/inc", Session: 1, Generation: "one"},
		Op:     hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"},
	}}
	work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn")
	stop := errors.New("stop at restore")
	for _, tc := range []struct {
		version  string
		restored bool
	}{{"v1", false}, {"v2", true}} {
		restored := false
		a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
			Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil },
			Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
				s := claimedStartFor(obs, work, tc.version)
				s.Restore = func(context.Context) error { restored = true; return stop }
				return s, nil
			}}}}
		policy := startPolicy{MaxStarts: 1, Window: time.Second, Pin: &Gen{Build: "v2"}}
		err := a.startAdmitted(context.Background(), &actEntry{host: obs.Host}, obs, policy)
		if restored != tc.restored {
			t.Fatalf("prepared %s: restored = %v, want %v (err %v)", tc.version, restored, tc.restored, err)
		}
		if !tc.restored && !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("prepared %s: %v, want hostwrite.ErrUnavailable", tc.version, err)
		}
		if tc.restored && !errors.Is(err, stop) {
			t.Fatalf("prepared %s: %v, want the restore's own error", tc.version, err)
		}
	}
}

// A foreign row keeps its canonical socket, so an overlap attempt on it
// runs as a hold (plan 1's reading): the old process stops, then the staged
// build starts on the same socket. Refusing it (errSwapLayout) left a
// foreign row unable to take any release with the flag on.
func TestSwapOverlapOnAForeignRowRunsAsAHold(t *testing.T) {
	f := newSwapFixture(t)
	f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutForeign, "hand-set socket" }
	incumbent := f.b.lastProc()
	res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
	if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged || res.Mode != ModeHold {
		t.Fatalf("Swap = %+v, want done through the hold, reported as a hold (Mode)", res)
	}
	ev := f.b.events()
	if i, j := slices.Index(ev, "stop:instance"), indexPrefix(ev, "startGen:"+f.staged); i < 0 || j < 0 || i > j {
		t.Fatalf("order %v: a foreign row stops before its replacement starts", ev)
	}
	if indexPrefix(ev, "raise") >= 0 {
		t.Fatalf("events %v: a foreign row never raises an overlap parent", ev)
	}
	if !incumbent.isStopped() {
		t.Fatal("the old process outlived the hold")
	}
}

// Task 5, plan review #2: the scheduler writes an attempt's intent in the
// mode Swap will run it in, which EffectiveMode answers with Swap's own
// override: an overlap on a foreign row is a hold, anything else is as given.
func TestEffectiveModeIsTheModeSwapRuns(t *testing.T) {
	f := newSwapFixture(t)
	host := f.attempt(ModeOverlap).Host
	if m := f.a.EffectiveMode(host, ModeOverlap); m != ModeOverlap {
		t.Fatalf("flat row: %s, want overlap", m)
	}
	f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutForeign, "hand-set socket" }
	if m := f.a.EffectiveMode(host, ModeOverlap); m != ModeHold {
		t.Fatalf("foreign row: %s, want hold", m)
	}
	if m := f.a.EffectiveMode(host, ModeHold); m != ModeHold {
		t.Fatalf("foreign row, hold: %s", m)
	}
}

// A nested row can hold-swap now that root mounts the staged build (Task 9):
// the refusal Task 6 made for it is gone, and the hold runs to SwapDone.
func TestSwapHoldAcceptsANestedRow(t *testing.T) {
	f := newSwapFixture(t)
	f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
	res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
	if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged {
		t.Fatalf("Swap = %+v, want a done hold on the staged build", res)
	}
}

// zdd-readiness Task 13 (Astra R1 T12, R3 §5 item 9): a hold stops its
// incumbent before its own build execs, so the build has to be in the
// generations file before that stop begins. Published only at the pinned
// wake's exec, root could sweep or restage the staged build in the window
// where nothing serves at all.

// The generations file names the hold's build when the incumbent's stop
// begins. The fake records what the file said at that moment, before the
// stop's own gate.
func TestSwapHoldPublishesItsBuildBeforeTheIncumbentStops(t *testing.T) {
	f := newSwapFixture(t)
	f.a.GenerationsFile = release.GenerationsPath(t.TempDir())
	incumbent := f.b.lastProc()
	incumbent.mu.Lock()
	incumbent.onStopEnter = func() {
		f.b.record(fmt.Sprintf("stop-begins:file=%s", strings.Join(hostsOf(f.a.GenerationsFile)[f.host], ",")))
	}
	incumbent.mu.Unlock()
	res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
	if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged {
		t.Fatalf("Swap = %+v, want a done hold on the staged build", res)
	}
	evs := f.b.events()
	began := slices.IndexFunc(evs, func(ev string) bool { return strings.HasPrefix(ev, "stop-begins:") })
	if began < 0 {
		t.Fatalf("events %q: the incumbent's stop never began", evs)
	}
	if evs[began] != "stop-begins:file=v1,v2" {
		t.Fatalf("when the incumbent's stop began the file named %q, want the hold's build v2 beside v1", evs[began])
	}
	if i := slices.Index(evs, "startGen:"+f.staged); i < began {
		t.Fatalf("events %q: the hold's exec at %d, before its incumbent's stop at %d", evs, i, began)
	}
	// The pending registration went with the swap: the file names what runs.
	waitForGenerations(t, f.a.GenerationsFile, map[string][]string{f.host: {"v2"}})
	f.a.mu.Lock()
	pending := len(f.a.genPending)
	f.a.mu.Unlock()
	if pending != 0 {
		t.Fatalf("%d pending generations left after the hold finished", pending)
	}
}

// A hold whose build cannot be published stops nothing: the incumbent keeps
// serving and the attempt is abandoned with nothing executed.
func TestSwapHoldUnpublishedLeavesTheIncumbentServing(t *testing.T) {
	f := newSwapFixture(t)
	f.a.GenerationsFile = unwritableGenerationsFile(t)
	incumbent := f.b.lastProc()
	res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
	if res.Outcome != SwapAbandoned || res.Executed || !strings.Contains(res.Reason, errStagedBuildNotPublished.Error()) {
		t.Fatalf("Swap = %+v, want abandoned, not executed: unpublished", res)
	}
	if n := incumbent.stopEntered(); n != 0 || incumbent.isStopped() {
		t.Fatalf("the incumbent's stop was entered %d times for a hold whose build was never published", n)
	}
	f.a.mu.Lock()
	e := f.a.entries[f.host]
	awake := e != nil && e.state == actAwake && e.gen.N == f.incumbent.N && e.holdNext == nil
	pending := len(f.a.genPending)
	f.a.mu.Unlock()
	if !awake {
		t.Fatal("want the incumbent's entry still awake on its generation, with no hold on it")
	}
	if pending != 0 {
		t.Fatalf("%d pending generations left by an abandoned hold", pending)
	}
	if gens, _, _ := f.b.calls(); gens != 0 {
		t.Fatalf("StartGen calls = %d, want 0", gens)
	}
}

// The publication runs off a.mu, so the admission is asked again once it is
// written. An attempt superseded while the file was written withdraws its
// pending build and stops nothing.
func TestSwapHoldReadmitsAfterThePublication(t *testing.T) {
	f := newSwapFixture(t)
	f.a.GenerationsFile = release.GenerationsPath(t.TempDir())
	incumbent := f.b.lastProc()
	at := f.attempt(ModeHold)
	// newest is asked under a.mu at each admission. The newer attempt
	// arrives exactly when the hold's build is in the file: after the
	// first admission, before the second.
	newest := func(host string) uint64 {
		if slices.Contains(hostsOf(f.a.GenerationsFile)[host], "v2") {
			return at.SchedRev + 1
		}
		return at.SchedRev
	}
	res := f.a.Swap(context.Background(), at, newest)
	if res.Outcome != SwapAbandoned || res.Executed || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
		t.Fatalf("Swap = %+v, want abandoned, not executed: superseded", res)
	}
	if n := incumbent.stopEntered(); n != 0 {
		t.Fatalf("the incumbent's stop was entered %d times for a superseded hold", n)
	}
	f.a.mu.Lock()
	e := f.a.entries[f.host]
	awake := e != nil && e.state == actAwake && e.holdNext == nil
	pending := len(f.a.genPending)
	f.a.mu.Unlock()
	if !awake || pending != 0 {
		t.Fatalf("awake with no hold = %v, pending generations = %d", awake, pending)
	}
	// The withdrawn build is no longer published.
	waitForGenerations(t, f.a.GenerationsFile, map[string][]string{f.host: {"v1"}})
	if gens, _, _ := f.b.calls(); gens != 0 {
		t.Fatalf("StartGen calls = %d, want 0", gens)
	}
}