package hostlife

import (
	"encoding/json"
	"fmt"
	"math/rand"
	"os"
	"strconv"
	"strings"
	"testing"
)

const (
	chaosSteps = 2000
	calmSteps  = 6000
)

func newWorld(seed int64, verbose bool) *world {
	w := &world{seed: seed, rng: &counting{r: rand.New(rand.NewSource(seed))}, verbose: verbose, perRegion: boxesPerRegion,
		stores: map[string]bool{app0: true, app1: true}, cfgVer: map[string]int{app0: 1, app1: 1}, purgeRes: map[string]bool{}, purgeStep: map[string]int{},
		control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, markerAlerted: map[string]bool{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}}
	for r := 0; r < nRegions; r++ {
		w.buckets[r] = newBucket()
	}
	w.admit = map[string]*admission{app0: {Reg: map[string]regEntry{}}, app1: {Reg: map[string]regEntry{}}}
	for _, f := range []string{"partition", "freeze", "pause", "destroy", "disk", "recreate", "pausecreate", "lostreturn", "markerhold", "freezetgt", "retention"} {
		w.fail[f] = w.rng.Intn(2) == 0
	}
	for r := 0; r < nRegions; r++ {
		b := &box{ID: r, Region: r, HasKey: r == 0 || w.rng.Intn(4) != 0, Skew: w.rng.Intn(3) - 1, Nested: w.rng.Intn(2) == 0, Enrolled: true}
		if r != 0 && w.rng.Intn(4) == 0 {
			b.FailApp = true
		}
		if w.rng.Intn(10) == 0 {
			b.FailArchive = true
		}
		if r != 0 && w.rng.Intn(2) == 0 {
			b.FailSettle = true
		}
		b.forget()
		w.boxes = append(w.boxes, b)
	}
	if w.rng.Intn(3) == 0 {
		w.legacyStart()
	} else if w.fail["retention"] {
		w.retentionStart()
	}
	// the two-box shape (region HA L18): a second enrolled box in every
	// region, drawn after every one-box draw of this world, so the one-box
	// stream is untouched; id nRegions+r, as addBox gives a witness
	for r := 0; r < nRegions && w.perRegion > 1; r++ {
		b := &box{ID: nRegions + r, Region: r, HasKey: w.rng.Intn(4) != 0, Skew: w.rng.Intn(3) - 1, Nested: w.rng.Intn(2) == 0, Enrolled: true}
		if w.rng.Intn(4) == 0 {
			b.FailApp = true
		}
		if w.rng.Intn(10) == 0 {
			b.FailArchive = true
		}
		if w.rng.Intn(2) == 0 {
			b.FailSettle = true
		}
		b.forget()
		w.boxes = append(w.boxes, b)
	}
	if w.perRegion > 1 {
		w.fail["terminate"] = w.rng.Intn(2) == 0
		for _, b := range w.boxes {
			b.DeleteOnTermination = w.rng.Intn(8) != 0 // one box in eight has a volume without the flag (§5.9, H19)
		}
	}
	return w
}

// retentionStart: the host has been recreated more times than
// retired_incarnations remembers (claim rev 6.5 §2 caps it at 32), and a
// replication object of an incarnation the list forgot is still in a
// bucket (rev 7.5, R4 §3: the retention-boundary fixture). Only a sweep
// that discovers locations from the namespace parks it.
func (w *world) retentionStart() {
	w.claim = &claim{Ver: w.newVer()}
	old := 1 // an incarnation older than every one the list remembers
	w.nextInc = old
	for i := 0; i < retiredCap+1; i++ {
		w.nextInc++
		w.claim.Retired = retire(w.claim.Retired, w.nextInc) // 2 to 33: the cap drops nothing but 1 was never listed
	}
	if contains(w.claim.Retired, old) || len(w.claim.Retired) != retiredCap {
		panic("retention fixture: the injected incarnation must be absent from a full list (R5-6)")
	}
	r := w.rng.Intn(nRegions)
	k := ltxKey(locationFor(old, false), "", 0, 1, 1)
	// a request that completed long after its incarnation was retired: a
	// straggler, which only the sweep retires (INV-10 at an operation's end
	// exempts it, as it exempts every completion after H)
	w.buckets[r].objs[k] = []*version{{ID: w.newVer(), Straggler: true, B: &blob{Kind: "ltx", Level: 0, Min: 1, Max: 1, Inc: old, Pages: map[string]int{"k0": 1}}}}
	w.logf("retention fixture: incarnation %d's replica in region %d, past the cap", old, r)
}

