package rollout

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"log"
	"os"
	"slices"
	"strconv"
	"strings"
	"sync"
	"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"
)

// Activator is the part of *activator.Service the scheduler drives. Swap is
// plan 3's, for instances; ReplaceWorker is Task 11's same-entry
// stop-then-start for workers and native sidecar members (spec §7.9), and
// PublishHeldServing releases a worker fallback's serving identity once
// done() has recorded the failure it falls back from (spec §8.1).
// EffectiveMode is the mode Swap will run an attempt of mode m on host in
// (an overlap on a foreign-layout row runs as a hold, spec §2.9), which the
// attempt's intent records before Swap runs (plan review finding 2).
type Activator interface {
	Current(host string) (activator.Gen, bool)
	Owns(host string) bool
	Swap(ctx context.Context, a activator.Attempt, newest func(host string) uint64) activator.SwapResult
	ReplaceWorker(ctx context.Context, a activator.Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) activator.SwapResult
	PublishHeldServing(host string)
	EffectiveMode(host string, m activator.SwapMode) activator.SwapMode
}

// Routes is the part of *registry.Registry the scheduler reads and writes.
type Routes interface {
	Get(host string) (*registry.Route, error)
	SetGoodBuild(host string, expect, next registry.GoodBuild) error
	SetServingVersion(host, ver string) error
}

// Deps is everything the manager needs from the box. Its funcs are always
// called without the manager's locks held.
type Deps struct {
	Act     Activator
	Routes  Routes
	Book    *FailureBook
	OptDir  string
	DataDir string
	// Box is this box's label (host.BoxLabel), named by the cohort-halted
	// event (spec §9.4: app, box, build, reason).
	Box string
	// Demand is release.CanonicalDemand bound to the deploy store and this
	// box's platform (spec §4.2).
	Demand func(ctx context.Context, acct, app, version, stagedPath string) (demand, digest string, err error)
	// Policy reads the app's default config bundle (spec §4.1).
	Policy func(ctx context.Context, acct, app string) (*appconfig.Manifest, error)
	// Pointer reads a channel's signed pointer, verified (release.Channel).
	// It is the retry signal (spec rev 14 §8.4) and the PromotedAt a failure
	// record keeps. Nil: records clear only by a newer installed build. It
	// is the raw signed read, not the per-box digest ratchet: only its
	// Version and PromotedAt are used, to clear failure records, and root's
	// install, which the ratchet guards, decides which build is installed.
	Pointer func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error)
	Events  activator.EventSink
	// Touch asks root for a stage pass (the stage relay, spec §7.7).
	Touch func()
	// Materialize installs and stamps a cold managed host's build (the exec
	// backend's Materialize hook) so a wake is planned against it (Task 10,
	// spec §6.4, §8.4). Nil: a cold managed host's wake is refused.
	Materialize func(ctx context.Context, host string) error
	// OnChange, when set, is told after every rollout transition (an
	// attempt's outcome, a wake's event, a newly installed build, a retry),
	// so the box republishes its status within seconds instead of at the
	// next tick (plan 6's deploy watch). It is a nudge: it is called with no
	// Manager or scheduler lock held, and it must not block.
	OnChange func()
	Now      func() time.Time
	Logf     func(format string, args ...any)
}

// pointerReadEvery bounds how often an app with a standing failure record
// has its channel pointers read for a retry. An app with no record costs no
// read at all.
const pointerReadEvery = 30 * time.Second

// stagingDeferral is how long an attempt refused as a staging error waits:
// one stage-relay window, for root to restage (spec §4.2, §7.7).
const stagingDeferral = 15 * time.Second

// abandonDeferral is how long a host waits after an attempt was abandoned
// while it was still the host's newest: a hold whose stop withdrew, a lost
// lease, a re-check that refused. Running the same attempt again at once
// would only repeat the stop cycle (plan 3's deferral to this plan). An
// attempt abandoned because a newer desire superseded it does not wait: the
// newer attempt goes as soon as it is admitted.
const abandonDeferral = 15 * time.Second

// Manager is the box's rollout scheduler: one goroutine per (account, app),
// fed by the agent pass, the adopt pass, `carlos restart` and config sync.
//
// Lock order: goodMu, then activator.Service.mu, then Manager.mu, then
// scheduler.mu, then FailureBook.mu. The activator (Current, Owns, Swap,
// ReplaceWorker) is never called while Manager.mu or a scheduler.mu is held.
// goodMu comes first because served, publishServing and writeGood call
// Act.Current under it, and observeGood takes Manager.mu and then a
// scheduler.mu under it. So never call served, publishServing or writeGood
// (or OnGen, which reaches them) while holding activator.Service.mu: that
// caller would hold Service.mu and wait for goodMu while another holds goodMu
// and waits for Service.mu.
type Manager struct {
	d        Deps
	ctx      context.Context
	policies *policyCache

	mu          sync.Mutex
	apps        map[appKey]*scheduler
	hostApp     map[string]appKey
	cfg         map[string]*configTrack
	retried     map[string]string // host -> build a retry cleared: its next attempt's cause is retry
	pointerRead map[appKey]time.Time
	// nudging holds the hosts a Nudge is working on; true asks it to look
	// once more for a nudge that arrived meanwhile.
	nudging map[string]bool
	// observing holds the hosts whose first wake is inside its 10 s
	// observation, each with the token of the wake it observes: opened by
	// that wake's healthy first-wake event, closed by an event carrying the
	// same token (OnGen). Status reports them as settling.
	observing map[string]uint64
	// planned holds each host whose planned first wake is in flight, and the
	// build it boots: from its Plan to its outcome (an OnGen event carrying
	// the plan's token), its next Plan, or the activator letting the host go
	// (a Pass that finds Owns false). prune keeps those builds' failure
	// records, because a wake's last selection check (planCheck) reads only
	// the book: a record pruned after the plan, once a promotion moved every
	// route off the build, would let the wake boot a build that failed on
	// this box. plannedSeq tells one registration from the next for that
	// Pass, which asks Owns off every lock.
	planned    map[string]plannedWake
	plannedSeq uint64
	// wakeTok is each host's newest wake token: the token of the last plan
	// made for it (final review F4). Tokens come from one counter, so they
	// are unique and only grow. An event carrying any other token is an older
	// wake's, and acts on nothing that belongs to the newer one.
	wakeTok    map[string]uint64
	tokenCount uint64
	// wakeAt holds, per host, the start of each planned wake that may still
	// report (wakeStart, by token): its plan time, which a hold-ran marker
	// the wake leaves carries as its At (ruling 205, Task 3 fix round 1), and
	// the failure book's snapshot taken with it, which orders the wake's
	// outcome against the book (Astra R3 N6, ruling 191). Kept per token,
	// not only the newest plan's, because neither a marker nor a failure is
	// token-scoped: a late event of an older wake still acts, as of its own
	// start. An entry goes with its wake's last event, with the host
	// (forget), or when the host has wakeStartsKept newer ones.
	wakeAt map[string][]wakeStart
	// execCount numbers the scheduled attempts' executions (nextExecID,
	// dispatch.ExecID).
	execCount uint64
	// inc is this manager's incarnation: 8 random bytes, hex, drawn once per
	// process start (NewManager). Every execution identity carries it
	// (nextExecID, wakeExecID), because the counters above start again at
	// zero in a new process while the markers and intents they named
	// survive in the book: without it the first wake after a restart could
	// be "w1" again, and markLocked would take the new execution for the
	// old one and keep its Seq and At, so a grant or a rollback that matched
	// the old execution would clear the new one (final review round 1,
	// finding 1, Astra).
	inc string

	// ackSeq numbers the config acks the tracks hand back, under mu, in the
	// critical section of the transition that releases each one: the
	// writer's order (ackLocked, plan review #10).
	ackSeq uint64
	// acks writes those acks, off the scheduler's slot and loop (Task 11).
	acks *ackWriter

	// goodMu serialises every serving- and good-build write with its
	// acknowledgement check (served, publishServing, writeGood), so an older
	// generation's write can never land after a newer one's (spec §8.5).
	goodMu sync.Mutex

	// The shutdown join (Close, spec §8.6). joinMu is a leaf: nothing else
	// is taken while it is held, so it may be taken under any other lock
	// (CoalesceRecycle takes it under activator.Service.mu and a
	// scheduler.mu). inflight counts the attempts' executions, the Plan and
	// OnGen calls under way, and the planned wakes in wakes; closed stops
	// admission; joined, when set, is closed once inflight is zero.
	joinMu   sync.Mutex
	closed   bool
	inflight int
	joined   chan struct{}
	// wakes holds the token of each planned wake whose intent beginWake
	// wrote and whose terminal event has not yet reached OnGen (Task 11 fix
	// round 1). Each is one count in inflight, taken and given back exactly
	// once whatever the order of plans and events.
	wakes map[uint64]struct{}
}

