package activator

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"math/rand"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/registry"
)

// TestSwapInvariants drives random schedules of requests, overlap and hold
// swaps, hibernation, lease loss, restarts, maintenance fences, root
// repoints, supersession, crashes, re-stop cadences and a final Shutdown
// against the real Service and a simulated backend. It checks I1, I2 and I3
// (plan 3 task 13; spec §12).
//
// The simulated processes do not all stop when asked. Some drain slowly,
// some answer ErrStillRunning or a failed kill and exit on their own later,
// some generations fork a process that outlives its leader, KillGen can
// fail before it heals, and a start can fail with its kill unproven
// (errGenUnproven), with or without a process handle. So the
// retain-until-proven paths (pending stops, the member and wake reapers)
// run under the sweep too. Each seed also picks whether generations are on
// (the dormant switch) and whether generation containment is.
//
// Seeds: SWAP_SIM_SEEDS (default 100). SWAP_SIM_SEED=<n> replays one seed.
// Goroutine interleaving is not deterministic, so a replay is best effort.
// The event log printed with a failure is the evidence.
//
// The mutation checks (testdata/swapsim/mutate.sh) run this sweep and the
// TestSwapWitness* tests against each `// swapsim:Mn` mutant.
func TestSwapInvariants(t *testing.T) {
	first, last := 0, 100
	if v := os.Getenv("SWAP_SIM_SEEDS"); v != "" {
		n, err := strconv.Atoi(v)
		if err != nil {
			t.Fatal(err)
		}
		last = n
	}
	if v := os.Getenv("SWAP_SIM_SEED"); v != "" {
		n, err := strconv.Atoi(v)
		if err != nil {
			t.Fatal(err)
		}
		first, last = n, n+1
	}
	oldSettle, oldJoin := swapSettle, opJoinBound
	swapSettle, opJoinBound = 2*time.Millisecond, 3*time.Second
	t.Cleanup(func() { swapSettle, opJoinBound = oldSettle, oldJoin })
	// The Service's own log lines go into the seed's event log, so a failure
	// prints them in order with the model's events, and a passing sweep
	// prints nothing.
	sink := &simLogSink{}
	oldOut, oldFlags := log.Writer(), log.Flags()
	log.SetOutput(sink)
	log.SetFlags(0)
	t.Cleanup(func() { log.SetOutput(oldOut); log.SetFlags(oldFlags) })
	staged := stageSimBuilds(t)
	sockDir := t.TempDir()
	failed := 0
	tally := map[string]int{}
	t.Cleanup(func() {
		keys := make([]string, 0, len(tally))
		for k := range tally {
			keys = append(keys, k)
		}
		sort.Strings(keys)
		lines := make([]string, len(keys))
		for i, k := range keys {
			lines[i] = fmt.Sprintf("%6d  %s", tally[k], k)
		}
		t.Logf("seeds %d..%d: seed settings and Swap results (mode/outcome: reason, normalized):\n%s", first, last-1, strings.Join(lines, "\n"))
	})
	for seed := first; seed < last; seed++ {
		violations, events, outcomes := simulateSwapOnce(seed, staged, sockDir, sink)
		for k, n := range outcomes {
			tally[k] += n
		}
		if len(violations) == 0 {
			continue
		}
		failed++
		if len(events) > 200 {
			events = events[len(events)-200:]
		}
		// A seed whose only failure is liveness (a Swap that never returned)
		// says so: it is not an invariant violation.
		kind := "liveness failed"
		for _, v := range violations {
			if !strings.HasPrefix(v, "liveness:") {
				kind = "invariant violated"
			}
		}
		t.Errorf("%s at seed %d (replay: SWAP_SIM_SEED=%d): %s\n--- last events ---\n%s", kind, seed, seed, strings.Join(violations, "; "), strings.Join(events, "\n"))
		if failed >= 3 {
			t.FailNow()
		}
	}
}

type simArtifact struct{ path, digest string }