// legacyStart: a host that exists before the release. Its data is untagged,
// a best-effort-placed descendant may be running, and an earlier
// (pre-release) move left a replica in another region.
func (w *world) legacyStart() {
	r := w.rng.Intn(nRegions)
	b := w.regionBox(r)
	w.nextInc++
	inc := w.nextInc
	w.rec = &record{Ver: w.newVer(), Inc: inc, App: app0, Region: r}
	w.legacyIncs[inc] = true
	key := 7
	b.Row = &row{Inc: inc, App: app0, Status: "ready"}
	b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}, Epoch: b.Epoch}
	for i := 0; i < 3; i++ {
		w.nextW++
		w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, Session: 0, Gen: "", At: 0, Kind: "write"}) // the pre-release writer: the legacy identity, session 0 (claim rev 6.24)
		w.writeInc[w.nextW] = inc
		w.commit(b.DB, w.nextW)
	}
	b.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true}
	b.Config, b.CfgVer, b.Live, b.LiveInc = inc, 1, app0, inc
	// the pre-release replica: one level-0 file of everything, untagged
	full := w.deltaOf(b.DB, 0)
	full.Inc = 0
	b.DB.Pos, b.DB.Pending = 1, nil
	k := ltxKey(b.DB.Loc, "", 0, 1, 1)
	w.buckets[r].objs[k] = []*version{{ID: w.newVer(), B: full}}
	w.markDurable(k, full)
	// an earlier move's replica in another region: the first transaction only
	first := w.txs[b.DB.Log[0]]
	part := &blob{Kind: "ltx", Level: 0, Min: 1, Max: 1, Inc: 0, SealKey: key, Pages: map[string]int{first.Key: first.Val}, TXs: []int{first.W}}
	other := (r + 1 + w.rng.Intn(nRegions-1)) % nRegions
	w.buckets[other].objs[ltxKey(b.DB.Loc, "", 0, 1, 1)] = []*version{{ID: w.newVer(), B: part}}
	if w.rng.Intn(2) == 0 {
		// the pre-release processes: the legacy identity, session 0 in the origin generation (claim rev 6.24; rev 7.13's seed 44450 found them acknowledging under no identity at all)
		b.Procs = []*proc{{Main: true, Strict: false, Inc: inc, Session: 0, Box: b.ID}, {Main: false, Strict: false, Inc: inc, Session: 0, Box: b.ID}}
	}
	if w.rng.Intn(2) == 0 {
		// that earlier move also left the old box's local state behind:
		// a database, a files directory and env files, and no row
		ob := w.regionBox(other)
		ob.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: cloneState(part.Pages), TX: 1, Pos: 1, Epoch: ob.Epoch}
		ob.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true}
	}
}

// ---------- the scheduler ----------

func (w *world) step() {
	w.now++
	w.completeDue(false)
	if p := w.recPend; p != nil && p.at <= w.now {
		w.recPend = nil
		if w.rec == nil { // create-only: it lands only where no record is
			w.rec = p.rec
			w.logf("a late record PUT for inc %d landed", p.rec.Inc)
		}
	}
	w.runStalled()
	if w.perRegion > 1 {
		w.twoBoxStep() // the two-box shape's own draws, after the one-box ones (ruling 1)
	}
	bs := w.liveBoxes()
	b := bs[w.rng.Intn(len(bs))]
	n := w.rng.Intn(100)
	switch {
	case !w.calm && n < 6:
		switch w.rng.Intn(10) {
		case 9:
			// a config or secrets update publishes a new bundle version
			app := []string{app0, app1}[w.rng.Intn(2)]
			w.cfgVer[app]++
			w.logf("config bundle %s v%d published", app, w.cfgVer[app])
		case 0, 1:
			w.userCreate()
		case 2, 7, 8:
			w.userMove()
		case 3:
			w.userDelete()
		case 4:
			w.userPush()
		case 5:
			if w.rng.Intn(3) == 0 {
				w.userPurge()
			}
		default:
			w.doorAuthorize()
		}
	case !w.calm && n < 12:
		w.fault()
	case n < 20:
		w.consoleStep()
	case n < 24:
		w.recoveryPass()
	case n < 25:
		w.doorOpen()
	case n < 26:
		if !w.calm {
			w.doorAuthorize()
		}
	case n < 28:
		if w.calm || w.rng.Intn(4) == 0 {
			w.operator() // operators act during the incident too, not only after
		}
	case n < 56:
		w.worker(b)
	case n < 62:
		w.obligationRunner(b)
	case n < 67:
		w.reconcile(b)
	case n < 73:
		w.traffic(b)
		w.wake(b)
	case n < 77:
		w.replicate(b)
	case n < 78:
		w.idleStop(b)
	case n < 81:
		w.passPlan(b, "config")
	case n < 84:
		w.passPlan(b, "adopt")
	case n < 89:
		w.passExec(b)
	case n < 94:
		w.runner(b)
	default:
		w.watchdog(b)
	}
	w.checkStep()
}

// heal ends the chaos. On the first call some faults linger, so the
// operator meets a frozen, paused or partitioned participant, and destroyed
// boxes stay down; on the final call everything heals, and each stopped box
// either comes back (an instance stopped, then started) or is replaced.
func (w *world) heal(final bool) {
	w.calm = true
	c := w.claim
	for _, b := range w.boxes {
		linger := !final && c != nil && (c.Op.Kind == "landing" || c.Op.Kind == "restoring") && b.ID == c.Op.Tgt && w.rng.Intn(2) == 0
		if linger {
			continue
		}
		b.Partitioned, b.Frozen, b.AgentPaused, b.AgentDown, b.PurgeFails = false, false, false, false, false
		if b.hostFailed {
			b.hostFailed, b.failCounted, b.startFails, b.nextStart = false, false, 0, 0 // the operator cleared the failed host
			w.stats["failed hosts cleared"]++
		}
	}
	if final {
		for _, b := range w.boxes {
			if b.Destroyed && !b.Terminated && w.regionBox(b.Region) == nil && w.rng.Intn(2) == 0 {
				b.Destroyed = false // started again, with its old disk
				b.forget()
				w.logf("stopped box %d started again", b.ID)
			}
		}
		w.replaceDestroyed()
	}
	// a box with no key gets one: ops distributes it (the approved fix)
	for _, b := range w.boxes {
		b.HasKey = true
		b.FailArchive = false
	}
}