// NewManager builds the manager. Its schedulers run until ctx ends.
func NewManager(ctx context.Context, d Deps) *Manager {
	if d.Now == nil {
		d.Now = time.Now
	}
	if d.Logf == nil {
		d.Logf = log.Printf
	}
	m := &Manager{d: d, ctx: ctx, inc: newIncarnation(),
		apps: map[appKey]*scheduler{}, hostApp: map[string]appKey{}, cfg: map[string]*configTrack{},
		retried: map[string]string{}, pointerRead: map[appKey]time.Time{}, nudging: map[string]bool{}, observing: map[string]uint64{},
		planned: map[string]plannedWake{}, wakeTok: map[string]uint64{}, wakeAt: map[string][]wakeStart{}}
	m.policies = &policyCache{load: d.Policy, ttl: 30 * time.Second, now: d.Now, m: map[appKey]*policyEntry{}}
	m.acks = newAckWriter(ctx, d.Logf)
	return m
}

// Close is the scheduler's part of a graceful stop (spec §8.6: shutdown
// cancels and joins the whole operation set; Task 5 review). The agent calls
// it after the activator's Shutdown, which has stopped every entry and so
// cancelled any swap in flight. It stops admitting attempts, then waits,
// until ctx ends, for every attempt's execution to finish its done() (the
// outcome's book writes and the intent's End), for every Plan and OnGen
// call under way to return, and for every planned wake whose intent is on
// disk to deliver the terminal event that ends it, and then writes the acks
// still pending and stops the ack writer. A planned restart in the middle
// of a swap or a wake so leaves the outcome on disk and no intent behind,
// and the next agent reconciles nothing. A Close that runs out of ctx
// returns anyway: what was not finished is what OpenFailureBook reconciles.
//
// The planned wakes are counted here, not left to Shutdown (Task 11 fix
// round 1): Shutdown waits for each entry's e.ready, and wake() closes it
// before afterWake sends the terminal event, while the first-wake
// observer's cancellation is sent from its own goroutine. A wake already
// past doWake's context check when Shutdown began can also still plan and
// write its intent. Each of those reaches OnGen after Shutdown returns.
func (m *Manager) Close(ctx context.Context) {
	m.joinMu.Lock()
	m.closed = true
	var wait chan struct{}
	if m.inflight > 0 {
		if m.joined == nil {
			m.joined = make(chan struct{})
		}
		wait = m.joined
	}
	n := m.inflight
	m.joinMu.Unlock()
	if wait != nil {
		select {
		case <-wait:
		case <-ctx.Done():
			m.d.Logf("rollout: stopping with %d attempt(s), plan(s), wake(s) or wake event(s) still running: %v", n, ctx.Err())
		}
	}
	m.acks.close(ctx)
}

// enter counts one operation Close joins. An admission (admit true) is
// refused once Close has begun; a Plan or OnGen call is always counted, so
// one that arrives while Close waits is waited for too.
func (m *Manager) enter(admit bool) bool {
	m.joinMu.Lock()
	defer m.joinMu.Unlock()
	if admit && m.closed {
		return false
	}
	m.inflight++
	return true
}

// leave ends an operation enter counted.
func (m *Manager) leave() {
	m.joinMu.Lock()
	defer m.joinMu.Unlock()
	m.leaveLocked()
}

func (m *Manager) leaveLocked() {
	m.inflight--
	if m.inflight == 0 && m.joined != nil {
		close(m.joined)
		m.joined = nil
	}
}

// enterWake counts the planned wake with token tok, whose intent beginWake
// has just written, until leaveWake (its terminal event). Always counted,
// even once Close has begun: the intent is on disk, and only that event's
// OnGen ends it. A token already counted is not counted again.
func (m *Manager) enterWake(tok uint64) {
	m.joinMu.Lock()
	defer m.joinMu.Unlock()
	if _, ok := m.wakes[tok]; ok {
		return
	}
	if m.wakes == nil {
		m.wakes = map[uint64]struct{}{}
	}
	m.wakes[tok] = struct{}{}
	m.inflight++
}

// leaveWake gives back the count enterWake took for tok, once. Every
// terminal event of a planned wake ends here; one whose wake wrote no
// intent (a refused plan, a failed Begin, an ordinary wake), or one
// delivered again, finds nothing and changes nothing.
func (m *Manager) leaveWake(tok uint64) {
	m.joinMu.Lock()
	defer m.joinMu.Unlock()
	if _, ok := m.wakes[tok]; !ok {
		return
	}
	delete(m.wakes, tok)
	m.leaveLocked()
}

// closing reports whether Close has begun.
func (m *Manager) closing() bool {
	m.joinMu.Lock()
	defer m.joinMu.Unlock()
	return m.closed
}

// changed tells OnChange a transition happened. Every caller has released
// Manager.mu and any scheduler.mu first, so the callback may read Status.
func (m *Manager) changed() {
	if m.d.OnChange != nil {
		m.d.OnChange()
	}
}

func (m *Manager) scheduler(k appKey) *scheduler {
	m.mu.Lock()
	defer m.mu.Unlock()
	s := m.apps[k]
	if s == nil {
		book := m.d.Book
		s = &scheduler{m: m, key: k, kick: make(chan struct{}, 1),
			q: newAppQueue(k.acct, k.app, func(build, host string, served bool) bool {
				return book.Blocks(k.acct, k.app, build, host, served)
			})}
		m.apps[k] = s
		go s.loop(m.ctx)
	}
	return s
}

func (m *Manager) schedulerOf(host string) *scheduler {
	m.mu.Lock()
	defer m.mu.Unlock()
	k, ok := m.hostApp[host]
	if !ok {
		return nil
	}
	return m.apps[k]
}

// track is host's config track; the caller holds m.mu.
func (m *Manager) track(host string) *configTrack {
	t := m.cfg[host]
	if t == nil {
		t = &configTrack{}
		m.cfg[host] = t
	}
	return t
}

// Pass is the agent pass of spec §6.1: each managed route's installed build,
// stamped by root, against the build its current generation serves. routes
// must be the box's whole list, because Pass also prunes failure records no
// route installs any more, and lifts the halts those records held.
func (m *Manager) Pass(routes []registry.Route) {
	keys := m.observe(routes)
	m.forget(routes)
	m.applyPointers(routes, keys)
	m.dropUnownedWakes()
	m.prune(routes)
	m.liftClearedHalts()
}

// heldOnBox reports whether rt is a route this manager keeps state for:
// Managed, but with a passing status (draining, moving, landing) read as
// ready. A row passing through one of those for a Pass keeps its place in
// the queue, its pending restart or config, and its held ack; a row that is
// gone, or that stopped being an exec route this scheduler moves, does not.
func heldOnBox(rt registry.Route) bool {
	return Restrictable(rt)
}

