package rollout

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"math/rand"
	"os"
	"path/filepath"
	"runtime"
	"runtime/debug"
	"slices"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// TestSimInvariants is plan 5's invariant simulation (spec §12
// "Concurrency"). It drives the real Manager, scheduler, appQueue,
// FailureBook, failCohort, decideWake and planCheck through random schedules
// of promotions, agent passes, admissions, attempt outcomes, planned wakes and
// their first-wake observations, hibernation, restarts, config deliveries,
// `carlos promote --retry`, `carlos instances retry` and restaging. Only the
// activator is modelled (simAct, and the wake steps below), along with root's
// install (a promotion writes the rows' installed build) and the scheduler's
// goroutine, whose loop body (tick) runs as a step of its own. Each attempt
// runs the real scheduler.execute and scheduler.done on a goroutine of its
// own, handed off so that exactly one goroutine runs at any moment: a seed is
// one deterministic interleaving, and every invariant check sees a state no
// transition is halfway through. An ordinary wake's healthy event may reach
// the agent some steps after its generation serves (sim.late), and one seed
// in four keeps its failure book on disk, where windows of failing saves
// open and close (sim.fault).
//
// The invariants, checked against the model's own ground truth (what it
// promoted, what failed, what it retried), never against the book:
//
//   - I1: a release-cause failure never leaves an instance that has a good
//     build unable to boot one: every wake of such an instance boots
//     something, a first wake in overlap mode carries the good build as its
//     fallback, and no wake boots, for an instance that has a good build and
//     has not served it, a build whose failure on this box stands. Spec
//     §6.4's hold-down (a hold build that ran and failed on the instance) is
//     the one refusal it allows, and spec §8.5's bootstrap rule exempts an
//     instance with no good build. An instance where a hold build ran
//     without passing (the hold-ran marker, final review wave 2, N2) is
//     restricted, not failed: its wakes carry no fallback, and may be
//     refused for a build a record holds for it (the marked build included:
//     the marker lifts no record, Astra R3 N7) or for its installed good
//     build until a signed rollback lifts the marker (ruling 205). I5 is the
//     restriction's own rule.
//   - I2: swap_concurrency is never exceeded.
//   - I3: a failure record never blocks an instance already serving that
//     build. With capacity free, its restart and config attempts onto that
//     build run unless their own cohort's canary is still unsettled on
//     another host, and an attempt by an instance that serves (or has as its
//     good build) the build it failed on never writes or changes a record.
//   - I4: a swap never switches to a superseded build: at the switch, an
//     adoption's build is the instance's installed build, and no failure of
//     it stands against the instance.
//   - I5: once a hold build has run on an instance without passing there
//     (an adoption abandoned or failed after its exec, not held down; a hold
//     first wake ended with no verdict after its exec), no wake of that
//     instance boots its good build in place of another installed build,
//     as a recovery or a fallback, until the instance records a good build
//     from an execution that began after the hold build ran (Astra R3 N6)
//     or a host retry names it (spec §6.4). A Pass may also lift the
//     restriction, but only on rollback evidence (ruling 205): the
//     instance's installed build is its good build, and its channel's
//     pointer names that build signed after the execution that ran the hold
//     build started (its dispatch, or its wake's plan). A marker the Pass
//     cleared without it is an I5 violation. (The book is read here only to
//     see what the Pass cleared; the evidence is the model's.) When the
//     marked build is itself the installed good build, it boots under the
//     marked-build rule, and I1 requires its wake to boot. A good write
//     that clears a marker the model's restriction still stands for (one
//     set after the write's execution began) is an I5 violation at the
//     write (goodWrite).
//   - I6: every execution's intent (Task 5) ends with its outcome: once the
//     run has drained, an intent stands only for a first wake still in
//     flight, and it is that wake's.
//   - I7: a retry is never undone by an outcome that began before it
//     (ruling 191, plan review finding 3): a host's `carlos instances
//     retry` stands in the record until that host fails the build again in
//     a later execution, and a record a re-signed pointer cleared is
//     neither written again nor its adoption cohort halted again but by a
//     failure of an execution that began after the clear.
//   - I8: a failure outlives an agent restart even when no save of the book
//     lands (Task 5, F5): at every failure, before the code records it, an
//     agent opening the book from disk holds the build for the instance,
//     and holds it down when the failure does.
//   - I9: a stale grant never clears a later failure (the crash-epoch hard
//     limit, readiness round 2; fix wave 1, item 3). The model numbers every
//     restriction transition it causes or observes, in the order they land,
//     apart from the book (sim.tseq): each one that adds a restriction is a
//     fact (sim.facts: a record created, or merged with a change; a
//     hold-down; a hold-ran marker set, renewed or replacing another; an
//     intent a restart reconciles), and each one that removes one is a
//     clear (sim.clears: a grant's applied retry or marker clear, a record
//     a pointer clears or a prune drops, a marker lifted or replaced). A
//     grant captures the number when it is issued from published status.
//     When a grant is accepted (RetryHostAt clears anything), no fact of its
//     (build, host), nor a new record of its build (which holds every host,
//     spec §8.4), may have landed after that capture, unless a clear of that
//     restriction was applied after the fact. Facts are ordered by when they
//     land, not by when their execution began: a late outcome (an execution
//     begun before the grant whose failure lands after it, not PreRetry) and
//     a change in the grant's own step are inside the check. Issuing a grant
//     answers nothing; only an applied clear does. The facts are the
//     model's, never the book's verdicts.
//   - I10: durable before use: every non-zero generation PublishedEpoch
//     returns is the app's generation in the book's last landed file,
//     re-read from disk at that moment, under the identity the file holds.
//
// I9 and I10 run on the seeds whose book is on disk (seed%4 == 3, not as
// root, where these seeds keep their book in memory and check neither),
// with I8's read-only windows, and what only those seeds draw (sim.gr):
// grants, queued and delivered later, again and again, as an agent
// re-reads its pending grants; agent restarts, which drop every execution
// in flight and reopen the book from disk under a new Manager; box
// rebuilds, which remove the book's file and open a new book; and a Retry
// on an instance its hold-ran marker holds, followed by a wake, of that
// instance or of another that never served the marked build, stopped
// before it settles (rewakeArm). The grant queue and the facts outlive
// restarts and rebuilds.
//
// Seeds: ROLLOUT_SIM_SEEDS (default 200; the per-revision sweep is 3000).
// ROLLOUT_SIM_SEED=<n> replays one seed. The production code iterates some
// maps, so a replay is best effort; the action log printed with a failure is
// the evidence. Every violation names its invariant ("I2 violated at seed"),
// which is what testdata/rolloutsim/mutate.sh reads.
func TestSimInvariants(t *testing.T) {
	first, last := int64(0), int64(200)
	if v := os.Getenv("ROLLOUT_SIM_SEEDS"); v != "" {
		n, err := strconv.ParseInt(v, 10, 64)
		if err != nil || n < 1 {
			t.Fatalf("ROLLOUT_SIM_SEEDS=%q", v)
		}
		last = n
	}
	if v := os.Getenv("ROLLOUT_SIM_SEED"); v != "" {
		n, err := strconv.ParseInt(v, 10, 64)
		if err != nil {
			t.Fatalf("ROLLOUT_SIM_SEED=%q", v)
		}
		first, last = n, n+1
	}
	opt, books := t.TempDir(), t.TempDir()
	staged := map[string]bool{}
	failed := 0
	for seed := first; seed < last; seed++ {
		v := runSimSeed(seed, opt, books, staged, simSteps)
		if v == nil {
			continue
		}
		failed++
		t.Errorf("rolloutsim: %s\n--- last actions ---\n%s", v.line, strings.Join(v.trace, "\n"))
		if failed >= 3 {
			t.FailNow()
		}
	}
}

const simSteps = 400

// TestSimSeedsEndTheirGoroutines: an agent restart drops the attempts parked
// in Swap (agentRestart), and the seed's teardown (runSimSeed) must end their
// goroutines. A goroutine left parked holds its sim, so a 3000-seed sweep
// would keep every seed with a restart for the life of the process.
func TestSimSeedsEndTheirGoroutines(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("as root no seed keeps its book on disk, so no seed restarts its agent")
	}
	opt, books := t.TempDir(), t.TempDir()
	staged := map[string]bool{}
	before := runtime.NumGoroutine()
	// The seeds whose book is on disk (seed%4 == 3) are the ones that restart.
	for seed := int64(3); seed < 16; seed += 4 {
		if v := runSimSeed(seed, opt, books, staged, simSteps); v != nil {
			t.Fatalf("rolloutsim: %s", v.line)
		}
	}
	// An ended Manager's own goroutines leave on their cancelled context, a
	// moment after it.
	deadline := time.Now().Add(5 * time.Second)
	for runtime.NumGoroutine() > before && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	if after := runtime.NumGoroutine(); after > before {
		t.Fatalf("%d goroutines before the seeds, %d after them", before, after)
	}
}

// The activator's first-wake window, as spawnPlanned reads it: B's health
// window ends firstWakeBudget less firstWakeKillReserve after the wake's
// start, and B is exec'd only with firstWakeMinVerdict of it left.
const (
	simHealthWindow = 55*time.Second - 25*time.Second
	simMinVerdict   = 20 * time.Second
)

// simViolation is an invariant violation, panicked by violate and recovered
// per seed. Any other panic is a crash of the simulation or the code under
// test, reported apart so the mutation script never counts it as a kill.
type simViolation struct {
	inv, msg string
}

type simCrash struct {
	msg   string
	stack []byte
}

type simFailure struct {
	line  string
	trace []string
}

func runSimSeed(seed int64, opt, books string, staged map[string]bool, steps int) (out *simFailure) {
	var s *sim
	defer func() {
		if s != nil && s.cancel != nil {
			s.cancel()
		}
		if s != nil {
			// Ends every attempt still parked in Swap (one an agent restart
			// dropped, or one a violation left), without returning from Swap,
			// so none of them touches the book or the model, and nothing holds
			// the sim after the seed.
			close(s.dead)
		}
		if s != nil && s.dir != "" {
			// Writable again, so the seed's book can go (and t.TempDir's
			// own cleanup never meets a read-only directory).
			os.Chmod(s.dir, 0o700)
			os.RemoveAll(s.dir)
		}
		r := recover()
		if r == nil {
			return
		}
		step, trace := -1, []string(nil)
		if s != nil {
			step, trace = s.step, s.trace
		}
		switch v := r.(type) {
		case simViolation:
			out = &simFailure{line: fmt.Sprintf("%s violated at seed %d step %d: %s", v.inv, seed, step, v.msg), trace: trace}
		case simCrash:
			out = &simFailure{line: fmt.Sprintf("crash at seed %d step %d: %s\n%s", seed, step, v.msg, v.stack), trace: trace}
		default:
			out = &simFailure{line: fmt.Sprintf("crash at seed %d step %d: %v\n%s", seed, step, r, debug.Stack()), trace: trace}
		}
	}()
	s = newSim(seed, opt, books, staged)
	s.run(steps)
	return nil
}

func violate(inv, format string, args ...any) {
	panic(simViolation{inv: inv, msg: fmt.Sprintf(format, args...)})
}

// simPhase is an instance's state as the modelled activator holds it.
type simPhase int

const (
	simAsleep    simPhase = iota
	simWaking             // planned, not yet exec'd: owned, nothing current
	simObserving          // a first wake inside its 10 s observation: owned, Current hides it
	simAwake              // serving gen
	simStopping           // hibernating: owned, nothing current
)

