package rollout

import (
	"context"
	"errors"
	"fmt"
	"time"

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

// wakeBook is the failure book's read half the planner needs.
type wakeBook interface {
	Blocks(acct, app, build, host string, served bool) bool
	HeldDown(acct, app, build, host string) bool
	HoldRanOn(host string) (HoldRan, bool)
}

// holdRanFor is host's hold-ran marker when it is one of (acct, app)'s: a
// marker left by another app's route under the same host name says nothing
// about this one's database.
func holdRanFor(b wakeBook, acct, app, host string) (HoldRan, bool) {
	h, ok := b.HoldRanOn(host)
	if !ok || h.Account != acct || h.App != app {
		return HoldRan{}, false
	}
	return h, true
}

type wakeInputs struct {
	Route  registry.Route
	OptDir string
	Policy AppPolicy
	Book   wakeBook
	// Demand is the canonical check for an adoption (spec §4.2), returning a
	// digest always. An error is a staging refusal: the staged file is
	// missing or is not the release's artifact.
	Demand func(version, stagedPath string) (demand, digest string, err error)
}

// HeldDownRefusal is the wake refusal for an instance where build failed in
// hold mode (spec §6.4, §8.3). It reaches the member's "wake failed" event,
// so it names the build only. A box that declined CARLOS_SWAP=overlap but
// honours the failure records refuses with the same words (internal/host).
func HeldDownRefusal(build string) string {
	return fmt.Sprintf("build %s failed in hold mode on this instance, so it stays down until you roll back or retry the build", build)
}

// BookUnreadableRefusal is the wake refusal when this box's failure records
// cannot be read or written: an unreadable book at arming (internal/host's
// refuseTexts), or a planned first wake whose intent could not be saved
// (Task 5). Approved text, shared rather than copied.
const BookUnreadableRefusal = "this box cannot start instances right now: its record of failed builds cannot be read"

// decideWake is spec §8.2's "when the wake picks a build", as a pure
// function. The installed build (root's stamp) boots unless a failure record
// holds it for this instance, in which case the good build boots from its
// staged path with its retained identity, and nothing is repointed. A build
// this instance has never served is a first wake; in overlap mode it carries
// the good build as its fallback. A build that failed in hold mode on this
// instance refuses the wake: nothing falls back automatically once a hold
// build has run against the database (spec §6.4, §8.3). An instance with no
// good build is never refused by a record alone (spec §8.5). An error is a
// staging refusal for the caller to report.
func decideWake(in wakeInputs) (activator.WakePlan, error) {
	rt := in.Route
	acct, app, host := release.RouteAccount(rt), rt.App, rt.Host
	installed := rt.Version
	if installed == "" {
		// Plan installs a cold host first; a plan for a host with nothing
		// installed refuses rather than let the live link run unplanned.
		return activator.WakePlan{Refuse: "no build is installed for this instance yet"}, nil
	}
	hasGood := rt.GoodVersion != "" && rt.GoodDigest != ""
	good := release.StagedIdentity{Version: rt.GoodVersion, StagedPath: release.VersionPath(in.OptDir, acct, rt.GoodVersion, app),
		Digest: rt.GoodDigest, Demand: rt.GoodDemand}
	recovery := activator.WakePlan{Boot: good, Mode: RecoveryMode(in.Policy, good.Demand)}
	if in.Book.HeldDown(acct, app, installed, host) {
		return activator.WakePlan{Refuse: HeldDownRefusal(installed)}, nil
	}
	if mark, ok := holdRanFor(in.Book, acct, app, host); ok {
		return decideHoldRan(in, mark, hasGood)
	}
	if hasGood && installed == rt.GoodVersion {
		return recovery, nil
	}
	// A record holds the installed build for this instance: boot the good
	// build instead. With no good build (every instance, until it first
	// serves after the upgrade) there is nothing to fall back to, which is
	// today's behaviour and not a regression (spec §8.5, §8.2): the wake
	// plans the installed build as a first wake with no fallback, as it
	// would had no record been written. Refusing it would turn one
	// sibling's failure into an outage for every sleeping sibling. A
	// hold-down was refused above and stays refused (spec §6.4).
	if hasGood && in.Book.Blocks(acct, app, installed, host, false) { // rolloutsim:M5
		return recovery, nil
	}
	path := release.VersionPath(in.OptDir, acct, installed, app)
	demand, digest, err := in.Demand(installed, path)
	if err != nil {
		if hasGood {
			return recovery, err
		}
		return activator.WakePlan{Refuse: fmt.Sprintf("build %s is not staged correctly on this box", installed)}, err
	}
	mode := AttemptMode(in.Policy, demand)
	plan := activator.WakePlan{Boot: release.StagedIdentity{Version: installed, StagedPath: path, Digest: digest, Demand: demand},
		Mode: mode, FirstWake: true}
	if hasGood && mode == activator.ModeOverlap {
		plan.Fallback = &good // rolloutsim:M6
	}
	return plan, nil
}

// decideHoldRan is decideWake for an instance with a hold-ran marker (final
// review wave 2, N2; spec §6.4: once a hold build has executed against an
// instance's database, nothing falls back automatically on that instance).
// The good build never boots in place of the marked build, as a recovery
// or as a fallback, and the marker never lifts a failure record.
//   - the installed build is the marked build: it boots, in hold mode with
//     no fallback, unless a failure record holds it for the instance (Astra
//     R3 N7: spec §8.4 exempts an instance that passed on the build, not one
//     where it merely ran). That holds even when the marked build is also
//     the good build (Task 3 fix round 1): the instance has served it, so a
//     sibling's record does not hold it (Blocks with served), and booting
//     it is the marked-build rule, not a fallback.
//
// With a good build, for any other installed build:
//   - the installed build is the good build: refused. An installed good
//     build is no proof of an explicit rollback, because root's automatic
//     sibling repoint installs it too (ruling 205, Astra R3). The proof is
//     a signed pointer naming the good build, newer than the marker, and the
//     pointer pass (applyPointers) clears the marker when it sees one; the
//     next wake then boots the good build as an ordinary wake. The planner
//     itself reads no pointer.
//   - the installed build is held for the instance by a failure record:
//     refused. With the good build barred as well, the instance stays down
//     until a retry or a newer build.
//
// Otherwise the installed build boots as a first wake with no fallback: the
// marked build in hold mode, whose failure here is a hold-down, any other
// build in its own mode. With no good build a record alone refuses nothing
// (spec §8.5), so the marked build boots even when a record blocks it.
//
// The hold-down refusal above this call still comes first. The refusal
// words are HeldDownRefusal's for the marked build: no new member text.
func decideHoldRan(in wakeInputs, mark HoldRan, hasGood bool) (activator.WakePlan, error) {
	rt := in.Route
	acct, app, host := release.RouteAccount(rt), rt.App, rt.Host
	installed := rt.Version
	switch {
	case installed == mark.Build:
		if hasGood && in.Book.Blocks(acct, app, installed, host, installed == rt.GoodVersion) { // rolloutsim:M7
			return activator.WakePlan{Refuse: HeldDownRefusal(mark.Build)}, nil
		}
	case hasGood && installed == rt.GoodVersion:
		return activator.WakePlan{Refuse: HeldDownRefusal(mark.Build)}, nil
	case hasGood && in.Book.Blocks(acct, app, installed, host, false):
		return activator.WakePlan{Refuse: HeldDownRefusal(mark.Build)}, nil
	}
	path := release.VersionPath(in.OptDir, acct, installed, app)
	demand, digest, err := in.Demand(installed, path)
	if err != nil {
		// A staging refusal cannot boot the good build instead here, as it
		// does for an instance with no marker: that would be the fallback.
		return activator.WakePlan{Refuse: fmt.Sprintf("build %s is not staged correctly on this box", installed)}, err
	}
	mode := AttemptMode(in.Policy, demand)
	if installed == mark.Build {
		mode = activator.ModeHold
	}
	return activator.WakePlan{Boot: release.StagedIdentity{Version: installed, StagedPath: path, Digest: digest, Demand: demand},
		Mode: mode, FirstWake: true}, nil
}

// Plan is the activator's Planner (spec §8.2). It is called without
// activator.Service.mu held, before the wake's lease. It also snapshots the
// instance's delivered config revision into the plan (WakePlan.ConfigRev).
// The wake's spawn reads the config file after this moment, so activating
// that revision when the wake is healthy (OnGen, from the revision its
// event carries) can only under-claim, which costs one more config attempt,
// never a missed one.
//
// Every plan gets a token of its own (WakePlan.Token, final review F4),
// taken before anything is registered for it. The activator carries it
// onto every event of the wake, and OnGen acts on a host's wake state
// (Manager.planned, observing, the config activation) only for the token
// that state belongs to, so a late event of an older wake never touches a
// newer one's.
//
// A Refuse text reaches the member's "wake failed" event (the activator's
// wake error), so it names builds only, never an error's text.
//
// Every plan for a managed host, refused or not, replaces the host's entry
// in Manager.planned, so prune keeps the failure record of a first wake's
// build until that wake is over. The installed build is registered there
// before the planner's slow reads (planRow), so a Pass that runs while they
// do keeps a record they could no longer see.
//
// The row is read again after those reads. A Pass whose snapshot preceded
// the registration could have pruned the build's record only if no route
// installed the build when that Pass listed them, so this host's row had
// already moved off it, and the read after the registration sees that. Such
// a plan is made once more against the fresh row. A row that moves again
// during the replan refuses the wake rather than chase the row: two installs
// inside one plan are installs still landing, and the activator's next wake
// plans against whatever settled.
func (m *Manager) Plan(ctx context.Context, rt registry.Route) (plan activator.WakePlan) {
	if !Managed(rt) {
		return activator.WakePlan{}
	}
	// Joined by Close for its whole length (spec §8.6, Task 11 fix round 1):
	// a plan under way may still write its wake's intent. Deferred first, so
	// it is given back last, after beginWake below has counted the wake
	// whose intent it wrote: the count never touches zero in between.
	m.enter(false)
	defer m.leave()
	tok, planAt := m.newWakeToken(rt.Host)
	defer func() {
		plan.Token = tok
		// Write-ahead (Task 5, Astra R1 #5): a first wake's intent is on disk
		// before the activator takes the lease, with no activator lock held
		// here. It ends on the wake's terminal event (OnGen). A plan whose
		// intent could not be written boots nothing, with the approved
		// refusal for a failure book this box cannot keep, and is
		// registered as the refusal it now is.
		plan = m.beginWake(rt, plan, tok, planAt)
		m.notePlanned(rt, plan, tok)
		// The book's snapshot is taken here, at the plan's end, after every
		// book read the planner made (decideWake, the replan): the wake's
		// failure is stale only after a retry the plan could not have seen
		// (ruling 191, Task 4 fix round 1). A retry landing during the slow
		// reads (Materialize, the policy load) let this plan boot the build,
		// so the build's failure in it is a real one. Its good write clears
		// only markers set before here (Astra R3 N6): the exec follows the
		// plan, and the plan's check re-reads the marker there.
		m.noteWakeStart(rt.Host, tok, planAt, m.d.Book.Snapshot(), plan)
	}()
	if m.attemptRunning(rt.Host) {
		// The instance's attempt has not reported yet (spec §6.4, §8.1): a
		// hold that went down, or a swap whose entry stopped under it, is
		// recorded by done() before anything wakes the instance, so no wake
		// can boot the good build ahead of a hold-down record.
		return activator.WakePlan{Refuse: movingRefusal}
	}
	if rt.Version == "" {
		// A cold managed host (spec §6.4, §8.4): its build is installed and
		// stamped first, and the wake is planned against that installed
		// identity, with its failure records and its first-wake canary.
		// Nothing managed ever runs through an empty plan: the live-link
		// wake would materialize the channel head and exec it unplanned.
		if m.d.Materialize == nil {
			return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
		}
		if err := m.d.Materialize(ctx, rt.Host); err != nil {
			m.d.Logf("rollout: %s: install before its wake: %v", rt.Host, err)
			return activator.WakePlan{Refuse: "its build could not be installed"}
		}
		fresh, err := m.d.Routes.Get(rt.Host)
		if err != nil || fresh == nil || fresh.Version == "" {
			if err != nil {
				m.d.Logf("rollout: %s: read the row after its install: %v", rt.Host, err)
			}
			return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
		}
		rt = *fresh
	}
	for again := false; ; again = true {
		plan = m.planRow(ctx, rt, tok)
		fresh, err := m.d.Routes.Get(rt.Host)
		if err != nil || fresh == nil {
			// Nothing confirms the row the plan was made against, so nothing
			// may boot from it: the same answer as an unreadable row after a
			// cold install.
			if err != nil {
				m.d.Logf("rollout: %s: read the row after its plan: %v", rt.Host, err)
			}
			return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
		}
		if fresh.Version == rt.Version {
			return plan
		}
		if again {
			m.d.Logf("rollout: %s: its installed build moved twice while its wake was planned (%s, now %s); refusing the wake", rt.Host, rt.Version, fresh.Version)
			return activator.WakePlan{Refuse: "no build is installed for this instance yet"}
		}
		m.d.Logf("rollout: %s: its installed build moved from %s to %s while its wake was planned; planning again", rt.Host, rt.Version, fresh.Version)
		rt = *fresh
	}
}

// movingRefusal is a wake's refusal while the instance's attempt has not
// reported (Plan's attemptRunning check, and beginWake meeting the
// attempt's intent between its slot freeing and its End).
const movingRefusal = "this instance is still moving to another build"

// beginWake writes the intent of plan, a first wake planned with token tok
// at planAt (Task 5 rule 1), and returns the plan to hand the activator:
// plan itself, or a refusal when the intent could not be written. Any
// other plan writes nothing. The intent's mode is the plan's: the activator
// runs a planned wake in the mode it was planned in (a foreign-layout row
// gets one generation at a time, so no override applies; its events carry
// plan.Mode), and that is the mode a failure of the wake records.
func (m *Manager) beginWake(rt registry.Route, plan activator.WakePlan, tok uint64, planAt time.Time) activator.WakePlan {
	if !plan.FirstWake || plan.Boot.Version == "" || plan.Refuse != "" {
		return plan
	}
	in := Intent{Account: release.RouteAccount(rt), App: rt.App, Build: plan.Boot.Version, Mode: string(plan.Mode),
		ID: m.wakeExecID(tok), Start: planAt}
	err := m.d.Book.Begin(rt.Host, in)
	if err == nil {
		// The intent is on disk: Close now waits for the terminal event
		// that ends it (OnGen, leaveWake). A refused plan, and a Begin that
		// failed, wrote nothing and is never counted.
		m.enterWake(tok)
		return plan
	}
	m.d.Logf("rollout: %s: write the intent of its first wake of %s: %v; refusing the wake", rt.Host, plan.Boot.Version, err)
	if errors.Is(err, ErrIntentStands) {
		// The instance's attempt finished and has not yet ended its intent:
		// it is still, for a moment, moving. Not a book fault.
		return activator.WakePlan{Refuse: movingRefusal, Token: tok}
	}
	return activator.WakePlan{Refuse: BookUnreadableRefusal, Token: tok}
}

// planRow is one plan against rt (Plan's body, after the cold install), for
// the plan whose token is tok. It registers rt's installed build as the
// host's in-flight planned build before any slow read (the policy, the
// book's reading in decideWake, the canonical demand, the failing pointer);
// Plan's deferred notePlanned replaces or clears that entry with what the
// plan finally boots.
func (m *Manager) planRow(ctx context.Context, rt registry.Route, tok uint64) activator.WakePlan {
	m.notePlanning(rt, tok)
	acct, app := release.RouteAccount(rt), rt.App
	in := wakeInputs{Route: rt, OptDir: m.d.OptDir, Policy: m.policies.get(ctx, acct, app), Book: m.d.Book,
		Demand: func(ver, path string) (string, string, error) {
			// The same reading an adoption attempt gets (resolve): only a
			// staging fault is an error, and anything unverifiable is hold.
			return m.stagedDemand(ctx, rt, ver, path)
		}}
	// The delivered revision now, before the wake's spawn reads the file.
	// It travels with the plan and comes back on the wake's events, so no
	// later plan for the host can change what this wake activates.
	m.mu.Lock()
	rev := m.track(rt.Host).delivered
	m.mu.Unlock()
	plan, err := decideWake(in)
	plan.ConfigRev = rev
	if err != nil {
		m.stagingRefused(rt, rt.Version, err)
	}
	if plan.FirstWake && plan.Boot.Version != "" {
		// Read now, before the wake runs, so a failure of the build is
		// published with no blocking read (buildFailed, spec §8.1).
		plan.FailedUnder = m.failedUnder(ctx, rt, plan.Boot.Version)
	}
	if plan.Boot.StagedPath != "" {
		plan.Check = m.planCheck(rt, plan)
	}
	return plan
}

// planCheck is a plan's selection gate (WakePlan.Check), asked by the
// activator at each planned exec, after the lease and the restore (spec
// §7.4, §8.1–8.4): a first wake's build that a failure record now holds for
// this instance (a sibling's failure since the plan, or a hold-down) may not
// boot. The good build always may: it is what a held build falls back to.
// An instance with no good build is refused only by a hold-down: a record
// alone would not move it onto anything, and it has no fallback by design
// until it first serves (spec §8.5), so refusing it would be an outage.
//
// A hold-ran marker (final review wave 2, N2), set since the plan or
// before it, applies decideHoldRan's rule at the exec:
//   - the marked build may not boot while it is held down there or, with a
//     good build, while a failure record holds it for the instance (Astra
//     R3 N7: the marker lifts no record). When it is also the good build it
//     is under this rule only as the installed build (Task 3 fix round 1),
//     and a record does not hold a build the instance served;
//   - otherwise the good build may not boot, whatever the route installs:
//     as a recovery, a fallback, an untried first wake's stand-in, or the
//     installed build itself (ruling 205: only the pointer pass's clearing
//     of the marker on a signed rollback lets it boot again, and the check
//     reads no pointer);
//   - any other build is checked as without a marker.
//
// The book is in memory; nothing here blocks.
func (m *Manager) planCheck(rt registry.Route, plan activator.WakePlan) func(build string) error {
	acct, app, host := release.RouteAccount(rt), rt.App, rt.Host
	boot := plan.Boot.Version
	hasGood := rt.GoodVersion != "" && rt.GoodDigest != ""
	// The texts reach the member's "wake failed" event (memberSafeWakeReason
	// passes a planned wake's error through).
	held := func(build string) error {
		return fmt.Errorf("build %s failed on this box while this instance was starting, so it was not started", build)
	}
	return func(build string) error {
		if mark, ok := holdRanFor(m.d.Book, acct, app, host); ok {
			isGood := hasGood && build == rt.GoodVersion
			switch {
			case build == mark.Build && (!isGood || rt.Version == mark.Build):
				if m.d.Book.HeldDown(acct, app, build, host) || (hasGood && m.d.Book.Blocks(acct, app, build, host, isGood)) {
					return held(build)
				}
				return nil
			case isGood:
				return errors.New(HeldDownRefusal(mark.Build))
			}
		}
		if !plan.FirstWake || build != boot {
			return nil
		}
		if m.d.Book.HeldDown(acct, app, build, host) || (hasGood && m.d.Book.Blocks(acct, app, build, host, false)) {
			return held(build)
		}
		return nil
	}
}

// attemptRunning reports whether host has an attempt in flight: admitted
// and not yet finished by done().
func (m *Manager) attemptRunning(host string) bool {
	s := m.schedulerOf(host)
	if s == nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	h := s.q.hosts[host]
	return h != nil && h.running != nil
}

// OnGen is the activator's OnGen: a planned wake's outcome (spec §6.4, §8.2,
// §8.5). A healthy wake records the serving build and activates the config
// it started with; a first wake becomes the good build only once it has
// survived its 10 s; a failed one goes through the cohort-failure
// transition, held down when it was a hold build. It is never called with
// activator.Service.mu held (emitGen), which served and publishServing need:
// they take goodMu and call back into Current.
//
// The host's wake state is the event's only when the token matches (final
// review F4): the planned entry and the observation each end only by an
// event carrying their own token, and only the host's newest wake (the
// token of its last plan) opens an observation or activates a config
// revision, the one its event carries. A late event of an older wake (an
// observer's delayed GenCancelled, say) leaves a newer wake's registration,
// observation and config alone. What an event says about its build is
// still acted on whoever's it is: a failure is recorded, and a serving or
// good build is published only while its generation is the one the
// activator serves (acknowledged). A token of zero matches nothing.
func (m *Manager) OnGen(ev activator.GenEvent) {
	// Joined by Close (spec §8.6): a wake's terminal event writes its outcome
	// and ends its intent, which must be on disk before the agent exits.
	m.enter(false)
	defer m.leave()
	rt := ev.Route
	// The wake's start, its plan: a hold-ran marker it leaves carries the
	// plan time (ruling 205), and its good write and its failure are ordered
	// against the book by the plan's snapshot (Astra R3 N6, ruling 191).
	// Every kind but a first wake's healthy is the wake's last event, which
	// lets the entry go: an ordinary wake (an untried first wake's good
	// build included) reports healthy once and nothing after.
	ended := ev.Kind != activator.GenHealthy || !ev.FirstWake
	started := m.wakeStarted(rt.Host, ev.Token, ended)
	m.mu.Lock()
	newest := ev.Token != 0 && m.wakeTok[rt.Host] == ev.Token
	// The wake has an outcome: its build is no longer only planned, so its
	// failure record need not outlive the install for its sake.
	if w, ok := m.planned[rt.Host]; ok && ev.Token != 0 && w.token == ev.Token {
		delete(m.planned, rt.Host)
	}
	switch {
	case ev.Kind == activator.GenHealthy && ev.FirstWake:
		if newest {
			m.observing[rt.Host] = ev.Token
		}
	case ev.Kind != activator.GenHealthy:
		// GenSettled, GenFailed, GenFellBack and GenCancelled (an
		// intentional stop ended the observation with no verdict) all close
		// the observation their wake opened. The switch below has no
		// recording arm for GenCancelled, so a cancelled observation records
		// nothing (spec §6.4).
		if tok, ok := m.observing[rt.Host]; ok && ev.Token != 0 && tok == ev.Token {
			delete(m.observing, rt.Host)
		}
	}
	m.mu.Unlock()
	defer m.changed() // after the event's own work, with no lock held
	switch ev.Kind {
	case activator.GenHealthy:
		// A wake that is not a first wake includes the good build booted
		// untried under Task 9's verdict floor: an ordinary wake of the good
		// build, recorded as served, with nothing against the build it skipped.
		if !ev.FirstWake {
			m.served(rt.Host, ev.Gen, started.snap)
		}
		// A first wake is published when it settles: until then the
		// activator hides it (Current), so it is not yet acknowledged.
		m.wakeConfigActivated(rt.Host, ev.Token, ev.ConfigRev)
	case activator.GenSettled:
		m.served(rt.Host, ev.Gen, started.snap)
	case activator.GenFellBack:
		// The failure hold first (spec §8.1): nothing that reads the
		// serving build or the active config may see the fallback before the
		// failed build is held.
		m.buildFailed(rt, ev, started)
		m.publishServing(rt.Host, ev.Gen)
		m.wakeConfigActivated(rt.Host, ev.Token, ev.ConfigRev)
		m.emit(rt, EvFallbackServed, "warn", textFallback(ev.Failed.Version, ev.Gen.Build))
	case activator.GenFailed:
		m.buildFailed(rt, ev, started)
	case activator.GenCancelled:
		// An intentional stop ended the first wake's observation, or the
		// wake's lifecycle ended it after its build ran (spec §6.4): no
		// verdict on the build, so nothing is recorded or published. A hold
		// build that ran leaves the instance's hold-ran marker (final review
		// wave 2, N2): its database may be the build's now, so the next wake
		// must not boot the good build. A fact about the instance, like a
		// failure, so it is not token-scoped.
		if ev.Executed && ev.FirstWake && ev.Mode == activator.ModeHold && ev.Gen.Build != "" {
			// At is the wake's plan time, its execution's start (ruling 205):
			// a rollback signed while the wake ran is newer than the marker.
			if started.at.IsZero() {
				m.d.Logf("rollout: %s: the start of the wake that ran %s is not known; its hold-ran marker is stamped now", rt.Host, ev.Gen.Build)
			}
			mark := HoldRan{Account: release.RouteAccount(rt), App: rt.App, Build: ev.Gen.Build, Exec: m.wakeExecID(ev.Token), At: started.at}
			if err := m.d.Book.MarkHoldRan(rt.Host, mark); err != nil {
				m.d.Logf("rollout: %s: write hold-ran marker for %s: %v (held in memory)", rt.Host, mark.Build, err)
			}
			m.pointerReadDue(appKey{mark.Account, mark.App})
		}
	}
	// The wake's intent ends with the first event that settles its planned
	// build's fate (Task 5 rule 2, plan review #1), after that event's own
	// book write above (buildFailed's record, the hold-ran marker): settled,
	// failed, fell back, cancelled (with or without a run), and healthy when
	// it is not a first wake's, which is the good build booted under the
	// verdict floor with the planned build never run. A first wake's healthy
	// event opens its observation and is not the end. An older wake's event
	// carries its own token, so it never ends a newer wake's intent. Not
	// under activator.Service.mu (emitGen's rule).
	if ended {
		if err := m.d.Book.End(rt.Host, m.wakeExecID(ev.Token)); err != nil {
			m.d.Logf("rollout: %s: end the intent of its wake: %v (it stays on disk until a save lands)", rt.Host, err)
		}
		// The wake's count in Close's join goes with the End (beginWake
		// took it). This call is itself counted until it returns, so the
		// join cannot see zero before the End above is written. A wake
		// replaced by a newer plan has its intent replaced on disk, but its
		// own terminal event still ends its count here: the activator sends
		// exactly one per planned wake (afterWake, cancelPlanned).
		m.leaveWake(ev.Token)
	}
}

// wakeConfigActivated is configActivated for a planned wake's event (final
// review wave 2, N3): the revision activates only while tok is still the
// host's newest wake token, and that check and the activation are one
// critical section of m.mu. A check made earlier (OnGen's newest) can go
// stale while the event does its other work (served's registry write): a
// newer wake could plan and activate its own revision meanwhile, and this
// older one would then put its revision back. The held ack is written, and
// the queue's pending config attempt cleared, after the lock is released.
func (m *Manager) wakeConfigActivated(host string, tok uint64, rev string) {
	if rev == "" || tok == 0 {
		return
	}
	m.mu.Lock()
	if m.wakeTok[host] != tok {
		m.mu.Unlock()
		return
	}
	write, _ := m.track(host).activate(rev)
	ack := m.ackLocked(write)
	m.mu.Unlock()
	m.activationDone(host, rev, ack)
}

// buildFailed is a failed first wake's cohort-failure transition (spec
// §8.1, §8.2 step 4). started is the wake's start (wakeStarted): a wake
// planned before a retry of its build fails stale (ruling 191), and runs
// none of the transition.
func (m *Manager) buildFailed(rt registry.Route, ev activator.GenEvent, started wakeStart) {
	// No blocking read before the record (spec §8.1): the failing pointer's
	// PromotedAt was read when the wake was planned (WakePlan.FailedUnder).
	rec := FailureRecord{Account: release.RouteAccount(rt), App: rt.App, Build: ev.Failed.Version, Host: rt.Host,
		Cause: string(activator.CauseRelease), Mode: string(ev.Mode), Reason: ev.Reason,
		PromotedAt: ev.FailedUnder, At: m.d.Now(),
		// The wake's start for a stale failure's marker: Record sets one in
		// place of a hold-down when the failure is pre-retry (ruling 191).
		Seq: started.snap, HoldRanAt: started.at, HoldRanExec: m.wakeExecID(ev.Token)}
	if rec.PromotedAt.IsZero() {
		// No pointer named the build when the wake was planned: the
		// failure's own time, so only a pointer signed after it clears the
		// record (as done() does for an attempt).
		rec.PromotedAt = rec.At
	}
	if ev.Mode == activator.ModeHold {
		// A hold build executed against this instance's database: nothing
		// falls back on it automatically (spec §6.4, §8.3).
		rec.HeldDown = []string{rt.Host}
	}
	s := m.schedulerOf(rt.Host)
	if s == nil {
		// No scheduler for the host's app: the record still halts the
		// cohort, and the halt is still announced, on the same condition
		// cohortFailed announces it (a newly recorded failure, never a
		// pre-retry one or one merged into a record that stands). Without
		// it a member whose first wake failed here saw the fallback event
		// and never the halt (Astra R3 flag triage T11).
		res, err := m.d.Book.Record(rec)
		if err != nil {
			m.d.Logf("rollout: write failure record %s: %v (held in memory)", rec.Build, err)
		}
		if res == Recorded {
			m.d.Logf("rollout: %s/%s build %s failed on %s: %s", rec.Account, rec.App, rec.Build, rec.Host, rec.Reason)
			m.emit(rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, reasonWord(rec.Reason)))
		}
		return
	}
	m.cohortFailed(s, rt, rec)
}