// forget drops what the manager holds for hosts no longer on the box: a
// route removed or moved away, or one that now belongs to another app or
// stopped being managed. Without it such a host stayed in Status for the
// life of the process, possibly as queued, and could hold a deploy watch
// open (Task 13). A host with an attempt running is kept: the attempt's
// done() still needs its slot, and the next Pass after it finishes forgets
// it. Failure records and cohort halts are not per host and are not
// touched; prune and liftClearedHalts own those.
//
// Manager.mu, then each scheduler.mu: the documented order. Nothing here
// calls the activator.
func (m *Manager) forget(routes []registry.Route) {
	present := make(map[string]appKey, len(routes))
	for _, rt := range routes {
		if heldOnBox(rt) {
			present[rt.Host] = appKey{release.RouteAccount(rt), rt.App}
		}
	}
	m.mu.Lock()
	running := map[string]bool{}
	var gone []string
	for k, s := range m.apps {
		s.mu.Lock()
		for host, h := range s.q.hosts {
			if pk, ok := present[host]; ok && pk == k {
				continue
			}
			if h.running != nil {
				running[host] = true
				continue
			}
			delete(s.q.hosts, host)
			gone = append(gone, host)
		}
		s.mu.Unlock()
	}
	for _, host := range gone {
		if _, ok := present[host]; ok || running[host] {
			continue // moved to another app on this box, or still running there
		}
		delete(m.hostApp, host)
		delete(m.cfg, host)
		delete(m.retried, host)
		delete(m.observing, host)
		// A later event of the host's last wake then matches no token: the
		// host is gone, and there is nothing of its wake left to act on.
		delete(m.wakeTok, host)
		delete(m.wakeAt, host)
	}
	m.mu.Unlock()
	if len(gone) > 0 {
		m.d.Logf("rollout: forgot %d host(s) no longer on this box: %v", len(gone), gone)
		m.changed() // the hosts left the report
	}
}

// Nudge re-reads one host and wakes its scheduler. The activator sweep's
// drift arm calls it (spec §10, OnDrift) on a goroutine of its own, for every
// drifted entry on every sweep, with no dedupe of its own. So nudges for one
// host collapse here: while one is reading and observing, the others only
// ask it for one more look and return, and a slow read (the row, the
// policy) never collects a goroutine per sweep. A nudge for the build the
// host already wants changes nothing (wantBuild), so a queued or running
// attempt is neither queued again nor superseded.
func (m *Manager) Nudge(host string) {
	m.mu.Lock()
	if _, busy := m.nudging[host]; busy {
		m.nudging[host] = true
		m.mu.Unlock()
		return
	}
	m.nudging[host] = false
	m.mu.Unlock()
	for {
		if rt, err := m.d.Routes.Get(host); err == nil && rt != nil {
			m.observe([]registry.Route{*rt})
		}
		m.mu.Lock()
		again := m.nudging[host]
		if !again {
			delete(m.nudging, host)
			m.mu.Unlock()
			return
		}
		m.nudging[host] = false
		m.mu.Unlock()
	}
}

// Adopted replaces BounceAdopted as the adopt pass's restart callback when
// the flag is on: the pass has repointed the live symlink, and the scheduler
// takes the move from here, paced. It reports whether an awake instance will
// be attempted, so makeRestartRoute does not fall through to systemctl.
func (m *Manager) Adopted(host string) bool {
	if _, awake := m.d.Act.Current(host); !awake {
		return false
	}
	m.Nudge(host)
	return true
}

func (m *Manager) observe(routes []registry.Route) []appKey {
	type seen struct {
		rt      registry.Route
		serving string
		known   bool // false while waking or stopping: the slot is left alone
	}
	byApp := map[appKey][]seen{}
	for _, rt := range routes {
		if !Managed(rt) {
			continue
		}
		s := seen{rt: rt, known: true}
		if g, ok := m.d.Act.Current(rt.Host); ok && g.Build != "" {
			s.serving = g.Build
		} else if m.d.Act.Owns(rt.Host) {
			s.known = false
		}
		k := appKey{release.RouteAccount(rt), rt.App}
		byApp[k] = append(byApp[k], s)
	}
	now := m.d.Now()
	keys := make([]appKey, 0, len(byApp))
	moved := false
	for k, list := range byApp {
		keys = append(keys, k)
		policy := m.policies.get(m.ctx, k.acct, k.app)
		s := m.scheduler(k)
		retried := map[string]string{}
		m.mu.Lock()
		for _, x := range list {
			m.hostApp[x.rt.Host] = k
			if b, ok := m.retried[x.rt.Host]; ok {
				retried[x.rt.Host] = b
			}
		}
		m.mu.Unlock()
		rearm := false
		s.mu.Lock()
		s.q.concurrency = policy.Concurrency
		for _, x := range list {
			if x.known {
				// Serving state only from a stable entry. The desired build
				// and its revision are updated either way (spec §6.3): a
				// build root installs while a hold drains or a wake runs
				// supersedes the attempt's, so its gates (pinCheck, the
				// result check) refuse the older build. Admission still
				// waits for a stable entry (admit's view leaves a waking or
				// stopping host out).
				s.q.observe(x.rt.Host, x.serving, x.rt.GoodVersion, now)
			}
			cause := activator.CauseRelease
			if retried[x.rt.Host] == x.rt.Version {
				cause = activator.CauseRetry
			}
			if s.q.wantBuild(x.rt.Host, cause, x.rt.Version, now) {
				moved = true
				// A marked host's installed build moved: it may be a rollback
				// onto its good build, so the app's pointers are read at the
				// next applyPointers rather than up to pointerReadEvery later
				// (ruling 205). The book's lock is the last in the order.
				if _, ok := holdRanFor(m.d.Book, k.acct, k.app, x.rt.Host); ok {
					rearm = true
				}
			}
		}
		s.mu.Unlock()
		if rearm {
			m.pointerReadDue(k)
		}
		s.poke()
	}
	if moved {
		// A sleeper has no attempt to finish, so the pass that first sees
		// its new build is the news for the deploy watch.
		m.changed()
	}
	return keys
}

// pointerReadDue makes the next applyPointers read k's pointers at once
// rather than up to pointerReadEvery later (ruling 205). observe calls it
// when a marked host's installed build moves; done() and OnGen call it when
// they set a marker, because the rollback that superseded the hold build is
// signed and installed while that build runs, before any marker exists, so
// the pass that saw the install could not make the read due. The caller
// holds no lock (Manager.mu is taken here).
func (m *Manager) pointerReadDue(k appKey) {
	m.mu.Lock()
	delete(m.pointerRead, k)
	m.mu.Unlock()
}

// routeChannel is the channel a route follows: "" means stable, as
// release's own (unexported) routeChannel reads it.
func routeChannel(rt registry.Route) string {
	if rt.Channel == "" {
		return "stable"
	}
	return rt.Channel
}

// applyPointers is the retry signal of spec rev 14 §8.4. For each app with a
// standing failure record, at most every pointerReadEvery, it reads the
// signed pointer of every channel the app's managed routes on this box
// follow, and clears a record when a pointer names its build with a newer
// PromotedAt (`carlos promote --retry`). The pointer the build failed under,
// read again, clears nothing. A second retry of the same build is the same
// rule again: the retry's own failure recorded the newer PromotedAt, so only
// a pointer signed after it clears the record.
//
// It is also the rollback signal for hold-ran markers (ruling 205, Astra
// R3). An app with a marked host whose route installs its good build is read
// too, on the same schedule, and that host's marker is cleared when its
// channel's pointer names the good build with a PromotedAt newer than the
// marker: `carlos rollback` re-signs the pointer, root's automatic sibling
// repoint does not, so only an explicit rollback lets the good build boot
// over the marker. The signature is the pointer read's to verify
// (release.Channel). The marker's At is the start of the execution that set
// it (an attempt's dispatch, a planned wake's plan), so the rollback that
// superseded a running hold build is newer than it. The read is made due at
// once when a marked host's installed build moves and when a marker is set
// (pointerReadDue), so a rollback is not left waiting out pointerReadEvery.
func (m *Manager) applyPointers(routes []registry.Route, keys []appKey) {
	if m.d.Pointer == nil {
		return
	}
	now := m.d.Now()
	for _, k := range keys {
		w := pointerWorkFor(m.d.Book, routes, k)
		if w.empty() {
			continue
		}
		m.mu.Lock()
		due := now.Sub(m.pointerRead[k]) >= pointerReadEvery
		if due {
			m.pointerRead[k] = now
		}
		m.mu.Unlock()
		if !due {
			continue
		}
		// Each cleared record's build is requeued as a retry at once, before
		// the app's next channel is read (Task 10 review minor 1).
		_, unmarked := w.apply(m.ctx, m.d.Book, m.d.Pointer, routes, k, m.d.Logf, func(build string) { m.retryBuild(k, build, "") })
		if unmarked {
			m.changed()
		}
	}
}