func stageSimBuilds(t *testing.T) map[string]simArtifact {
	t.Helper()
	out := map[string]simArtifact{}
	for _, b := range []string{"v3", "v4"} {
		dir := filepath.Join(t.TempDir(), "versions", b)
		if err := os.MkdirAll(dir, 0o755); err != nil {
			t.Fatal(err)
		}
		p := filepath.Join(dir, "app")
		body := []byte("build " + b)
		if err := os.WriteFile(p, body, 0o755); err != nil {
			t.Fatal(err)
		}
		sum := sha256.Sum256(body)
		out[b] = simArtifact{p, hex.EncodeToString(sum[:])}
	}
	return out
}

// simLogSink routes the log package into the current seed's world. A
// goroutine a previous seed left behind may log into the next seed's
// events; the lines say what they are.
type simLogSink struct{ w atomic.Pointer[simWorld] }

func (s *simLogSink) Write(p []byte) (int, error) {
	if w := s.w.Load(); w != nil {
		w.logf("log: %s", strings.TrimRight(string(p), "\n"))
	}
	return len(p), nil
}

// simWorld is the model: which simulated processes run on each host, the
// live link, the staged builds, and the violations seen. Lock order: a
// caller may hold a.mu when it takes w.mu (the backend is called under the
// entry lock), never the reverse; a process's own p.mu nests inside w.mu
// (die), never the reverse.
type simWorld struct {
	mu         sync.Mutex
	rng        *rand.Rand
	events     []string
	running    map[string]map[*simProc]bool
	live       map[string]string
	staged     map[string]string
	violations []string
	seen       map[string]bool
	nextID     int
	hosts      []string
	// gensOn is this seed's dormant switch (Service.gens). contained is
	// generation containment: with it, each generation has its own child
	// that KillGen empties, and a start can fail with its kill unproven;
	// without it, KillGen answers errNoGenContainment, the parent cannot be
	// raised, and a generation's leader is all there is of it.
	gensOn, contained bool
	// killFails is how many more KillGen calls fail for host#n before the
	// kill heals.
	killFails map[string]int
	// shutdown is set once Service.Shutdown has returned: nothing may spawn
	// after it (I2).
	shutdown bool
	// outcomes counts the seed's settings and its Swap results by mode,
	// outcome and reason, so a sweep shows what it reached.
	outcomes map[string]int
	// dying, when set, runs once a death is published, with no model lock
	// held: the moment a reaper that saw Exited() could act. The model's
	// own interleaving test sets it; the sweep never does.
	dying func(p *simProc)
}

func (w *simWorld) logf(format string, args ...any) {
	w.mu.Lock()
	w.events = append(w.events, fmt.Sprintf(format, args...))
	w.mu.Unlock()
}

func (w *simWorld) violatef(format string, args ...any) {
	msg := fmt.Sprintf(format, args...)
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.seen[msg] {
		return
	}
	w.seen[msg] = true
	w.violations = append(w.violations, msg)
	w.events = append(w.events, "VIOLATION "+msg)
}

func (w *simWorld) intn(n int) int {
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.rng.Intn(n)
}

func (w *simWorld) jitter() {
	if d := w.intn(4); d > 0 {
		time.Sleep(time.Duration(d) * 250 * time.Microsecond)
	}
}

// spawn starts one simulated process of generation n. A leader gets a
// random stop behaviour; a descendant (a process the app forked) stops
// only when killed.
func (w *simWorld) spawn(host, build string, n uint64, leader bool) *simProc {
	w.mu.Lock()
	w.nextID++
	p := &simProc{w: w, id: w.nextID, host: host, build: build, n: n, leader: leader}
	if leader {
		switch r := w.rng.Intn(20); {
		case r < 2:
			p.mode = procStubborn
		case r < 3:
			// A failed kill that is not ErrStillRunning is a claims-on
			// process's answer (KillAndEmpty), and the generations-on stop
			// handles it through releasable. With generations off the exec
			// backends' Stop (childProc, unitProc) answers ErrStillRunning
			// for a process that may remain, so that is what it gets.
			p.mode = procStubborn
			if w.gensOn {
				p.mode = procKillFails
			}
		case r < 7:
			p.mode = procSlow
		}
	}
	if w.running[host] == nil {
		w.running[host] = map[*simProc]bool{}
	}
	w.running[host][p] = true
	what := "spawn"
	if !leader {
		what = "fork"
	}
	w.events = append(w.events, fmt.Sprintf("%s p%d %s build=%s gen=%d mode=%s", what, p.id, host, build, n, p.mode))
	after := w.shutdown && w.gensOn
	w.mu.Unlock()
	if after {
		w.violatef("I2: p%d (%s build %s) spawned after Shutdown returned", p.id, host, build)
	}
	return p
}