type simHost struct {
	name  string
	ch    int
	phase simPhase
	gen   activator.Gen // the generation serving (awake) or observed (observing)
	// good is the model's good build: the last build that settled here. It
	// is ground truth, kept apart from the row the code writes.
	good   string
	served map[string]bool
	// A wake in flight: its plan, the route the activator holds for it, and
	// the FailedUnder the model expects for its build.
	plan      activator.WakePlan
	planRoute registry.Route
	wakeStart time.Time
	// planAt is when the plan the wake runs was made (the first plan, or
	// its replan at the exec): the start of its execution, which a hold-ran
	// marker the wake leaves is stamped with (ruling 205).
	planAt    time.Time
	stamp     time.Time
	failLater bool // observing: the build exits inside its observation
	// underMark: the wake's plan was made while a hold-ran restriction
	// stood (N2), so it carries no fallback even if a retry lifts the
	// restriction before its exec.
	underMark bool
}

type simBuild struct {
	bad        bool // fails on every instance that has not served it
	flaky      bool // fails half the time on an instance that has not served it
	holdDemand bool // the release demands hold
	misstaged  bool // the staged file is not the canonical artifact until root restages
}

// simSwap is one attempt inside the modelled Swap, parked until a complete
// step lets it finish.
type simSwap struct {
	a       activator.Attempt
	newest  func(string) uint64
	release chan struct{}
	stamp   time.Time // the FailedUnder the model expects
	// started is when the attempt was dispatched (scheduler.execute's
	// d.Started): the At of a hold-ran marker it leaves (ruling 205).
	started time.Time
	// Set by finishSwap for the I3 record check: whether the instance
	// serves the attempt's build, and the book's record for it beforehand.
	serves bool
	before string
}

// simLate is a healthy event on its way (sim.late): the event, and the
// start of the wake's execution (its plan), which the model's good write
// is ordered by.
type simLate struct {
	ev    activator.GenEvent
	start time.Time
}

// simGrant is a signed `carlos instances retry` grant (I9): the build and
// host it names, the book and generation status published for them, when
// it was issued (the step, and its time, for messages), and the model's
// transition number at that moment (cap, sim.tseq), which every fact is
// ordered against.
type simGrant struct {
	build, host, book string
	epoch             uint64
	step              int
	at                time.Time
	cap               uint64
}

// simFact is one of I9's facts: a transition (seq, sim.tseq) that added a
// restriction of build on host (kind says what), from an execution that
// began at start (for messages only: facts are ordered by seq, when they
// landed). app is a new record, made by a failure on host: it holds every
// host of the app (spec §8.4: "the build's next failure writes a new record,
// which holds every host"), so the fact is every host's, not host's alone.
type simFact struct {
	build, host, kind string
	start             time.Time
	seq               uint64
	marker            bool // a hold run: its host's hold-ran restriction
	app               bool // a new record: every host of the app
}

// simClear is a transition (seq) that removed a restriction, as applied:
// never a grant's issue, only its clear. marker is the hold-ran
// restriction of host (build "" is whichever build it named: a lift);
// otherwise build's record restriction of host, or of every host when host
// is "" (the record left the book: a pointer's clear or a prune).
type simClear struct {
	build, host, kind string
	seq               uint64
	marker            bool
}

// simRec is the model's own failure record of a build (sim.grec): the hosts
// a host retry cleared, and those held down.
type simRec struct {
	retried, down map[string]bool
}

type simEvent struct {
	host  string
	swap  *simSwap // the attempt reached Swap and is parked
	crash any
	stack []byte
}

type sim struct {
	r     *rand.Rand
	seed  int64
	step  int
	now   time.Time
	trace []string

	opt    string
	staged map[string]bool
	m      *Manager
	sched  *scheduler
	book   *FailureBook
	rows   *fakeRoutes

	hosts       []*simHost
	byName      map[string]*simHost
	channels    [2]string
	installed   [2]string
	pointers    map[string]release.ChannelPointer // channel -> its signed pointer
	builds      map[string]*simBuild
	nb, nreq    int
	hold        bool // the app's swap policy is hold
	concurrency int
	genN        uint64

	// Ground truth. A failure of build b stands while no pointer naming b
	// has been signed with a PromotedAt newer than b's newest failure stamp
	// (the FailedUnder its attempt or wake read, else the failure's time):
	// spec rev 14 §8.4. exempt is `carlos instances retry` per host until
	// that host fails the build again; heldDown is a hold build that ran and
	// failed on the host, until a host retry.
	lastStamp  map[string]time.Time
	lastSigned map[string]time.Time
	failedEver map[string]bool
	exempt     map[string]map[string]bool
	heldDown   map[string]map[string]bool
	// holdRan is the hold-ran restriction per host: the hold build that ran
	// there without passing, until the host records a good build (goodAt),
	// a host retry names it, or a Pass lifts it on rollback evidence.
	// holdRanAt is the start of the execution that set it (an attempt's
	// dispatch, a planned wake's plan; Task 3 fix round 1), which a rollback
	// pointer must be newer than.
	holdRan   map[string]string
	holdRanAt map[string]time.Time
	// holdRanSet is when the restriction was last set (the step), which a
	// good write must come after the start of: a good build whose execution
	// began before a hold build ran on the host never lifts the restriction
	// (Astra R3 N6).
	holdRanSet map[string]time.Time
	// Retry ordering (ruling 191). retryAt is each host's latest `carlos
	// instances retry` of a build, clearedAt the latest Pass that cleared a
	// build's record on a `carlos promote --retry` pointer. A failure of an
	// execution that started before either is stale: it changes nothing a
	// retry cleared and holds nothing down.
	retryAt   map[string]map[string]time.Time
	clearedAt map[string]time.Time
	// failedAt is when each build last failed in an execution that began
	// after every retry of it (a failure the retry ordering lets stand): I7
	// reads it against clearedAt.
	failedAt map[string]time.Time

	// The book on disk (I8): one seed in four keeps the failure book in a
	// directory of its own, and fault makes that directory read-only for
	// a while, so every save fails as a full or broken disk would. fr draws
	// the fault windows, apart from r, so the seeds that keep their book in
	// memory run exactly the schedules they ran before I8 existed.
	dir   string
	fault bool
	fr    *rand.Rand

	// The crash-epoch hard limit (I9, I10), on the seeds whose book is on
	// disk. gr draws the grants, their deliveries, the agent restarts, the
	// box rebuilds and the rewakes (rewakeArm), apart from r and fr. cancel
	// ends the running agent's Manager (its ack writer).
	gr     *rand.Rand
	cancel context.CancelFunc
	// grants are the signed host retries the console issued, oldest first,
	// each delivered again at every delivery step (an agent re-reads its
	// pending grants), as the agent does until a grant expires. They
	// survive restarts and rebuilds.
	grants []simGrant
	// tseq numbers every restriction transition the model causes or
	// observes, in the order they land (I9), kept apart from the book: facts
	// are the ones that added a restriction, clears the applied ones that
	// removed one. Facts survive restarts and rebuilds. Clears do not: a
	// crash rolls back whatever clear the book had not saved, and the model
	// cannot tell which had landed, so after a restart no earlier clear
	// discharges anything (the strict direction).
	tseq   uint64
	facts  []simFact
	clears []simClear
	// grec is the model's own failure record per build (I9): which hosts a
	// retry clears and which are held down, so a failure is a fact only when
	// it changes what the spec says restricts its host (plan finding 7: an
	// unchanged record or a pre-retry outcome changes nothing). Kept from the
	// model's events. It takes three things from the book: a record leaving
	// it (a pass), a restart's reread of the disk, and a grant's clear, where
	// deliver marks the host retried once the book's Lookup shows the record
	// now lists it (a clear I9 has vetted first).
	grec map[string]*simRec
	// rewake is the host rewakeRun wakes and stops at the next step: the one
	// rewakeArm granted a retry to, or another on its channel.
	rewake string
	// rewakeFail is the build whose first wake on rewake fails its health
	// check (another instance's wake, M17's witness), or "".
	rewakeFail string

	// late holds an ordinary wake's healthy event (GenHealthy, not a first
	// wake) that the activator has not yet delivered, per host: its
	// generation serves, and its good write waits for the event (Astra R3
	// N6). Such a wake has no intent, so an adoption may run on the host
	// meanwhile; a hold one that fails leaves a marker the late write must
	// not clear (I5).
	late map[string]simLate

	swaps    map[string]*simSwap
	events   chan simEvent
	draining bool
	// dead is closed when the seed ends (runSimSeed): an attempt goroutine
	// still parked in Swap, or about to report on events, exits instead.
	dead chan struct{}
}

func newSim(seed int64, opt, books string, staged map[string]bool) *sim {
	s := &sim{r: rand.New(rand.NewSource(seed)), seed: seed, now: t0, opt: opt, staged: staged,
		book: &FailureBook{}, rows: &fakeRoutes{rows: map[string]registry.Route{}},
		byName: map[string]*simHost{}, channels: [2]string{"", "beta"},
		pointers: map[string]release.ChannelPointer{}, builds: map[string]*simBuild{},
		lastStamp: map[string]time.Time{}, lastSigned: map[string]time.Time{}, failedEver: map[string]bool{},
		exempt: map[string]map[string]bool{}, heldDown: map[string]map[string]bool{}, holdRan: map[string]string{},
		holdRanAt: map[string]time.Time{}, holdRanSet: map[string]time.Time{}, retryAt: map[string]map[string]time.Time{},
		clearedAt: map[string]time.Time{}, failedAt: map[string]time.Time{}, late: map[string]simLate{}, swaps: map[string]*simSwap{},
		grec:   map[string]*simRec{},
		events: make(chan simEvent), dead: make(chan struct{})}
	if seed%4 == 3 && os.Geteuid() != 0 {
		// Root writes into a read-only directory, so it gets no faults, and
		// none of I8, I9 and I10: its book stays in memory.
		dir, err := os.MkdirTemp(books, "book")
		if err != nil {
			panic(simCrash{msg: "make the book's directory: " + err.Error()})
		}
		s.dir = dir
		if s.book, err = OpenFailureBook(dir); err != nil {
			panic(simCrash{msg: "open the book: " + err.Error()})
		}
		// The agent saves a new book before anything runs (internal/host).
		if err := s.book.Flush(); err != nil {
			panic(simCrash{msg: "save the new book: " + err.Error()})
		}
		s.fr = rand.New(rand.NewSource(seed*7919 + 1))
		s.gr = rand.New(rand.NewSource(seed*7919 + 2))
	}
	s.book.now = func() time.Time { return s.now }
	s.hold = s.r.Intn(4) == 0
	s.concurrency = 1 + s.r.Intn(3)
	s.addBuild("b0", &simBuild{})
	s.installed = [2]string{"b0", "b0"}
	for _, ch := range s.channels {
		s.sign(ch, "b0")
	}
	n := 3 + s.r.Intn(4)
	for i := range n {
		h := &simHost{name: "h" + strconv.Itoa(i), ch: i % 2, served: map[string]bool{}}
		row := registry.Route{Host: h.name, Kind: "instance", App: "blog", Account: "bab", Status: "ready",
			Hibernate: true, Version: "b0", Channel: s.channels[h.ch]}
		switch {
		case s.r.Intn(2) == 0:
			// Awake: Bootstrap records the build it serves as good (spec §8.5).
			h.phase, h.gen = simAwake, s.newGen(s.identity("b0"))
			h.good, h.served["b0"] = "b0", true
		case s.r.Intn(2) == 0:
			id := s.identity("b0")
			row.GoodVersion, row.GoodDigest, row.GoodDemand = id.Version, id.Digest, id.Demand
			h.good, h.served["b0"] = "b0", true
		default:
			// Asleep with no good build: the upgrade's bootstrap case.
		}
		s.rows.put(row)
		s.hosts = append(s.hosts, h)
		s.byName[h.name] = h
	}
	s.startAgent()
	return s
}