// pointerWork is what one app's channel pointers could clear on this box
// (applyPointers): its failure records, and its marked hosts whose route
// installs the good build, by channel. An app with neither costs no read.
type pointerWork struct {
	recs     []FailureRecord
	rollback map[string][]registry.Route
}

func (w pointerWork) empty() bool { return len(w.recs) == 0 && len(w.rollback) == 0 }

// pointerWorkFor is k's pointerWork against book and the box's routes.
func pointerWorkFor(book *FailureBook, routes []registry.Route, k appKey) pointerWork {
	w := pointerWork{recs: book.ForApp(k.acct, k.app), rollback: map[string][]registry.Route{}}
	for _, rt := range routes {
		if !Managed(rt) || release.RouteAccount(rt) != k.acct || rt.App != k.app || rt.GoodVersion == "" || rt.Version != rt.GoodVersion {
			continue
		}
		if _, ok := holdRanFor(book, k.acct, k.app, rt.Host); ok {
			ch := routeChannel(rt)
			w.rollback[ch] = append(w.rollback[ch], rt)
		}
	}
	return w
}

// apply reads the signed pointer of every channel k's managed routes follow
// and applies applyPointers' two rules to the book: a record whose build a
// pointer names with a newer PromotedAt is cleared (ClearOnPointer), and so
// is the marker of a host whose channel's pointer names its installed good
// build with a PromotedAt after the marker's At (ClearHoldRanOnPointer). A
// read error is logged and that channel skipped; each read is bounded by
// readBound (Task 11 rule 2). onCleared is called with each cleared record's
// build as soon as the book has cleared it, before the next record or
// channel, with no lock held: the armed pass requeues the build there, so
// the retry is not left waiting on the rest of the app's reads (Task 10
// review minor 1). It reports how many records it cleared and whether it
// cleared a marker; what follows from a marker is the caller's.
func (w pointerWork) apply(ctx context.Context, book *FailureBook, read PointerRead, routes []registry.Route, k appKey,
	logf func(format string, args ...any), onCleared func(build string)) (cleared int, unmarked bool) {
	channels := map[string]bool{}
	for _, rt := range routes {
		if Managed(rt) && release.RouteAccount(rt) == k.acct && rt.App == k.app {
			channels[routeChannel(rt)] = true
		}
	}
	for ch := range channels {
		rctx, cancel := context.WithTimeout(ctx, readBound)
		ptr, err := read(rctx, k.acct, k.app, ch)
		cancel()
		if err != nil {
			logf("rollout: %s/%s channel %s: %v (retry not checked this pass)", k.acct, k.app, ch, err)
			continue
		}
		if ptr == nil {
			continue
		}
		for _, r := range w.recs {
			if ptr.Version != r.Build {
				continue
			}
			ok, err := book.ClearOnPointer(k.acct, k.app, r.Build, ptr.PromotedAt)
			if err != nil {
				logf("rollout: %s/%s clear record for %s: %v (cleared in memory)", k.acct, k.app, r.Build, err)
			}
			if ok {
				cleared++
				onCleared(r.Build)
			}
		}
		for _, rt := range w.rollback[ch] {
			if ptr.Version != rt.GoodVersion {
				continue
			}
			ok, err := book.ClearHoldRanOnPointer(rt.Host, k.acct, k.app, ptr.PromotedAt)
			if err != nil {
				logf("rollout: %s: clear hold-ran marker on a rollback to %s: %v (cleared in memory)", rt.Host, rt.GoodVersion, err)
			}
			if ok {
				logf("rollout: %s: a signed rollback to %s (pointer of %s) cleared its hold-ran marker", rt.Host, rt.GoodVersion, ptr.PromotedAt.UTC().Format(time.RFC3339))
				unmarked = true
			}
		}
	}
	return cleared, unmarked
}

// PointerRead reads a channel's signed pointer, verified (release.Channel
// with the box's store and root key): Deps.Pointer's shape.
type PointerRead = func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error)

// PointerReadEvery is pointerReadEvery, for the pointer pass of a box with
// no Manager (ApplyPointers), which paces itself.
const PointerReadEvery = pointerReadEvery

// ApplyPointers is applyPointers for a box with no Manager: one that
// declined CARLOS_SWAP=overlap but honours the records and hold-ran markers
// an armed run left (internal/host's failureHolds; Astra R2 ruling 194
// "startup-only records", Task 10). Every app with a managed route in routes
// and something its pointers could clear is read now, under the same two
// rules on the same book; the caller paces the pass, at most every
// PointerReadEvery. There is no scheduler to requeue a cleared build, so a
// cleared record is only logged. It reports whether anything was cleared.
func ApplyPointers(ctx context.Context, book *FailureBook, read PointerRead, routes []registry.Route, logf func(format string, args ...any)) bool {
	changed := false
	seen := map[appKey]bool{}
	for _, rt := range routes {
		k := appKey{release.RouteAccount(rt), rt.App}
		if !Managed(rt) || seen[k] {
			continue
		}
		seen[k] = true
		w := pointerWorkFor(book, routes, k)
		if w.empty() {
			continue
		}
		cleared, unmarked := w.apply(ctx, book, read, routes, k, logf, func(build string) {
			logf("rollout: %s/%s: a signed pointer re-promoted %s, so its failure record is cleared", k.acct, k.app, build)
		})
		changed = changed || cleared > 0 || unmarked
	}
	return changed
}

// retryBuild follows a cleared record: the next attempt onto build carries
// cause retry and runs as a fresh canary. host == "" is an app-wide retry,
// which reopens the halted cohort: its next attempt is a canary whatever
// wantBuild reports, because unhalt starts the cohort again. A host names
// `carlos instances retry`, whose attempt is that host's own canary;
// RetryHost has already bound the host's retry flag, in the step that
// cleared the book, and the flag is not set again here, where a forced
// attempt may already have started and spent it.
func (m *Manager) retryBuild(k appKey, build, host string) {
	s := m.scheduler(k)
	now := m.d.Now()
	s.mu.Lock()
	var hosts []string
	for name, h := range s.q.hosts {
		if h.wantBuild == build && (host == "" || name == host) {
			hosts = append(hosts, name)
			s.q.wantBuild(name, activator.CauseRetry, build, now)
		}
	}
	if host == "" {
		s.q.unhalt(build)
	}
	s.mu.Unlock()
	m.mu.Lock()
	for _, name := range hosts {
		m.retried[name] = build
	}
	m.mu.Unlock()
	m.d.Logf("rollout: %s/%s build %s retried (%v)", k.acct, k.app, build, hosts)
	s.poke()
	m.changed()
}