// spawnGen starts generation n's leader. With generations and containment
// on, one generation in four also forks a process that outlives its
// leader, which only KillGen (or a crash) ends.
func (w *simWorld) spawnGen(host, build string, n uint64) *simProc {
	p := w.spawn(host, build, n, true)
	if w.gensOn && w.contained && n != 0 && w.intn(4) == 0 {
		w.spawn(host, build, n, false)
	}
	return p
}

func (w *simWorld) runningOn(host string) int {
	w.mu.Lock()
	defer w.mu.Unlock()
	return len(w.running[host])
}

func (w *simWorld) runningCount() int {
	w.mu.Lock()
	defer w.mu.Unlock()
	n := 0
	for _, ps := range w.running {
		n += len(ps)
	}
	return n
}

func (w *simWorld) leaseFreed(host, by string) {
	if n := w.runningOn(host); n > 0 {
		w.violatef("I2: lease of %s released by %s while %d generation process(es) run", host, by, n)
	}
	w.logf("lease %s released by %s", host, by)
}

func (w *simWorld) setLive(host, build string) {
	w.mu.Lock()
	w.live[host] = build
	w.events = append(w.events, "repoint "+host+" -> "+build)
	w.mu.Unlock()
}

func (w *simWorld) crashOne(host string) {
	w.mu.Lock()
	var victims []*simProc
	for p := range w.running[host] {
		victims = append(victims, p)
	}
	// Map order is random; sort so the seed picks the victim.
	sort.Slice(victims, func(i, j int) bool { return victims[i].id < victims[j].id })
	var p *simProc
	if len(victims) > 0 {
		p = victims[w.rng.Intn(len(victims))]
	}
	w.mu.Unlock()
	if p != nil {
		p.die("crash")
	}
}

var simDigits = regexp.MustCompile(`[0-9]+`)

// tally counts one Swap result. The reason is normalized (host names and
// numbers out, length capped) so that like results group.
func (w *simWorld) tally(res SwapResult) {
	key := string(res.Mode) + "/" + string(res.Outcome)
	if res.Outcome != SwapDone && res.Reason != "" {
		r := res.Reason
		for _, h := range w.hosts {
			r = strings.ReplaceAll(r, h, "<host>")
		}
		r = simDigits.ReplaceAllString(r, "N")
		if len(r) > 110 {
			r = r[:110] + "..."
		}
		key += ": " + r
	}
	w.mu.Lock()
	w.outcomes[key]++
	w.mu.Unlock()
}

func (w *simWorld) snapshot() ([]string, []string, map[string]int) {
	w.mu.Lock()
	defer w.mu.Unlock()
	outcomes := make(map[string]int, len(w.outcomes))
	for k, n := range w.outcomes {
		outcomes[k] = n
	}
	return append([]string(nil), w.violations...), append([]string(nil), w.events...), outcomes
}

// simProcMode is how a simulated leader answers Stop.
type simProcMode int

const (
	// procPrompt ends at once.
	procPrompt simProcMode = iota
	// procSlow drains for a few milliseconds, then ends.
	procSlow
	// procStubborn answers ErrStillRunning: SIGKILL did not prove it gone.
	// It exits on its own a few milliseconds after it was first asked.
	procStubborn
	// procKillFails answers a kill error that is not ErrStillRunning (a
	// claims-on KillAndEmpty failure), and exits on its own later, as
	// procStubborn does. Generations-on seeds only (see spawn).
	procKillFails
)

func (m simProcMode) String() string {
	return [...]string{"prompt", "slow", "stubborn", "killfails"}[m]
}

var errSimKillFailed = errors.New("sim: kill failed")