// startAgent is the agent's start with s.book (internal/host): a new
// Manager, Bootstrap, and its first pass.
func (s *sim) startAgent() {
	ctx, cancel := context.WithCancel(context.Background())
	s.cancel = cancel
	s.m = NewManager(ctx, Deps{Act: simAct{s}, Routes: s.rows, Book: s.book, OptDir: s.opt, Box: "box-1",
		Demand: s.demand, Policy: s.policy, Pointer: s.pointer, Now: func() time.Time { return s.now },
		Logf: func(string, ...any) {}})
	// The app's scheduler, registered by hand so its goroutine never starts:
	// tick is its loop body, run as a step. The queue is built as
	// Manager.scheduler builds it.
	key := appKey{"bab", "blog"}
	book := s.book
	s.sched = &scheduler{m: s.m, key: key, kick: make(chan struct{}, 1),
		q: newAppQueue(key.acct, key.app, func(build, host string, served bool) bool {
			return book.Blocks(key.acct, key.app, build, host, served)
		})}
	s.m.apps[key] = s.sched
	rs, _ := s.rows.List()
	s.m.Bootstrap(rs)
	s.pass()
}

func (s *sim) logf(format string, args ...any) {
	if len(s.trace) >= 60 {
		s.trace = s.trace[1:]
	}
	s.trace = append(s.trace, fmt.Sprintf("%4d %s ", s.step, s.now.Sub(t0))+fmt.Sprintf(format, args...))
}

func (s *sim) addBuild(b string, sb *simBuild) {
	s.builds[b] = sb
	if !s.staged[b] {
		if err := release.StageArtifact(s.opt, "bab", b, "blog", []byte(b)); err != nil {
			panic(simCrash{msg: "stage " + b + ": " + err.Error()})
		}
		s.staged[b] = true
	}
}

func (s *sim) identity(b string) release.StagedIdentity {
	id := release.StagedIdentity{Version: b, StagedPath: release.VersionPath(s.opt, "bab", b, "blog"), Digest: "d-" + b}
	if s.builds[b].holdDemand {
		id.Demand = release.DemandHold
	}
	return id
}

func (s *sim) newGen(id release.StagedIdentity) activator.Gen {
	s.genN++
	return activator.Gen{N: s.genN, Build: id.Version, StagedPath: id.StagedPath, Digest: id.Digest, Demand: id.Demand}
}

func (s *sim) sign(channel, b string) {
	s.pointers[channel] = release.ChannelPointer{Schema: 1, Account: "bab", App: "blog", Channel: channel, Version: b, PromotedAt: s.now}
	s.lastSigned[b] = s.now
}

// demand is release.CanonicalDemand's contract, as the harness has it.
func (s *sim) demand(_ context.Context, _, _, ver, _ string) (string, string, error) {
	b := s.builds[ver]
	if b == nil || b.misstaged {
		return "", "", fmt.Errorf("%w: the staged executable's digest differs from the canonical manifest", release.ErrStagedDigestMismatch)
	}
	if b.holdDemand {
		return release.DemandHold, "d-" + ver, nil
	}
	return "", "d-" + ver, nil
}

func (s *sim) policy(context.Context, string, string) (*appconfig.Manifest, error) {
	swap := appconfig.SwapOverlap
	if s.hold {
		swap = appconfig.SwapHold
	}
	return &appconfig.Manifest{Schema: 1, App: "blog", Swap: swap, SwapConcurrency: s.concurrency}, nil
}

func (s *sim) pointer(_ context.Context, _, _, channel string) (*release.ChannelPointer, error) {
	p, ok := s.pointers[channel]
	if !ok {
		return nil, nil
	}
	return &p, nil
}

// stampFor is the FailedUnder the model expects for build on host, read when
// its attempt or wake starts: the PromotedAt of the host's channel pointer
// when it names build, else zero (the failure's own time stands in).
func (s *sim) stampFor(h *simHost, build string) time.Time {
	if p, ok := s.pointers[s.channels[h.ch]]; ok && p.Version == build {
		return p.PromotedAt
	}
	return time.Time{}
}

func (s *sim) expectHold(build string) bool {
	return s.hold || s.builds[build].holdDemand
}

// failure is the ground truth of a build that ran and failed on host, in the
// execution that started at start. One that started before a retry of the
// build since (the host's, or a Pass's clear on a re-signed pointer) is
// stale (ruling 191): it stands for nothing the retry cleared, and a
// hold-down it would have made leaves the hold-ran restriction instead (the
// hold build ran against the database; Task 4 fix round 1). A restriction a
// run leaves without a hold-down is the caller's, as for any run.
func (s *sim) failure(build, host string, stamp time.Time, heldDown bool, start time.Time) {
	s.failureOf(build, host, stamp, heldDown, start, false)
}

// failureOf is failure, and with reconciled an intent a restart turned into
// a failure (OpenFailureBook): never stale, since nothing of the retry
// ordering survives a restart, and always a fact, since the record it makes
// carries the open time, newer than any.
func (s *sim) failureOf(build, host string, stamp time.Time, heldDown bool, start time.Time, reconciled bool) {
	if !reconciled && s.stale(build, host, start) {
		s.logf("ground truth: %s failed on %s in an execution started at %s, before a retry of it: stale", build, host, start.Sub(t0))
		if heldDown {
			s.holdRanOn(s.byName[host], build, start)
		}
		return
	}
	kind := "a failure"
	if heldDown {
		kind = "a hold-down"
	}
	if reconciled {
		kind = "an intent the restart reconciled"
	}
	switch changed, created := s.recFailed(build, host, heldDown); {
	case created:
		s.addFact(simFact{build: build, host: host, kind: kind, start: start, app: true})
	case changed || reconciled:
		s.addFact(simFact{build: build, host: host, kind: kind, start: start})
	}
	s.checkI8(build, host, heldDown)
	if stamp.IsZero() {
		stamp = s.now
	}
	s.failedEver[build] = true
	s.failedAt[build] = s.now
	if stamp.After(s.lastStamp[build]) {
		s.lastStamp[build] = stamp
	}
	delete(s.exempt[host], build)
	if heldDown {
		if s.heldDown[host] == nil {
			s.heldDown[host] = map[string]bool{}
		}
		s.heldDown[host][build] = true
	}
	s.logf("ground truth: %s failed on %s (stamp %s, held down %v)", build, host, stamp.Sub(t0), heldDown)
}

// stale reports whether an execution of build on host that started at start
// began before a retry of it since (ruling 191): the host's `carlos
// instances retry`, or a Pass's clear on a re-signed pointer.
func (s *sim) stale(build, host string, start time.Time) bool {
	return !start.IsZero() && (start.Before(s.retryAt[host][build]) || start.Before(s.clearedAt[build]))
}

// recFailed is a failure of build on host, not stale, in the model's own
// record of build (grec). It reports whether the failure made a new record
// (created), which restricts every host of the app, and otherwise whether it
// changed what restricts host (changed): host's retry spent, or host newly
// held down (I9). A failure that changes none of these moves no generation
// (rule 1 in FailureBook's doc), so it is no fact; a newer failure stamp
// alone is not counted either, the permissive direction.
func (s *sim) recFailed(build, host string, heldDown bool) (changed, created bool) {
	r := s.grec[build]
	if r == nil {
		r = &simRec{retried: map[string]bool{}, down: map[string]bool{}}
		s.grec[build] = r
		created = true
	} else {
		changed = r.retried[host] || (heldDown && !r.down[host])
	}
	delete(r.retried, host)
	if heldDown {
		r.down[host] = true
	}
	return changed, created
}

// addFact files f, a transition that added a restriction (I9), at the
// model's next transition number.
func (s *sim) addFact(f simFact) {
	s.tseq++
	f.seq = s.tseq
	s.facts = append(s.facts, f)
}

// addClear files c, an applied transition that removed a restriction (I9),
// at the model's next transition number.
func (s *sim) addClear(c simClear) {
	s.tseq++
	c.seq = s.tseq
	s.clears = append(s.clears, c)
	s.logf("transition %d: %s cleared (build %q, host %q, marker %v)", c.seq, c.kind, c.build, c.host, c.marker)
}

// discharged reports whether a clear applied after f landed removed f's
// restriction of host (I9): for a hold run, a lift or replacement of its
// host's hold-ran marker; for a record fact, a grant's retry of its build
// on host, or its build's record leaving the book. host is f's own host
// but for a new record (f.app), which holds every host: there it is the
// host of the grant I9 checks. A grant's issue never discharges anything.
func (s *sim) discharged(f simFact, host string) bool {
	return slices.ContainsFunc(s.clears, func(c simClear) bool {
		return c.seq > f.seq && c.marker == f.marker && (c.build == "" || c.build == f.build) && (c.host == "" || c.host == host)
	})
}

// diskFault opens or closes a window in which every save of the book fails
// (I8), for a seed whose book is on disk: its directory is made read-only,
// so WriteFileAtomic cannot create its temp file there.
func (s *sim) diskFault() {
	if s.dir == "" {
		return
	}
	switch {
	case !s.fault && s.fr.Intn(10) == 0:
		s.fault = true
	case s.fault && s.fr.Intn(4) == 0:
		s.fault = false
	default:
		return
	}
	mode := os.FileMode(0o700)
	if s.fault {
		mode = 0o500
	}
	if err := os.Chmod(s.dir, mode); err != nil {
		panic(simCrash{msg: "the book's directory: " + err.Error()})
	}
	s.logf("disk: saves of the failure book now fail: %v", s.fault)
}

// checkI8 is I8 at a failure of build on host, called before the code has
// written it: an agent restarting now must still find it. The restart is
// simulated by a second book opened from the disk as the agent opens it at
// startup (OpenFailureBook, which turns intents into records). It must hold
// build for host, and hold host down when the failure does. Only the
// execution's intent, saved before the build ran (Task 5, F5), can carry
// that across a window in which every save fails.
func (s *sim) checkI8(build, host string, heldDown bool) {
	if s.dir == "" {
		return
	}
	b, err := OpenFailureBook(s.dir)
	if err != nil {
		panic(simCrash{msg: "reopen the book: " + err.Error()})
	}
	if !b.Blocks("bab", "blog", build, host, false) || (heldDown && !b.HeldDown("bab", "blog", build, host)) {
		violate("I8", "%s failed on %s (held down %v), and an agent restarting now would not hold it there (saves failing: %v)", build, host, heldDown, s.fault)
	}
}

// checkI7 is I7 after every step, against the model's retries: a failure of
// an execution that began before a retry never takes the retry back.
//   - A `carlos instances retry` of host for build b (exempt) stands until
//     host fails b again in an execution that began after it (failure
//     deletes exempt) or b's record goes: the record still lists host as
//     retried.
//   - A record a re-signed pointer cleared (clearedAt) is written again, and
//     b's adoption cohort is halted again, only by a failure of an execution
//     that began after the clear (failedAt). A stale one's halt is caught
//     here before the next Pass's liftClearedHalts could hide it.
func (s *sim) checkI7() {
	for _, b := range sortedKeys(s.clearedAt) {
		if !s.failedAt[b].Before(s.clearedAt[b]) {
			continue
		}
		if r, ok := s.book.Lookup("bab", "blog", b); ok {
			violate("I7", "%s's record, cleared by a re-signed pointer at %s, stands again with no failure begun after the clear: %+v",
				b, s.clearedAt[b].Sub(t0), r)
		}
		s.sched.mu.Lock()
		c := s.sched.q.cohorts[cohortKey(b, classAdopt)]
		halted := c != nil && c.halted
		s.sched.mu.Unlock()
		if halted {
			violate("I7", "%s's cohort, reopened by a re-signed pointer at %s, is halted again with no failure begun after the clear", b, s.clearedAt[b].Sub(t0))
		}
	}
	for _, host := range sortedKeys(s.exempt) {
		for _, b := range sortedKeys(s.exempt[host]) {
			if r, ok := s.book.Lookup("bab", "blog", b); ok && !slices.Contains(r.Retried, host) {
				violate("I7", "%s's retry of %s was taken back with no failure there begun after it: %+v", host, b, r)
			}
		}
	}
}