func (w *world) quiescent() bool {
	if len(w.pending) > 0 || len(w.console) > 0 || w.recPend != nil || !w.atRest() {
		return false
	}
	return len(w.residue()) == 0
}

// whyNotQuiescent names, for a non-convergence report, what quiescent()
// found unfinished.
func (w *world) whyNotQuiescent() string {
	var out []string
	for _, r := range w.pending {
		out = append(out, fmt.Sprintf("pending %v %s r%d by %s issued %d due %d beyond=%v", r.kind, r.key, r.region, r.by, r.at, r.doneAt, r.beyond))
	}
	if len(w.console) > 0 {
		out = append(out, fmt.Sprintf("%d console plans", len(w.console)))
	}
	if w.recPend != nil {
		out = append(out, "a record PUT pending")
	}
	if w.h2.Kind != "" {
		out = append(out, "h2 "+w.h2.Kind)
	}
	for app, ad := range w.admit {
		if ad.Closed {
			out = append(out, "admission closed for "+app)
		}
		for id, e := range ad.Reg {
			if w.now <= e.Until || !on("registration-recovery") {
				out = append(out, fmt.Sprintf("registration %s of %s live until %d", id, app, e.Until))
			}
		}
	}
	if c := w.claim; c != nil && c.Holder != nil {
		if c.Op.Kind != "serving" {
			out = append(out, "op "+c.Op.Kind)
		}
		for _, o := range c.Obl {
			out = append(out, fmt.Sprintf("obligation %s %s %s", o.ID, o.Kind, o.Prefix))
		}
		if e := w.execBox(c); e != nil && (e.Config != c.Holder.Inc || e.Live != c.Holder.App || e.LiveInc != c.Holder.Inc) {
			out = append(out, "config or adoption owed")
		}
		for _, b := range w.liveBoxes() {
			if e := w.execBox(c); b != e && (b.Row != nil || len(b.Procs) > 0) {
				out = append(out, fmt.Sprintf("box %d has a row or processes", b.ID))
			}
			if b.open != nil {
				out = append(out, fmt.Sprintf("box %d admitted into %q, start in progress at stage %s", b.ID, b.open.Gen, b.open.stage))
			}
		}
	}
	if r := w.residue(); len(r) > 0 {
		out = append(out, fmt.Sprintf("residue %v", r))
	}
	if len(out) == 0 {
		return "quiescent"
	}
	return strings.Join(out, "; ")
}

// atRest: the claim, record and boxes have converged.
func (w *world) atRest() bool {
	if w.h2.Kind == "creating" || w.h2.Kind == "purging" || w.h2.Kind == "fenced" {
		return false
	}
	for _, ad := range w.admit {
		if ad.Closed {
			return false
		}
		for _, e := range ad.Reg {
			if w.now <= e.Until || !on("registration-recovery") {
				return false // a live registration: its create is still in flight
			}
		}
	}
	c := w.claim
	if c == nil {
		return w.rec == nil
	}
	if c.Holder == nil {
		return c.Op.Kind == "" && w.rec == nil && len(c.Obl) == 0
	}
	if c.Op.Kind != "serving" || len(c.Obl) > 0 || w.rec == nil || w.rec.Inc != c.Holder.Inc || w.rec.MoveID != "" {
		return false
	}
	e := w.execBox(c)
	if e == nil || e.Row == nil || e.Row.Status != "ready" {
		return false
	}
	if d := c.Op.Completed; d != nil && !d.Discharged || e.actFence != "" {
		return false // a completion not yet marked, or a fence not yet lifted (L18 finding 2)
	}
	if e.Config != c.Holder.Inc || e.Live != c.Holder.App || e.LiveInc != c.Holder.Inc {
		return false // config sync and adoption still owe this incarnation
	}
	for _, b := range w.liveBoxes() {
		if b != e && (b.Row != nil || len(b.Procs) > 0) {
			return false
		}
		if b.open != nil {
			return false // an admitted writer still finishing its start
		}
	}
	return true
}

// ---------- invariants ----------