// simProc is one simulated process of a generation: its leader, or a
// process the leader forked (leader false), which nothing but KillGen or a
// crash ends.
type simProc struct {
	w      *simWorld
	id     int
	host   string
	build  string
	n      uint64
	leader bool
	mode   simProcMode
	mu     sync.Mutex
	dead   bool
	later  bool
}

func (p *simProc) Stop(context.Context) error {
	p.w.jitter()
	if p.Exited() {
		return nil
	}
	switch p.mode {
	case procSlow:
		time.Sleep(time.Duration(1+p.w.intn(6)) * time.Millisecond)
	case procStubborn:
		p.exitLater()
		return ErrStillRunning
	case procKillFails:
		p.exitLater()
		return errSimKillFailed
	}
	p.die("stop")
	return nil
}

// exitLater arms the one delayed exit of a process that did not stop when
// asked.
func (p *simProc) exitLater() {
	p.mu.Lock()
	armed := p.later
	p.later = true
	p.mu.Unlock()
	if armed {
		return
	}
	d := time.Duration(1+p.w.intn(15)) * time.Millisecond
	time.AfterFunc(d, func() { p.die("late exit") })
}

func (p *simProc) Exited() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.dead
}

// die is one transition of the model: the process is marked dead and
// leaves running in the same critical section (w.mu, then p.mu, the
// model's lock order), so no observer can see it Exited() and still
// running. A concurrent die waits on w.mu and finds it already dead.
func (p *simProc) die(why string) {
	p.w.mu.Lock()
	p.mu.Lock()
	was := p.dead
	p.dead = true
	p.mu.Unlock()
	if !was {
		delete(p.w.running[p.host], p)
		p.w.events = append(p.w.events, fmt.Sprintf("exit p%d (%s)", p.id, why))
	}
	p.w.mu.Unlock()
	if !was && p.w.dying != nil {
		p.w.dying(p)
	}
}

type simRep struct {
	w    *simWorld
	host string
	mu   sync.Mutex
	err  error
}

func (r *simRep) Err() error {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.err
}

func (r *simRep) Close(context.Context) error {
	r.w.jitter()
	r.w.leaseFreed(r.host, "rep.Close")
	return nil
}

type simLease struct {
	b    *simBackend
	host string
}

func (l *simLease) Replicate(context.Context, string, time.Duration) (Rep, error) {
	r := &simRep{w: l.b.w, host: l.host}
	l.b.mu.Lock()
	l.b.reps[l.host] = r
	l.b.mu.Unlock()
	return r, nil
}

func (l *simLease) Release(context.Context) error {
	l.b.w.leaseFreed(l.host, "lease.Release")
	return nil
}

// simBackend is Backend, GenBackend, SwapBackend and genKiller, as the
// exec backend is. It has no hostKiller or sidecarKiller (no workers, no
// sidecars). The live-link wake goes through StartGen, so every generation
// has a number KillGen can name.
type simBackend struct {
	w    *simWorld
	dir  string
	mu   sync.Mutex
	reps map[string]*simRep
}

var (
	_ Backend     = (*simBackend)(nil)
	_ GenBackend  = (*simBackend)(nil)
	_ SwapBackend = (*simBackend)(nil)
	_ genKiller   = (*simBackend)(nil)
	_ DueStore    = simDue{}
)

func (b *simBackend) loseLease(host string) {
	b.mu.Lock()
	r := b.reps[host]
	b.mu.Unlock()
	if r != nil {
		r.mu.Lock()
		r.err = errors.New("sim: lease lost")
		r.mu.Unlock()
	}
}

func (b *simBackend) Acquire(_ context.Context, rt registry.Route) (Lease, error) {
	b.w.jitter()
	if n := b.w.runningOn(rt.Host); n > 0 {
		b.w.violatef("I2: lease of %s acquired while %d process(es) of the previous holder's generations run", rt.Host, n)
	}
	return &simLease{b: b, host: rt.Host}, nil
}
func (b *simBackend) Restore(context.Context, registry.Route) (bool, error) {
	b.w.jitter()
	return false, nil
}
func (b *simBackend) Seed(context.Context, registry.Route) error { return nil }