func (s *sim) stands(build string) bool {
	return s.failedEver[build] && !s.lastSigned[build].After(s.lastStamp[build])
}

func (s *sim) blockedFor(host, build string) bool {
	return s.stands(build) && !s.exempt[host][build]
}

// snap is the book's record for build, as text, for the I3 record check.
func (s *sim) snap(build string) string {
	r, ok := s.book.Lookup("bab", "blog", build)
	return fmt.Sprintf("%v %+v", ok, r)
}

// simAct is the modelled activator. Only one goroutine runs at a time (the
// handoffs in launch and complete), so it reads the model without a lock.
type simAct struct{ s *sim }

func (f simAct) Current(host string) (activator.Gen, bool) {
	if h := f.s.byName[host]; h != nil && h.phase == simAwake {
		return h.gen, true
	}
	return activator.Gen{}, false
}

func (f simAct) Owns(host string) bool {
	h := f.s.byName[host]
	return h != nil && h.phase != simAsleep
}

// Swap parks the attempt until a complete step releases it. When the seed
// ends first (sim.dead), the goroutine exits from here: Swap never returns,
// so done() never runs and the attempt's Manager never touches the book.
func (f simAct) Swap(_ context.Context, a activator.Attempt, newest func(string) uint64) activator.SwapResult {
	sw := &simSwap{a: a, newest: newest, release: make(chan struct{})}
	select {
	case f.s.events <- simEvent{host: a.Host, swap: sw}:
	case <-f.s.dead:
		runtime.Goexit()
	}
	select {
	case <-sw.release:
	case <-f.s.dead:
		runtime.Goexit()
	}
	return f.s.finishSwap(sw)
}

func (f simAct) ReplaceWorker(context.Context, activator.Attempt, *release.StagedIdentity, func(string) uint64) activator.SwapResult {
	panic(simCrash{msg: "the simulation runs no workers"})
}

func (f simAct) PublishHeldServing(string) {}

// EffectiveMode: the simulation has no foreign-layout rows.
func (f simAct) EffectiveMode(_ string, m activator.SwapMode) activator.SwapMode { return m }

// finishSwap is the modelled Swap's end (plan 3), run on the attempt's
// goroutine while the simulation waits: the re-check against newest, then
// the build's health, then the switch. It is one transition: nothing else
// runs until done() has returned.
func (s *sim) finishSwap(sw *simSwap) activator.SwapResult {
	a := sw.a
	h := s.byName[a.Host]
	sw.serves = h.good == a.Build || (h.phase == simAwake && h.gen.Build == a.Build)
	sw.before = s.snap(a.Build)
	res := activator.SwapResult{Mode: a.Mode}
	if s.draining || sw.newest(a.Host) != a.SchedRev || s.r.Intn(30) == 0 {
		// Superseded, or a re-check that refused (a lost lease, a stop).
		res.Outcome, res.Reason = activator.SwapAbandoned, "superseded"
		if !s.draining && s.r.Intn(3) == 0 {
			// After the build's exec (SwapResult.Executed). A hold stopped the
			// old generation first, and the abandoned build is stopped too:
			// nothing serves. Not seen to fail, so nothing is recorded, but an
			// adoption's hold build ran against the database (N2).
			res.Executed = true
			if a.Mode == activator.ModeHold {
				h.phase, h.gen = simAsleep, activator.Gen{}
				if !sw.serves {
					s.holdRanOn(h, a.Build, sw.started)
				}
			}
		}
		s.logf("swap %s %s %s: abandoned (executed %v)", a.Host, a.Cause, a.Build, res.Executed)
		return res
	}
	res.Executed = true
	fresh := !h.served[a.Build]
	b := s.builds[a.Build]
	fails := false
	switch {
	case fresh:
		fails = b.bad || (b.flaky && s.r.Intn(2) == 0)
	case h.phase == simAwake && a.Build == h.gen.Build:
		fails = s.r.Intn(8) == 0 // a restart or config attempt that fails
	}
	if fails {
		res.Reason = "health check failed"
		if a.Mode == activator.ModeHold && s.r.Intn(4) != 0 {
			// The old generation was stopped first: nothing serves.
			res.Outcome = activator.SwapDown
			h.phase, h.gen = simAsleep, activator.Gen{}
		} else {
			res.Outcome = activator.SwapFailed
		}
		s.logf("swap %s %s %s (%s): %s", a.Host, a.Cause, a.Build, a.Mode, res.Outcome)
		if fresh {
			s.failure(a.Build, a.Host, sw.stamp, res.Outcome == activator.SwapDown, sw.started)
		} else if !sw.serves && !s.stale(a.Build, a.Host, sw.started) {
			// An adoption onto a build the instance served once, and serves no
			// more: not a failure the invariants count, but the code records
			// it (done's transition), so the model's own record follows it
			// (I9). It is no fact for its own host, unless it made a new
			// record: that holds every host of the app, so it is the app's
			// fact (simFact.app).
			if _, created := s.recFailed(a.Build, a.Host, res.Outcome == activator.SwapDown); created {
				s.addFact(simFact{build: a.Build, host: a.Host, kind: "an adoption failure", start: sw.started, app: true})
			}
		}
		if res.Outcome == activator.SwapFailed && a.Mode == activator.ModeHold && !sw.serves {
			// A hold adoption that ran, failed and was not held down.
			s.holdRanOn(h, a.Build, sw.started)
		}
		return res
	}
	if a.Cause == activator.CauseRelease || a.Cause == activator.CauseRetry {
		if want := s.installed[h.ch]; a.Build != want {
			violate("I4", "%s switched to %s (cause %s) while its installed build is %s", a.Host, a.Build, a.Cause, want)
		}
		if a.Build != h.good && !h.served[a.Build] && s.blockedFor(a.Host, a.Build) {
			violate("I4", "%s switched to %s (cause %s) after its failure on this box was recorded", a.Host, a.Build, a.Cause)
		}
	}
	g := s.newGen(release.StagedIdentity{Version: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand})
	h.phase, h.gen = simAwake, g
	s.goodAt(h, a.Build, sw.started)
	h.served[a.Build] = true
	res.Outcome, res.Gen = activator.SwapDone, g
	s.logf("swap %s %s %s (%s): done", a.Host, a.Cause, a.Build, a.Mode)
	return res
}

// await is the handoff: the one running attempt goroutine has parked in Swap
// or returned.
func (s *sim) await() simEvent {
	select {
	case ev := <-s.events:
		if ev.crash != nil {
			if v, ok := ev.crash.(simViolation); ok {
				panic(v)
			}
			if c, ok := ev.crash.(simCrash); ok {
				panic(c)
			}
			panic(simCrash{msg: fmt.Sprint(ev.crash), stack: ev.stack})
		}
		return ev
	case <-time.After(30 * time.Second):
		panic(simCrash{msg: "an attempt neither reached Swap nor returned within 30s"})
	}
}

// launch runs the real scheduler.execute for d up to its Swap (or to its
// return, for a staging refusal).
func (s *sim) launch(d dispatch) {
	stamp := s.stampFor(s.byName[d.ID.Host], d.ID.Build)
	go func() {
		defer func() {
			ev := simEvent{host: d.ID.Host}
			if r := recover(); r != nil {
				ev.crash, ev.stack = r, debug.Stack()
			}
			select {
			case s.events <- ev:
			case <-s.dead:
				// The seed ended (Swap's Goexit, or no one left to await).
			}
		}()
		s.sched.execute(context.Background(), d)
	}()
	ev := s.await()
	if ev.swap == nil {
		s.logf("launch %s %s %s: returned before Swap", d.ID.Host, d.ID.Cause, d.ID.Build)
		return
	}
	ev.swap.stamp = stamp
	// execute ran up to Swap within this step, so its dispatch time is now.
	ev.swap.started = s.now
	s.swaps[d.ID.Host] = ev.swap
	s.logf("launch %s %s %s rev %d canary %v class %d", d.ID.Host, d.ID.Cause, d.ID.Build, d.SchedRev, d.Canary, d.Class)
}

func (s *sim) run(steps int) {
	for s.step = 0; s.step < steps; s.step++ {
		s.now = s.now.Add(time.Duration(1+s.r.Intn(3)) * time.Second)
		s.diskFault()
		if s.gr != nil {
			// Before the step's own action, so an execution it starts begins
			// after a restart's pass and after every grant delivered here, as
			// the code's snapshot sees it.
			switch {
			case s.gr.Intn(150) == 0:
				s.agentRestart(true)
			case s.gr.Intn(40) == 0:
				s.agentRestart(false)
			}
			if s.rewake != "" {
				s.rewakeRun()
			} else if s.gr.Intn(6) == 0 {
				s.rewakeArm()
			}
			if s.gr.Intn(3) == 0 {
				s.deliverGrants()
			}
			if s.gr.Intn(4) == 0 {
				s.retryHeld()
			}
		}
		if hosts := sortedKeys(s.late); len(hosts) > 0 && s.r.Intn(12) == 0 {
			s.deliverLate(hosts[s.r.Intn(len(hosts))])
		}
		switch k := s.r.Intn(26); {
		case k < 2:
			s.promote()
		case k < 4:
			s.pass()
		case k < 10:
			s.tick()
		case k < 14:
			s.complete()
		case k < 16:
			s.wakeBegin()
		case k < 18:
			s.wakeExec()
		case k < 19:
			s.observeEnd()
		case k < 20:
			s.sleep()
		case k < 21:
			s.stopEnd()
		case k < 22:
			s.restart()
		case k < 23:
			s.config()
		case k < 24:
			s.pointerRetry()
		case k < 25:
			s.hostRetry()
		default:
			s.restage()
		}
		s.checkI2()
		s.checkI7()
		s.checkI10()
	}
	s.draining = true
	for _, host := range sortedKeys(s.swaps) {
		sw := s.swaps[host]
		delete(s.swaps, host)
		close(sw.release)
		s.await()
	}
	for _, host := range sortedKeys(s.late) {
		s.deliverLate(host)
	}
	s.checkIntents()
}

func sortedKeys[V any](m map[string]V) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

func (s *sim) pick(ok func(h *simHost) bool) *simHost {
	var c []*simHost
	for _, h := range s.hosts {
		if ok(h) {
			c = append(c, h)
		}
	}
	if len(c) == 0 {
		return nil
	}
	return c[s.r.Intn(len(c))]
}