// checkStep runs after every step.
func (w *world) checkStep() {
	// one writer per host, across every region
	var writers []int
	for _, b := range w.boxes {
		if !b.Destroyed && !b.Frozen && len(b.Procs) > 0 {
			writers = append(writers, b.ID)
		}
	}
	if len(writers) > 1 {
		w.violate("INV-1: boxes %v run the host at once", writers)
	}
	// an app store exists while any instance of the app is held for serving
	if w.h2.Kind == "serving" && !w.stores[w.h2.App] {
		w.violate("purge: store of app %s destroyed while its second host serves", w.h2.App)
	}
	if c := w.claim; c != nil && c.Holder != nil && !w.stores[c.Holder.App] {
		switch c.Op.Kind {
		case "purging", "deleting", "parking":
		default:
			w.violate("purge: store of app %s destroyed while its host is %s", c.Holder.App, c.Op.Kind)
		}
	}
	if c := w.claim; c != nil && c.Op.Kind == "restoring" && c.Op.SrcPhase == "stuck" {
		w.violate("stuck: source could not recover %s", c.Op.ID)
	}
	if c := w.claim; c != nil && c.Op.Kind == "reverting" && c.Op.RevPhase == "stuck" {
		w.violate("stuck: source could not revert %s", c.Op.ID)
	}
	// INV-27 (region HA HA1, HA7): while relocating, op.exec is none or an
	// enrolled box that is not on lost; execution authority never names a
	// fenced or unenrolled reference
	if c := w.claim; c != nil && c.Op.Kind == "relocating" && c.Op.Exec >= 0 {
		released := c.Op.Release != nil && c.Op.Release.Box == c.Op.Exec // a released executor fenced before PinAfterRelease (FenceSource after the release, rev 4)
		if e := w.boxByID(c.Op.Exec); e == nil || !e.Enrolled || contains(c.Lost, e.ID) && !released {
			w.violate("INV-27: relocation %s names box %d as op.exec, which is unenrolled or lost", c.Op.ID, c.Op.Exec)
		}
	}
	w.checkCompletion() // INV-31 (L18 finding 2): only a relocation sets owedDischarge
	if w.perRegion > 1 && !w.noPinWatch {
		w.checkFilesPreserved() // INV-30 (joint round 2): only the two-box shape has files keys
	}
}

func (w *world) boxesState() string {
	s := ""
	for _, b := range w.liveBoxes() {
		s += fmt.Sprintf(" [box %d r%d row=%+v procs=%d db=%v files=%v cfg=%d live=%s/%d old=%v]", b.ID, b.Region, b.Row, len(b.Procs), b.DB != nil, b.Files != nil, b.Config, b.Live, b.LiveInc, b.Old)
	}
	return s
}

// residue lists live-namespace data that should not exist: everything
// under the host's namespace in a region the host does not serve from, and
// in the serving region anything but the claim's location and generation
// (and the parkers' own locks). Nothing here writes.
func (w *world) residue() []string { return w.residueOf(false) }

// residueAtEnd is INV-10 as the spec states it, checked when an operation
// ends (a release to serving, a free): no copy outside the live set, except
// a straggler that completed after H, which only the sweep can retire.
func (w *world) residueAtEnd() []string { return w.residueOf(true) }

func (w *world) residueOf(exceptStragglers bool) []string {
	var out []string
	c := w.claim
	live := map[string]bool{}
	serving := -1
	if c != nil && c.Holder != nil {
		if e := w.execBox(c); e != nil {
			serving = e.Region
			loc, gen := c.Holder.Location, c.Holder.Gen
			live[dataKey(loc, gen, "")] = true // the generation's prefix: every LTX file of the generation
			live[genMarkerKey(loc)] = true
			live[dataKey(loc, gen, "closed/")] = true // the live generation's certificates, every session's (claim rev 6.8 §6.4a)
			// an unfinished admission's predecessor is its baseline until the
			// marker names the new generation (claim rev 6.10, R10-2)
			if bg, _ := w.bucketGen(e, loc); bg != gen {
				if chain, ok := w.unfinishedChain(c, bg); ok {
					for _, g := range append(chain, bg) {
						live[dataKey(loc, g, "")] = true
						live[dataKey(loc, g, "closed/")] = true
					}
				}
			}
		}
	}
	for r := 0; r < nRegions; r++ {
		bk := w.buckets[r]
		for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} {
			for _, k := range bk.keys(p) {
				if strings.HasSuffix(k, "lock.json") || (r == serving && (live[k] || isDataKey(k) && live[genPrefixOf(k)] || isClosedKey(k) && live[k[:strings.LastIndex(k, "closed/")+7]] || w.filesLiveSet(c, r, k))) {
					continue
				}
				if exceptStragglers && w.owed(r, k) {
					continue // a pending chore of another operation covers it
				}
				for _, v := range bk.objs[k] {
					if !v.Marker && !(exceptStragglers && (v.Straggler || v.SweepOwed)) {
						out = append(out, fmt.Sprintf("r%d:%s", r, k))
						break
					}
				}
			}
		}
	}
	return out
}

// owed: an open park obligation names this region and a prefix of k.
func (w *world) owed(r int, k string) bool {
	c := w.claim
	if c == nil {
		return false
	}
	for _, o := range c.Obl {
		if o.Region == r && o.Prefix != "" && strings.HasPrefix(k, o.Prefix) {
			return true
		}
	}
	// a cold admission whose writer died after its marker landed and before
	// it queued the retire chores (rev 7.11): the writers journal still
	// shows its predecessor uncovered, and the next admission's marker
	// retires it (§5.1 step 9); until then the predecessor is owed, as a
	// pending chore's keys are (seeds 4159, 26817, 34163 / 100,000)
	if h := c.Holder; h != nil && strings.HasPrefix(k, h.Location) {
		g := genOf(h.Location, k)
		for i := range c.Holder.Admissions {
			if e := &c.Holder.Admissions[i]; e.Mode == "cold" && !e.Covered && e.Predecessor.Gen == g {
				return true
			}
		}
	}
	return false
}