// Start is the pre-generations spawn. The Service never calls it on a
// GenBackend; it is here for the Backend interface.
func (b *simBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
	p, g, err := b.StartGen(ctx, rt, 0, false)
	return p, g.Build, err
}

// StartGen is the wake's spawn of the live link (which only ever names v1
// or v2).
func (b *simBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
	b.w.jitter()
	b.w.mu.Lock()
	build := b.w.live[rt.Host]
	b.w.mu.Unlock()
	g := canonicalGen(rt, n, build)
	if perGen {
		g.Socket = filepath.Join(b.dir, fmt.Sprintf("%s.%d.sock", rt.Host, n))
		g.Control = g.Socket + ".control"
	}
	if err := ctx.Err(); err != nil {
		return nil, g, err
	}
	p, err := b.startOutcome(rt.Host, build, n)
	return p, g, err
}

// startOutcome is a spawn's result. Most run. Some fail before the exec,
// with nothing running. With containment, some fail after it with their
// kill unproven (errGenUnproven, as plan 2's startUnitAt answers): the
// generation runs on, with its handle returned or not.
func (b *simBackend) startOutcome(host, build string, n uint64) (Proc, error) {
	w := b.w
	r := w.intn(30)
	switch {
	case r == 0:
		return nil, errors.New("sim: exec failed")
	case r <= 2 && w.contained:
		p := w.spawnGen(host, build, n)
		err := fmt.Errorf("sim: health wait failed; %w: kill unproven", errGenUnproven)
		w.logf("start of %s gen %d failed with its kill unproven (handle returned: %v)", host, n, r == 1)
		if r == 1 {
			return p, err
		}
		return nil, err
	}
	return w.spawnGen(host, build, n), nil
}

func (b *simBackend) NextDueVia(context.Context, registry.Route, string) (int64, error) {
	return 0, nil
}
func (b *simBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
func (b *simBackend) HasSidecar(registry.Route) bool                         { return false }
func (b *simBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
	return nil, errors.New("sim: no sidecars")
}
func (b *simBackend) StartWorker(context.Context, registry.Route, map[string]string) (Proc, string, error) {
	return nil, "", errors.New("sim: no workers")
}

// StartStaged checks I3 at the exec: the staged path a generation runs is
// the one its build was staged at. Its sockets live in the run's own
// directory, because a hold's pinned wake links the canonical path to the
// generation's socket for real (swapCanonicalLink).
func (b *simBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) {
	b.w.mu.Lock()
	want := b.w.staged[g.Build]
	b.w.mu.Unlock()
	if g.StagedPath != want {
		b.w.violatef("I3: generation %d of %s execs %q for build %s (staged at %q)", g.N, rt.Host, g.StagedPath, g.Build, want)
	}
	b.w.jitter()
	if err := ctx.Err(); err != nil {
		return g, nil, err
	}
	// The caller's exec gate at the launch point, after the preparation
	// the jitter stands for (final review F1).
	if err := runExecGate(ctx); err != nil {
		return g, nil, err
	}
	g.Socket = filepath.Join(b.dir, fmt.Sprintf("%s.%d.sock", rt.Host, g.N))
	g.Control = g.Socket + ".control"
	p, err := b.startOutcome(rt.Host, g.Build, g.N)
	return g, p, err
}

// KillGen is plan 2's generation kill: it empties generation n's child
// and returns once the child is empty. One generation in six fails its
// first one or two kills, then heals. A process that does not die when
// killed (procStubborn, procKillFails) keeps the child from emptying until
// it exits on its own. Without containment it answers errNoGenContainment,
// as ExecBackend.KillGen does.
func (b *simBackend) KillGen(_ context.Context, rt registry.Route, n uint64) error {
	w := b.w
	if !w.contained {
		return fmt.Errorf("sim: %w", errNoGenContainment)
	}
	w.jitter()
	key := fmt.Sprintf("%s#%d", rt.Host, n)
	w.mu.Lock()
	fails, seen := w.killFails[key]
	if !seen && w.rng.Intn(6) == 0 {
		fails = 1 + w.rng.Intn(2)
	}
	var procs []*simProc
	for p := range w.running[rt.Host] {
		if p.n == n {
			procs = append(procs, p)
		}
	}
	if fails > 0 {
		w.killFails[key] = fails - 1
		w.events = append(w.events, fmt.Sprintf("KillGen %s gen %d fails (%d more)", rt.Host, n, fails-1))
		w.mu.Unlock()
		return errors.New("sim: cgroup.kill: the child is not empty")
	}
	w.killFails[key] = 0
	w.mu.Unlock()
	stuck := 0
	for _, p := range procs {
		if p.mode == procStubborn || p.mode == procKillFails {
			p.exitLater()
			stuck++
			continue
		}
		p.die("cgroup.kill")
	}
	if stuck > 0 {
		return fmt.Errorf("sim: g%d still holds %d process(es)", n, stuck)
	}
	return nil
}

func (b *simBackend) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault }
func (b *simBackend) GenReady(registry.Route, Gen, Proc)          {}