// pass is the agent pass (Manager.Pass) over every row. The only marker a
// Pass may clear is one a signed rollback lifts (applyPointers, ruling 205),
// so each marker it cleared is checked against the model's evidence, then
// lifted in the model too.
//
// A record the Pass cleared while a row still installs its build was cleared
// by a re-signed pointer (a prune drops only builds no row installs): that
// is the retry ordering's app-wide clear (ruling 191), noted as clearedAt.
// (Read from the book like the markers; the model's stands rule already
// says whether it may go.)
func (s *sim) pass() {
	marked := map[string]bool{}
	for _, h := range s.hosts {
		if _, ok := s.book.HoldRanOn(h.name); ok {
			marked[h.name] = true
		}
	}
	recorded := map[string]bool{}
	for _, r := range s.book.List() {
		recorded[r.Build] = true
	}
	rs, _ := s.rows.List()
	s.m.Pass(rs)
	installed := map[string]bool{}
	for _, r := range rs {
		installed[r.Version] = true
	}
	still := map[string]bool{}
	for _, r := range s.book.List() {
		still[r.Build] = true
	}
	// The model's own records (I9) leave with the book's: a pointer's clear
	// or a prune, a clear of the record for every host.
	for _, b := range sortedKeys(s.grec) {
		if !still[b] {
			delete(s.grec, b)
			s.addClear(simClear{build: b, kind: "the record (a pointer's clear or a prune)"})
		}
	}
	for _, b := range sortedKeys(recorded) {
		if still[b] {
			continue
		}
		// The record is gone, cleared or pruned, and with it every host
		// retry's exemption from it: the build's next failure writes a new
		// record, which holds every host (spec §8.4).
		for _, host := range sortedKeys(s.exempt) {
			delete(s.exempt[host], b)
		}
		if installed[b] {
			s.clearedAt[b] = s.now
			s.logf("ground truth: a pass cleared %s's record on a re-signed pointer", b)
		}
	}
	// The Pass's prune drops the retry clears of builds no row installs and
	// no planned first wake still carries (the hosts waking on a first-wake
	// plan: Manager.planned, once a pass has let go of the unowned ones). A
	// failure arriving after is fresh again (Task 4 fix round 1).
	for _, h := range s.hosts {
		if h.phase == simWaking && h.plan.FirstWake {
			installed[h.plan.Boot.Version] = true
		}
	}
	for _, b := range sortedKeys(s.clearedAt) {
		if !installed[b] {
			delete(s.clearedAt, b)
		}
	}
	for _, host := range sortedKeys(s.retryAt) {
		for _, b := range sortedKeys(s.retryAt[host]) {
			if !installed[b] {
				delete(s.retryAt[host], b)
			}
		}
	}
	for _, h := range s.hosts {
		if _, still := s.book.HoldRanOn(h.name); !marked[h.name] || still {
			continue
		}
		if mark := s.holdRan[h.name]; mark != "" {
			p, ok := s.pointers[s.channels[h.ch]]
			if h.good == "" || s.installed[h.ch] != h.good || !ok || p.Version != h.good || !p.PromotedAt.After(s.holdRanAt[h.name]) {
				violate("I5", "a pass cleared %s's hold-ran restriction (hold build %s ran at %s) with no signed rollback onto its good build %q (installed %s, pointer %+v)",
					h.name, mark, s.holdRanAt[h.name].Sub(t0), h.good, s.installed[h.ch], p)
			}
			delete(s.holdRan, h.name)
			s.addClear(simClear{host: h.name, marker: true, kind: "a signed rollback's lift"})
			s.logf("ground truth: a signed rollback onto %s lifted %s's hold-ran restriction (%s)", h.good, h.name, mark)
		}
	}
}

// promote is a promotion and root's install onto one channel's hosts, then
// the agent pass that sees it: a fresh build, or the other channel's build
// promoted here (which signs a newer pointer for it, spec rev 14 §8.4).
func (s *sim) promote() {
	ch := s.r.Intn(2)
	var b string
	if other := s.installed[1-ch]; s.r.Intn(4) == 0 && other != s.installed[ch] {
		b = other
	} else {
		s.nb++
		b = "b" + strconv.Itoa(s.nb)
		s.addBuild(b, &simBuild{bad: s.r.Intn(4) == 0, flaky: s.r.Intn(5) == 0, holdDemand: s.r.Intn(6) == 0, misstaged: s.r.Intn(10) == 0})
	}
	s.installed[ch] = b
	s.sign(s.channels[ch], b)
	for _, h := range s.hosts {
		if h.ch == ch {
			row := s.rows.get(h.name)
			row.Version = b
			s.rows.put(row)
		}
	}
	s.logf("promote %s onto channel %q %+v", b, s.channels[ch], *s.builds[b])
	s.pass()
}

// tick is scheduler.loop's body: the view, configSlept, admit, and the
// attempts it hands out run up to their Swap.
func (s *sim) tick() {
	view, pendingConfig := s.sched.view()
	for _, host := range sortedKeys(pendingConfig) {
		if serving, seen := view[host]; seen && serving == "" {
			s.m.configSlept(host, pendingConfig[host])
		}
	}
	s.sched.mu.Lock()
	ds := s.sched.q.admit(view, s.now)
	s.sched.mu.Unlock()
	// Liveness is judged on the admission itself, before any attempt runs:
	// one that returns at once (a staging refusal) frees its slot after
	// admit stopped at the limit, and the loop's next pass admits the rest.
	s.checkI2()
	s.checkI3Liveness(view)
	for _, d := range ds {
		s.launch(d)
	}
	s.checkI2()
}

func (s *sim) checkI2() {
	s.sched.mu.Lock()
	running := s.sched.q.running
	s.sched.mu.Unlock()
	if len(s.swaps) > s.concurrency || running > s.concurrency {
		violate("I2", "%d attempts in Swap (%d running in the queue) with swap_concurrency %d", len(s.swaps), running, s.concurrency)
	}
}

// checkI3Liveness runs after an admission that left capacity free: every
// awake instance's restart or config attempt onto the build it serves was
// admitted, unless it waits out its own deferral or its own cohort's
// unsettled canary on another host. A failure record (of the build it serves
// or of the one it wants) never holds it.
func (s *sim) checkI3Liveness(view map[string]string) {
	q := s.sched.q
	s.sched.mu.Lock()
	defer s.sched.mu.Unlock()
	if q.running >= s.concurrency {
		return
	}
	for _, h := range s.hosts {
		if serving, seen := view[h.name]; !seen || serving == "" {
			continue
		}
		slot := q.hosts[h.name]
		if slot == nil || slot.running != nil {
			continue
		}
		d, ok := q.pending(slot)
		if !ok || d.ID.Build != slot.serving {
			continue
		}
		if s.now.Before(slot.notBefore) || (d.ID.Cause == activator.CauseConfig && s.now.Before(slot.configAfter)) {
			continue
		}
		if c := q.cohorts[cohortKey(slot.serving, classRecover)]; c != nil && c.canary != "" && c.canary != h.name && !c.settled {
			continue
		}
		violate("I3", "%s's %s attempt onto %s, the build it serves, was held back with capacity free (cohort %s)", h.name, d.ID.Cause, d.ID.Build, d.Cohort)
	}
}

// complete lets one parked attempt finish: the modelled Swap's end, then the
// real done().
func (s *sim) complete() {
	hosts := sortedKeys(s.swaps)
	if len(hosts) == 0 {
		return
	}
	host := hosts[s.r.Intn(len(hosts))]
	sw := s.swaps[host]
	delete(s.swaps, host)
	close(sw.release)
	if ev := s.await(); ev.swap != nil {
		panic(simCrash{msg: "a second attempt reached Swap while " + host + "'s finished"})
	}
	if sw.serves && s.snap(sw.a.Build) != sw.before {
		violate("I3", "%s's %s attempt onto %s, a build it serves, changed its failure record: %s -> %s", host, sw.a.Cause, sw.a.Build, sw.before, s.snap(sw.a.Build))
	}
}

// heldDownOn reports whether the model let host be held down on build: a
// hold build that ran and failed there, with no host retry since.
func (s *sim) heldDownOn(host, build string) bool {
	return s.heldDown[host][build]
}

// holdRanOn is the ground truth of a hold build that ran on h without
// passing or being held down there (N2), in the execution that started at
// start. Each execution renews the restriction (plan review finding 4): its
// stamp is the latest start of the runs since it was last lifted, whatever
// their builds (Task 4 fix round 1), so a rollback signed between two runs
// lifts nothing, and it is set again now, so a good write decided before now
// lifts nothing either.
func (s *sim) holdRanOn(h *simHost, build string, start time.Time) {
	if s.holdRan[h.name] == "" || start.After(s.holdRanAt[h.name]) {
		s.holdRanAt[h.name] = start
	}
	if old := s.holdRan[h.name]; old != "" && old != build {
		// The marker now names build: old's restriction of h is gone (I9).
		s.addClear(simClear{build: old, host: h.name, marker: true, kind: "a marker replaced"})
	}
	s.holdRan[h.name] = build
	s.holdRanSet[h.name] = s.now
	// Every run is a new execution, which sets or renews the marker (I9).
	s.addFact(simFact{build: build, host: h.name, kind: "a hold run", start: start, marker: true})
	s.logf("ground truth: hold build %s ran on %s without a verdict", build, h.name)
}

// goodAt is h recording build as its good build (the code's good-build
// write) for the execution that started at start. It lifts a hold-ran
// restriction set before that start, and only such a one (Astra R3 N6): a
// hold build that ran on the host after the passing execution began is not
// one the instance has passed over.
func (s *sim) goodAt(h *simHost, build string, start time.Time) {
	h.good = build
	if s.holdRanSet[h.name].Before(start) {
		if s.holdRan[h.name] != "" {
			s.addClear(simClear{host: h.name, marker: true, kind: "a good write's lift"})
		}
		delete(s.holdRan, h.name)
	}
}

// checkI5 is I5 at the exec of build on h, for a wake planned against rt.
// The installed good build is barred too (ruling 205): root's sibling
// repoint installs it without any rollback, and only a Pass that saw a
// signed one lifts the restriction. The marked build itself is exempt even
// when it is the good build (Task 3 fix round 1): booting it is the
// marked-build rule, not a boot of the good build in its place.
func (s *sim) checkI5(h *simHost, rt registry.Route, build string) {
	if mark := s.holdRan[h.name]; mark != "" && build == h.good && build != mark {
		violate("I5", "%s boots its good build %s (installed %s) after hold build %s ran there, with no signed rollback since", h.name, build, rt.Version, mark)
	}
}

// bootable is the plan's Check for build at its exec (planGate): nil when
// the exec may go ahead.
func bootable(plan activator.WakePlan, build string) error {
	if plan.Check == nil {
		return nil
	}
	return plan.Check(build)
}

// wakeBegin is a wake's planning (Manager.Plan, as the activator calls it
// before the lease).
func (s *sim) wakeBegin() {
	if h := s.pick(func(h *simHost) bool { return h.phase == simAsleep }); h != nil {
		s.wakeBeginOn(h)
	}
}

// wakeBeginOn is wakeBegin for h, which is asleep.
func (s *sim) wakeBeginOn(h *simHost) {
	// The activator's outcome fence (awaitOutcomes): an earlier wake's
	// events reach the agent before the host plans again.
	if _, ok := s.late[h.name]; ok {
		s.deliverLate(h.name)
	}
	row := s.rows.get(h.name)
	h.phase = simWaking
	plan := s.m.Plan(context.Background(), row)
	if plan.Refuse != "" || plan.Boot.StagedPath == "" {
		h.phase = simAsleep
		s.logf("wake %s: refused at planning: %s", h.name, plan.Refuse)
		s.checkWakeRefused(h, row.Version, "its plan refused: "+plan.Refuse)
		return
	}
	h.plan, h.planRoute, h.wakeStart, h.planAt, h.stamp = plan, row, s.now, s.now, s.stampFor(h, plan.Boot.Version)
	h.underMark = s.holdRan[h.name] != ""
	s.logf("wake %s planned: boot %s first %v fallback %v mode %s", h.name, plan.Boot.Version, plan.FirstWake, plan.Fallback != nil, plan.Mode)
	s.checkPlanned(h, plan.Boot.Version)
}

func (s *sim) checkWakeRefused(h *simHost, installed, why string) {
	if h.good == "" || s.heldDownOn(h.name, installed) {
		return
	}
	if s.fault && strings.Contains(why, BookUnreadableRefusal) {
		// A first wake whose intent could not be saved boots nothing (Task 5):
		// its failure could not outlive a restart (I8).
		return
	}
	if mark := s.holdRan[h.name]; mark != "" {
		if installed == mark && installed == h.good {
			// The marked build is the installed good build (Task 3 fix round
			// 1): it boots under the marked-build rule. A record does not
			// hold a build the instance served, and the hold-down returned
			// above, so only a staging refusal may refuse it.
			if s.builds[installed].misstaged {
				return
			}
			violate("I1", "%s has its good build %s installed, the hold build that ran there, but its wake boots nothing: %s", h.name, installed, why)
		}
		// Restricted (N2): refused for an installed build held for the
		// instance, the marked build included (Astra R3 N7: the marker lifts
		// no record), for its installed good build until a signed rollback
		// lifts the restriction (ruling 205; pass checks the lifting), or for
		// a build not staged correctly. "Held" is read loosely here (a
		// failure of it was recorded at some point and no host retry exempts
		// the instance): the book clears a record only at its next due
		// pointer read, after the model's re-sign.
		held := s.failedEver[installed] && !s.exempt[h.name][installed]
		if held || installed == h.good || s.builds[installed].misstaged {
			return
		}
		violate("I1", "%s (good build %s, hold build %s ran there) has %s installed but its wake boots nothing: %s", h.name, h.good, mark, installed, why)
	}
	violate("I1", "%s has good build %s but its wake boots nothing: %s", h.name, h.good, why)
}