func hasNonMarker(vs []*version) bool {
	for _, v := range vs {
		if !v.Marker {
			return true
		}
	}
	return false
}

// canStart is the start check and admission, computed without starting
// anything: the final oracle never writes.
func (w *world) canStart(e *box, c *claim) string {
	c = c.clone()
	// the wake's first step is the record's initialization from this fresh
	// read (claim rev 6.5 §6.4): a box with no record for the incarnation
	// that the claim names as op.exec takes holder.stamp. On a copy of the
	// box: the oracle never writes (R6-1)
	eb := *e
	eb.Stamps = map[int]*stamps{}
	for k, v := range e.Stamps {
		cp := *v
		cp.Retired = append([]int(nil), v.Retired...)
		eb.Stamps[k] = &cp
	}
	inc := c.Holder.Inc
	if eb.stampOf(inc) == 0 && c.Holder.Stamp != 0 && c.Op.Exec == eb.ID && !(eb.Stamps[inc] != nil && contains(eb.Stamps[inc].Retired, c.Holder.Stamp)) {
		eb.install(inc, c.Holder.Stamp)
	}
	c.readStamp = eb.stampOf(inc) // as a fresh read by e would carry it
	if !w.admits(&eb, c, "") {
		return "not admitted"
	}
	h := c.Holder
	bg, binc := w.bucketGen(e, h.Location)
	if bg != h.Gen || (bg != "" && binc != h.Inc) {
		return fmt.Sprintf("generation.json %q, claim %q", bg, h.Gen)
	}
	if e.Files != nil && (e.Files.Gen != bg || e.Files.Inc != h.Inc) {
		return fmt.Sprintf(".files-gen %q inc %d", e.Files.Gen, e.Files.Inc)
	}
	return ""
}

// finalFreed is the end state of a freed host: no record, and no box holds
// local state for it.
func (w *world) finalFreed() {
	if w.rec != nil {
		w.violate("claim/record: claim free but record inc %d exists", w.rec.Inc)
	}
	for _, b := range w.liveBoxes() {
		if b.DB != nil || b.Files != nil {
			w.violate("INV-10: box %d still holds local state for a freed host", b.ID)
		}
	}
}

func (w *world) final() {
	if !w.quiescent() {
		if len(w.pending) == 0 && len(w.console) == 0 && w.atRest() {
			w.violate("INV-10: residue never retired: %v", w.residue())
			return
		}
		w.violate("INV-3: no convergence after %d calm steps: %s; %s%s", calmSteps, w.whyNotQuiescent(), w.state(), w.boxesState())
		return
	}
	c := w.claim
	if c == nil || c.Holder == nil {
		w.finalFreed()
		return
	}
	e := w.execBox(c)
	if !hasMain(e) && c.Op.Kind == "serving" && w.certificate(e, c) == nil {
		// the host is at rest after a session that did not certify: its
		// next wake is a cold admission that declares the acknowledged,
		// unuploaded writes (R10-6). Rev 7.11's oracle exempted them as a
		// prospective declaration, recording and alerting nothing; the
		// recovery admission is run for real now, then everything it
		// queued, and the writes are judged by what it declared
		w.stats["final: recovery admission run"]++
		w.drive(400, func() bool { return hasMain(e) && e.open == nil }, func() { w.wake(e) }, w.actAll(), w.runConsole, w.recoveryPass, w.operator)
		if len(w.violations) > 0 {
			return
		}
		if !(hasMain(e) && e.open == nil) {
			w.stats["final: unverified prospective declaration"]++
			w.logf("final: the recovery admission on box %d did not complete; its declaration is unverified", e.ID)
		} else {
			w.drive(calmSteps, w.quiescent, w.actAll(), w.runConsole, w.recoveryPass, w.operator)
			if len(w.violations) > 0 {
				return
			}
			if !w.quiescent() {
				w.violate("INV-3: no convergence after the final recovery admission: %s; %s%s", w.whyNotQuiescent(), w.state(), w.boxesState())
				return
			}
		}
		c = w.claim
		if c == nil || c.Holder == nil {
			// the final drive ran the console too: a purge may have freed
			// the host (seed 2547 / 3,000 under admission-registration)
			w.finalFreed()
			return
		}
		e = w.execBox(c)
	}
	if e == nil || w.rec == nil {
		w.violate("claim/record: the claim's holder has no record or no executing box at the end: %s", w.state())
		return
	}
	if w.rec.Region != e.Region {
		w.violate("claim/record: record region %d, claim serves from box %d in region %d", w.rec.Region, e.ID, e.Region)
	}
	for _, b := range w.liveBoxes() {
		if b != e && (b.DB != nil || b.Files != nil) {
			w.violate("INV-10: box %d holds local state; the host serves from box %d", b.ID, e.ID)
		}
	}
	if !hasMain(e) {
		if why := w.canStart(e, c); why != "" {
			w.violate("final: the serving box %d cannot start the host: %s (%s)", e.ID, why, w.state())
			return
		}
	}
	// the serving box's database as it stands, or, if it has none, the
	// cold restore it would do: computed, never written
	before := w.fingerprint()
	defer w.pure("final oracle", before)
	local := e.DB
	if local == nil {
		local = w.restoreAsNext(e, c)
	}
	if local == nil {
		w.violate("final: the serving box %d has no database and nothing to restore", e.ID)
		return
	}
	if w.stats["final: unverified prospective declaration"] == 0 || w.certificate(e, c) != nil {
		w.checkLocal(c, e, local)
	}
	w.checkDurable(c, "final")
	if w.lostAlert != w.stats["writes_lost declared"] {
		w.violate("INV-14: %d writes_lost records, %d alerts delivered (R10-6)", w.stats["writes_lost declared"], w.lostAlert)
	}
}