// RaiseGenParent refuses without containment, as plan 2's ParentLimit
// does, so such a seed never overlaps.
func (b *simBackend) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error {
	if !b.w.contained {
		return fmt.Errorf("sim: %w", errNoGenContainment)
	}
	return nil
}
func (b *simBackend) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error {
	return nil
}
func (b *simBackend) GenSwitched(rt registry.Route, g Gen, _ Proc) error {
	b.w.logf("switch %s -> gen %d (%s)", rt.Host, g.N, g.Build)
	return nil
}

type simDue struct{}

func (simDue) SetNextDue(string, int64) error               { return nil }
func (simDue) ClearNextDueIf(string, int64) error           { return nil }
func (simDue) DueInstances(int64) ([]registry.Route, error) { return nil, nil }
func (simDue) AlwaysOnWorkers() ([]registry.Route, error)   { return nil, nil }
func (simDue) SetVersions(string, string, string) error     { return nil }
func (simDue) EarliestTimers() (map[string]int64, error)    { return nil, nil }

// checkSwap is I3 on the result: a SwapDone reports the attempt's build and
// staged path. The live link only ever names v1 or v2 and attempts only
// target v3 or v4, so a swap that ran the live link is a visible mismatch.
func (w *simWorld) checkSwap(at Attempt, res SwapResult) {
	w.logf("swap %s %s -> %s: %s %s", at.Host, at.Mode, at.Build, res.Outcome, res.Reason)
	w.tally(res)
	if res.Outcome != SwapDone {
		return
	}
	if res.Gen.Build != at.Build || res.Gen.StagedPath != at.StagedPath {
		w.violatef("I3: %s swap of %s to %s reported generation %d running %s from %q", at.Mode, at.Host, at.Build, res.Gen.N, res.Gen.Build, res.Gen.StagedPath)
	}
}

// checkGenerations is I1, polled against the entry's state under a.mu
// until stop closes.
func (w *simWorld) checkGenerations(a *Service, hosts []string, stop <-chan struct{}) {
	retired := map[string]map[uint64]bool{}
	for {
		for _, h := range hosts {
			cur, members := a.genStateForTest(h)
			if retired[h] == nil {
				retired[h] = map[uint64]bool{}
			}
			attempts := 0
			for n, role := range members {
				if role == roleRetiring {
					retired[h][n] = true
				} else {
					attempts++
				}
				if cur.N != 0 && n == cur.N {
					w.violatef("I1: generation %d of %s is current and also %v", n, h, role)
				}
			}
			if attempts > 1 {
				w.violatef("I1: %d attempts in flight for %s", attempts, h)
			}
			if cur.N != 0 && retired[h][cur.N] {
				w.violatef("I1: retired generation %d of %s is current again", cur.N, h)
			}
		}
		select {
		case <-stop:
			return
		case <-time.After(200 * time.Microsecond):
		}
	}
}

func onOff(b bool) string {
	if b {
		return "on"
	}
	return "off"
}