// checkBoot is I1 at the exec of build on h. A hold-ran restriction exempts
// nothing (Astra R3 N7): spec §8.4 exempts an instance that served the
// build, not one where it merely ran.
func (s *sim) checkBoot(h *simHost, build string) {
	if h.good != "" && build != h.good && !h.served[build] && s.blockedFor(h.name, build) {
		violate("I1", "%s (good build %s) boots %s, whose failure on this box stands", h.name, h.good, build)
	}
}

// checkPlanned is I1 at a wake's plan, its first gate: the planner never
// names a build whose failure on this box stands for an instance with a
// good build (decideWake, and decideHoldRan for the marked build, Astra R3
// N7). The plan's Check at the exec is the second gate, for records written
// since the plan, and is not what keeps a held build out of a plan.
func (s *sim) checkPlanned(h *simHost, build string) {
	if h.good != "" && build != h.good && !h.served[build] && s.blockedFor(h.name, build) {
		violate("I1", "%s (good build %s) plans a wake of %s, whose failure on this box stands", h.name, h.good, build)
	}
}

// wakeExec is the activator's checkPlan and exec (spec §7.4, §8.2): the
// plan's Check, one replan if it refuses, the verdict floor, and the build's
// outcome reported through the real OnGen.
func (s *sim) wakeExec() {
	if h := s.pick(func(h *simHost) bool { return h.phase == simWaking }); h != nil {
		s.wakeExecOn(h, "")
	}
}

// wakeExecOn is wakeExec for h, which is waking. A first wake of failBuild
// that runs fails its health check at once, as a flaky build may (rewakeRun's
// scenario); "" fails nothing more than the build itself does.
func (s *sim) wakeExecOn(h *simHost, failBuild string) {
	rt, plan := h.planRoute, h.plan
	if s.r.Intn(20) == 0 {
		// The lease was refused (a previous incarnation still holds it): the
		// wake ends before any exec. The activator reports it as cancelled
		// with nothing run (Task 5), which ends the plan's registration and
		// its intent. Not a build failure, so no I1 check.
		h.phase = simAsleep
		s.logf("wake %s: refused at its lease, cancelled with nothing run", h.name)
		s.cancelUnrun(rt, plan)
		return
	}
	if plan.Check != nil && plan.Check(plan.Boot.Version) != nil {
		next := s.m.Plan(context.Background(), rt)
		if next.Refuse != "" || next.Boot.StagedPath == "" || (next.Check != nil && next.Check(next.Boot.Version) != nil) {
			h.phase = simAsleep
			s.logf("wake %s: abandoned at exec (plan stale, replan %q)", h.name, next.Refuse)
			// The activator's checkPlan cancels a replan its own check
			// refused, then afterWake the wake's plan (Task 5): neither ran.
			if next.Refuse == "" && next.Boot.StagedPath != "" {
				s.cancelUnrun(rt, next)
			}
			s.cancelUnrun(rt, plan)
			// The replan read the row afresh (Plan replans a moved row).
			s.checkWakeRefused(h, s.rows.get(h.name).Version, "the plan went stale and the replan refused: "+next.Refuse)
			return
		}
		// The replan replaces the plan, which checkPlan cancels: it never ran.
		s.cancelUnrun(rt, plan)
		plan, h.plan, h.planAt, h.stamp = next, next, s.now, s.stampFor(h, next.Boot.Version)
		h.underMark = s.holdRan[h.name] != ""
		s.logf("wake %s replanned: boot %s first %v", h.name, plan.Boot.Version, plan.FirstWake)
		s.checkPlanned(h, plan.Boot.Version)
	}
	canFall := plan.FirstWake && plan.Fallback != nil && plan.Mode == activator.ModeOverlap
	if left := h.wakeStart.Add(simHealthWindow).Sub(s.now); canFall && left < simMinVerdict {
		// The verdict floor (spawnPlanned): B's health window ends 30 s
		// after the wake's start (55 s budget less the 25 s kill reserve),
		// and with less than 20 s of it left at the exec, the good build
		// boots untried and nothing is recorded against B.
		good := *plan.Fallback
		if err := bootable(plan, good.Version); err != nil {
			// The good build's own gate refused it at the exec: nothing ran.
			h.phase = simAsleep
			s.logf("wake %s: %s untried, and the good build %s was refused at its exec: %v", h.name, plan.Boot.Version, good.Version, err)
			s.cancelUnrun(rt, plan)
			s.checkWakeRefused(h, rt.Version, "the untried first wake's good build was refused at its exec")
			return
		}
		s.checkBoot(h, good.Version)
		s.checkI5(h, rt, good.Version)
		h.phase, h.gen = simAwake, s.newGen(good)
		h.served[good.Version] = true
		s.logf("wake %s: %s untried, booted the good build %s", h.name, plan.Boot.Version, good.Version)
		// served: the good build is written again, when the event lands.
		s.healthy(h, activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})
		return
	}
	boot := plan.Boot
	s.checkBoot(h, boot.Version)
	s.checkI5(h, rt, boot.Version)
	marked := h.underMark || s.holdRan[h.name] != ""
	if h.good != "" && plan.FirstWake && !s.expectHold(boot.Version) && (plan.Fallback == nil || plan.Fallback.Version != h.good) && !marked {
		violate("I1", "%s (good build %s) first-wakes %s in overlap mode without its good build behind it", h.name, h.good, boot.Version)
	}
	if !plan.FirstWake {
		h.phase, h.gen = simAwake, s.newGen(boot)
		h.served[boot.Version] = true
		s.logf("wake %s: booted %s", h.name, boot.Version)
		s.healthy(h, activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})
		return
	}
	if s.r.Intn(12) == 0 {
		// The wake's lifecycle ended it after the build's exec (a stop, a
		// fence, a step after the spawn): no verdict. The activator reports
		// it as cancelled after a run (GenCancelled, Executed): a hold one
		// leaves the hold-ran restriction (N2), an overlap one only ends the
		// wake (Task 5).
		h.phase, h.gen = simAsleep, activator.Gen{}
		s.logf("wake %s: first wake of %s ended after its exec, no verdict (mode %s)", h.name, boot.Version, plan.Mode)
		if plan.Mode == activator.ModeHold {
			s.holdRanOn(h, boot.Version, h.planAt)
		}
		s.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: rt, Gen: activator.Gen{Build: boot.Version}, Mode: plan.Mode,
			FirstWake: true, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev, Executed: true})
		return
	}
	b := s.builds[boot.Version]
	forced := boot.Version == failBuild
	fails := forced || b.bad || (b.flaky && s.r.Intn(2) == 0)
	if fails && !forced && s.r.Intn(2) == 0 {
		// It passed its health check and exits inside its observation.
		fails, h.failLater = false, true
	} else {
		h.failLater = false
	}
	switch {
	case !fails:
		h.phase, h.gen = simObserving, s.newGen(boot)
		s.logf("wake %s: first wake of %s healthy, observing", h.name, boot.Version)
		s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})
	case canFall && bootable(plan, plan.Fallback.Version) != nil:
		// B failed, and its fallback's own gate refused the good build at
		// its exec: B's failure is recorded, nothing fell back.
		s.failure(boot.Version, h.name, h.stamp, false, h.planAt)
		h.phase, h.gen = simAsleep, activator.Gen{}
		s.logf("wake %s: first wake of %s failed, its fallback %s refused at its exec", h.name, boot.Version, plan.Fallback.Version)
		s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: boot, Mode: plan.Mode,
			FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})
		if h.good != "" && s.holdRan[h.name] == "" {
			violate("I1", "%s (good build %s) failed a first wake of %s and its fallback was refused with no hold-ran restriction", h.name, h.good, boot.Version)
		}
	case canFall:
		s.checkI5(h, rt, plan.Fallback.Version)
		s.failure(boot.Version, h.name, h.stamp, false, h.planAt)
		h.phase, h.gen = simAwake, s.newGen(*plan.Fallback)
		s.logf("wake %s: first wake of %s failed, fell back to %s", h.name, boot.Version, plan.Fallback.Version)
		s.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, Gen: h.gen, Failed: boot, Mode: plan.Mode,
			FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})
	default:
		hold := plan.Mode == activator.ModeHold
		s.failure(boot.Version, h.name, h.stamp, hold, h.planAt)
		h.phase, h.gen = simAsleep, activator.Gen{}
		s.logf("wake %s: first wake of %s failed, nothing fell back (mode %s)", h.name, boot.Version, plan.Mode)
		s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: boot, Mode: plan.Mode,
			FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})
		if h.good != "" && !hold && !marked {
			violate("I1", "%s (good build %s) failed a first wake of %s in overlap mode and nothing fell back", h.name, h.good, boot.Version)
		}
	}
}

// cancelUnrun is the activator's report of a planned wake that ended before
// its build ran (Task 5): refused at its lease or its exec, or replaced by a
// replan. It records nothing, and ends the plan's registration and intent.
func (s *sim) cancelUnrun(rt registry.Route, plan activator.WakePlan) {
	s.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: rt, Mode: plan.Mode, FirstWake: plan.FirstWake, At: s.now,
		Token: plan.Token, ConfigRev: plan.ConfigRev})
}

// healthy reports an ordinary wake's healthy event (not a first wake's),
// which writes the good build (served). Half of them are delivered late
// (sim.late): the activator serves the generation at once, but its event,
// and with it the good write, reaches the agent some steps later (OnGen
// takes a late event as a matter of course). The good write is then
// decided by a wake whose execution began before whatever ran on the host
// meanwhile: an adoption, admitted because the generation serves, whose
// hold build ran and failed leaves a marker the write must not clear (Astra
// R3 N6, ClearHoldRanBefore).
func (s *sim) healthy(h *simHost, ev activator.GenEvent) {
	s.late[h.name] = simLate{ev: ev, start: h.planAt}
	if s.r.Intn(2) == 0 {
		s.logf("wake %s: its healthy event is late", h.name)
		return
	}
	s.deliverLate(h.name)
}

// deliverLate delivers host's healthy event. The good write follows only
// while the event's generation is the one serving (Manager.acknowledged),
// and it lifts only a restriction set before the wake's execution began
// (goodAt).
func (s *sim) deliverLate(host string) {
	l, ok := s.late[host]
	if !ok {
		return
	}
	delete(s.late, host)
	h := s.byName[host]
	s.logf("wake %s: healthy event of %s delivered", host, l.ev.Gen.Build)
	s.goodWrite(h, l.ev.Gen.Build, l.start, h.phase == simAwake && h.gen.N == l.ev.Gen.N, func() { s.m.OnGen(l.ev) })
}

// goodWrite is a good-build write of build on h by an execution that began
// at start (when acked: the code writes only while the generation serves),
// made by deliver, the real event. It is I5 at the write: a write may lift
// only a restriction set before its execution began (goodAt), so a marker
// that stood in the book before the event and is gone after it, while the
// model's restriction stands, was cleared by a write decided before its
// hold build ran (Astra R3 N6). The book is read only to see what the write
// cleared, as pass reads it for a Pass.
func (s *sim) goodWrite(h *simHost, build string, start time.Time, acked bool, deliver func()) {
	_, had := s.book.HoldRanOn(h.name)
	if acked {
		s.goodAt(h, build, start)
	}
	deliver()
	if _, still := s.book.HoldRanOn(h.name); had && !still && s.holdRan[h.name] != "" {
		violate("I5", "%s's good write of %s, by an execution begun at %s, cleared the hold-ran restriction of %s, set at %s after it began",
			h.name, build, start.Sub(t0), s.holdRan[h.name], s.holdRanSet[h.name].Sub(t0))
	}
}