// checkDurable is the cold-restore promise, checked at every release to
// serving and at the end: a restore of the claim's location and generation
// in the serving region holds the effect of every acknowledged write that
// a completed replica object ever covered, and equals the database the
// acknowledged writes up to its TXID produce. It reads the bucket and
// writes nothing.
func (w *world) checkDurable(c *claim, when string) {
	if c == nil || c.Holder == nil {
		return
	}
	e := w.execBox(c)
	if e == nil {
		return
	}
	before := w.fingerprint()
	defer w.pure("INV-2 oracle", before)
	db := w.restoreAsNext(e, c)
	need := map[int]bool{}
	seen := map[string]bool{}
	for p := dataKey(c.Holder.Location, c.Holder.Gen, ""); p != "" && !seen[p]; p = w.genParent[p] {
		seen[p] = true
		for x := range w.durable[p] {
			need[x] = true
		}
	}
	have := map[int]bool{}
	if db != nil {
		for _, x := range db.Log {
			have[x] = true
		}
	}
	for _, a := range w.acks {
		if a.Inc != c.Holder.Inc || a.Exempt || a.Kind != "write" || !need[a.W] {
			continue
		}
		// coverage by write identity (R7-2): a required write is present
		// only if a file the restore applied carried it; a maximum TXID
		// says nothing after a fork
		if db == nil || !have[a.W] {
			w.violate("INV-2 (cold restore, %s): durable write %d (box %d, t=%d) lost (%s)", when, a.W, a.Box, a.At, w.ltxShape(e.Region, dataKey(c.Holder.Location, c.Holder.Gen, "")))
			return
		}
	}
	if db == nil {
		return
	}
	w.foreignWrites(db, c.Holder.Inc, "cold restore, "+when)
	// the restored state must be the replay, in acknowledgement order, of
	// exactly the writes its files carried (R7-2): a write declared lost
	// with a disk that landed anyway is replayed because it is there, a
	// required write that is missing was caught above, and no key is ever
	// masked; a later required write re-establishes its key's value
	if why, ok := statesEqual(w.expectFrom(c.Holder.Inc, have), db.State); !ok {
		w.violate("INV-2 (cold restore, %s): the restored database differs from the replay of the writes it carries: %s (restored TXID %d; %s; %s; %s)", when, why, db.TX, w.keyHistory(c.Holder.Inc, why, have), w.ltxShape(e.Region, dataKey(c.Holder.Location, c.Holder.Gen, "")), w.fileDetail(e.Region, dataKey(c.Holder.Location, c.Holder.Gen, ""), why))
	}
}

// fileDetail lists, for diagnosis, each current file of the generation
// with its kind, range, TXs and the page it holds for the key named in a
// statesEqual message.
func (w *world) fileDetail(r int, prefix, why string) string {
	key := ""
	if i := strings.Index(why, "key "); i >= 0 {
		rest := why[i+4:]
		if j := strings.Index(rest, " "); j > 0 {
			key = rest[:j]
		}
	}
	bk := w.buckets[r]
	var out []string
	for _, k := range bk.keys(prefix) {
		v := bk.current(k)
		if v == nil || v.B == nil || !isDataKey(k) {
			continue
		}
		val, ok := v.B.Pages[key]
		out = append(out, fmt.Sprintf("%s{full=%v txs=%v %s=%d/%v}", strings.TrimPrefix(k, prefix), v.B.Full, v.B.TXs, key, val, ok))
	}
	return "detail " + strings.Join(out, " ")
}

// keyHistory lists, for diagnosis, the acknowledged writes of inc that
// touched the key named in a statesEqual message: W, lineage TXID, value,
// whether the restored files carried it, whether it is exempt.
func (w *world) keyHistory(inc int, why string, have map[int]bool) string {
	key := ""
	if i := strings.Index(why, "key "); i >= 0 {
		rest := why[i+4:]
		if j := strings.Index(rest, " "); j > 0 {
			key = rest[:j]
		}
	}
	var out []string
	for _, a := range w.acks {
		if a.Inc != inc || a.Kind != "write" {
			continue
		}
		t := w.txs[a.W]
		k := t.Key
		if k == "" {
			k = fmt.Sprintf("k%d", a.W)
		}
		if k != key {
			continue
		}
		out = append(out, fmt.Sprintf("W%d:TX%d=%d have=%v exempt=%v session=%d gen=%q", a.W, t.TX, t.Val, have[a.W], a.Exempt, a.Session, a.Gen))
	}
	return "history of " + key + ": " + strings.Join(out, " ")
}