func simulateSwapOnce(seed int, staged map[string]simArtifact, sockDir string, sink *simLogSink) ([]string, []string, map[string]int) {
	hosts := []string{"one.sim.example.com", "two.sim.example.com"}
	w := &simWorld{rng: rand.New(rand.NewSource(int64(seed))), running: map[string]map[*simProc]bool{},
		live: map[string]string{}, staged: map[string]string{}, seen: map[string]bool{}, outcomes: map[string]int{},
		killFails: map[string]int{}, hosts: hosts}
	// One seed in five runs with the dormant switch off (today's paths,
	// where stop()'s own ErrStillRunning check is the only retention), and
	// one in five without generation containment.
	w.gensOn, w.contained = w.rng.Intn(5) != 0, w.rng.Intn(5) != 0
	w.outcomes[fmt.Sprintf("seeds with generations %s, containment %s", onOff(w.gensOn), onOff(w.contained))]++
	w.events = append(w.events, fmt.Sprintf("seed %d: generations %s, containment %s", seed, onOff(w.gensOn), onOff(w.contained)))
	sink.w.Store(w)
	defer sink.w.Store(nil)
	for b, art := range staged {
		w.staged[b] = art.path
	}
	for _, h := range hosts {
		w.live[h] = "v1"
	}
	sb := &simBackend{w: w, dir: sockDir, reps: map[string]*simRep{}}
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	a := New(Config{Bucket: "sim", Idle: time.Hour}, sb, simDue{})
	a.gens = w.gensOn
	a.now = clk.now
	a.genHealth = func(ctx context.Context, _ Gen) error {
		w.jitter()
		if err := ctx.Err(); err != nil {
			return err
		}
		if w.intn(6) == 0 {
			return errors.New("sim: unhealthy")
		}
		return nil
	}
	revs := map[string]*atomic.Uint64{}
	for _, h := range hosts {
		revs[h] = new(atomic.Uint64)
	}
	newest := func(h string) uint64 { return revs[h].Load() }

	stopCheck, checked := make(chan struct{}), make(chan struct{})
	go func() { defer close(checked); w.checkGenerations(a, hosts, stopCheck) }()

	var reqs, swaps sync.WaitGroup
	var fenceGen atomic.Uint64
	// The sweeper. The real Service ticks it on its own goroutine, so a
	// pending stop or an unproven wake is retried without an event to
	// trigger it. Here it ticks every 5ms of wall time (the fake clock does
	// not move, so nothing goes idle by it) and is serialized with the
	// events' own sweeps: there is one sweeper. It stops before Shutdown, as
	// the real one does.
	var sweepMu sync.Mutex
	sweep := func() {
		sweepMu.Lock()
		a.sweepOnce()
		sweepMu.Unlock()
	}
	stopSweeper, sweeperDone := make(chan struct{}), make(chan struct{})
	go func() {
		defer close(sweeperDone)
		tick := time.NewTicker(5 * time.Millisecond)
		defer tick.Stop()
		for {
			select {
			case <-stopSweeper:
				return
			case <-tick.C:
				sweep()
			}
		}
	}()
	for i := 0; i < 25; i++ {
		host := hosts[w.intn(len(hosts))]
		rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(sockDir, host+".sock"),
			DBPath: filepath.Join(sockDir, host+".db"), Hibernate: true, Status: "ready"}
		switch w.intn(12) {
		case 0, 1:
			reqs.Add(1)
			go func() {
				defer reqs.Done()
				ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
				defer cancel()
				if release, err := a.beginRequest(ctx, rt); err == nil {
					w.jitter()
					release()
				}
			}()
		case 2, 3, 4:
			mode := ModeOverlap
			if w.intn(3) == 0 {
				mode = ModeHold
			}
			build := []string{"v3", "v4"}[w.intn(2)]
			at := Attempt{Host: host, Cause: CauseRelease, Build: build, StagedPath: staged[build].path, Digest: staged[build].digest,
				Mode: mode, SchedRev: revs[host].Add(1), Canary: w.intn(2) == 0}
			// A scheduler swaps awake instances. Three attempts in four wake
			// their host first, so most reach a spawn, a switch or a hold's
			// pinned wake; the rest meet whatever state the host is in.
			wakeFirst := w.intn(4) != 0
			swaps.Add(1)
			go func() {
				defer swaps.Done()
				if wakeFirst {
					ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
					if release, err := a.beginRequest(ctx, rt); err == nil {
						release()
					}
					cancel()
				}
				w.checkSwap(at, a.Swap(context.Background(), at, newest))
			}()
		case 5:
			clk.advance(2 * time.Hour)
			sweep()
		case 6:
			sb.loseLease(host)
			sweep()
		case 7:
			a.Bounce(host)
		case 8:
			reqs.Add(1)
			go func() {
				defer reqs.Done()
				g := fenceGen.Add(1)
				ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
				defer cancel()
				_, _ = a.FenceAndStop(ctx, host, g)
				w.jitter()
				a.LiftFence(host, g)
			}()
		case 9:
			w.setLive(host, []string{"v1", "v2"}[w.intn(2)])
			revs[host].Add(1) // the scheduler's newer desired build supersedes in-flight attempts
		case 10:
			w.crashOne(host)
			sweep()
		case 11:
			// A stop-retry interval passes: the sweep's re-stop cadences
			// (pending stops, unproven members) come due, and nothing goes
			// idle.
			clk.advance(actStopTimeout + time.Second)
			sweep()
		}
		w.jitter()
	}
	reqs.Wait()
	close(stopSweeper)
	<-sweeperDone
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	a.Shutdown(ctx)
	cancel()
	w.mu.Lock()
	w.shutdown = true
	w.mu.Unlock()
	w.logf("Shutdown returned")
	// "Nothing runs after Shutdown returns" is the generations-on Shutdown
	// (spec §7.4: it cancels and joins waking entries, mutant M6's line).
	// With the switch off, today's rule holds by design (global constraints:
	// Shutdown and wake() byte for byte as today): a wake already in flight
	// resolves on its own and the lease TTL bounds it. Flag-off seeds check
	// every lease release and acquire, not this clause.
	if n := w.runningCount(); n > 0 && w.gensOn {
		w.violatef("I2: %d generation process(es) still run after Shutdown returned", n)
	}
	done := make(chan struct{})
	go func() { swaps.Wait(); close(done) }()
	select {
	case <-done:
	case <-time.After(10 * time.Second):
		w.violatef("liveness: a Swap did not return after Shutdown")
	}
	if n := w.runningCount(); n > 0 && w.gensOn {
		w.violatef("I2: %d generation process(es) run once every Swap has returned after Shutdown", n)
	}
	close(stopCheck)
	<-checked
	return w.snapshot()
}