// RetryHost is the box half of `carlos instances retry <host>` (spec §8.4,
// §11): a grant the console signed (release.HostRetryGrant), which the
// caller has verified, naming the build whose restriction of host it lifts
// and that restriction's epoch and book as the box last reported them
// (plan review findings 5 and 13; final review round 1, finding 3). The
// book clears build's record for host, when one holds it, and host's
// hold-ran marker, when the marker names build, and only while the app's
// restriction generation is still epoch, the book owes no save, and the
// grant names this box's book (RetryHostAt; the crash-epoch hard limit in
// FailureBook's doc): a grant from before any later change of the app's
// restrictions, one replayed after it was spent, or one for another book's
// restriction at the same value, clears nothing. True means either was
// cleared.
//
// When the record was cleared and host's slot wants build, its next attempt
// is a canary. The book entry is cleared and the host's retry flag bound in
// one hold of the scheduler's lock (scheduler.mu before FailureBook.mu), the
// book first. Admission cannot run between the two, so the adoption onto
// the cleared build is the attempt the flag lets through, as a canary, and
// a restart or config attempt cannot start in between and spend the flag.
// The book also notes the retry's place in its sequence there (ruling 191):
// an outcome of an execution that began before it is stale, and takes the
// retry back no more than it halts the cohort (failCohort). No write is
// added under the scheduler's lock: the book saves as it did before.
//
// A grant naming another build than the slot's (the marker's build B while
// the slot wants A) clears B's restriction and leaves A's attempt alone; a
// cleared record of B still binds the host's canary entitlement for B,
// which the adoption onto B spends when the slot comes to want it, through
// B's halt if one stands (final review round 1, finding 5).
func (m *Manager) RetryHost(host, build, book string, epoch uint64) bool {
	m.mu.Lock()
	k, ok := m.hostApp[host]
	m.mu.Unlock()
	if !ok || build == "" {
		return false
	}
	s := m.scheduler(k)
	s.mu.Lock()
	want := ""
	if h := s.q.hosts[host]; h != nil {
		want = h.wantBuild
	}
	cleared, unmarked, cur, err := m.d.Book.RetryHostAt(k.acct, k.app, build, host, book, epoch)
	if cleared {
		// The canary entitlement is build's whatever the slot wants now
		// (final review round 1, finding 5): bound here, in the step that
		// cleared the book, and spent only by the adoption onto build.
		s.q.retryHost(host, build)
	}
	requeue := cleared && want == build
	if requeue {
		s.q.wantBuild(host, activator.CauseRetry, build, m.d.Now())
	}
	s.mu.Unlock()
	if err != nil && (cleared || unmarked) {
		m.d.Logf("rollout: retry %s on %s: %v (cleared in memory)", build, host, err)
	} else if err != nil {
		// The book owed a save and it failed again: nothing is cleared until
		// one lands (the crash-epoch hard limit, rule 2).
		m.d.Logf("rollout: retry %s on %s: the failure book cannot be saved (%v), so nothing was cleared", build, host, err)
	} else if !cleared && !unmarked && cur != epoch {
		m.d.Logf("rollout: retry %s on %s: the grant is for epoch %d, the restriction's is %d, so nothing was cleared", build, host, epoch, cur)
	} else if !cleared && !unmarked && book != m.d.Book.ID() {
		m.d.Logf("rollout: retry %s on %s: the grant is for book %q, not this box's, so nothing was cleared", build, host, book)
	}
	if requeue {
		m.retryBuild(k, build, host)
	} else if cleared || unmarked {
		m.changed()
	}
	return cleared || unmarked
}

// failedUnder is the PromotedAt a failure record keeps (spec rev 14 §8.4):
// the pointer that named the build when it failed. If that pointer cannot be
// read, or already names another build, it is zero, and the failure's own
// time stands in when the record is written: only a pointer signed after the
// failure can then clear the record, which is the safe direction.
//
// ctx is the caller's: the wake's in planRow, the attempt's in execute
// (Task 11 rule 2, Astra R1 Flag triage T10). A cancelled wake no longer
// waits on the store for its pointer, and no read waits past readBound.
func (m *Manager) failedUnder(ctx context.Context, rt registry.Route, build string) time.Time {
	if m.d.Pointer != nil {
		ctx, cancel := context.WithTimeout(ctx, readBound)
		defer cancel()
		ptr, err := m.d.Pointer(ctx, release.RouteAccount(rt), rt.App, routeChannel(rt))
		if err == nil && ptr != nil && ptr.Version == build {
			return ptr.PromotedAt
		}
	}
	return time.Time{}
}

// plannedWake is a planned first wake in flight (Manager.planned). token is
// its plan's (WakePlan.Token): only an event carrying it ends the entry.
type plannedWake struct {
	acct, app, build string
	seq              uint64
	token            uint64
}

// newWakeToken starts a plan for host: a fresh token, recorded as the
// host's newest. It is taken before the plan registers anything (Plan), so
// the plan's registration and every event of its wake carry it.
// It also returns the plan's time, the start of the wake's execution
// (noteWakeStart).
func (m *Manager) newWakeToken(host string) (uint64, time.Time) {
	now := m.d.Now()
	m.mu.Lock()
	defer m.mu.Unlock()
	m.tokenCount++
	m.wakeTok[host] = m.tokenCount
	return m.tokenCount, now
}

// newIncarnation draws a manager's incarnation (Manager.inc). crypto/rand
// does not fail on the platforms the agent runs on; if it ever did, the
// clock's nanoseconds still tell one process start from the next.
func newIncarnation() string {
	var b [8]byte
	if _, err := rand.Read(b[:]); err != nil {
		return strconv.FormatInt(time.Now().UnixNano(), 16)
	}
	return hex.EncodeToString(b[:])
}

// nextExecID is a scheduled attempt's execution identity (dispatch.ExecID):
// "a<incarnation>-<n>", n from this manager's counter. The "a" is what
// Begin's wake rule reads (wakeIntent).
func (m *Manager) nextExecID() string {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.execCount++
	return "a" + m.inc + "-" + strconv.FormatUint(m.execCount, 10)
}

// wakeExecID is a planned wake's execution identity (HoldRan.Exec):
// "w<incarnation>-<token>". Zero, an event of no known plan, has none. The
// incarnation is fixed for the manager's life, so every event of one wake
// names the identity its plan's intent was written with.
func (m *Manager) wakeExecID(tok uint64) string {
	if tok == 0 {
		return ""
	}
	return "w" + m.inc + "-" + strconv.FormatUint(tok, 10)
}

// wakeStart is a planned wake's start, by its token (Manager.wakeAt): its
// plan time (the plan's start) and the failure book's snapshot taken at the
// plan's end, after the planner's book reads (Plan).
type wakeStart struct {
	token uint64
	at    time.Time
	snap  uint64
}

// wakeStartsKept bounds Manager.wakeAt per host. Every planned wake ends
// with a terminal event (Task 5), which drops its entry; a plan the
// activator never ran (one Plan returned to a caller that then gave up)
// leaves its entry until this many newer plans push it out; a live wake is
// never that far behind its host's newest plan, because the activator waits
// for a planned wake's outcome before it plans again (outcomeFenceWait).
const wakeStartsKept = 4

// noteWakeStart keeps the start of a planned wake that boots a build (tok):
// a first wake's end can leave a hold-ran marker or a failure (OnGen), and
// any wake's healthy event writes the good build, which clears only markers
// older than its snapshot (Astra R3 N6). Any mode is kept: the activator
// reports the mode the wake ran in, which the plan does not fix.
func (m *Manager) noteWakeStart(host string, tok uint64, at time.Time, snap uint64, plan activator.WakePlan) {
	if tok == 0 || plan.Refuse != "" || plan.Boot.StagedPath == "" {
		return
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	starts := append(m.wakeAt[host], wakeStart{token: tok, at: at, snap: snap})
	if len(starts) > wakeStartsKept {
		starts = slices.Delete(starts, 0, len(starts)-wakeStartsKept)
	}
	m.wakeAt[host] = starts
}

// wakeStarted is the start of the wake whose token is tok, and drops its
// entry when ended (the wake's last event). A zero result means the start
// is not known (a token pushed out by newer plans, or a host forgotten): the
// book then stamps a marker with its own clock, later than the real start,
// so only a pointer signed later still lifts it; the good write clears no
// marker; and a failure is never taken as stale. Each is the safe
// direction.
func (m *Manager) wakeStarted(host string, tok uint64, ended bool) wakeStart {
	m.mu.Lock()
	defer m.mu.Unlock()
	if tok == 0 {
		return wakeStart{}
	}
	starts := m.wakeAt[host]
	i := slices.IndexFunc(starts, func(w wakeStart) bool { return w.token == tok })
	if i < 0 {
		return wakeStart{}
	}
	w := starts[i]
	if ended {
		if starts = slices.Delete(starts, i, i+1); len(starts) == 0 {
			delete(m.wakeAt, host)
		} else {
			m.wakeAt[host] = starts
		}
	}
	return w
}

// notePlanned records what host's newest plan boots: an entry for a first
// wake, none for anything else. Either way it replaces the host's previous
// plan, whose wake is over or is being planned again. A plan that is no
// longer the host's newest (tok) changes nothing: the newer plan owns the
// entry.
func (m *Manager) notePlanned(rt registry.Route, plan activator.WakePlan, tok uint64) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.wakeTok[rt.Host] != tok {
		return
	}
	if !plan.FirstWake || plan.Boot.Version == "" || plan.Refuse != "" {
		delete(m.planned, rt.Host)
		return
	}
	m.plannedSeq++
	m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: plan.Boot.Version,
		seq: m.plannedSeq, token: tok}
}