// expectFrom replays, in acknowledgement order, the acknowledged writes of
// inc that the restored files carried (have), into the state a correct
// restore of those files must equal.
func (w *world) expectFrom(inc int, have map[int]bool) map[string]int {
	state := map[string]int{}
	for _, a := range w.acks {
		if a.Inc != inc || a.Kind != "write" || !have[a.W] {
			continue
		}
		t := w.txs[a.W]
		if t.Key == "" {
			t.Key, t.Val = fmt.Sprintf("k%d", a.W), a.W
		}
		if t.Val == 0 {
			delete(state, t.Key)
		} else {
			state[t.Key] = t.Val
		}
	}
	return state
}

// counting is the sweep's random source, counting its draws so the oracle
// purity check can show an oracle consumed none (R6-1).
type counting struct {
	r *rand.Rand
	n int
}

func (c *counting) Intn(n int) int {
	c.n++
	return c.r.Intn(n)
}

// fingerprint is everything an oracle must leave alone: the buckets (the
// version counter and every version), the pending requests, the boxes'
// upload queues and anchors, the console's plans and the random source's
// position (R6-1). An oracle that changes any of it is repairing or
// disturbing the system it checks.
func (w *world) fingerprint() string {
	objs := 0
	for _, bk := range w.buckets {
		for _, vs := range bk.objs {
			objs += len(vs)
		}
	}
	queue, anchors := 0, 0
	for _, b := range w.boxes {
		if b.DB != nil {
			queue += len(b.DB.Queue)
		}
		for _, rs := range b.anchors {
			anchors += len(rs)
		}
	}
	draws := -1
	if c, ok := w.rng.(*counting); ok {
		draws = c.n
	}
	return fmt.Sprintf("ver=%d objs=%d pending=%d queue=%d anchors=%d console=%d draws=%d", w.ver, objs, len(w.pending), queue, anchors, len(w.console), draws)
}

// pure records a violation if an oracle changed the system (R6-1): the
// fingerprint taken before it must equal the one after.
func (w *world) pure(what, before string) {
	if after := w.fingerprint(); after != before {
		w.violate("oracle purity: the %s changed the system: before %s, after %s", what, before, after)
	}
}

// ltxShape describes the current LTX files under prefix, for diagnosis.
func (w *world) ltxShape(r int, prefix string) string {
	var parts []string
	bk := w.buckets[r]
	for _, k := range bk.keys(prefix) {
		if !isDataKey(k) {
			continue
		}
		var vs []string
		for _, v := range bk.objs[k] {
			if v.Marker {
				vs = append(vs, "M")
			} else {
				vs = append(vs, fmt.Sprintf("v%d", v.ID))
			}
		}
		parts = append(parts, strings.TrimPrefix(k, prefix)+":"+strings.Join(vs, ","))
	}
	return "files " + strings.Join(parts, " ")
}

// foreignWrites: a database restored or started for incarnation inc holds
// no key another incarnation wrote (C1, INV-4, claim C3a).
func (w *world) foreignWrites(db *localDB, inc int, how string) {
	for k, v := range db.State {
		if v == 0 {
			continue
		}
		if i, ok := w.writeInc[keyID(k)]; ok && i != inc {
			w.violate("C3a (%s): incarnation %d's database holds key %s of incarnation %d", how, inc, k, i)
			return
		}
	}
}

// checkLocal: the serving box's database holds the effect of every
// acknowledged, non-exempt write of the incarnation (INV-2), and every
// acknowledged timer is delivered or still registered.
func (w *world) checkLocal(c *claim, e *box, db *localDB) {
	// the serving database is the replay of every acknowledged write that
	// was never declared lost, and of every write declared lost that came
	// back: an upload in flight at a lost disk that landed anyway, which a
	// later cold admission's remote baseline carried (R7-2's identity rule,
	// as checkDurable applies it; seed 5863 / 30,000)
	have := map[int]bool{}
	for _, x := range db.Log {
		have[x] = true
	}
	want := map[int]bool{}
	for _, a := range w.acks {
		if a.Inc == c.Holder.Inc && a.Kind == "write" && (!a.Exempt || have[a.W]) {
			want[a.W] = true
		}
	}
	if why, ok := statesEqual(w.expectFrom(c.Holder.Inc, want), db.State); !ok {
		w.violate("INV-2 (local): the serving database differs from the acknowledged writes: %s (%s)", why, w.keyHistory(c.Holder.Inc, why, have))
		return
	}
	timers := map[int]bool{}
	for _, x := range e.Timers {
		timers[x] = true
	}
	for _, a := range w.acks {
		if a.Inc != c.Holder.Inc || a.Exempt || a.Kind != "timer" {
			continue
		}
		if !w.delivered[a.W] && !timers[a.W] {
			w.violate("INV-2: acknowledged timer %d (box %d, t=%d) lost", a.W, a.Box, a.At)
			return
		}
	}
}