// checkIntents is I6 once the run has drained (Task 5): every execution's
// intent ends with its outcome, so the only intents left are those of first
// wakes still in flight (planned or under observation), each the host's
// current plan's. A leaked intent would refuse the host's adoptions, and an
// agent restart would turn it into a failure nothing failed.
func (s *sim) checkIntents() {
	s.book.mu.Lock()
	intents := maps.Clone(s.book.intents)
	s.book.mu.Unlock()
	for _, host := range sortedKeys(intents) {
		in, h := intents[host], s.byName[host]
		inFlight := h != nil && (h.phase == simWaking || h.phase == simObserving) && h.plan.FirstWake && in.ID == s.m.wakeExecID(h.plan.Token)
		if !inFlight {
			violate("I6", "%s's intent %+v outlived its execution", host, in)
		}
	}
}

// observeEnd ends a first wake's observation: it settles, or the build exits.
func (s *sim) observeEnd() {
	h := s.pick(func(h *simHost) bool { return h.phase == simObserving })
	if h == nil {
		return
	}
	rt, plan := h.planRoute, h.plan
	if h.failLater {
		s.failure(plan.Boot.Version, h.name, h.stamp, plan.Mode == activator.ModeHold, h.planAt)
		h.phase, h.gen = simAsleep, activator.Gen{}
		s.logf("wake %s: %s exited inside its observation", h.name, plan.Boot.Version)
		s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true,
			Reason: "exited within 10s of becoming healthy", At: s.now, FailedUnder: plan.FailedUnder, Token: plan.Token, ConfigRev: plan.ConfigRev})
		return
	}
	h.phase = simAwake
	h.served[h.gen.Build] = true
	s.logf("wake %s: %s settled", h.name, h.gen.Build)
	s.goodWrite(h, h.gen.Build, h.planAt, true, func() {
		s.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now, Token: plan.Token, ConfigRev: plan.ConfigRev})
	})
}

// sleep hibernates an instance with no attempt in Swap: at once, or through a
// stopping phase the view leaves out. An observation it ends is cancelled.
func (s *sim) sleep() {
	if h := s.pick(func(h *simHost) bool {
		_, inSwap := s.swaps[h.name]
		return !inSwap && (h.phase == simAwake || h.phase == simObserving)
	}); h != nil {
		s.sleepOn(h)
	}
}

// sleepOn is sleep for h, which is awake or observing with no attempt in
// Swap.
func (s *sim) sleepOn(h *simHost) {
	observing := h.phase == simObserving
	g := h.gen
	h.phase, h.gen = simAsleep, activator.Gen{}
	if s.r.Intn(2) == 0 {
		h.phase = simStopping
	}
	s.logf("sleep %s (stopping phase %v, observation cancelled %v)", h.name, h.phase == simStopping, observing)
	if observing {
		// The observed build was healthy, so it ran: a hold one leaves the
		// hold-ran restriction (N2).
		if h.plan.Mode == activator.ModeHold {
			s.holdRanOn(h, g.Build, h.planAt)
		}
		s.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.planRoute, Gen: g, Mode: h.plan.Mode, FirstWake: true, At: s.now,
			Token: h.plan.Token, ConfigRev: h.plan.ConfigRev, Executed: true})
	}
}

func (s *sim) stopEnd() {
	if h := s.pick(func(h *simHost) bool { return h.phase == simStopping }); h != nil {
		h.phase = simAsleep
		s.logf("stopped %s", h.name)
	}
}

func (s *sim) restart() {
	h := s.hosts[s.r.Intn(len(s.hosts))]
	s.nreq++
	id := "r" + strconv.Itoa(s.nreq)
	ok := s.m.Restart(h.name, id)
	s.logf("restart %s %s: %v", h.name, id, ok)
}

func (s *sim) config() {
	h := s.hosts[s.r.Intn(len(s.hosts))]
	s.nreq++
	rev := "c" + strconv.Itoa(s.nreq)
	s.m.ConfigDelivered(h.name, rev)
	s.logf("config %s %s", h.name, rev)
}

// pointerRetry is `carlos promote --retry`: the channel's head is signed again
// with a newer PromotedAt. The next due agent pass clears the record.
func (s *sim) pointerRetry() {
	ch := s.r.Intn(2)
	b := s.installed[ch]
	if !s.stands(b) {
		return
	}
	s.sign(s.channels[ch], b)
	s.logf("promote --retry %s on channel %q", b, s.channels[ch])
}

// hostRetry is `carlos instances retry <host>` for a host whose installed
// build's failure stands.
func (s *sim) hostRetry() {
	h := s.hosts[s.r.Intn(len(s.hosts))]
	if s.gr != nil {
		// A seed whose book is on disk queues the grant (I9): delivered at
		// later steps, again and again (deliverGrants).
		s.issueGrant(h, "")
		return
	}
	b := s.installed[h.ch]
	if !s.stands(b) {
		return
	}
	// The retry orders later failures only when it un-retries the host (Task
	// 4 fix round 1): a record of b stands and does not already exempt h.
	// Whether the code's record still lists h as retried is read from the
	// book, like the clears pass() sees: the model's exempt outlives a record
	// that was cleared and written again.
	rec, recorded := s.book.Lookup("bab", "blog", b)
	clears := recorded && !slices.Contains(rec.Retried, h.name)
	// The console's grant names the build and the generation the box
	// publishes for it (Task 8; readiness round 2 Task 1): the sim reads it
	// just as the console would, and delivers it at once. This book is in
	// memory, so it never owes a save; the seeds whose book is on disk queue
	// their grants instead (issueGrant, deliverGrants).
	epoch := s.book.PublishedEpoch("bab", "blog", b, h.name)
	ok := s.m.RetryHost(h.name, b, s.book.ID(), epoch)
	// The retry exempts h from the record that stands now, and from no
	// record written after that one goes (pass): with no record there is
	// nothing to exempt it from.
	if recorded {
		if s.exempt[h.name] == nil {
			s.exempt[h.name] = map[string]bool{}
		}
		s.exempt[h.name][b] = true
	}
	if clears {
		if s.retryAt[h.name] == nil {
			s.retryAt[h.name] = map[string]time.Time{}
		}
		s.retryAt[h.name][b] = s.now
	}
	delete(s.heldDown[h.name], b)
	// A grant lifts the marker only when the marker names its build (plan
	// review finding 5).
	if s.holdRan[h.name] == b {
		delete(s.holdRan, h.name)
	}
	s.logf("instances retry %s onto %s: %v", h.name, b, ok)
}

// heldBuilds are the builds that may hold h as status shows it: the
// installed build of h's channel, and its hold-ran marker's build.
func (s *sim) heldBuilds(h *simHost) []string {
	builds := []string{s.installed[h.ch]}
	if m, ok := s.book.HoldRanOn(h.name); ok && m.Build != builds[0] {
		builds = append(builds, m.Build)
	}
	return builds
}

// retryHeld is an operator pressing Retry on a held instance, on a seed
// whose book is on disk: one (host, build) the box publishes a generation
// for, at random, gets a grant (issueGrant).
func (s *sim) retryHeld() {
	var held [][2]string
	for _, h := range s.hosts {
		for _, b := range s.heldBuilds(h) {
			if s.published(b, h.name) != 0 {
				held = append(held, [2]string{h.name, b})
			}
		}
	}
	if len(held) == 0 {
		return
	}
	p := held[s.gr.Intn(len(held))]
	s.issueGrant(s.byName[p[0]], p[1])
}

// rewakeArm and rewakeRun, on a seed whose book is on disk, are an operator
// pressing Retry on an instance its hold-ran marker holds (asleep, the
// marker's build X still installed), and the instance then woken again and
// stopped before that wake settles, before the grant is read. The wake
// boots X under the marker as a hold first wake (decideHoldRan): a run of
// X that ends with no verdict renews the marker as a new execution's, and
// a run that fails holds it down. Either is a fact of (X, host) after the
// grant was issued, which must leave the grant clearing nothing (I9; M14's
// witness: a renewal that moved no generation). The random schedule meets
// this only rarely, as an instance is seldom woken again under its marker
// before another restriction of its app moves the generation; here the two
// halves are one step apart.
//
// When X has no failure record yet, the wake is instead another asleep
// instance's on the same channel, one that never served X, when there is one,
// and its first wake of X, if it runs, fails its health check (rewakeFail,
// as a flaky build may: a build with a marker and no record has seldom
// failed anywhere). That failure writes X's first record, which holds every
// host of the app, the granted one too (spec §8.4): a fact of every host
// after the grant was issued (M17's witness: a new record that moved no
// generation). The book is read here only to pick the scenario, never for
// a verdict.
//
// rewakeArm issues the grant and remembers the host to wake.
func (s *sim) rewakeArm() {
	var c []*simHost
	for _, h := range s.hosts {
		m, ok := s.book.HoldRanOn(h.name)
		if h.phase == simAsleep && ok && m.Build == s.installed[h.ch] && s.published(m.Build, h.name) != 0 {
			c = append(c, h)
		}
	}
	if len(c) == 0 {
		return
	}
	h := c[s.gr.Intn(len(c))]
	x := s.installed[h.ch]
	s.issueGrant(h, x)
	s.rewake = h.name
	if _, recorded := s.book.Lookup("bab", "blog", x); !recorded {
		var o []*simHost
		for _, h2 := range s.hosts {
			if h2 != h && h2.ch == h.ch && h2.phase == simAsleep && !h2.served[x] {
				o = append(o, h2)
			}
		}
		if len(o) > 0 {
			s.rewake, s.rewakeFail = o[s.gr.Intn(len(o))].name, x
		}
	}
	s.logf("rewake %s: retry pressed under its hold-ran marker; %s wakes next", h.name, s.rewake)
}

// rewakeRun wakes the host rewakeArm chose, if it is still asleep: its plan
// and exec, then its stop while the first wake is under observation. Then
// the agent reads its grants.
func (s *sim) rewakeRun() {
	h, fail := s.byName[s.rewake], s.rewakeFail
	s.rewake, s.rewakeFail = "", ""
	if h.phase != simAsleep {
		return
	}
	s.wakeBeginOn(h)
	if h.phase == simWaking {
		s.wakeExecOn(h, fail)
	}
	if _, inSwap := s.swaps[h.name]; h.phase == simObserving && !inSwap {
		s.sleepOn(h)
	}
	s.deliverGrants()
}

// issueGrant is the console signing `carlos instances retry` for h onto
// build b ("" picks one of heldBuilds), on a seed whose book is on disk:
// it reads, as status does, the (book, generation) the box publishes for
// b's restriction of h, and queues a grant for it, stamped with the step.
// A build the box publishes no generation for gets no grant (the console
// refuses the retry).
func (s *sim) issueGrant(h *simHost, b string) {
	if b == "" {
		builds := s.heldBuilds(h)
		b = builds[s.gr.Intn(len(builds))]
	}
	epoch := s.published(b, h.name)
	if epoch == 0 {
		return
	}
	g := simGrant{build: b, host: h.name, book: s.book.ID(), epoch: epoch, step: s.step, at: s.now, cap: s.tseq}
	// Pending for the rest of the run: a grant expires after 24 h, and a
	// run is minutes long.
	s.grants = append(s.grants, g)
	s.logf("grant: instances retry %s onto %s issued (book %.8s, generation %d, transition %d)", h.name, b, g.book, epoch, g.cap)
}