// notePlanning registers rt's installed build as host's in-flight planned
// build while its plan (tok) is still being made (Plan, planRow), so a Pass
// during the planner's slow reads keeps the build's failure record. A
// plan's end (notePlanned) replaces or clears the entry, and a newer seq
// keeps a Pass that snapshotted the old one (dropUnownedWakes) from
// dropping it.
func (m *Manager) notePlanning(rt registry.Route, tok uint64) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.wakeTok[rt.Host] != tok {
		return
	}
	if rt.Version == "" {
		delete(m.planned, rt.Host)
		return
	}
	m.plannedSeq++
	m.planned[rt.Host] = plannedWake{acct: release.RouteAccount(rt), app: rt.App, build: rt.Version,
		seq: m.plannedSeq, token: tok}
}

// dropUnownedWakes forgets the planned wakes of hosts the activator no
// longer owns: a planned wake of an older activator, or one whose terminal
// event is still on its way, leaves its entry behind, and only the entry's
// own end may let the record go. Owns is asked with no lock held (the lock
// order on Manager), and an entry is dropped only if no newer plan replaced
// it meanwhile.
//
// It never ends a wake's intent (plan review #1): a failed wake's entry goes
// before its GenFailed is delivered (activator's wake, then afterWake), so
// Owns false is no proof the outcome is written, and ending the intent there
// could lose a failure across a failed save and a restart. The intent ends
// on the wake's own terminal event (OnGen); every planned wake sends one.
func (m *Manager) dropUnownedWakes() {
	m.mu.Lock()
	seen := make(map[string]uint64, len(m.planned))
	for host, w := range m.planned {
		seen[host] = w.seq
	}
	m.mu.Unlock()
	gone := map[string]uint64{}
	for host, seq := range seen {
		if !m.d.Act.Owns(host) {
			gone[host] = seq
		}
	}
	if len(gone) == 0 {
		return
	}
	m.mu.Lock()
	for host, seq := range gone {
		if w, ok := m.planned[host]; ok && w.seq == seq {
			delete(m.planned, host)
		}
	}
	m.mu.Unlock()
}

// prune drops failure records for builds no route installs. A held-down
// instance's route still names its failed build, so its record stays (spec
// §8.4). Every route counts, whatever its status or shape: a row passing
// through "draining", "moving" or "landing" for one Pass, or one the
// scheduler does not manage, still names its build, and dropping the record
// then would lift the halt and let a later wake boot the failed hold build
// against its database. Pruning errs towards keeping. An empty route list is
// never authority to drop anything: registry.Open creates an empty database
// on a wrong path. A build a planned first wake still carries counts as
// installed (Manager.planned): the wake was planned before the install moved
// off it, and its exec may yet boot it. The installed callback runs under the
// book's lock and reads only this map, built first.
//
// The pass also retries a save the book owes (the crash-epoch hard limit,
// rule 6; SaveIfDirty): while one is owed status publishes no generation
// and no retry grant clears anything, so a box whose disk recovered must
// not wait for another change, pointer or grant to save. Prune retries it
// when it runs; a pass with no routes still calls SaveIfDirty, and so does
// one whose Prune landed, for a save owed since.
func (m *Manager) prune(routes []registry.Route) {
	if err := m.pruneBook(routes); err != nil {
		m.d.Logf("rollout: prune failure records: %v", err)
	} else if err := m.d.Book.SaveIfDirty(); err != nil {
		m.d.Logf("rollout: save the failure book: %v", err)
	}
}

// pruneBook is prune's Prune call, nil when there are no routes to judge by.
func (m *Manager) pruneBook(routes []registry.Route) error {
	if len(routes) == 0 {
		return nil
	}
	installed := map[[3]string]bool{}
	for _, rt := range routes {
		if rt.Version != "" {
			installed[[3]string{release.RouteAccount(rt), rt.App, rt.Version}] = true
		}
	}
	m.mu.Lock()
	for _, w := range m.planned {
		installed[[3]string{w.acct, w.app, w.build}] = true
	}
	m.mu.Unlock()
	return m.d.Book.Prune(func(acct, app, build string) bool { return installed[[3]string{acct, app, build}] })
}

// liftClearedHalts reopens every halted adoption cohort whose failure record
// no longer stands (spec §8.2, §8.4: a newer build, a retry or an install by
// hand lifts the hold). A halt lives in memory beside its record, so a record
// pruned after an install by hand would otherwise leave the build refused for
// good if it were installed again. failCohort writes the record and the halt
// under the same scheduler lock this reads them under. In the same hold it
// drops every retry entitlement the book no longer lists (final review round
// 2, finding 5): RetryHost binds one in the step that lists the host, so an
// entitlement and its listing go together.
func (m *Manager) liftClearedHalts() {
	m.mu.Lock()
	scheds := make([]*scheduler, 0, len(m.apps))
	for _, s := range m.apps {
		scheds = append(scheds, s)
	}
	m.mu.Unlock()
	for _, s := range scheds {
		var lifted []string
		s.mu.Lock()
		for key, c := range s.q.cohorts {
			build, adopt := strings.CutPrefix(key, "adopt:")
			if !adopt || !c.halted {
				continue
			}
			if _, held := m.d.Book.Lookup(s.key.acct, s.key.app, build); !held {
				s.q.unhalt(build)
				lifted = append(lifted, build)
			}
		}
		s.q.dropRetries(func(build, host string) bool {
			return m.d.Book.Retried(s.key.acct, s.key.app, build, host)
		})
		s.mu.Unlock()
		if len(lifted) > 0 {
			m.d.Logf("rollout: %s/%s: halt lifted for %v, their failure records are gone", s.key.acct, s.key.app, lifted)
			s.poke()
			m.changed()
		}
	}
}

// Bootstrap is spec §8.5's first start: every awake instance records the
// build it serves as good. Nothing is inferred from Version or PrevVersion.
// No execution of this agent started the generation, so the write has no
// book snapshot and clears no hold-ran marker (Astra R3 N6).
func (m *Manager) Bootstrap(routes []registry.Route) {
	for _, rt := range routes {
		if !Managed(rt) || rt.GoodVersion != "" {
			continue
		}
		if g, ok := m.d.Act.Current(rt.Host); ok && g.Build != "" {
			m.writeGood(rt.Host, g, 0)
		}
	}
}

// Restart is a `carlos restart` request for host (spec §10). id is the
// request's identity: the restart generation's etag, or a narrowed request's
// id. A sleeping instance is never woken. One that is waking, stopping or
// inside its first-wake observation keeps the request, which runs once it
// is stable (spec §6.2).
func (m *Manager) Restart(host, id string) bool {
	s := m.schedulerOf(host)
	if s == nil {
		return false
	}
	if !m.d.Act.Owns(host) {
		return false
	}
	s.mu.Lock()
	changed := s.q.wantRestart(host, id, m.d.Now())
	s.mu.Unlock()
	if changed {
		s.poke()
		m.changed()
	}
	return changed
}