func (w *world) state() string {
	c := w.claim
	if c == nil {
		return fmt.Sprintf("claim absent, record %+v", w.rec)
	}
	obl := ""
	for _, o := range c.Obl {
		obl += fmt.Sprintf(" %s:%s/r%d/b%d/%s", o.ID, o.Kind, o.Region, o.Box, o.Prefix)
	}
	adm := ""
	for app, ad := range w.admit {
		adm += fmt.Sprintf(" %s:closed=%v/reg=%d", app, ad.Closed, len(ad.Reg))
	}
	return fmt.Sprintf("op=%s/%s phase=%s land=%s src=%s rev=%s exec=%d obl=%d holder=%+v rec=%+v pending=%d console=%d obls=[%s] quiet=%v h2=%+v admission=[%s] residue=%v",
		c.Op.Kind, c.Op.ID, c.Op.Phase, c.Op.LandPhase, c.Op.SrcPhase, c.Op.RevPhase, c.Op.Exec, len(c.Obl), c.Holder, w.rec, len(w.pending), len(w.console), obl, c.Quiet, w.h2, adm, w.residue())
}

// ---------- running ----------

func runSeed(seed int64, verbose bool) *world {
	w := newWorld(seed, verbose)
	for i := 0; i < chaosSteps; i++ {
		w.step()
	}
	w.heal(false)
	for i := 0; i < calmSteps && !w.quiescent(); i++ {
		w.step()
		if i == 400 || i%400 == 399 {
			w.heal(true)
			if w.verbose {
				w.logf("calm step %d: %s", i, w.whyNotQuiescent())
			}
		}
	}
	w.completeDue(true)
	w.final()
	return w
}

func seedsFromEnv(def int) int {
	if s := os.Getenv("HOSTLIFE_SEEDS"); s != "" {
		if n, err := strconv.Atoi(s); err == nil {
			return n
		}
	}
	return def
}

// TestHostLifecycle runs the joint model. HOSTLIFE_SEEDS sets the seed
// count (the gate's default is short); HOSTLIFE_SEED=n runs one seed and
// prints its trace; HOSTLIFE_BOXES=2 runs the two-box shape (region HA
// L18) in place of the one-box one.
func TestHostLifecycle(t *testing.T) {
	sweep(t)
}

// TestHostLifecycleTwoBox runs the joint model with two boxes per region
// (region HA L18) beside the one-box sweep, so the gate covers both
// shapes. With HOSTLIFE_BOXES set, TestHostLifecycle runs that shape and
// this test stands aside.
func TestHostLifecycleTwoBox(t *testing.T) {
	if os.Getenv("HOSTLIFE_BOXES") != "" {
		t.Skip("HOSTLIFE_BOXES is set: TestHostLifecycle runs that shape")
	}
	saved := boxesPerRegion
	boxesPerRegion = 2
	defer func() { boxesPerRegion = saved }()
	sweep(t)
}

func sweep(t *testing.T) {
	if s := os.Getenv("HOSTLIFE_SEED"); s != "" {
		seed, _ := strconv.ParseInt(s, 10, 64)
		w := runSeed(seed, true)
		for _, line := range w.trace {
			t.Log(line)
		}
		for _, v := range w.violations {
			t.Errorf("seed %d: %s", seed, v)
		}
		return
	}
	n := seedsFromEnv(300)
	failed := 0
	kinds := map[string]int{}
	firstSeed := map[string]int64{}
	ops := 0
	stats := map[string]int{}
	for seed := int64(1); seed <= int64(n); seed++ {
		w := runSeed(seed, false)
		ops += w.opsDone
		for k, v := range w.stats {
			stats[k] += v
		}
		if len(w.violations) > 0 {
			failed++
			k := kind(w.violations[0])
			kinds[k]++
			if _, ok := firstSeed[k]; !ok {
				firstSeed[k] = seed
				t.Logf("seed %d: %s", seed, w.violations[0])
			}
		}
	}
	t.Logf("%d seeds, %d failed, %d completed operations; mutated=%q boxes=%d", n, failed, ops, mutated, boxesPerRegion)
	t.Logf("outcomes: %v", stats)
	if js, err := json.Marshal(stats); err == nil {
		t.Logf("COUNTERS %s", js) // structured, for the spec's generator (R10-9)
	}
	for k, c := range kinds {
		t.Logf("  %4d × %s (first seed %d)", c, k, firstSeed[k])
	}
	if failed > 0 {
		t.Fatalf("%d of %d seeds violated an invariant", failed, n)
	}
}

func kind(v string) string {
	if i := strings.Index(v, ": "); i >= 0 {
		v = v[i+2:]
	}
	// strip ids and numbers so a family groups together
	var b strings.Builder
	skip := false
	for _, r := range v {
		if r >= '0' && r <= '9' {
			if !skip {
				b.WriteByte('#')
			}
			skip = true
			continue
		}
		skip = false
		b.WriteRune(r)
	}
	v = b.String()
	if i := strings.Index(v, " holder="); i >= 0 {
		v = v[:i]
	}
	if len(v) > 110 {
		v = v[:110]
	}
	return v
}