// The model's death is one transition (final review F2): a reaper that
// sees a process Exited() and releases its lease never finds the same
// process still listed as running, so the sweep cannot report a false I2.
// The dying hook runs at the one point a reaper could act during a death;
// a second, concurrent die of the same process returns there too.
func TestSimDeathIsOneTransition(t *testing.T) {
	for _, concurrent := range []bool{false, true} {
		w := &simWorld{rng: rand.New(rand.NewSource(1)), running: map[string]map[*simProc]bool{},
			live: map[string]string{}, staged: map[string]string{}, seen: map[string]bool{}, outcomes: map[string]int{},
			killFails: map[string]int{}}
		p := w.spawn("h.example.com", "v1", 1, true)
		fired := false
		w.dying = func(q *simProc) {
			if fired {
				return
			}
			fired = true
			if concurrent {
				q.die("concurrent")
			}
			// A reaper: the process reads as exited, so it gives the lease back.
			if q.Exited() {
				w.leaseFreed(q.host, "reaper")
			}
		}
		p.die("first")
		if !fired {
			t.Fatalf("concurrent %v: the dying hook never ran", concurrent)
		}
		if v, ev, _ := w.snapshot(); len(v) != 0 {
			t.Fatalf("concurrent %v: the model reported %v\n%s", concurrent, v, strings.Join(ev, "\n"))
		}
		if n := w.runningOn("h.example.com"); n != 0 {
			t.Fatalf("concurrent %v: %d process(es) still listed after the death", concurrent, n)
		}
	}
}