// ConfigDelivered is config sync's report that host's config file now holds
// revision rev (spec §8.1). An instance with a process (awake, waking or
// stopping) gets a pending config attempt; if its wake turns out to have read
// rev, the wake's activation clears the attempt before it runs. A sleeping
// one needs none, because its next wake starts with the file: rev counts as
// what that wake will run, and config sync's ack goes out at once with
// today's hibernating acknowledgement (configTrack.deferAck). A host this
// manager does not manage is not tracked, so its ack goes out as before.
func (m *Manager) ConfigDelivered(host, rev string) {
	s := m.schedulerOf(host)
	if s == nil {
		return
	}
	// Asked before m.mu: the activator is never called under the manager's
	// lock (the lock order on Manager).
	awake := m.d.Act.Owns(host)
	m.mu.Lock()
	t := m.track(host)
	t.delivered, t.fails, t.lastErr = rev, 0, ""
	t.sleeping = !awake
	if !awake {
		t.active = rev
	}
	m.mu.Unlock()
	if !awake {
		// A sleeper needs no attempt, including one still pending for an
		// older revision it was awake for: its next wake starts with this
		// file. Left in place, that attempt would restart a generation
		// already running rev and then mark the older revision active.
		s.mu.Lock()
		s.q.dropConfig(host)
		s.mu.Unlock()
		return
	}
	s.mu.Lock()
	changed := s.q.wantConfig(host, rev, m.d.Now())
	s.mu.Unlock()
	if changed {
		s.poke()
		m.changed() // the revision is delivered and not yet active, which Status shows
	}
}

// DeferAck wraps config sync's ack writer: a revision is acknowledged only
// once it is active (spec §8.1). Every ack, one written at once included,
// goes through the manager's writer (Manager.acks), numbered here under m.mu
// with the transitions that release held ones: written in line, an ack at
// delivery could land before an older held ack the writer still has, and be
// overwritten by it (plan review #10). write is handed a context bounded by
// ackWriteBound.
func (m *Manager) DeferAck(write func(context.Context, appconfig.ConfigAck)) func(appconfig.ConfigAck) {
	return func(a appconfig.ConfigAck) {
		m.mu.Lock()
		now := m.ackLocked(m.track(a.Host).deferAck(a, write))
		m.mu.Unlock()
		m.submitAck(a.Host, now)
	}
}

// ackLocked numbers an ack a config track handed back (nil when none). The
// caller holds m.mu and is in the critical section of the transition that
// released it, so the numbers follow the order of the transitions, which
// the writer keeps per host (plan review #10).
func (m *Manager) ackLocked(write func(context.Context)) ackJob {
	if write == nil {
		return ackJob{}
	}
	m.ackSeq++
	return ackJob{seq: m.ackSeq, write: write}
}

// submitAck hands an ack to the writer; it never blocks.
func (m *Manager) submitAck(host string, j ackJob) {
	if j.write != nil {
		m.acks.submit(host, j)
	}
}

// resolve turns a dispatch into the Attempt Swap runs: its artifact and mode.
// Recovery onto the serving build uses the generation's captured identity,
// and a move back onto the good build uses the retained one; neither is
// checked against the canonical manifest (spec §4.2). An adoption checks the
// staged file exists and then the canonical demand. A missing staged file,
// or a staged file that is not the release's canonical artifact
// (release.ErrStagedDigestMismatch), is a staging refusal, never a build
// failure. Any other trouble reading the demand runs the attempt in hold:
// missing or unverifiable policy data never permits an overlap.
func (m *Manager) resolve(ctx context.Context, d dispatch, rt registry.Route) (activator.Attempt, error) {
	acct, app := release.RouteAccount(rt), rt.App
	policy := m.policies.get(ctx, acct, app)
	a := activator.Attempt{Host: d.ID.Host, Cause: d.ID.Cause, Build: d.ID.Build, ConfigRev: d.ID.ConfigRev,
		RestartID: d.ID.RestartID, SchedRev: d.SchedRev, Canary: d.Canary}
	if g, ok := m.d.Act.Current(rt.Host); ok && g.Build == d.ID.Build {
		a.StagedPath, a.Digest, a.Demand = g.StagedPath, g.Digest, g.Demand
		a.Mode = RecoveryMode(policy, g.Demand)
		return a, nil
	}
	path := release.VersionPath(m.d.OptDir, acct, d.ID.Build, app)
	if d.ID.Build == rt.GoodVersion && rt.GoodDigest != "" {
		a.StagedPath, a.Digest, a.Demand = path, rt.GoodDigest, rt.GoodDemand
		a.Mode = RecoveryMode(policy, rt.GoodDemand)
		return a, nil
	}
	demand, digest, err := m.stagedDemand(ctx, rt, d.ID.Build, path)
	if err != nil {
		return a, err
	}
	a.StagedPath, a.Digest, a.Demand = path, digest, demand
	a.Mode = AttemptMode(policy, demand)
	return a, nil
}

// stagedDemand is the canonical demand and digest of build's staged file at
// path, for an adoption attempt (resolve) and a first wake (Plan) alike
// (spec §4.2). A missing staged file, or one that is not the release's
// canonical artifact (release.ErrStagedDigestMismatch), is an error: a
// staging refusal, never a build failure. Any other trouble reading the
// demand is hold, on the staged bytes hashed here.
func (m *Manager) stagedDemand(ctx context.Context, rt registry.Route, build, path string) (demand, digest string, err error) {
	if _, err := os.Stat(path); err != nil {
		return "", "", err
	}
	// Bounded (Task 11 rule 2): a demand read that times out is a read
	// error, so the attempt or the wake runs in hold, as below.
	rctx, cancel := context.WithTimeout(ctx, readBound)
	demand, digest, err = m.d.Demand(rctx, release.RouteAccount(rt), rt.App, build, path)
	cancel()
	if errors.Is(err, release.ErrStagedDigestMismatch) {
		return "", "", err
	}
	if err != nil {
		// The canonical manifest could not be read or compared: hold, on
		// the bytes hashed below.
		m.d.Logf("rollout: %s: canonical demand of %s: %v (runs in hold)", rt.Host, build, err)
		demand, digest = release.DemandHold, ""
	}
	if digest == "" {
		// Nothing compared the staged bytes with the manifest (plan 4:
		// DemandHold with no digest), so the demand read describes nothing
		// verified: hold. The build still runs only on the bytes hashed
		// here, because plan 3's verifyStaged refuses an empty digest.
		demand = release.DemandHold
		if digest, err = release.FileSHA256(path); err != nil {
			return "", "", err
		}
	}
	return demand, digest, nil
}

// served records a generation that became current after its checks: the
// serving build, and the good build (spec §8.5). Both only while g is the
// generation the activator serves (acknowledged). snap is the book's
// snapshot from the start of the execution that produced g (writeGoodLocked).
func (m *Manager) served(host string, g activator.Gen, snap uint64) {
	if g.Build == "" {
		return
	}
	m.goodMu.Lock()
	defer m.goodMu.Unlock()
	if !m.acknowledged(host, g) {
		m.d.Logf("rollout: %s: generation %d (%s) is no longer served; not recording it", host, g.N, g.Build)
		return
	}
	m.setServingLocked(host, g)
	m.writeGoodLocked(host, g, snap)
}

// servedOneShot records the good build of a replacement whose build exited
// cleanly as a one-shot (spec §6.5, §8.5; Astra R3 flag triage T9). Nothing
// of it keeps running, so the activator has no current generation for the
// host to acknowledge, and served would drop the write. It skips only the
// acknowledged check: done() calls it before the attempt's slot is freed,
// so no newer attempt for the host can have been admitted, let alone
// completed, and the result is the host's newest outcome. Everything else
// is served's: goodMu, the generation's identity, and the book snapshot of
// the execution's start (snap), so only a hold-ran marker older than the
// attempt is cleared. A one-shot has no serving build to publish, and a
// worker's is the activator's alone (setServingLocked).
func (m *Manager) servedOneShot(host string, g activator.Gen, snap uint64) {
	if g.Build == "" {
		return
	}
	m.goodMu.Lock()
	defer m.goodMu.Unlock()
	m.writeGoodLocked(host, g, snap)
}

// setServingLocked writes an instance's serving build. A worker's is the
// activator's alone (Task 11, publishWorkerServing, ordered by incarnation),
// so two writers never race on one row. The caller holds m.goodMu.
func (m *Manager) setServingLocked(host string, g activator.Gen) {
	if g.Worker {
		return
	}
	if err := m.d.Routes.SetServingVersion(host, g.Build); err != nil {
		m.d.Logf("rollout: %s: record serving build %s: %v", host, g.Build, err)
	}
}