// deliverGrants is the agent reading its pending grants (internal/host,
// host-retry.json): each one, oldest first, through RetryHost, which applies
// it with RetryHostAt. A grant spent or stale clears nothing (the plan's
// ruling: grants are never skipped, only refused), so every pending one is
// delivered again at every read, before and after restarts and rebuilds.
func (s *sim) deliverGrants() {
	for _, g := range slices.Clone(s.grants) {
		s.deliver(g)
	}
}

// deliver applies g. When it clears anything, I9 is checked first, then the
// model follows what it cleared: a record that now lists the host as
// retried (exempt, and the retry's place in the ordering), a marker of g's
// build gone (the hold-ran restriction lifted), each an applied clear
// (sim.clears).
func (s *sim) deliver(g simGrant) {
	h := s.byName[g.host]
	rec, recorded := s.book.Lookup("bab", "blog", g.build)
	wasRetried := recorded && slices.Contains(rec.Retried, g.host)
	mk, marked := s.book.HoldRanOn(g.host)
	marked = marked && mk.Build == g.build
	if !s.m.RetryHost(g.host, g.build, g.book, g.epoch) {
		return
	}
	s.checkI9(g)
	after, still := s.book.Lookup("bab", "blog", g.build)
	recCleared := recorded && still && !wasRetried && slices.Contains(after.Retried, g.host)
	_, markStill := s.book.HoldRanOn(g.host)
	if s.retryAt[g.host] == nil {
		s.retryAt[g.host] = map[string]time.Time{}
	}
	s.retryAt[g.host][g.build] = s.now
	if recCleared {
		if s.exempt[g.host] == nil {
			s.exempt[g.host] = map[string]bool{}
		}
		s.exempt[g.host][g.build] = true
		delete(s.heldDown[g.host], g.build)
		if r := s.grec[g.build]; r != nil {
			r.retried[g.host] = true
			delete(r.down, g.host)
		}
		s.addClear(simClear{build: g.build, host: g.host, kind: "a grant's retry"})
	}
	if marked && !markStill {
		s.addClear(simClear{build: g.build, host: g.host, marker: true, kind: "a grant's marker clear"})
		if s.holdRan[h.name] == g.build {
			delete(s.holdRan, h.name)
		}
	}
	s.logf("grant: instances retry %s onto %s from step %d (generation %d) cleared it (record %v, marker %v)",
		g.host, g.build, g.step, g.epoch, recCleared, marked && !markStill)
}

// checkI9 is I9 at a grant the code has just accepted (it cleared
// something): no fact of its (build, host) landed after the grant's capture
// (a seq above g.cap), unless a clear of that restriction was applied after
// the fact (discharged). A new record is a fact of every host of the app
// (simFact.app), so one made by a failure on another host counts too,
// discharged only by a retry of g's build on g's host or the record leaving
// the book. Facts are ordered by when they landed, so a late outcome and a
// change in the grant's own step after its issue are inside the check, and
// one that landed before the issue, in the same step or earlier, is not:
// the published generation the grant names already held it.
func (s *sim) checkI9(g simGrant) {
	for _, f := range s.facts {
		if f.build != g.build || (!f.app && f.host != g.host) || f.seq <= g.cap || s.discharged(f, g.host) {
			continue
		}
		what := ""
		if f.app {
			what = " (a new record, which holds every host)"
		}
		violate("I9", "a grant for %s on %s issued at step %d (%s, book %.8s, generation %d, transition %d) cleared it after %s of %s on %s%s (transition %d, an execution begun at %s), which no applied clear has discharged",
			g.build, g.host, g.step, g.at.Sub(t0), g.book, g.epoch, g.cap, f.kind, f.build, f.host, what, f.seq, f.start.Sub(t0))
	}
}

// published is PublishedEpoch for build on host, as status reads it, with
// I10 checked on every non-zero value.
func (s *sim) published(build, host string) uint64 {
	e := s.book.PublishedEpoch("bab", "blog", build, host)
	if e != 0 && s.dir != "" {
		s.checkI10At(build, host, e)
	}
	return e
}

// checkI10 is I10 after every step, on a seed whose book is on disk: what
// status would publish now for each host (its installed build, and its
// hold-ran marker's) is on disk.
func (s *sim) checkI10() {
	if s.dir == "" {
		return
	}
	for _, h := range s.hosts {
		s.published(s.installed[h.ch], h.name)
		if m, ok := s.book.HoldRanOn(h.name); ok && m.Build != s.installed[h.ch] {
			s.published(m.Build, h.name)
		}
	}
}

// checkI10At is I10 at one published value: the book's file, read now, is
// the last save that landed, and it must hold epoch as the app's generation,
// under the identity a grant for it would name.
func (s *sim) checkI10At(build, host string, epoch uint64) {
	f, err := s.diskBook()
	if err != nil {
		violate("I10", "%s on %s published generation %d, and the book's file cannot be read: %v", build, host, epoch, err)
	}
	id := s.book.ID()
	if got := f.Gens[genKey("bab", "blog")]; got != epoch || f.Book != id {
		violate("I10", "%s on %s published generation %d under book %.8s, and the last landed file holds generation %d under book %.8s (saves failing: %v)",
			build, host, epoch, id, got, f.Book, s.fault)
	}
}

// diskBook reads the book's file as the agent would at its start. A missing
// file is an empty book.
func (s *sim) diskBook() (failureFile, error) {
	var f failureFile
	raw, err := os.ReadFile(filepath.Join(s.dir, FailureFile))
	if errors.Is(err, os.ErrNotExist) {
		return f, nil
	}
	if err != nil {
		return f, err
	}
	err = json.Unmarshal(raw, &f)
	return f, err
}

// agentRestart is the agent process ending and starting again, on a seed
// whose book is on disk (I9). Everything in flight goes with it. Each
// attempt parked in Swap stays parked until the seed ends, when its
// goroutine exits from Swap (sim.dead) without returning, so nothing of the
// old Manager touches the book again, as a killed process cannot. A healthy
// event on its way is lost, and every instance stops, the agent's children
// with it. A window of failing saves closes first: an agent that cannot save
// its book at startup refuses managed starts (internal/host), so the box
// runs again only once the disk takes a save. The new agent opens the book
// from disk (OpenFailureBook, which reconciles the intents left there),
// saves it, and starts a Manager with it.
//
// rebuild is a box rebuilt: the book's file is removed first, so the new
// book is empty under a new identity, and the model forgets what the old
// book held (forgetBook). Grants and failure facts outlive both.
func (s *sim) agentRestart(rebuild bool) {
	s.cancel()
	dropped := len(s.swaps)
	s.swaps = map[string]*simSwap{}
	s.late = map[string]simLate{}
	for _, h := range s.hosts {
		h.phase, h.gen, h.plan, h.planRoute, h.failLater, h.underMark = simAsleep, activator.Gen{}, activator.WakePlan{}, registry.Route{}, false, false
	}
	if s.fault {
		if err := os.Chmod(s.dir, 0o700); err != nil {
			panic(simCrash{msg: "the book's directory: " + err.Error()})
		}
		s.fault = false
	}
	if rebuild {
		if err := os.Remove(filepath.Join(s.dir, FailureFile)); err != nil && !errors.Is(err, os.ErrNotExist) {
			panic(simCrash{msg: "remove the book: " + err.Error()})
		}
		s.forgetBook()
	} else {
		s.reread()
	}
	// The retry ordering is in memory only (FailureBook.cleared): no
	// execution from before the restart is left for it to order.
	s.retryAt, s.clearedAt = map[string]map[string]time.Time{}, map[string]time.Time{}
	// A clear the book had not saved is rolled back, and the model keeps
	// none across a restart (sim.tseq): no earlier clear discharges a fact.
	s.clears = nil
	b, err := OpenFailureBook(s.dir)
	if err != nil {
		panic(simCrash{msg: "reopen the book: " + err.Error()})
	}
	b.now = func() time.Time { return s.now }
	if err := b.Flush(); err != nil {
		panic(simCrash{msg: "save the reopened book: " + err.Error()})
	}
	s.book = b
	s.logf("agent restart (box rebuilt %v): %d attempt(s) dropped, book %.8s", rebuild, dropped, b.ID())
	s.startAgent()
}

// reread is the model's side of a restart's open (agentRestart). A crash
// rolls back what the book held in memory and never saved: a clear made
// while saves failed (a host retry's, a good write's marker clear) is undone
// on disk, and the restriction is back. So the model rereads, as the agent
// does, the restrictions on disk: each record's retried and held-down hosts
// (exempt, heldDown, its own record grec) and the hold-ran markers. Nothing
// is lost the other way: a failure or a marker set while saves failed has
// its execution's intent on disk still (its End failed too), and the open
// turns each intent into the failure it may have been, which the model
// takes as a failure here (failureOf, reconciled), and as a hold run when
// the intent was a hold's.
func (s *sim) reread() {
	f, err := s.diskBook()
	if err != nil {
		panic(simCrash{msg: "read the book's file: " + err.Error()})
	}
	s.grec = map[string]*simRec{}
	for _, r := range f.Records {
		g := &simRec{retried: map[string]bool{}, down: map[string]bool{}}
		s.grec[r.Build] = g
		for _, h := range s.hosts {
			delete(s.exempt[h.name], r.Build)
		}
		for _, host := range r.Retried {
			g.retried[host] = true
			if s.exempt[host] == nil {
				s.exempt[host] = map[string]bool{}
			}
			s.exempt[host][r.Build] = true
		}
		for _, host := range r.HeldDown {
			g.down[host] = true
			if g.retried[host] {
				continue
			}
			if s.heldDown[host] == nil {
				s.heldDown[host] = map[string]bool{}
			}
			s.heldDown[host][r.Build] = true
		}
	}
	for _, h := range s.hosts {
		m, ok := f.HoldRan[h.name]
		switch {
		case !ok:
			delete(s.holdRan, h.name)
		case s.holdRan[h.name] != m.Build:
			// A marker the crash put back (its clear was never saved).
			s.holdRan[h.name], s.holdRanAt[h.name], s.holdRanSet[h.name] = m.Build, m.At, s.now
		}
	}
	for _, host := range sortedKeys(f.Intents) {
		in := f.Intents[host]
		h := s.byName[host]
		hold := in.Mode == string(activator.ModeHold)
		s.logf("restart: %s's intent %s (%s, %s) is a failure now", host, in.ID, in.Build, in.Mode)
		s.failureOf(in.Build, host, s.now, hold, in.Start, true)
		if hold {
			start := in.Start
			if start.IsZero() {
				start = s.now
			}
			s.holdRanOn(h, in.Build, start)
		}
	}
	// Every restriction standing now was set before the open, and so before
	// anything the new agent starts (its snapshots are newer than every
	// marker's Seq): a good write by an execution starting in this very
	// step lifts it, as the code's ClearHoldRanBefore does.
	for _, host := range sortedKeys(s.holdRan) {
		if !s.holdRanSet[host].Before(s.now) {
			s.holdRanSet[host] = s.now.Add(-time.Nanosecond)
		}
	}
}

// forgetBook is the model's side of a box rebuild (agentRestart): the book
// is gone, and with it every failure, retry, hold-down and hold-ran
// restriction the model held for it. The facts and grants stay.
func (s *sim) forgetBook() {
	s.failedEver, s.lastStamp, s.failedAt = map[string]bool{}, map[string]time.Time{}, map[string]time.Time{}
	s.exempt, s.heldDown = map[string]map[string]bool{}, map[string]map[string]bool{}
	s.holdRan, s.holdRanAt, s.holdRanSet = map[string]string{}, map[string]time.Time{}, map[string]time.Time{}
	s.grec = map[string]*simRec{}
}

// restage is root restaging a build whose staged file was wrong.
func (s *sim) restage() {
	for _, b := range sortedKeys(s.builds) {
		if s.builds[b].misstaged {
			s.builds[b].misstaged = false
			s.logf("restaged %s", b)
			return
		}
	}
}