// publishServing records the serving build alone (a healthy first wake, a
// fallback), on the same terms as served.
func (m *Manager) publishServing(host string, g activator.Gen) {
	if g.Build == "" {
		return
	}
	m.goodMu.Lock()
	defer m.goodMu.Unlock()
	if !m.acknowledged(host, g) {
		return
	}
	m.setServingLocked(host, g)
}

// acknowledged reports whether g is the generation the activator serves
// for host right now (spec §8.5: the good build is written expecting the
// current one). A late completion or event of an older generation is not,
// so it can never overwrite what a newer one wrote. Generation numbers
// decide when both sides have one; a worker's (no number) is compared by
// its staged artifact.
func (m *Manager) acknowledged(host string, g activator.Gen) bool {
	cur, ok := m.d.Act.Current(host)
	if !ok {
		return false
	}
	if cur.N != 0 && g.N != 0 {
		return cur.N == g.N
	}
	return cur.StagedPath == g.StagedPath && cur.Digest == g.Digest
}

// writeGood is writeGoodLocked for a caller outside served: Bootstrap.
// Same terms: g must be acknowledged.
func (m *Manager) writeGood(host string, g activator.Gen, snap uint64) {
	m.goodMu.Lock()
	defer m.goodMu.Unlock()
	if !m.acknowledged(host, g) {
		return
	}
	m.writeGoodLocked(host, g, snap)
}

// writeGoodLocked replaces host's good build with g's identity,
// conditionally on the row still holding what was just read. goodMu
// serialises every writer and each has just proven g current, so a retry
// only covers a write from outside this manager; a second loss is logged
// and the next serve tries again. The caller holds m.goodMu.
//
// A good build recorded (written, or found already written) clears the
// host's hold-ran marker (final review wave 2, N2): the instance passed on
// g, so g is what a later wake may fall back to. Only a marker set before
// g's execution began (snap, Astra R3 N6): the write can be delayed after
// its acknowledgement check, and a hold build that ran on the host
// meanwhile (an attempt done() records without goodMu) left a marker the
// instance has not passed over.
func (m *Manager) writeGoodLocked(host string, g activator.Gen, snap uint64) {
	if g.Build == "" || g.Digest == "" {
		return
	}
	next := registry.GoodBuild{Version: g.Build, Digest: g.Digest, Demand: g.Demand}
	for range 2 {
		cur, err := m.d.Routes.Get(host)
		if err != nil || cur == nil {
			return
		}
		if cur.Good() == next {
			m.goodRecorded(host, next.Version, snap)
			return
		}
		err = m.d.Routes.SetGoodBuild(host, cur.Good(), next)
		if err == nil {
			m.goodRecorded(host, next.Version, snap)
			return
		}
		if !errors.Is(err, registry.ErrGoodBuildChanged) {
			m.d.Logf("rollout: %s: record good build %s: %v", host, g.Build, err)
			return
		}
	}
	m.d.Logf("rollout: %s: good build moved twice under this write, leaving it", host)
}

// goodRecorded follows a good-build write that landed (writeGoodLocked):
// a hold-ran marker older than the write's execution (snap) goes, and the
// queue learns the build.
func (m *Manager) goodRecorded(host, ver string, snap uint64) {
	if _, err := m.d.Book.ClearHoldRanBefore(host, snap); err != nil {
		m.d.Logf("rollout: %s: clear hold-ran marker after good build %s: %v (cleared in memory)", host, ver, err)
	}
	m.observeGood(host, ver)
}

func (m *Manager) observeGood(host, ver string) {
	s := m.schedulerOf(host)
	if s == nil {
		return
	}
	s.mu.Lock()
	if h := s.q.hosts[host]; h != nil {
		h.good = ver
	}
	s.mu.Unlock()
}

func (m *Manager) configActivated(host, rev string) {
	if rev == "" {
		return
	}
	m.mu.Lock()
	write, _ := m.track(host).activate(rev)
	ack := m.ackLocked(write)
	m.mu.Unlock()
	m.activationDone(host, rev, ack)
}

// activationDone is the rest of an activation of rev, after the config
// track has recorded it under m.mu: the queue's pending config attempt for
// rev is cleared, and the held ack (ack, empty when none), numbered in that
// critical section, is handed to the writer. Called with no lock held. The
// ack is not written here (Task 11 rule 3, Astra R1 Flag triage T8): done()
// calls this before the attempt's slot frees, and configSlept from the
// scheduler loop, and neither may wait on the store.
func (m *Manager) activationDone(host, rev string, ack ackJob) {
	if s := m.schedulerOf(host); s != nil {
		s.mu.Lock()
		s.q.configActive(host, rev)
		s.mu.Unlock()
	}
	m.submitAck(host, ack)
}

// configSlept is configActivated for an instance seen asleep with rev
// pending (spec §8.1): its next wake starts with the delivered file, so rev
// is what that wake runs and no config attempt is left to make. A held ack
// goes out as a sleeper's does at delivery, marked Hibernating. rev was read
// before the activator reported the instance asleep, so the file already
// held it when the instance stopped: this can only under-claim.
func (m *Manager) configSlept(host, rev string) {
	m.mu.Lock()
	t := m.track(host)
	if t.ack != nil && t.ackRev == rev {
		t.ack.Hibernating, t.ack.Restarted = true, false
	}
	write, _ := t.activate(rev)
	ack := m.ackLocked(write)
	m.mu.Unlock()
	m.activationDone(host, rev, ack)
}

// configFailure counts a failed attempt that carried config revision rev
// and returns its backoff. Only a failure of the delivered revision counts:
// one of an older revision says nothing about the newer one's attempts. The
// caller sets the queue's backoff (appQueue.configFailed) in the step that
// frees the attempt's slot.
func (m *Manager) configFailure(host, rev, reason string) (time.Duration, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	t := m.track(host)
	if rev == "" || rev != t.delivered {
		return 0, false
	}
	t.fails++
	t.lastErr = reason
	return configBackoff(t.fails), true
}

// cohortFailed runs the cohort-failure transition for s's app. A failure of
// an execution that began before a retry of its build runs none of it
// (PreRetry, ruling 191); a hold-down it asked for is a hold-ran marker
// instead (FailureBook.Record), whose rollback read is then due.
func (m *Manager) cohortFailed(s *scheduler, rt registry.Route, rec FailureRecord) {
	s.mu.Lock()
	res, superseded, err := failCohort(m.d.Book, s.q, rec)
	s.mu.Unlock()
	if err != nil {
		m.d.Logf("rollout: write failure record %s/%s %s: %v (held in memory)", rec.Account, rec.App, rec.Build, err)
	}
	if res == PreRetry {
		m.d.Logf("rollout: %s/%s build %s failed on %s, but in an execution that began before a retry of the build: the retry stands", rec.Account, rec.App, rec.Build, rec.Host)
		if slices.Contains(rec.HeldDown, rec.Host) {
			m.pointerReadDue(s.key)
		}
		s.poke()
		return
	}
	if len(superseded) > 0 {
		m.d.Logf("rollout: %s/%s build %s failed: superseding attempts past spawn on %v", rec.Account, rec.App, rec.Build, superseded)
	}
	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)))
	}
	s.poke()
}

func (m *Manager) stagingRefused(rt registry.Route, build string, err error) {
	m.d.Logf("rollout: %s: build %s refused as a staging error: %v", rt.Host, build, err)
	m.emit(rt, EvStagingRefused, "warn", textStagingRefused(build))
	if m.d.Touch != nil {
		m.d.Touch()
	}
}

// emit logs an event and publishes it to the member's log. text is built
// from build names and reason words only; RedactPaths stays as a second
// guard against a box path in a build name.
func (m *Manager) emit(rt registry.Route, code EventCode, level, text string) {
	m.d.Logf("rollout: %s %s: %s", code, rt.Host, text)
	if m.d.Events != nil {
		m.d.Events.Event(release.RouteAccount(rt), rt.App, rt.Host, level, activator.RedactPaths(text))
	}
}