// Adoption — one check-and-apply sweep over the registry, shaped after
// Eleven updater.go's updatePass, simplified: channel pointers replace the
// GitHub feed, the Store replaces HTTP downloads, and per-host update prefs
// are the registry row itself (Channel, Hibernate).
package release

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"errors"
	"fmt"
	"io"
	"os"
	"runtime"
	"slices"
	"strings"
	"syscall"
	"time"

	"github.com/carlosframework/platform/internal/caps"
	"github.com/carlosframework/platform/internal/controlname"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/safefile"
	"github.com/carlosframework/platform/internal/updatewords"
)

// adoptPlatform picks which manifest artifact a box installs — a var so
// tests can exercise resolution without faking runtime.GOOS/GOARCH.
var adoptPlatform = runtime.GOOS + "-" + runtime.GOARCH

// PassOption configures one call to Pass or PassFilter beyond their required
// arguments — NagExpiredCanaries and WithCapabilities today. The type exists
// so a call site opts in explicitly and a future option composes the same
// way, rather than every option becoming its own trailing parameter.
type PassOption func(*passConfig)

type passConfig struct {
	openRootUpdates                RootUpdateOpener
	rootUpdates                    *RootUpdateInputs
	legacyRecovery                 func() (bool, error)
	recovery                       func(context.Context, registry.Route) (bool, error)
	rootHistory                    *RootHistoryPass
	requireRootHistory             bool
	historySelected                map[string]historySelection
	runningVersion                 func(context.Context, registry.Route) (string, error)
	propagateRestarts              bool
	restartAttemptSettled          func(string) error
	restartErrors                  *[]error
	hostWrites                     hostwrite.Config
	selectedRestart, selectedStart hostwrite.Callback
	// selected is this pass's release work per adoptable host, captured once
	// in PassFilter. D1's group rollback reuses a sibling's entry, so the
	// sibling's rollback writes are checked against the same selection its
	// forward adoption ran under, never a fresh one that could be a
	// successor's.
	selected map[string]hostwrite.Work
	// ratchet is #253's per-box ratchet (WithDigestRatchet); nil admits
	// every verified pointer, as before.
	ratchet *DigestRatchet
	// nag is where the canary-TTL nag (spec §5) is printed; nil (the
	// default — every call site that does not pass NagExpiredCanaries)
	// means the nag never fires at all, not even to a discarded writer.
	nag io.Writer

	// capsOps arms the capability converge inside adoptHost; nil (the
	// default — every call site that does not pass WithCapabilities, and
	// every existing test) means converge never runs and this package's
	// behaviour is byte-identical to before the capability design landed.
	capsOps *caps.Ops

	// bakeDir, when non-empty, is the directory holding the bake clock
	// (BakeDir option); empty keeps the legacy /opt-dir location.
	bakeDir string

	// stageBlocked is called when staging fails because this process cannot
	// write the version store (OnStageBlocked). nil — every existing caller
	// and every existing test — means the pass reports such a failure exactly
	// as it always has and nothing else happens.
	stageBlocked func(host, ver string, err error)

	// restartHibernating is the callback hibernate-flagged routes get
	// (RestartHibernating). nil — the default, and RunAdopt — keeps the
	// original split: sleepers are repointed with restart=nil and no callback
	// is consulted for them at all.
	restartHibernating func(host string) error

	// unitsToRoot hands every route that names its own unit to the root pass
	// before anything is staged or repointed (RootRestartsUnits). false, the
	// default and RunAdopt's, restarts the named unit in this pass.
	unitsToRoot bool

	// startUnit is consulted ONLY on adoptHost's restart-failure rollback
	// path, after the symlink has gone back (WithStartUnit). nil — every
	// call site that does not pass the option, and every existing test —
	// keeps today's behaviour byte-identically: the rollback repoints and
	// stops.
	startUnit func(host string) error

	// rollbackRestart replaces startUnit on D1's path only — a named unit
	// whose restart returned *UnitNotUpError. It must REPLACE whatever is
	// running (reset-failed + restart), because a deadline failure can leave
	// the rejected build running unready and `start` is a no-op on it.
	rollbackRestart func(host string) error

	// proof is D3's gate (WithProofGate, proofgate.go). nil, the default for
	// every existing caller and test, means no proof is read or required.
	proof *ProofGate

	// pins is WithPinSource's provider: staged copies that something beyond
	// the registry rows, the live links and the generations file still
	// refers to. nil adds nothing.
	pins           PinSource
	generationPins PinSource

	// stage is what ensureStagedChecked needs to judge a staged copy it is
	// about to reuse (spec §4.2). PassFilter sets it once per pass.
	stage stageCheck

	// settings arms the instance update settings (settings spec §4,
	// WithUpdateSettings): adoptHost asks Target which build to install for
	// each exec instance. resolve is root's alone: only root writes the
	// consumed file. openFailed opens the zero-downtime failure book;
	// PassFilter calls it once per pass into failed (or failedErr), so a
	// long-lived option never reads a stale book. managed is
	// rollout.Managed: an unreadable book holds those routes too.
	settings   bool
	resolve    bool
	openFailed func() (FailedBuild, error)
	managed    func(registry.Route) bool
	failed     FailedBuild
	failedErr  error
	// route reads one registry row now, for root's re-check under the carry
	// fence (publishConsumed, ruling R-CARRY-FENCE); PassFilter sets it.
	route func(host string) (*registry.Route, error)
	// pub verifies promotion records (§4.1 evidence); PassFilter sets it.
	pub ed25519.PublicKey
	// tags is "<acct>/<app>" -> release version -> its newest semver tag, read
	// once per pass for apps with a route under a major setting.
	tags map[string]map[string]string

	// pointerGate is WithPointerGate's gate; nil (every caller but the
	// agent's fast sweep) reads every group's pointer, as before.
	pointerGate PointerGate
}

// PointerGate lets a caller skip a channel pointer read that cannot have
// anything new (S3 change-index design §5). PassFilter asks it once per
// (account, app, channel) group, with the group's routes this pass would
// adopt. skip true leaves the whole group out of this pass: no pointer read
// and no adoption, exactly as if extra had refused its routes. Otherwise
// read, when non-nil, is called once with the outcome of the group's read:
// the verified pointer, or the read's error or the ratchet's refusal.
type PointerGate func(acct, app, channel string, routes []registry.Route) (skip bool, read func(ptr *ChannelPointer, err error))

// WithPointerGate arms g for this pass. The agent's fast sweep passes one
// that skips settled canary groups whose app scope has not moved.
func WithPointerGate(g PointerGate) PassOption {
	return func(c *passConfig) { c.pointerGate = g }
}

// gatePointers asks cfg.pointerGate about every group adopts admits, and
// returns adopts narrowed to the groups it did not skip, with each kept
// group's read callback.
func (cfg *passConfig) gatePointers(routes []registry.Route, adopts func(registry.Route) bool) (func(registry.Route) bool, map[[3]string]func(*ChannelPointer, error)) {
	if cfg.pointerGate == nil {
		return adopts, nil
	}
	groupOf := func(rt registry.Route) [3]string { return [3]string{RouteAccount(rt), rt.App, routeChannel(rt)} }
	groups := map[[3]string][]registry.Route{}
	var order [][3]string
	for _, rt := range routes {
		if !adopts(rt) {
			continue
		}
		k := groupOf(rt)
		if _, ok := groups[k]; !ok {
			order = append(order, k)
		}
		groups[k] = append(groups[k], rt)
	}
	reads := map[[3]string]func(*ChannelPointer, error){}
	skipped := map[[3]string]bool{}
	for _, k := range order {
		skip, read := cfg.pointerGate(k[0], k[1], k[2], groups[k])
		switch {
		case skip:
			skipped[k] = true
		case read != nil:
			reads[k] = read
		}
	}
	if len(skipped) == 0 {
		return adopts, reads
	}
	return func(rt registry.Route) bool { return adopts(rt) && !skipped[groupOf(rt)] }, reads
}

// WithRunningVersion retains root's verified running build through a repoint.
// It must already be named by current or previous; PointLive checks that again.
func WithRunningVersion(running func(context.Context, registry.Route) (string, error)) PassOption {
	return func(c *passConfig) { c.runningVersion = running }
}

// FailedBuild reports whether host may not be moved onto build: the
// zero-downtime failure book (rollout.FailureBook), taken as a func because
// rollout imports this package. served is build == the route's GoodVersion.
type FailedBuild func(acct, app, build, host string, served bool) bool

// ErrInstallRequestNeedsRoot is what an agent pass hands OnStageBlocked's
// callback when it stops for an install request only root may resolve
// (settings spec §4.3 rule 2; ruling R-AGENT-RELAY). Nothing failed: the
// callback only asks for root's pass, and the version is the request's T as
// this pass sees it. Never shown to members.
var ErrInstallRequestNeedsRoot = errors.New("release: an install request waits for root's pass to resolve it")

// ErrUnitRestartNeedsRoot is what a RootRestartsUnits pass hands
// OnStageBlocked's callback for a route that names its own unit: nothing
// failed, and the version is the build root's pass should install. Never
// shown to members.
var ErrUnitRestartNeedsRoot = errors.New("release: a route's named unit is restarted by root's pass")

// RootRestartsUnits is for a pass whose process cannot restart a unit an
// operator named (issue #315): the agent's in-process loops when they do not
// run as root. carlos-edge.service is User=carlos, and systemd refuses it
// ("Interactive authentication required") for console.service and every
// other route.Unit. Before this option such a pass repointed the live link,
// failed the restart, read the refusal as a broken build, rolled back,
// poisoned the version and tried to start the old build over a process that
// had never stopped.
//
// With it, a route that names its unit, and every route in that unit's
// process group (processGroup: same account, app, channel and upstream), is
// handed to root before anything is staged or repointed, through
// OnStageBlocked with ErrUnitRestartNeedsRoot, the way a build this process
// cannot stage is. The live links keep naming the running build, so no
// sibling is stamped with a build the unit does not run. Nothing is poisoned
// and no refusal is recorded; root's
// pass stages, repoints, restarts and stamps. Every hold, refusal and proof
// check before the hand-off still applies, so root is only asked for a build
// this pass would have installed.
//
// Lifecycle host writes (WithHostWrites) restart through the selected
// receiver instead, so the option does nothing under them. Nor for a route
// this pass would not restart (restart nil: a sleeper on a pass without
// RestartHibernating, a sidecar).
func RootRestartsUnits() PassOption {
	return func(c *passConfig) { c.unitsToRoot = true }
}

// NagExpiredCanaries opts a Pass/PassFilter call into spec §5's nag: one
// line per expired canary/* route the call sees, written to w.
//
// DO NOT pass this to a fast/frequent adoption loop. It belongs to the
// HOURLY ROOT adopt pass ALONE — carlos-adopt.service's RunAdopt
// (internal/host/agent.go), the only caller that sets it (2026-08-03
// review), and every future caller must keep it that way. The same split
// PR #9's config sync uses, for the same reason: the in-process adoptLoop
// (hourly + jitter, inside the sandboxed carlos-edge.service) and
// fastConvergeLoop (every 2-3 SECONDS — the whole point of which is fast
// pickup of a freshly-promoted canary or a serial bump) must NEVER
// set this. An expired canary sitting still on that fast loop would
// otherwise nag every 2-3 seconds forever, turning a visibility feature
// into a journal-flooding one; gating it to the hourly root pass keeps the
// nag journal-friendly — once an hour, not a poll-storm of log lines. This
// is enforced by convention (which call sites pass the option), not by the
// type system — see agent.go's adoptLoop/fastPassOpts for the loud
// "never add this" comments at the two places it would be easiest to add
// by accident.
func NagExpiredCanaries(w io.Writer) PassOption {
	return func(c *passConfig) { c.nag = w }
}

// WithCapabilities arms the capability converge inside adoptHost (spec §4d):
// on a version change, a unit-backed route's drop-ins are materialized for
// the NEW version before its live symlink moves, and restored with the
// symlink when the restart fails. Unset — every existing caller and every
// existing test — means converge never runs and this package behaves exactly
// as it did before.
//
// The zero Ops is the production one: real /etc/systemd/system, real
// /var/lib/carlos-app, real systemctl, real useradd. A test injects temp
// dirs and recorders through the same value.
//
// DO NOT pass this to a fast/frequent adoption loop, and DO NOT pass it from
// anything running inside carlos-edge.service. It belongs to the HOURLY ROOT
// adopt pass ALONE — carlos-adopt.service's RunAdopt (internal/host/agent.go),
// the only caller that sets it — for a harder reason than the nag's: this
// option WRITES /etc/systemd/system, and carlos-edge.service is
// ProtectSystem=strict with ReadWritePaths=/data /run/carlos, so the
// in-process adoptLoop and fastConvergeLoop physically cannot. Arming it
// there would not converge anything; it would log a permission denial every
// 2-3 seconds forever. Enforced by convention (which call sites pass the
// option), not by the type system — see agent.go's adoptLoop/fastPassOpts
// for the loud "never add this" comments at the two places it would be
// easiest to add by accident.
//
// The consequence, ruled on and accepted (2026-08-03): a freshly promoted
// canary picked up by the fast loop runs the new binary against the unit
// shape the PREVIOUS version left, until the hourly root pass converges it.
// adoptHost prints one line per such repoint so the window is visible —
// see the capsOps == nil branch there.
func WithCapabilities(o caps.Ops) PassOption {
	return func(c *passConfig) { c.capsOps = &o }
}

// BakeDir moves this pass's bake clock into dir — in production the data
// dir (CARLOS_DATA_DIR, default /data), the one place both the sandboxed
// edge unit (ReadWritePaths=/data) and the root adopt one-shot can write.
// Reads migrate: a box whose clock still lives at the legacy
// /opt/carlos/channel-first-seen.json keeps every observation (issue #13).
// Unset, the clock stays at its legacy path — existing tests and callers
// are unaffected.
//
// The migration is durable after the FIRST pass this option is set on a box
// still on the legacy path: PassFilter forces a Save even when no new
// channel head was observed, so the new path is fully populated (and the
// legacy file provably safe to delete) one pass later, not whenever the
// fleet next happens to see a new head.
func BakeDir(dir string) PassOption {
	return func(c *passConfig) { c.bakeDir = dir }
}

// OnStageBlocked reports a version this pass could not stage or point live
// because the process cannot WRITE the install directory — never because
// the bucket, the signature or the manifest was bad. It is the release half
// of the relay carlos-adopt.path already runs for config: the agent's
// in-process loops are
// sandboxed (carlos-edge.service sets ProtectSystem=strict with no /opt write),
// so they can see a promotion within seconds and stage nothing, while the only
// process that CAN stage runs hourly off carlos-adopt.timer. Without a relay a
// promote is invisible for up to an hour on a box that noticed it immediately.
//
// Deliberately narrow. A read failure, a bad signature or a poisoned version
// must NOT fire this: none of them is fixed by re-running as root, and a
// callback that fired on any staging error would relay a bucket outage into a
// root pass every tick. Only EROFS/EPERM/EACCES — "this uid cannot write
// there" — describe a job a privileged pass would actually finish.
//
// Never fatal, and never a substitute for the timer: the hourly pass remains
// the floor, exactly as it is for config. A dropped or throttled request costs
// latency, not correctness.
func OnStageBlocked(fn func(host, ver string, err error)) PassOption {
	return func(c *passConfig) { c.stageBlocked = fn }
}

// RestartHibernating extends restart to hibernate-flagged routes, which
// otherwise get the pure repoint (restart=nil) and are never offered to the
// callback at all.
//
// It exists because the flag is not the question the restart decision
// actually wants answered. Hibernate=true means "this route MAY sleep", not
// "it is asleep right now" — an exec instance is minted hibernating
// (provision/reconcile.go) and then spends every serving minute awake. With
// the flag alone deciding, a busy exec child kept serving the binary its
// live symlink no longer names until it happened to go idle. Handing the
// callback the sleepers lets it answer the real question at the moment it
// matters.
//
// This does NOT weaken "an adopt pass never wakes a sleeping host" — it moves
// the guarantee from withholding the callback to the callback's own
// semantics, and only a callback with those semantics may be passed here.
// The agent's makeRestartRoute qualifies on both halves: activator Bounce
// returns false without starting anything when the host is not awake, and
// `systemctl try-restart` is defined as a no-op on a unit that is not
// running. Nothing on either path can start a process. A callback that could
// start one belongs on the awake loop only. (WithStartUnit narrows that
// invariant without breaking it: the ONE start this package can now perform
// fires only after this callback RETURNED AN ERROR — and no stopped unit's
// try-restart produces one, so a sleeper cannot arm it. See WithStartUnit.)
//
// DO NOT pass this from RunAdopt (carlos-adopt.service). That process is not
// the activator's parent: the exec children belong to the agent, so its only
// available restart is systemctl, which cannot reach them — offering it
// sleepers would buy nothing and cost a try-restart per sleeping host per
// version change. Set by the agent's in-process adoptLoop/fastConvergeLoop
// alone, the two loops that run inside the process owning the children.
//
// Sleepers are still adopted AFTER every awake host and still poison a bad
// build the same way, both unchanged: the option decides who the callback
// sees, never the order or the health gate.
func RestartHibernating(restart func(host string) error) PassOption {
	return func(c *passConfig) { c.restartHibernating = restart }
}

// WithStartUnit gives the pass a way to START a route's process after a
// FAILED restart has been rolled back — the one moment adoption knows the
// symlink is correct and the service behind it is down. Without it the
// rollback repoints the symlink and stops there: the unit systemd left
// failed stays failed, and because `systemctl try-restart` is by definition
// a no-op on a unit that is not running (the same property that lets a
// deliberately stopped service stay stopped), the NEXT pass's restart exits
// 0 and SetVersions stamps a build that has never run — the exact lying
// header the named-unit arm's returned error exists to prevent, now over a
// service that is not running at all.
//
// start fires ONLY on that rollback path, and only when the rollback
// actually HAPPENED: a first adoption whose restart fails has no previous
// version, and a rollback PointLive itself refused (the old version's
// directory gone — materialize.go's "link present, staged version gone"
// state) rolled nothing — in both the symlink still names the refused
// build, and starting there would run the exact build the failure refused.
// A start that also fails changes nothing about the refusal (the version
// stays poisoned and unstamped) but is printed, because at that point the
// box is most likely serving nothing and an operator must not read the
// rollback line as "recovered".
//
// This cannot WAKE a hibernating route, and the reasoning is the contract:
// the path is reached only after a restart that was ATTEMPTED and returned
// an error. Of the agent's dispatch arms (internal/host's makeRestartRoute)
// only the NAMED-unit arm returns one, and its try-restart exits 0 on a
// loaded-but-stopped unit — so an error means the unit was running and
// would not come back, or the name resolves to nothing (in which case the
// start reaches nothing either). The bounce and carlos-app@ arms swallow
// their failures and return nil, so a sleeping exec child can never route
// here. The converged-daemon arm (caps.Restart) is excluded at the guard
// itself: the restart that failed there was carlos-app@<host>'s, its
// rollback belongs to caps.Restore, and starting the route's NAMED unit
// there would start a unit that pass never touched.
//
// start must start ONLY a route that NAMES its unit (registry Unit column)
// and return nil for everything else — internal/host's makeStartRoute is
// the intended shape. It is `systemctl start`, not try-restart, because the
// entire premise is that the unit is down; on a unit something else already
// brought up (caps.Restore's socket sequence) start is an exit-0 no-op, so
// the two cannot fight.
func WithStartUnit(start func(host string) error) PassOption {
	return func(c *passConfig) { c.startUnit = start }
}

// WithRollbackRestart sets D1's rollback restart (see passConfig).
func WithRestartAttemptSettled(fn func(string) error) PassOption {
	return func(c *passConfig) { c.restartAttemptSettled = fn }
}

func WithRestartFailurePropagation() PassOption {
	return func(c *passConfig) { c.propagateRestarts = true }
}

func WithRollbackRestart(fn func(host string) error) PassOption {
	return func(c *passConfig) { c.rollbackRestart = fn }
}

// WithProofGate arms D3 (proofgate.go): a build of an app the policy file
// lists is adopted only with a verified proof for its bytes, or from this
// box's healthy history. Pass the SAME gate to every pass of a process, so
// its 30 s miss cache and its waiting list survive between ticks.
func WithProofGate(g *ProofGate) PassOption {
	return func(c *passConfig) { c.proof = g }
}

// WithDigestRatchet arms #253's per-box ratchet for this pass (spec §4.6):
// a pointer without digests is refused on a channel this box has seen with
// them, or everywhere under CARLOS_REQUIRE_POINTER_DIGEST. Root and the
// agent each pass their own (internal/host pointerRatchet).
func WithDigestRatchet(r *DigestRatchet) PassOption {
	return func(c *passConfig) { c.ratchet = r }
}

// WithUpdateSettings arms settings spec §4 for this pass: each exec
// instance installs what Target decides, with §4.1's evidence. book opens
// the failure book; it is called at the start of every pass, so pass the
// opener, never a book read once (nil: no failure records). managed is
// rollout.Managed (nil: none). resolve is set on root's pass only: root
// alone resolves install requests and writes the consumed file (§4.3 rule
// 2); the agent's passes apply the same decisions and write nothing.
//
// A route with default settings and no consumed file takes today's
// adoption unchanged and never consults the book. An unreadable book holds
// every route with settings and every managed route (the zero-downtime
// plans' refuseManaged posture).
func WithUpdateSettings(book func() (FailedBuild, error), managed func(registry.Route) bool, resolve bool) PassOption {
	return func(c *passConfig) {
		c.settings, c.openFailed, c.managed, c.resolve = true, book, managed, resolve
	}
}

// WithPinSource adds src's references to the pin set a reused staged copy is
// checked against (spec §4.2), on top of the live links, the registry rows
// and the generations file, which every pass with a data dir reads itself.
func WithPinSource(src PinSource) PassOption {
	return func(c *passConfig) { c.pins = src }
}

// stageBlocked reports whether err is the version store refusing a write to
// THIS uid, rather than any of the many other ways staging fails. errors.Is
// unwraps the *os.PathError that mkdir/openat return.
func stageBlocked(err error) bool {
	return errors.Is(err, syscall.EROFS) ||
		errors.Is(err, syscall.EPERM) ||
		errors.Is(err, syscall.EACCES)
}

// StageBlocked is stageBlocked for callers outside this package: the
// relay-or-fail decision a move's held stage makes (MaterializeHeld). EROFS
// is what ProtectSystem=strict answers a write with, and it is not
// fs.ErrPermission, so a caller must not substitute that test for this one.
func StageBlocked(err error) bool { return stageBlocked(err) }

// Pass is one full adoption sweep: group instance routes by (App, Channel),
// resolve each channel pointer once, verify signatures and artifact sha256s,
// download/stage missing versions, and repoint live/<host> symlinks.
//
// Die-free by design: per-host and per-channel problems log and continue —
// an unreachable channel or one broken host must not strand the rest of the
// fleet unchecked. Only foundation failures (registry unreadable, bake clock
// unwritable) fail the pass.
//
// restart cycles a host's process onto the new symlink target (the
// activator's proc converge); nil means symlink-only for every host. Hosts
// whose route has Hibernate=true get symlink-only by default and are NEVER
// woken: an hourly wake would defeat hibernation, and the next wake execs
// the new binary anyway since the spawn resolves live/<host> at exec time.
// The trade (from Eleven): no health check for a sleeper — a bad build
// surfaces at wake, recovered by Rollback, which repoints back. Hibernating
// hosts are processed AFTER every awake host, so an awake host discovers a
// boot-broken build (and poisons it) before any sleeper is moved onto it.
//
// RestartHibernating overrides only the FIRST half of that: it gives
// hibernate-flagged hosts a restart callback too, because the flag means "may
// sleep", not "is asleep" — a warm exec child must not go on serving a binary
// its live symlink no longer names. It never makes the pass wake anything;
// the option only accepts a callback that cannot start a process — and the
// start WithStartUnit adds is armed only by that callback returning an error,
// which a sleeper's exit-0 try-restart never does (see WithStartUnit). See
// its doc comment. Everything else above holds unchanged: sleepers still go
// last,
// still poison a bad build, still have no health gate of their own.
func Pass(ctx context.Context, reg *registry.Registry, s Store, pub ed25519.PublicKey, optDir string, restart func(host string) error, opts ...PassOption) error {
	return PassFilter(ctx, reg, s, pub, optDir, restart, nil, opts...)
}

// PassFilter is Pass narrowed to the routes extra additionally accepts
// (nil extra makes it identical to Pass) — the seam a faster, narrower
// sweep composes on top of the full one without duplicating the group
// resolution, bake-clock, staging, or restart logic. Built for the
// fast-converge sweep (internal/host/agent.go's fastConvergeLoop): canary
// channels are individually created and inherently low-cardinality (one
// deliberate rollout at a time, never the whole fleet — see bake.go's
// "canary/* has no window"), and unstaged routes drain out of the set as
// they stage (with per-host backoff while they can't), so sweeping just
// those every few seconds instead of hourly can't turn into a fleet-wide
// poll storm the way sweeping every route that fast would.
func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed25519.PublicKey, optDir string, restart func(host string) error, extra func(registry.Route) bool, opts ...PassOption) error {
	var cfg passConfig
	for _, opt := range opts {
		opt(&cfg)
	}
	if cfg.propagateRestarts || cfg.requireRootHistory {
		cfg.restartErrors = new([]error)
	}
	if cfg.openRootUpdates != nil {
		var err error
		cfg.rootUpdates, err = cfg.openRootUpdates(ctx, s, pub)
		if err != nil {
			return err
		}
		if cfg.rootUpdates != nil && cfg.rootUpdates.Close != nil {
			defer cfg.rootUpdates.Close()
		}
	}
	cfg.pub = pub
	cfg.route = reg.Get
	if cfg.recovery != nil {
		routes, err := reg.List()
		if err != nil {
			return err
		}
		if cfg.rootUpdates != nil {
			routes, err = cfg.rootUpdates.routes(ctx, routes)
			if err != nil {
				return err
			}
		}
		handled := map[string]bool{}
		var failures []error
		remaining := false
		for _, rt := range routes {
			if extra != nil && !extra(rt) {
				continue
			}
			done, err := cfg.recovery(ctx, rt)
			handled[rt.Host] = done
			if err != nil {
				failures = append(failures, err)
			}
			if !done && (Adoptable(rt) || SidecarAdoptable(rt)) {
				remaining = true
			}
		}
		if !remaining && len(handled) > 0 {
			return errors.Join(failures...)
		}
		for _, done := range handled {
			if done {
				prior := extra
				extra = func(rt registry.Route) bool { return !handled[rt.Host] && (prior == nil || prior(rt)) }
				break
			}
		}
		cfg.recovery = nil
		opts = append(opts, func(c *passConfig) { c.recovery = nil })
		if cfg.legacyRecovery != nil {
			legacy := len(routes) == 0
			var err error
			if !legacy {
				legacy, err = cfg.legacyRecovery()
			}
			if err != nil {
				return errors.Join(errors.Join(failures...), err)
			}
			if legacy {
				opts = append(opts, func(c *passConfig) { c.requireRootHistory = false; c.rootHistory = nil })
			}
		}
		return errors.Join(errors.Join(failures...), PassFilter(ctx, reg, s, pub, optDir, restart, extra, opts...))
	}
	// Settings spec §4.3: the failure book is read fresh for every pass, so a
	// record the agent persisted since the last one counts now.
	if cfg.rootUpdates == nil && cfg.settings && cfg.openFailed != nil {
		cfg.failed, cfg.failedErr = cfg.openFailed()
		if cfg.failedErr != nil && cfg.resolve {
			fmt.Printf("  failure book unreadable (%v): routes with update settings and managed routes install nothing this pass\n", cfg.failedErr)
		}
	}
	// D3: re-read the proof policy at the start of every pass, so an edit
	// reaches the agent and root alike without a restart (spec §4.5).
	cfg.proof.BeginPass()
	routes, err := reg.List()
	if err != nil {
		return err
	}
	if cfg.rootUpdates != nil {
		routes, err = cfg.rootUpdates.routes(ctx, routes)
		if err != nil {
			return err
		}
	}
	adopts := func(rt registry.Route) bool {
		return (Adoptable(rt) || SidecarAdoptable(rt)) && (extra == nil || extra(rt))
	}
	// WithPointerGate: a skipped group leaves the pass before anything is
	// selected for it.
	adopts, pointerRead := cfg.gatePointers(routes, adopts)
	selected := make(map[string]hostwrite.Work, len(routes))
	cfg.historySelected = make(map[string]historySelection)
	for _, rt := range routes {
		if !adopts(rt) {
			continue
		}
		if cfg.requireRootHistory {
			sel, err := cfg.rootHistory.selectRoute(ctx, rt)
			if err != nil {
				return err
			}
			cfg.historySelected[rt.Host] = sel
		}
		work, selectErr := cfg.hostWrites.Select(ctx, rt.Host, "release")
		if selectErr != nil {
			return fmt.Errorf("%s: select release work: %w", rt.Host, selectErr)
		}
		selected[rt.Host] = work
	}
	cfg.selected = selected

	// The reuse check's view of what refers to a staged copy (spec §4.2). It
	// is asked only when a staged copy turns out to differ from its canonical
	// release, so a healthy pass never pays for it, and it reads fresh then,
	// so a host this pass has already repointed counts.
	cfg.stage = stageCheck{
		beforeWrite: func(path string) error { return cfg.generationWriteCheck(ctx, optDir, path) },
		verified:    map[string]string{},
		refs: func() (StagedPaths, error) {
			rows, err := reg.List()
			if err != nil {
				return nil, fmt.Errorf("list routes: %w", err)
			}
			// An app whose settings references are incomplete comes back
			// with the pins (*IncompleteRefsError); stageFrom refuses that
			// app's restage only.
			pins, incomplete := BoxPins(optDir, rows)
			var inc *IncompleteRefsError
			if incomplete != nil && (pins == nil || !errors.As(incomplete, &inc)) {
				return nil, incomplete
			}
			// Live and pending generations, from the file the agent's
			// activator writes into the data dir (index amendments; its
			// workers' captured builds since ruling R17). Every production
			// pass passes the data dir as BakeDir: root's RunAdopt, the
			// agent's loops, and cold materialization. A pass with no data
			// dir (tests) has no agent beside it to ask.
			if cfg.bakeDir != "" {
				gen, err := GenerationPins(GenerationsPath(cfg.bakeDir), func() ([]registry.Route, error) { return rows, nil }).StagedPins(ctx, optDir)
				if err != nil {
					return nil, err
				}
				for p := range gen {
					pins.add(p)
				}
			}
			if cfg.pins != nil {
				more, err := cfg.pins.StagedPins(ctx, optDir)
				if err != nil {
					return nil, err
				}
				for p := range more {
					pins.add(p)
				}
			}
			if inc != nil {
				return pins, inc
			}
			return pins, nil
		},
	}

	// Resolve every distinct (Account, App, Channel) once. Verification
	// failures kill the group (nil entry), not the pass.
	//
	// The account is part of the group key, not just the app: app names are
	// unique WITHIN an account, so two routes named the same app in different
	// accounts are two different apps following two different channel
	// pointers. Grouping on (app, channel) alone would resolve one of them and
	// silently install it on both.
	type groupKey struct{ acct, app, channel string }
	pointers := map[groupKey]*ChannelPointer{}
	// Whether an app has anything under its qualified prefix, cached for the
	// pass: one listing per (account, app), not one per channel. It decides
	// which of the two missing-pointer lines below an app gets.
	type acctApp struct{ acct, app string }
	// migratedState caches one pass's answer to "has this app anything under
	// its qualified prefix", together with the listing failure when there was
	// one — a failed listing must neither invent the migrate advice nor claim
	// the healthy skip (2026-08-03 review).
	type migratedState struct {
		has     bool
		listErr error
	}
	migrated := map[acctApp]migratedState{}
	for _, rt := range routes {
		if !adopts(rt) {
			continue
		}
		k := groupKey{RouteAccount(rt), rt.App, routeChannel(rt)}
		if _, done := pointers[k]; done {
			continue
		}
		ptr, err := Channel(ctx, s, pub, k.acct, k.app, k.channel)
		read := pointerRead[[3]string{k.acct, k.app, k.channel}]
		if err != nil {
			if read != nil {
				read(nil, err)
			}
			pointers[k] = nil // resolved-and-failed: don't retry per host
			if errors.Is(err, os.ErrNotExist) {
				// SKIP, never abort: one app with no pointer must not stop
				// every other app on the box from installing.
				//
				// Two deployments produce a missing pointer, and only one of
				// them should be told to migrate. A channel with nothing
				// promoted yet is ordinary — a migrated app waiting on its
				// first stable promotion sits here for days — and gets the
				// plain line. An app with NOTHING under its qualified prefix
				// upgraded without migrating, and gets the store's error,
				// which names the verb. Advice that fires every hour on a
				// healthy fleet is advice nobody reads.
				ak := acctApp{k.acct, k.app}
				st, cached := migrated[ak]
				if !cached {
					has, lerr := AppHasQualifiedObjects(ctx, s, k.acct, k.app)
					// A listing we could not perform suppresses the advice
					// rather than inventing it: a bucket this box cannot list
					// is not evidence that an app never migrated. It is not
					// evidence of a healthy skip either, so the failure rides
					// along and is printed instead of a claim this box never
					// established (realistic trigger: GetObject IAM without
					// ListBucket).
					st = migratedState{has: has || lerr != nil, listErr: lerr}
					migrated[ak] = st
				}
				switch {
				case st.listErr != nil:
					fmt.Printf("  %s/%s/%s: channel state unknown (could not list the app's prefix: %v)\n", k.acct, k.app, k.channel, st.listErr)
				case st.has:
					fmt.Printf("  %s/%s/%s: channel has no release yet\n", k.acct, k.app, k.channel)
				default:
					fmt.Printf("  %s/%s/%s: channel has no release yet (%v)\n", k.acct, k.app, k.channel, err)
				}
			} else {
				fmt.Printf("  %s/%s/%s: channel pointer rejected: %v\n", k.acct, k.app, k.channel, err)
			}
			continue
		}
		// Any error from Admit refuses the pointer, ErrRatchetUnpersisted
		// included, not only ErrPointerWithoutDigests.
		if rerr := cfg.ratchet.Admit(k.acct, ptr); rerr != nil {
			if read != nil {
				read(nil, rerr)
			}
			pointers[k] = nil // refused: the group keeps what it has
			fmt.Printf("  %s/%s/%s: channel pointer refused: %v\n", k.acct, k.app, k.channel, rerr)
			continue
		}
		if read != nil {
			read(ptr, nil)
		}
		pointers[k] = ptr
	}

	// Settings spec §3.1: a major setting holds a head whose semver tag is
	// absent or in another major. Tags are read once per app that has a route
	// under one. ListTags is semver-descending, so the first tag seen for a
	// release is its newest. An unreadable list leaves the app's map empty:
	// every head reads as untagged, so a major-held instance stays held.
	if cfg.settings {
		cfg.tags = map[string]map[string]string{}
		for _, rt := range routes {
			u, err := ParseUpdateSettingsLenient(rt.UpdateSettings)
			if err != nil || u.Major == nil || !adopts(rt) {
				continue
			}
			k := RouteAccount(rt) + "/" + rt.App
			if _, done := cfg.tags[k]; done {
				continue
			}
			m := map[string]string{}
			tags, err := ListTags(ctx, s, RouteAccount(rt), rt.App)
			if err != nil {
				fmt.Printf("  %s: tags unreadable (%v): every head reads as untagged\n", k, err)
			}
			for _, tg := range tags {
				if _, seen := m[tg.Release]; !seen {
					m[tg.Release] = tg.Version
				}
			}
			cfg.tags[k] = m
		}
	}

	// Resolve each app's pipeline once per pass — for every app with a
	// resolved NON-CANARY head, never one read per channel or per route.
	// canary/* lives outside the pipeline entirely (spec §2): no pipeline
	// can name it, its window is zero either way, so a canary-only
	// appearance has no verdict a read could change — which is also what
	// keeps the 2-3-second canary fast sweep at zero per-tick bucket reads
	// here, the same cost property the production-flag reads below hold.
	//
	// A read/verify ERROR is kept, and kept apart from verified-absent
	// (nil, nil): the errored app's windows fall back to DefaultHolds
	// FLOORED by the box's own ratchet memory (the appHolds loop below) —
	// every channel the memory carries enforces max(its DefaultHolds
	// resolution, the remembered window), so corrupting the object can
	// never make this box looser than it already was (spec §3: corruption
	// must never loosen; plain DefaultHolds would hand a tightened rung
	// name like edge its ZERO legacy window back). A channel with no memory
	// keeps the fail-closed DefaultHolds resolution — the old-binary
	// posture (spec §3's old-binary edge), whose production: true co-write
	// remains the mitigation for a MEMORY-LESS box only. Where memory DOES
	// exist, the floor is enforcement in full: the bake VERDICT rides on
	// the same memory (bakeIgnore below never ignores an app whose ratchet
	// is still live — which is a WIDER condition than "the memory raises a
	// window", deliberately, since an app-defined channel's ratchet
	// normally raises nothing at all), so nothing here depends on the
	// console having managed the co-write.
	// The error is NEVER folded into the ratchet as a nil: nil
	// means verified-absent only (ObservePipeline's contract), and a
	// corrupt object must not masquerade as a deletion — a deletion RATCHETS
	// each remembered channel down to its legacy default over one old-window
	// cycle (ObservePipeline's p == nil branch), where an error holds every
	// remembered window exactly where it is.
	type pipeState struct {
		p   *Pipeline
		err error // non-nil: the object exists but would not read or verify
	}
	pipes := map[acctApp]pipeState{}
	for k, ptr := range pointers {
		if ptr == nil {
			continue
		}
		if r, ok := Rung(k.channel); ok && r == RungCanary {
			continue
		}
		ak := acctApp{k.acct, k.app}
		if _, done := pipes[ak]; done {
			continue
		}
		p, perr := GetPipeline(ctx, s, pub, ak.acct, ak.app)
		pipes[ak] = pipeState{p: p, err: perr}
		if perr != nil {
			// SAY SO, like the production flag below — same realistic
			// non-tampering trigger too: on a GetObject-without-ListBucket
			// fleet a GET of an ABSENT key answers 403, not 404, and every
			// app would land here every pass.
			fmt.Printf("  %s/%s: pipeline unreadable (%v) — remembered windows still apply; anything unremembered falls to legacy holds\n", ak.acct, ak.app, perr)
		}
	}

	// The bake clock, loaded before any of this pass's reads are folded into
	// it. now is stamped once so every observation and verdict this pass
	// answers the same instant.
	now := time.Now()
	var bake *FirstSeen
	if cfg.requireRootHistory {
		bindings := make([]HistoryBinding, 0, len(cfg.historySelected))
		for _, sel := range cfg.historySelected {
			bindings = append(bindings, sel.binding)
		}
		bake, err = cfg.rootHistory.History.LoadBake(bindings, now)
		if err != nil {
			return err
		}
	} else if cfg.bakeDir != "" {
		bake = LoadFirstSeenMigrating(FirstSeenPath(cfg.bakeDir), FirstSeenPath(optDir))
	} else {
		bake = LoadFirstSeen(FirstSeenPath(optDir))
	}
	// bake.migrated forces this pass's Save even if nothing new is observed
	// below: otherwise a changed-gated Save could leave the new path absent
	// for days on a fleet that happens to see no new channel head, and the
	// legacy file would look safe to delete long before it actually is (see
	// LoadFirstSeenMigrating and BakeDir's doc comments, 2026-08-03 review).
	changed := bake.migrated

	// ORDER IS LOAD-BEARING: fold this pass's verified pipeline reads into
	// the ratchet memory HERE, before the per-app window maps below are
	// computed from it — a tightened window read this pass must be enforced
	// this pass, and a pending loosening whose old window has elapsed must
	// land before the verdicts that depend on it. Same same-pass rule as
	// ObserveProduction's fold further down. The err == nil guard is the
	// contract, not paranoia: nil reaches ObservePipeline only for a
	// verified-absent object (which never loosens anything, but does give
	// this app's pending promotions their chance to land); an errored read
	// never ticks the memory at all.
	for ak, ps := range pipes {
		if ps.err == nil && bake.ObservePipeline(ak.acct, ak.app, ps.p, now) {
			changed = true
		}
	}

	// One enforced-window map per app for this pass. A verified pipeline
	// resolves through the box's ratchet memory (HoldsForApp); an ERRORED
	// read keeps DefaultHolds floored by that same memory (never looser
	// than what this box already enforced — see the pipes comment above);
	// a VERIFIED-ABSENT pipeline takes the same memory floor, which is what
	// enforces the deletion ratchet-down for as long as the memory rides
	// above legacy; an app with no pipeline AND no memory keeps bare
	// DefaultHolds, today's behaviour byte-for-byte. canary/* is stitched back in at zero on the verified
	// path: it lives outside the pipeline (spec §2), so HoldsForApp's map
	// cannot carry it, and HoldFor's fallback for a name missing from the
	// map is the 72h fail-closed default — which would silently freeze
	// every canary deploy of a pipelined app. Every OTHER channel outside
	// the pipeline does take that 72h (spec §3: a channel with no pipeline
	// entry keeps the fail-closed default), including rung names like beta
	// whose DefaultHolds entry no longer applies.
	//
	// floored is the SEPARATE question the bake verdict asks: does this box
	// hold pipeline evidence for the app at all (see bakeIgnore)? It is a
	// membership set, never derived from the overlay above — a ratchet can
	// be live while raising nothing, which is the ordinary state of every
	// app-defined channel name (holdMemory.liveRatchet has the full
	// argument).
	appHolds := map[acctApp]map[string]time.Duration{}
	floored := map[acctApp]bool{}
	for ak, ps := range pipes {
		switch {
		case ps.err == nil && ps.p != nil:
			h := bake.HoldsForApp(ak.acct, ak.app, ps.p)
			h["canary/*"] = 0
			appHolds[ak] = h
			floored[ak] = true
		default:
			// The memory floor — for an ERRORED read and for VERIFIED
			// ABSENCE alike. Overrides overlay a COPY of DefaultHolds,
			// so a remembered channel is only ever RAISED above its legacy
			// resolution, an unremembered one resolves exactly as legacy
			// (HoldFor's 72h unknown-name fallback included), and canary/*
			// keeps its zero: no pipeline can name canary, so no ratchet
			// memory exists to raise it. With nothing remembered above
			// legacy, no map is built and holdsFor falls to DefaultHolds
			// itself. The prefix match is exact by construction: neither an
			// account, an app name nor a non-canary channel can contain
			// clockSep or "/" (channelClockKey's own injectivity argument).
			//
			// For verified absence the floor is half of what makes the
			// deletion ratchet (ObservePipeline's p == nil branch, folded
			// above) ENFORCED: while a remembered channel's window still
			// exceeds its legacy default, deleting the object cannot adopt
			// a held head this pass. That covers the names whose legacy
			// default is NARROWER than what was enforced (edge, 0); the
			// other half — an app-defined name, whose legacy default is the
			// 72h fallback and therefore lands instantly, raising nothing —
			// is carried by the verdict alone (floored below,
			// holdMemory.DeletedAt). Once every remembered channel has
			// finished ratcheting, nothing exceeds the default, no map is
			// built, and the app is back on the bare-DefaultHolds path. An
			// app with no memory at all — every legacy app, every pass —
			// takes that bare path byte-identically, and the len check
			// keeps the fleet-wide no-pipeline case allocation-free.
			//
			// The overlay is NOT the pass's answer to "is this app
			// pipelined as far as this box knows" — floored below is, and
			// the two are deliberately separate. A raised window is only
			// one shape a live ratchet takes: an app-defined channel like
			// "production" resolves to HoldFor's 72h fail-closed default
			// under DefaultHolds, so a declared 24h ratchets DOWN to a
			// window BELOW legacy and raises nothing at all. Deriving the
			// verdict from this overlay would have found no evidence for
			// exactly that app and fallen to the production flag.
			if len(bake.Holds) == 0 {
				break
			}
			var h map[string]time.Duration
			prefix := holdKeyPrefix(ak.acct, ak.app)
			for key, mem := range bake.Holds {
				// mem == nil: {"holds":{"x":null}} parses cleanly into a
				// nil entry — skip it, never dereference; a panic here
				// takes down the adopt pass on every box that loads the
				// file.
				if mem == nil {
					continue
				}
				name := strings.TrimPrefix(key, prefix)
				if name == key || name == "" {
					continue
				}
				if mem.liveRatchet(name, bake.Jitter, now) {
					floored[ak] = true
				}
				if mem.Window > HoldFor(DefaultHolds, name) {
					if h == nil {
						h = make(map[string]time.Duration, len(DefaultHolds)+1)
						for ch, d := range DefaultHolds {
							h[ch] = d
						}
					}
					h[name] = mem.Window
				}
			}
			if h != nil {
				appHolds[ak] = h
			}
		}
	}
	holdsFor := func(ak acctApp) map[string]time.Duration {
		if h, ok := appHolds[ak]; ok {
			return h
		}
		return DefaultHolds
	}

	// Resolve each app's production flag once per pass — but ONLY for apps
	// with a HELD head resolved this pass (spec §6), held now meaning held
	// under the app's OWN window map above. That filter is a cost property,
	// not a nicety: any pass pays at most one read per (account, app) with a
	// held head, never one per channel or per route. For the fast-converge
	// sweep (PassFilter every 2-3 SECONDS) that usually means zero flag
	// reads — canary/* has a zero window on every path, and a fully-staged
	// box's sweep carries nothing else — but an UNSTAGED held route in the
	// sweep's set does resolve its held channel and pays this read too.
	// What keeps that bounded is the sweep's own membership, not this
	// filter: an unstaged host backs off 2s→5min (internal/host's
	// fastMembers), so a held, never-staging route costs one flag read per
	// backoff window at worst, not one per tick. A PIPELINED app's held
	// head pays the read as well even though its verdict never consults the
	// flag (bakeIgnore below): the fold keeps the legacy latch fresh for
	// the day the pipeline is deleted AND its deletion ratchet has landed,
	// which is when the app falls back to the flag for real.
	type prodState struct {
		flag *ProductionFlag
		err  error // non-nil: the object exists but would not read or verify
	}
	bakeable := map[acctApp]bool{}
	for k, ptr := range pointers {
		if ptr != nil && HoldFor(holdsFor(acctApp{k.acct, k.app}), k.channel) > 0 {
			bakeable[acctApp{k.acct, k.app}] = true
		}
	}
	prod := map[acctApp]prodState{}
	for ak := range bakeable {
		// A missing flag is (nil, nil) — the default, not production. Any
		// OTHER failure is kept as an error and fails toward baking in
		// ProductionBake: an attacker who can corrupt but not sign must not
		// be able to shorten a window by breaking the object.
		f, err := GetProduction(ctx, s, pub, ak.acct, ak.app)
		prod[ak] = prodState{flag: f, err: err}
		if err != nil {
			// SAY SO. This failure re-imposes the full window on every held
			// head of the app, and the most likely cause is not tampering at
			// all: on a GetObject-without-ListBucket fleet (the same IAM shape
			// the missing-pointer branch above already handles) a GET of an
			// ABSENT key answers 403, not 404 — so the spec's default-OFF
			// silently inverts to production for every app with no flag. A
			// 72h hold nobody can see the reason for is the worst version of
			// this feature; log it like every other per-app problem here and
			// carry on.
			fmt.Printf("  %s/%s: production flag unreadable (%v) — treating as production, holds apply\n", ak.acct, ak.app, err)
		}
	}

	// Tick the bake clock: record every channel head this box hasn't seen at
	// that head before, so the supply-chain window is measured from the box's
	// own first sighting, never a feed timestamp (bake.go). The clock only
	// ever ticks for pointers that verified above.
	for k, ptr := range pointers {
		if ptr != nil && bake.Observe(k.acct, k.app, k.channel, ptr.Version, now) {
			changed = true
		}
	}
	// ORDER IS LOAD-BEARING: fold this pass's verified flag reads into the
	// box's production memory HERE, before any bake verdict below is taken.
	// That is what makes prodMemory's "a mixed-version rollout costs at most
	// one pass of this memory" true — a binary that wiped the latch re-latches
	// it from the bucket and honors it in the SAME pass, never a pass later.
	//
	// The err == nil guard is the second lock on a door ObserveProduction's
	// own nil-check already holds (GetProduction returns a nil flag on every
	// error path, and a nil flag is a no-op there). It stays because the rule
	// it states — the memory only ever ticks for VERIFIED documents — is
	// worth being true at the call site too, not only at the callee.
	for ak, ps := range prod {
		if ps.err == nil && bake.ObserveProduction(ak.acct, ak.app, ps.flag, now) {
			changed = true
		}
	}
	// bake.tainted forces a Save for a different reason than migrated: the
	// clock file existed but did not read or parse, so every held head on
	// this box is failing toward baking until that file is replaced
	// (ProductionBake). Left alone, the corrupt file would survive every
	// changed-gated pass and the box would bake forever.
	//
	// But ONLY a full pass may perform that repair. A NARROWED pass
	// (extra != nil — the fast-converge sweep, every 2-3 seconds) fetches
	// production flags only for the held heads inside its narrowed set —
	// often none (canary/* has no hold), sometimes a few (an unstaged
	// stable/beta member), never reliably ALL of the apps whose latches the
	// clock file carries: its in-memory Production map is PARTIAL, not
	// repaired. The pipeline ratchet's Holds map is partial for the same
	// reason (pipelines are read only for the narrowed set's own non-canary
	// apps), so this skip protects both memories at once. Letting it write
	// would overwrite the corrupt file with that partial view AND clear the
	// taint that is currently failing closed,
	// dropping the latch of every app outside its set before anything
	// re-observed those flags — and if a bucket flag had also been deleted,
	// that latch would be gone for good. So a tainted clock on a narrowed
	// pass skips the Save entirely, forced or ordinary: a pass that observed
	// only a slice of the flags cannot repair a file that carries them all.
	// Nothing is lost — the observations it would have recorded re-occur
	// within seconds, and the hourly pass does the repair with the full flag
	// set actually in hand.
	//
	// Residual, accepted: even the full pass only re-observes flags for apps
	// with a currently-resolved HELD head, so an app whose channel pointer is
	// simultaneously unreadable keeps its latch out of the repaired file
	// until that pointer heals. That is triple-tamper territory (corrupt
	// clock file AND unreadable pointer AND deleted flag) and sits inside the
	// box-file trust boundary LoadFirstSeen already documents as accepted.
	switch {
	case bake.tainted && extra != nil:
		// skip: see above.
	case changed || bake.tainted:
		if err := bake.Save(); err != nil {
			return fmt.Errorf("record bake clock: %v", err)
		}
	}

	// A version that failed its restart on one host isn't tried on the next
	// this pass — one bad build shouldn't bounce every host in turn.
	poisoned := map[string]bool{}

	// One definition of "does the window apply here", shared by the awake and
	// hibernating loops below — the same question must not be able to be
	// answered two ways for the same app depending on whether it sleeps.
	//
	// RouteAccount(rt), never the raw rt.Account column: the group and
	// bakeable keys were built from RouteAccount output, and an unmigrated
	// row can still carry an account NAME there, which resolves to ops.
	bakeIgnore := func(rt registry.Route) bool {
		ak := acctApp{RouteAccount(rt), rt.App}
		// A PIPELINED app never consults the flag: its bake policy is the
		// pipeline's alone (spec §3) — never ignore, and a zero-window
		// channel holds nothing anyway.
		//
		// "Pipelined" here is every app this pass found pipeline EVIDENCE
		// for (floored), not only the apps whose read verified this pass:
		// a verified pipeline, or a ratchet this box is still running for
		// the app (holdMemory.liveRatchet — a window that differs from the
		// name's legacy resolution in EITHER direction, a pending
		// loosening, or a deletion whose cycle has not run out). That
		// memory IS the evidence the app was pipelined, and it is
		// deliberately NOT the appHolds overlay: the overlay only ever
		// RAISES a window above legacy, and an app-defined channel name
		// resolves to HoldFor's 72h fail-closed default, so a declared 24h
		// on "production" raises nothing while being the very policy the
		// hold exists to enforce. Reading the overlay collapsed exactly
		// that app (PR #258 review, C1). Riding the
		// verdict on the console's production: true CO-WRITE instead would
		// make enforcement depend on a bucket write this box can never
		// confirm happened: an app pipelined before the co-write existed, an
		// app whose flag was deleted, a direct-bucket pipeline write — each
		// leaves a flagless app whose held heads would adopt the instant one
		// GET of pipeline.json errored, or the instant the object was
		// deleted, defeating the very floor computed above. The co-write
		// keeps its own job, which was always the MEMORY-LESS box (spec §3's
		// old-binary edge — any box that never saw the pipeline).
		//
		// The flag path is left to apps with no pipeline signal at all —
		// every legacy app, byte-identically — and to a once-pipelined app
		// whose every remembered channel has finished ratcheting back to
		// its legacy default: nothing is live then, the set is empty for
		// the app, and it is a legacy app again in full.
		if floored[ak] {
			return false
		}
		return !bake.ProductionBake(ak.acct, ak.app, prod[ak].err, bakeable[ak], DefaultHolds, now)
	}
	// routeHolds resolves the same per-app window map for a route that the
	// bakeable filter above used — shared for the same reason bakeIgnore is:
	// the same question must not be answerable two ways for one app.
	routeHolds := func(rt registry.Route) map[string]time.Duration {
		return holdsFor(acctApp{RouteAccount(rt), rt.App})
	}

	// Awake hosts first, hibernating second: repoints have no health gate, so
	// let an awake host discover (and poison) a broken build before any
	// sleeper is moved onto it.
	var hibernating, sidecars []registry.Route
	for _, rt := range routes {
		if !adopts(rt) {
			continue
		}
		if rt.IsSidecar() {
			sidecars = append(sidecars, rt)
			continue
		}
		if rt.Hibernate {
			hibernating = append(hibernating, rt)
			continue
		}
		ptr := pointers[groupKey{RouteAccount(rt), rt.App, routeChannel(rt)}]
		adoptHost(ctx, selected[rt.Host], reg, s, optDir, bake, poisoned, rt, ptr, restart, now, cfg, routeHolds(rt), bakeIgnore(rt))
	}
	for _, rt := range hibernating {
		ptr := pointers[groupKey{RouteAccount(rt), rt.App, routeChannel(rt)}]
		// cfg.restartHibernating is nil by default (and for RunAdopt): the
		// pure repoint, never woken. The capability converge still runs for
		// them (a drop-in is unit text, not a wake); see adoptHost.
		//
		// When it IS set — the agent's in-process loops — the callback is
		// consulted for sleepers too, and the "never woken" guarantee moves
		// into the callback: RestartHibernating only accepts one that cannot
		// start a process (Bounce is false-when-not-awake, try-restart is a
		// no-op on a stopped unit). The flag says a route MAY sleep, not that
		// it is asleep now, and a warm exec child must not keep serving a
		// binary its symlink no longer names.
		adoptHost(ctx, selected[rt.Host], reg, s, optDir, bake, poisoned, rt, ptr, cfg.restartHibernating, now, cfg, routeHolds(rt), bakeIgnore(rt))
	}
	// Native sidecars LAST, and never through a restart callback (spec
	// §7.9). The pass stages, repoints and stamps. The running worker moves
	// when the activator's drift check sees the link change, which plan 5
	// queues as a paced release attempt; a Bounce here would stop an always-on
	// worker with nothing to start it. Last for the reason sleepers go after
	// awake hosts: a repoint has no health gate, so a build an instance
	// poisons this pass never reaches one.
	for _, rt := range sidecars {
		ptr := pointers[groupKey{RouteAccount(rt), rt.App, routeChannel(rt)}]
		adoptHost(ctx, selected[rt.Host], reg, s, optDir, bake, poisoned, rt, ptr, nil, now, cfg, routeHolds(rt), bakeIgnore(rt))
	}
	if cfg.restartErrors != nil {
		return errors.Join(*cfg.restartErrors...)
	}
	return nil
}

// Adoptable filters the routes a Pass moves: lifecycle-managed instances of
// a known app that are ready (an app-less or mid-provision route has nothing
// to adopt or must not be disturbed), excluding platform-minted routes.
//
// A platform domain (internal/domains) is an ALIAS for an app's canonical
// route — a second name pointing at the same backend — not a host with a
// lifecycle of its own. Adopting one would stage a redundant copy of the
// artifacts, mint a live/<alias> symlink nothing ever execs, try-restart a
// carlos-app@<alias> unit that does not exist, and stamp Version onto a row
// the console then renders as though it were a real instance. The canonical
// route it aliases is adopted normally — that IS the deploy, and the alias
// serves the new version the moment it lands.
//
// Exported for host.fastMembers (the fast-converge sweep's membership scan),
// the same reason CapabilityRoute below is: the sweep must count as
// "outstanding work" exactly the routes a pass would consider, or a route
// the pass refuses on shape would be swept forever. A second copy of this
// predicate in another package is exactly the drift that would let the two
// disagree.
func Adoptable(rt registry.Route) bool {
	return rt.Kind == "instance" && rt.App != "" && !rt.Platform &&
		(rt.Status == "" || rt.Status == "ready")
}

// SidecarAdoptable is the pass's SECOND admission predicate (zero-downtime
// deploys spec §7.9): a native sidecar row (Kind "sidecar") of a known app,
// not platform-minted, and ready. Its build follows the channel on its own
// row, as sidecar_reconcile.go provisions it, through the same signed
// selection, bake holds, staging, repoint and stamp an instance gets.
//
// Adoptable is deliberately NOT widened: capability drop-ins, the fast
// sweep's membership, config sync, the restart passes and the console's
// Restartable flag all keep their own answer for a sidecar. The pass opts in
// by name, and only the pass.
func SidecarAdoptable(rt registry.Route) bool {
	return rt.IsSidecar() && rt.App != "" && !rt.Platform &&
		(rt.Status == "" || rt.Status == "ready")
}

// CapabilityRoute reports whether this route may materialize capabilities:
// a lifecycle-managed instance of a known app, unit-backed, non-platform,
// ready (spec §4e). It is Adoptable() plus the unit backing, deliberately —
// a route this pass would not adopt is not one whose unit shape this pass
// has any business rewriting.
//
// A platform ALIAS never materialises anything: an alias is a second name
// for one process, and a second carlos-app@ unit claiming :25 would break
// the real one. Adoptable() already excludes them and this depends on that.
//
// An EXEC-backed route (Backing == "") whose release declares capabilities
// materialises nothing here and is told so once per pass by RunAdopt's
// advisory, not by this function — that is the guided path through the §5
// migration.
//
// Exported for host.capabilityPass (the hourly grant-drift pass): the two
// converge entry points must agree, to the character, on which routes may
// have a root-owned unit drop-in written for them. A second copy of this
// predicate in another package is exactly the drift that would let one pass
// materialize a shape the other refuses.
func CapabilityRoute(rt registry.Route) bool {
	return Adoptable(rt) && rt.Backing == registry.BackingUnit
}

// routeChannel is the channel a route follows; "" means stable.
func routeChannel(rt registry.Route) string {
	if rt.Channel == "" {
		return "stable"
	}
	return rt.Channel
}

// RouteAccount is the account whose bucket keys a route's app lives under:
// the sqid stamped on the row, else the built-in ops.
//
// The ValidAccount gate is doing migration work, not paranoia. Before the
// multi-account change the account column held an account NAME ("ops",
// "paul@keymail.dev"), and a box can be running this code against a registry
// still carrying those. A name is not a sqid, so it reads as ops — whose keys
// are tried qualified first and then fall back to the legacy unqualified
// shape, which is exactly where an unmigrated app's objects still are.
func RouteAccount(rt registry.Route) string {
	if ValidAccount(rt.Account) {
		return rt.Account
	}
	return OpsAccount
}

// adoptHost converges one host onto its channel head ptr (nil = the group's
// pointer didn't resolve; skip quietly, the group phase already logged).
// restart==nil is the symlink-only path (a caller with no proc manager, or a
// hibernating host on a pass that did not set RestartHibernating): repoint
// and record, never wake, never health-gate. A hibernating host WITH that
// option set arrives here with a non-nil callback and takes the ordinary
// path — safe because the option only accepts a callback that cannot start a
// process, so "never woken" survives the callback being consulted. The one
// start this function itself can perform (cfg.startUnit, the restart-failure
// rollback below) does not weaken that: it is armed only by that callback
// RETURNING AN ERROR, which a stopped unit's exit-0 try-restart never
// produces — WithStartUnit's doc carries the full argument.
// cfg is the call's PassOptions, already applied: cfg.nag nil (the ordinary
// case) means the canary-TTL nag below never fires, never even builds its
// string, and cfg.capsOps nil means the capability converge never runs at
// all. See NagExpiredCanaries and WithCapabilities for who sets each and why.
//
// holds is the enforced per-channel window map the caller resolved for this
// route's app: a verified pipeline's ratcheted HoldsForApp map (canary/*
// stitched back in at zero), DefaultHolds floored by the ratchet memory for
// an app whose pipeline read errored or is verified-absent mid-ratchet, else
// DefaultHolds — never consulted app-blind.
//
// ignore is the app-level bake verdict from ProductionBake — the LEGACY
// production flag's doing, never a publisher field, and constant false for
// any app this box carries pipeline evidence for: a verified pipeline, or a
// ratchet this box is still running for it (holdMemory.liveRatchet, which is
// NOT the same question as "does the memory raise a window" — see
// bakeIgnore). Such an app's windows come from holds alone (spec §3). True
// means the supply-chain window does not CURRENTLY apply to this app on
// this box (not declared production, or declared off and past the ratchet);
// the caller computes it once per route from the flag it read this pass.
func adoptHost(ctx context.Context, work hostwrite.Work, reg *registry.Registry, s Store, optDir string, bake *FirstSeen, poisoned map[string]bool, rt registry.Route, ptr *ChannelPointer, restart func(host string) error, now time.Time, cfg passConfig, holds map[string]time.Duration, ignore bool) {
	// settings: this route's build is the instance update settings' decision
	// (settings spec §4.3). Rules 1-3 need no channel head, so a route with
	// settings is not skipped when the pointer is unreadable (Astra I1 #9).
	settings := cfg.settings && ((rt.Kind == "instance" && rt.Unit == "") || cfg.rootUpdates != nil)
	if ptr == nil && !settings {
		return
	}
	nag := cfg.nag
	// The nag (spec §5 "Canaries expire"): visibility only, and deliberately
	// BEFORE the up-to-date short-circuit below — an expired canary that
	// isn't moving (the common case: the branch shipped once and nobody
	// touched it since) must still be seen every pass it's part of, not just
	// the passes that happen to change something. Nothing here blocks adopt,
	// restart or promote; it only (optionally) prints. Gated on nag != nil
	// rather than always printing to os.Stdout: see NagExpiredCanaries — this
	// must stay silent on the 2-3-second canary fast loop and speak only on
	// the hourly root adopt pass, or an expired canary sitting still would
	// nag every few seconds forever instead of once an hour.
	if nag != nil && ptr != nil && CanaryExpired(ptr, now) {
		// FormatCanaryAge(CanaryAge(...)), NOT a raw now.Sub(ptr.PromotedAt):
		// that would print the pointer's TOTAL age (CanaryAge + CanaryTTL),
		// overstating how long it's been expired by exactly one lease length
		// (2026-08-03 review, IMPORTANT 2) — and disagreeing with the
		// console's badge, which renders the same CanaryAge/FormatCanaryAge
		// pair (internal/console/ladder.go's canaryExpiry). One formatter,
		// shared, so the two operator-facing surfaces of this feature cannot
		// drift apart on what a duration means.
		fmt.Fprintf(nag, "  %s: canary %s (%s) has been expired %s — still serving, nothing blocked\n",
			rt.Host, ptr.Channel, ptr.Version, FormatCanaryAge(CanaryAge(ptr, now)))
	}
	channel := routeChannel(rt)
	ver := ""
	if ptr != nil {
		ver = ptr.Version
	}
	cur, err := LiveVersion(optDir, rt.Host)
	if err != nil {
		fmt.Printf("  %s: read live link: %v — skipped\n", rt.Host, err)
		return
	}
	// Repair opportunity for the LIVE version's sibling, every pass, whether
	// or not this pass moves the symlink. C1 (2026-08-03 review):
	// ensureStaged's own back-fill only runs from the call below, which sits
	// behind the cur == ver short-circuit right after this — a version
	// that's already live never calls ensureStaged again, so a sibling write
	// that failed once (a full disk, a permissions hiccup) would otherwise
	// be stuck missing forever for exactly the version actually running.
	// This call never fails the pass: an absent or unreadable sibling is the
	// safe leave-alone state, so a repeated failure here is only a log line.
	if cur != "" && ptr != nil {
		backfillManifestDigest(ctx, s, optDir, RouteAccount(rt), rt.App, cur, rt.Host, ptr.DigestOf(cur), cfg.stage)
	}
	// Instance update settings (settings spec §4): one decision per route.
	// Everything below (bake for the decisions that take it, the proof gate,
	// refusals, D2, caps, restart) applies to whatever it chooses. A route
	// with default settings and no consumed file gets today's head.
	var dec TargetDecision
	if settings {
		var stop bool
		ver, dec, stop = settingsTarget(ctx, optDir, bake, rt, ptr, cur, now, cfg, holds, ignore)
		if stop {
			return
		}
	}
	// sel is the build this pass installs, as the gates below read it (Astra
	// I2 #8): the head itself when the decision is the head, else a pointer
	// naming only the selected build, the way heldProofAdmit's stand-in does
	// (materializeheld.go). The bake waiver, the proof gate (admit and bind),
	// D1, D2 and the refusal records judge the build that would run, never a
	// head the decision passed over. ptr stays the real head, nil or not, for
	// the staging digest below.
	sel := ptr
	if ptr == nil || ver != ptr.Version {
		sel = &ChannelPointer{Channel: channel, Version: ver}
	}
	if cfg.requireRootHistory && sel == ptr && ptr != nil {
		if err := cfg.recheckHistory(ctx, rt); err != nil {
			cfg.historyError(err)
			return
		}
		if err := cfg.rootHistory.History.RetireD1(cfg.historySelected[rt.Host].binding, ptr); err != nil {
			cfg.historyError(err)
			return
		}
	}
	if cur == ver {
		cfg.proof.clear(rt.Host) // live on its target: nothing left to wait for
		// A refusal that no longer blocks the pointer is stale even though
		// nothing moves: v2 was refused and rolled back to v1, then the
		// channel went back to v1. Without this every later pass returns
		// here and the watcher reports the v2 refusal forever (final review
		// F8). Best-effort, like the clear after a successful adoption: the
		// agent cannot remove root's file, so its failure is ignored and the
		// next root pass retries. A record that still blocks stays: the link
		// is on the refused build only when its rollback failed, and that is
		// worth showing. The record is judged against the build this route
		// selected (sel) and against the head (ptr): under a pin or a hold
		// a refusal of the head that is still current stays, so an unpin
		// does not retry a promotion this box already refused (Task 12a).
		if cfg.bakeDir != "" {
			if r, err := ReadRefusal(cfg.bakeDir, rt.Host); err == nil && r != nil && !r.Blocks(sel) && !r.Blocks(ptr) {
				_ = ClearRefusal(cfg.bakeDir, rt.Host)
			}
		}
		return // up to date — stay quiet on the periodic tick
	}
	// Two independent waivers, both root-signed in origin: ignore is the
	// app's legacy production verdict (§2/§4; constant false for a pipelined
	// app), ptr.FastTrack is this ONE pointer's per-promotion waiver (§5) —
	// deliberately unratcheted (pipelines spec §3 amendment: ratcheting it
	// would make it wait out the very window it exists to skip). Either one
	// alone releases the head.
	//
	// Each return from here to admit stops D3's gate being asked about this
	// head, so each one also takes the host off the proof waiting list
	// (cfg.proof.clear). A wait left behind for a head the channel has moved
	// off would stay due for ever, and the fast loop would sweep the host
	// every tick for the whole hold (plan 3 ruling 5).
	//
	// Under settings only rule 4 and a new held instance's first head bake
	// (dec.Bake): a pin, a resolved T and a recovery waive it (§4.3). Both
	// baking decisions name the head, so sel is ptr there.
	if !settings || dec.Bake {
		if hold := Hold(bake, RouteAccount(rt), rt.App, channel, ver, holds, ignore || sel.FastTrack, now); hold != "" {
			fmt.Printf("  %s: %s (%s) is %s\n", rt.Host, ver, channel, hold)
			cfg.proof.clear(rt.Host)
			return
		}
	}
	if poisoned[ver] {
		fmt.Printf("  %s: skipping %s (failed on another host this pass)\n", rt.Host, ver)
		cfg.proof.clear(rt.Host)
		return
	}
	// D1's refusal (refusal.go): this box rolled this exact promotion back
	// because the unit did not come up. Checked before ensureStaged so a
	// refused build is neither downloaded nor relayed to a root pass.
	//
	// A D3 refusal (RefusalNotProven) is the one kind that depends on the
	// policy: it stands only while this pass's policy still requires a proof
	// for the app, or failed to load (closed). Once the require entry or the
	// whole file is gone, the record is ignored and best-effort cleared, so
	// removing the requirement is a break glass that works after a binding
	// refusal too (final review H3). The agent cannot remove root's file;
	// that failure is ignored and the next root pass clears it. D1 and D2
	// records never depend on the policy.
	if cfg.bakeDir != "" && !cfg.requireRootHistory {
		r, err := ReadRefusal(cfg.bakeDir, rt.Host)
		if err == nil && r.Blocks(sel) && r.Kind == RefusalNotProven {
			if required, _ := cfg.proof.Requires(rt.App); !required {
				if ClearRefusal(cfg.bakeDir, rt.Host) == nil {
					fmt.Printf("  %s: cleared the not-proven refusal of %s: this box no longer requires a proof for %s\n", rt.Host, ver, rt.App)
				}
				r = nil
			}
		}
		if err != nil {
			fmt.Printf("  %s: reading refusal record: %v\n", rt.Host, err)
		} else if r.Blocks(sel) {
			switch r.Kind {
			case RefusalWrongProgram:
				// --retry would re-stage the same bytes and be refused again,
				// and re-shipping this version cannot help: its manifest is
				// immutable, so the consent has to ride on a new version.
				fmt.Printf("  %s: refused: %s is the wrong program for this route (%s); still on %s; ship a new version with carlos ship --program-change and promote it, if the change is intended\n",
					rt.Host, ver, r.Reason, r.RestoredVersion)
			case RefusalNotProven:
				// D3's binding check (proofgate.go bind). Nothing was started
				// or rolled back, and --retry would re-sign the pointer over
				// the same staged bytes and be refused again.
				fmt.Printf("  %s: refused: %s's staged bytes do not match dev's proof (%s); still on %s; roll the exact bytes dev proved\n",
					rt.Host, ver, strings.TrimPrefix(r.Reason, "not proven: "), orNothing(r.RestoredVersion))
			default:
				fmt.Printf("  %s: refused: %s did not come up on this box at %s (%s); retry with: carlos promote --retry %s %s\n",
					rt.Host, ver, r.RefusedAt.Format(time.RFC3339), r.Reason, ver, channel)
			}
			cfg.proof.clear(rt.Host)
			return
		}
	}
	if cfg.requireRootHistory {
		selHistory := cfg.historySelected[rt.Host]
		refs, err := cfg.rootHistory.History.Refusals(selHistory.binding)
		if err != nil {
			cfg.historyError(err)
			return
		}
		for _, ref := range refs {
			if ref.Kind == RefusalNotUp && ref.Blocks(sel) {
				return
			}
		}
	}
	// D3's cheap gate (proofgate.go, spec §4.5 check 1): before ensureStaged,
	// so an unproven build is neither downloaded nor relayed to root.
	verdict, admitted := cfg.proof.admit(ctx, s, optDir, rt, sel, cur)
	if !admitted {
		return
	}
	// A named unit this process cannot restart is root's (RootRestartsUnits),
	// and so is every route its process serves: ask for root's pass and touch
	// nothing. Silent on the tick; the requester logs each request it writes.
	if cfg.unitsToRoot && restart != nil && !cfg.hostWrites.Enabled && unitServed(reg, rt) {
		if cfg.stageBlocked != nil {
			cfg.stageBlocked(rt.Host, ver, ErrUnitRestartNeedsRoot)
		}
		return
	}
	// The evidence that authorizes ver (settings spec §4.1). Only today's
	// adoption of the head, and a request whose T is still the head, rest on
	// the signed pointer and its digest; everything else needs a promotion
	// record with a matching manifest digest (stageAuthorized). The pointer
	// branch is refused for any build but the head itself: a non-head build
	// has no pointer digest, and staging it unbound on a digested channel is
	// the #253 replay (invariant 3).
	var m *Manifest
	switch {
	case !settings || dec.Evidence == EvidencePointer:
		if ptr == nil || ver != ptr.Version {
			fmt.Printf("  %s: %s: not the signed head, and the decision named no other evidence; not staged\n", rt.Host, ver)
			cfg.proof.clear(rt.Host)
			return
		}
		m, err = ensureStagedDigest(ctx, s, optDir, RouteAccount(rt), rt.App, ver, rt.Host, ptr.DigestOf(ver), cfg.stage)
	case dec.Evidence == EvidenceRecord:
		m, err = stageAuthorized(ctx, s, cfg.pub, optDir, RouteAccount(rt), rt.App, ver, rt.Host, cfg.stage)
	default:
		fmt.Printf("  %s: %s: no evidence this pass can check; not staged\n", rt.Host, ver)
		cfg.proof.clear(rt.Host)
		return
	}
	if err != nil {
		fmt.Printf("  %s: %v\n", rt.Host, err)
		// A version this uid cannot write is the one staging failure a
		// privileged pass would finish; everything else (bucket, signature,
		// manifest) fails the same way as root. See OnStageBlocked.
		if cfg.stageBlocked != nil && stageBlocked(err) {
			// No manifest identity exists on this refusal path, so there is no
			// admitted host mutation to relay under lifecycle mode.
			if !cfg.hostWrites.Enabled {
				cfg.stageBlocked(rt.Host, ver, err)
			}
		}
		return
	}
	id, err := releaseIdentity(m)
	if err != nil {
		fmt.Printf("  %s: release identity: %v\n", rt.Host, err)
		return
	}

	// D2 (programhistory.go): the staged bytes, not the manifest, decide.
	// Skipped for non-Go binaries and when this pass has no bake dir.
	newProg := FileProgram(VersionPath(optDir, RouteAccount(rt), ver, rt.App))
	if (cfg.bakeDir != "" || cfg.requireRootHistory) && newProg != "" {
		var hist []string
		fresh := false
		if cfg.requireRootHistory {
			hist, err = cfg.rootHistory.History.Programs(cfg.historySelected[rt.Host].binding)
			if err != nil {
				cfg.historyError(err)
				return
			}
			sel := cfg.historySelected[rt.Host]
			fresh, err = cfg.rootHistory.History.FreshInstallation(sel.binding, sel.creation)
			if err != nil {
				cfg.historyError(err)
				return
			}
		} else {
			hist = ReadPrograms(cfg.bakeDir, rt.Host)
		}
		// The live binary is what the route is running, so its program belongs
		// in the history whatever the file already says: a history of [A] on a
		// route that is live on B (a change made before this check existed, or
		// a record lost) must not refuse the next B build.
		if cur != "" && !cfg.requireRootHistory {
			if live := FileProgram(VersionPath(optDir, RouteAccount(rt), cur, rt.App)); live != "" && !slices.Contains(hist, live) {
				_ = RecordProgram(cfg.bakeDir, rt.Host, live)
				hist = append(slices.Clone(hist), live)
			}
		}
		if (len(hist) > 0 || cfg.requireRootHistory && !fresh) && !slices.Contains(hist, newProg) && m == nil {
			// ensureStaged has the binary but could not obtain its manifest
			// (no sibling, and the store did not answer), so whether the
			// release carries --program-change is unknown. Unknown is not a
			// refusal: a record here would sit at the gate above and stop the
			// next pass from ever reading the manifest, so an authorised
			// change would stay blocked after the store recovered. Not pointed
			// live, and the next pass asks again.
			fmt.Printf("  %s: %s is %s, not in this route's history, and its manifest could not be read to check for --program-change; not pointed live, will retry\n",
				rt.Host, ver, newProg)
			return
		}
		programChange := m != nil && m.ProgramChange
		if cfg.requireRootHistory {
			programChange = programChange && ptr.DigestOf(ver) != ""
		}
		if (len(hist) > 0 || cfg.requireRootHistory && !fresh) && !slices.Contains(hist, newProg) && !programChange {
			fmt.Printf("  %s: %s is %s, but this route has only run %s. not pointed live (ship a new version with carlos ship --program-change and promote it, if the change is intended)\n",
				rt.Host, ver, newProg, strings.Join(hist, ", "))
			sha := m.Artifacts[adoptPlatform].SHA256
			// Nothing was restarted, so the box is still on its old build.
			if err := cfg.recordRefusal(ctx, rt, Refusal{Host: rt.Host, App: rt.App, Version: ver, SHA256: sha,
				Platform: adoptPlatform, PointerPromotedAt: sel.PromotedAt, RefusedAt: now.UTC(),
				Reason: "wrong program: " + newProg,
				Kind:   RefusalWrongProgram, Recovered: true, RestoredVersion: cur}); err != nil {
				cfg.historyError(err)
				fmt.Printf("  %s: writing refusal record: %v\n", rt.Host, err)
			}
			return
		}
	}

	// D3's binding check (spec §4.5 check 2): the staged bytes themselves
	// must be the proven ones, or a build this box has run well.
	if !cfg.bindProof(ctx, optDir, rt, sel, verdict, cur, now) {
		return
	}
	var historyAttempt string
	if cfg.requireRootHistory {
		if m == nil || m.Artifacts[adoptPlatform].SHA256 == "" {
			cfg.historyError(os.ErrPermission)
			return
		}
		historyIdentity := id
		if !cfg.hostWrites.Enabled {
			historyIdentity = hostwrite.Identity{}
		}
		historyAttempt, err = cfg.rootHistory.begin(ctx, rt, cfg.historySelected[rt.Host], sel, strings.ToLower(m.Artifacts[adoptPlatform].SHA256), now, historyIdentity)
		if err != nil {
			cfg.historyError(err)
			return
		}
	}

	// The capability converge for the NEW version, BEFORE PointLive (spec
	// §4d). m is the manifest ensureStaged has in hand — never a re-read of
	// the staged sibling, whose write is best-effort: a sibling that failed
	// to write reads back as (nil, nil), which converge would take as
	// "leave the drop-ins alone", and the new version would go live against
	// the previous version's unit shape silently. Here a manifest this pass
	// could not obtain REFUSES the version instead (capabilityPlan).
	//
	// A hibernating route converges here too: a drop-in is unit text, not a
	// wake, and nothing below starts it. (It reaches this with restart == nil
	// on an ordinary pass, or with the RestartHibernating callback on the
	// agent's loops — which cannot start anything either. In practice this
	// branch is root-only regardless: cfg.capsOps is set by RunAdopt alone,
	// and RunAdopt does not set RestartHibernating.)
	var (
		prev      caps.Snapshot
		plan      caps.Plan
		converged bool
	)
	if cfg.capsOps != nil && CapabilityRoute(rt) {
		// restart != nil is what makes the deferral safe: it says somebody is
		// going to start this daemon after the symlink moves. With no restart
		// callback at all (a caller with no proc manager) converge keeps
		// today's behaviour and brings the unit up itself.
		plan, err = capabilityPlan(rt, m, restart != nil)
		var res caps.Result
		if err == nil {
			err = withReleaseWork(ctx, work, id, func() error {
				if err := cfg.recheckHistory(ctx, rt); err != nil {
					cfg.historyError(err)
					return err
				}
				var convergeErr error
				res, convergeErr = caps.Converge(*cfg.capsOps, plan)
				return convergeErr
			})
			prev = res.Prev
		}
		if err != nil {
			// A converge REFUSAL refuses the VERSION on this box, in the same
			// direction and for the same reason as an ensureStaged failure:
			// nothing is pointed live, so the box keeps serving what it has
			// rather than run a new binary against a unit shape the platform
			// declined to write. Converge itself has already written nothing
			// and logged the operator's fix.
			//
			// The version is deliberately NOT poisoned here (2026-08-03
			// review). A refusal bounces nothing — no symlink moved, no unit
			// cycled — so there is no bad build to keep off the next host,
			// which is the only thing the poison exists for. And the refusal
			// is often HOST-shaped: a port another route's drop-in already
			// claims, a state-directory flip mid-migration. Poisoning on one
			// of those would let a single misconfigured route freeze every
			// other host on the box out of a version that is fine for them,
			// pass after pass, until an operator noticed.
			//
			// Converge's own contract: bytes are still moved only when it
			// reports Changed with the error — a failure AFTER its
			// daemon-reload, which it deliberately does not undo because the
			// version-change caller owns it. That is here. Without this the
			// NEW version's unit shape stays on disk under the OLD binary
			// permanently: the next pass finds no diff, snapshots those bytes
			// as its Prev, and every later "restore" restores the wrong
			// shape. Restore is NOT called for an ordinary refusal, where
			// Prev is empty and would be read as "this route had no drop-ins"
			// — that is a deletion, not a restore.
			if res.Changed {
				if err := withReleaseWork(ctx, work, id, func() error { return caps.Restore(*cfg.capsOps, prev) }); err != nil {
					fmt.Printf("  %s: restoring the previous version's capability drop-ins: %v\n", rt.Host, err)
				}
			}
			fmt.Printf("  %s: capabilities for %s were refused: %v — not pointed live\n", rt.Host, ver, err)
			return
		}
		converged = true
	}

	if err := withReleaseWork(ctx, work, id, func() error {
		if err := cfg.recheckHistory(ctx, rt); err != nil {
			return err
		}
		running := ""
		if cfg.runningVersion != nil {
			var err error
			running, err = cfg.runningVersion(ctx, rt)
			if err != nil {
				return err
			}
		}
		if err := PointLive(optDir, rt.Host, RouteAccount(rt), ver, rt.App, running); err != nil {
			return err
		}
		if running != "" {
			cur = running
		}
		return nil
	}); err != nil {
		cfg.historyError(err)
		// The symlink did not move, so the box is still serving cur — and the
		// unit shape on disk is the new version's. Put the old one back, or a
		// port the running binary never asked for stays publicly bound.
		// Unlike the restart-failure branch below there is no cur == ""
		// exception: nothing is live either way, so "restore what the symlink
		// names" means removing what this pass just wrote.
		rolled := ""
		if converged {
			if rerr := withReleaseWork(ctx, work, id, func() error { return caps.Restore(*cfg.capsOps, prev) }); rerr != nil {
				fmt.Printf("  %s: restoring the previous version's capability drop-ins: %v\n", rt.Host, rerr)
			} else {
				rolled = " — restored the capability drop-ins the symlink still names"
			}
		}
		fmt.Printf("  %s: repoint failed: %v%s\n", rt.Host, err, rolled)
		// A release staged by an earlier root pass can still need root to
		// swap its live symlink. Relay that write refusal too, or the fast
		// loop never requests another pass once the artifact exists.
		if cfg.stageBlocked != nil && stageBlocked(err) {
			if guardErr := withReleaseWork(ctx, work, id, func() error { return nil }); guardErr != nil {
				fmt.Printf("  %s: release authority changed before privileged staging relay: %v\n", rt.Host, guardErr)
			} else {
				cfg.stageBlocked(rt.Host, ver, err)
			}
		}
		return
	}
	if cfg.bakeDir != "" && newProg != "" && !cfg.requireRootHistory {
		if err := RecordProgram(cfg.bakeDir, rt.Host, newProg); err != nil {
			fmt.Printf("  %s: recording program history: %v\n", rt.Host, err)
		}
	}

	// The accepted window, made visible (2026-08-03 ruling). A pass with no
	// WithCapabilities — the in-process adoptLoop, and above all the 2-3
	// second fastConvergeLoop, neither of which can write /etc from inside
	// carlos-edge.service — has just moved a unit-backed route onto a version
	// that declares capabilities, without converging its unit shape.
	//
	// The signal is "the new manifest declares anything at all", not "the new
	// set differs from what is materialized": the honest difference test needs
	// the route's grants and a render of both sides, and this line exists to
	// mark a window, not to diff it. It costs nothing — m is already in hand —
	// and it errs toward one extra line, once per version change per host,
	// never per tick.
	if cfg.capsOps == nil && CapabilityRoute(rt) && m != nil && m.Capabilities != nil {
		fmt.Printf("  %s: %s declares capabilities and this pass does not write unit configuration — it runs against the drop-ins the previous version left until the root adopt pass (carlos-adopt.service, hourly) converges them\n", rt.Host, ver)
	}

	if restart != nil {
		if err := withReleaseWork(ctx, work, id, func() error { return cfg.recheckHistory(ctx, rt) }); err != nil {
			cfg.historyError(err)
			fmt.Printf("  %s: release authority changed before restart: %v\n", rt.Host, err)
			return
		}
		// A converged DAEMON's restart is caps.Restart, not the caller's
		// callback: this is the only call site that knows the plan, and it is
		// the only restart whose error must be real. rootRestartRoute's
		// try-restart (and its swallow-and-continue) keeps owning every other
		// route, byte-identically to before.
		//
		// plan.Set is non-nil whenever converged is true (capabilityPlan
		// guarantees it); the nil check is here anyway, because a nil deref
		// on this path would panic the root adopt pass for the whole box.
		if cfg.restartAttemptSettled != nil {
			defer func() {
				if err := cfg.restartAttemptSettled(rt.Host); err != nil && cfg.restartErrors != nil {
					*cfg.restartErrors = append(*cfg.restartErrors, err)
				}
			}()
		}
		var rerr error
		daemonRestart := converged && plan.Set != nil && plan.Set.Daemon
		if cfg.hostWrites.Enabled {
			if cfg.selectedRestart == nil {
				rerr = hostwrite.ErrUnavailable
			} else {
				rerr = cfg.selectedRestart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id})
			}
		} else if daemonRestart {
			rerr = caps.Restart(*cfg.capsOps, plan)
		} else {
			rerr = restart(rt.Host)
		}
		if rerr != nil {
			if cfg.restartErrors != nil {
				*cfg.restartErrors = append(*cfg.restartErrors, rerr)
			}
			// Poison the build for the rest of the pass and repoint back —
			// the host keeps running what it ran. The symlink goes back
			// FIRST: caps.Restore may start the service, and it must start
			// on the old binary.
			//
			// The repoint's error is CHECKED, where it was once discarded:
			// "link present, staged version gone" is a documented real state
			// (materialize.go — an operator's /opt cleanup, a partially
			// restored volume), and in it PointLive refuses and the symlink
			// STILL NAMES ver. Everything below that claims a rollback — the
			// restore, the summary line, above all the start — must key on
			// whether the rollback actually HAPPENED, not on whether one was
			// attempted.
			var temporary interface{ Temporary() bool }
			if !errors.As(rerr, &temporary) || !temporary.Temporary() {
				poisoned[ver] = true
			}
			var notUp *UnitNotUpError
			isNotUp := errors.As(rerr, &notUp)
			rolledBack := false
			var rollbackErr error
			if cur != "" {
				if perr := withReleaseWork(ctx, work, id, func() error { return PointLive(optDir, rt.Host, RouteAccount(rt), cur, rt.App) }); perr != nil {
					fmt.Printf("  %s: could not roll the symlink back to %s: %v\n", rt.Host, cur, perr)
					rollbackErr = perr
				} else {
					rolledBack = true
				}
			}
			// The rule is one line long: restore the unit shape that matches
			// whatever the symlink NOW names.
			//
			// With cur == "" there was no previous version and the symlink
			// above could not be rolled back, so it still names ver — and the
			// drop-ins already match it. Restoring here would hand Restore an
			// empty snapshot, which it reads as "this route had no generated
			// drop-ins" and honours by REMOVING the ones this pass wrote,
			// leaving the live version running with no capability shape at
			// all until the hourly grant-drift pass rebuilds it. A rollback
			// that itself FAILED lands in the same place for the other
			// reason: the symlink still names ver, and putting the previous
			// version's drop-ins under it would be exactly the shape/binary
			// mismatch this whole branch exists to prevent.
			rolled := ""
			if converged && rolledBack {
				// Bytes are not a binding: Restore re-runs the socket
				// sequence itself when the socket half moves, or the MX
				// stays off :25 after a rollback.
				if err := withReleaseWork(ctx, work, id, func() error { return caps.Restore(*cfg.capsOps, prev) }); err != nil {
					fmt.Printf("  %s: restoring the previous version's capability drop-ins: %v\n", rt.Host, err)
					rolled = " but could not restore the previous version's capability drop-ins"
				} else {
					rolled = " and restored the previous version's capability drop-ins"
				}
			} else if converged {
				if cur == "" {
					rolled = "; there is no previous version to roll back to, so its capability drop-ins stand as converged"
				} else {
					rolled = "; its capability drop-ins stand as converged with the version the symlink still names"
				}
			}
			action := "rolled the symlink back"
			if cur != "" && !rolledBack {
				action = "could not roll the symlink back"
			}
			fmt.Printf("  %s: update to %s failed: %v — %s%s\n", rt.Host, ver, rerr, action, rolled)
			// Start the service again on the version the symlink NOW names —
			// but ONLY when that is the restored one. The rollback above put
			// the bytes back; without this the box is left with a correct
			// symlink and no process — and because try-restart is a no-op on
			// a unit that is not running (which is what lets a
			// deliberately-stopped service stay stopped), the NEXT pass's
			// restart would exit 0, SetVersions would stamp the build that
			// has never run, and the header would claim it over a service
			// that is down. Start, not try-restart: the whole premise here is
			// that the unit is not running.
			//
			// Gated on rolledBack, not cur != "": a rollback that FAILED (the
			// old version's directory gone — PointLive refuses) leaves the
			// symlink naming ver, and a start there would launch the exact
			// build whose restart just failed. And not for a converged daemon
			// (daemonRestart): the restart that failed there was
			// carlos-app@<host>'s, via caps.Restart; its rollback belongs to
			// caps.Restore above, and this callback would start rt.Unit — a
			// unit that pass never touched. AFTER caps.Restore, deliberately:
			// the drop-in shape must match the binary before anything execs
			// it, and on a unit Restore's socket sequence already brought up,
			// start is an exit-0 no-op.
			//
			// A start that also fails changes nothing about the refusal — the
			// version stays poisoned and unstamped — but it is printed,
			// because at that point the box is very likely serving nothing
			// and the operator must not read the rollback line as
			// "recovered". "May": a try-restart that failed without stopping
			// anything (a permission-denied systemctl) fails its start the
			// same way while the old process still serves — the two errors on
			// the line tell them apart.
			//
			// Under lifecycle host writes (hostwrite) the start goes through
			// the selected receiver, as the forward restart does; D1's checked
			// replace (rollbackRestart) is the legacy path's start for a
			// not-up refusal.
			start := cfg.startUnit
			if isNotUp && cfg.rollbackRestart != nil {
				start = cfg.rollbackRestart
			}
			// D1's group (rollback.go): siblings follow bound onto the
			// restored build only when bound's own rollback happened, each
			// under ONE release work of its own: the pass's original
			// selection where it has one (the one the sibling's forward
			// adoption ran under), else a single Select made here. Never a
			// Select per write: a second one could capture a successor's
			// installation and stamp it.
			var group []string
			if isNotUp {
				restored := ""
				if rolledBack {
					restored = cur
				}
				group = rollbackGroup(reg, optDir, rt, ver, restored, func(h string) (func(func() error) error, error) {
					w, ok := cfg.selected[h]
					if !ok || w == nil {
						var err error
						if w, err = cfg.hostWrites.Select(ctx, h, "release"); err != nil {
							return nil, err
						}
					}
					return func(fn func() error) error { return withReleaseWork(ctx, w, id, fn) }, nil
				})
				if notUp.Journal != "" {
					fmt.Printf("  %s: %s's last lines:\n%s\n", rt.Host, notUp.Unit, indent(notUp.Journal, "      "))
				}
			}
			started := false
			var serr, authErr error
			if start != nil && rolledBack && !daemonRestart {
				// Authority lost here skips the start, and every process
				// mutation after it, but NOT the refusal records below: the
				// next pass has fresh authority and an empty poisoned map, and
				// without a record it would adopt the rejected build again.
				if guardErr := withReleaseWork(ctx, work, id, func() error { return nil }); guardErr != nil {
					fmt.Printf("  %s: release authority changed before rollback start: %v — not started\n", rt.Host, guardErr)
					authErr = guardErr
				}
			}
			if start != nil && rolledBack && !daemonRestart && authErr == nil {
				started = true
				if cfg.hostWrites.Enabled {
					if cfg.selectedStart == nil {
						serr = hostwrite.ErrUnavailable
					} else {
						serr = cfg.selectedStart(ctx, hostwrite.Selection{Host: rt.Host, Work: work, Identity: id})
					}
				} else {
					serr = start(rt.Host)
				}
				if serr != nil {
					if cfg.restartErrors != nil {
						*cfg.restartErrors = append(*cfg.restartErrors, serr)
					}
					if rt.Unit == "" && !isNotUp {
						fmt.Printf("  %s: rolled back to %s but could not start it: %v\n", rt.Host, cur, serr)
					} else {
						fmt.Printf("  %s: rolled back to %s but could not start it: %v — this host may be serving nothing\n",
							rt.Host, cur, serr)
					}
				}
			}
			if isNotUp {
				// The refusals go down AFTER the restart, carrying what it
				// did: a record written before it would tell the watcher
				// "rolled back" about a box whose restored build never came
				// up (final review F6).
				rec := recovery{recovered: true}
				switch {
				case cur == "":
					rec = recovery{err: "there is no previous version to roll back to"}
				case !rolledBack:
					rec = recovery{err: fmt.Sprintf("could not roll the symlink back to %s: %v", cur, rollbackErr)}
				case authErr != nil:
					rec = recovery{restored: cur, err: "release authority changed before rollback start: " + authErr.Error()}
				case !started:
					rec = recovery{restored: cur, err: "nothing on this pass restarts the unit on " + cur}
				case serr != nil:
					rec = recovery{restored: cur, err: serr.Error()}
				default:
					rec.restored = cur
				}
				sha := ""
				if m != nil {
					sha = m.Artifacts[adoptPlatform].SHA256
				}
				writeGroupRefusals(cfg, group, rt.App, sel, ver, sha, adoptPlatform, notUp, rec, now)
			}
			return
		}
	}
	if cfg.requireRootHistory {
		if err := cfg.rootHistory.complete(ctx, rt, cfg.historySelected[rt.Host], historyAttempt, ver); err != nil {
			cfg.historyError(err)
			return
		}
	}
	if err := withReleaseWork(ctx, work, id, func() error { return reg.SetVersions(rt.Host, ver, cur) }); err != nil {
		fmt.Printf("  %s: repointed but recording the version failed: %v\n", rt.Host, err)
		return
	}
	// A record left from an older promotion is stale once a build is live.
	// Best-effort: the agent cannot remove root's file, the root pass can.
	if cfg.bakeDir != "" {
		_ = ClearRefusal(cfg.bakeDir, rt.Host)
	}
	if rt.IsSidecar() {
		fmt.Printf("  %s: repointed %s -> %s (%s; sidecar, its worker moves onto it on the agent's drift check)\n", rt.Host, cur, ver, channel)
		return
	}
	if rt.Hibernate {
		// Two different facts, one grep. With no restart callback this pass
		// genuinely only moved a symlink, and the next wake is when the new
		// version runs. With one (RestartHibernating, the agent's loops) the
		// callback has ALREADY been consulted above: a warm exec child was
		// bounced onto ver a few lines ago, and a sleeping one was a no-op.
		// Which of the two happened is makeRestartRoute's to log — it prints
		// the bounce — so this line stops claiming the sleeping case for
		// both, rather than growing a third state to track.
		if restart != nil {
			fmt.Printf("  %s: repointed %s -> %s (%s; hibernating — bounced if awake, else runs it on next wake)\n", rt.Host, cur, ver, channel)
		} else {
			fmt.Printf("  %s: repointed %s -> %s (%s; hibernating — runs it on next wake)\n", rt.Host, cur, ver, channel)
		}
	} else {
		fmt.Printf("  %s: updated %s -> %s (%s)\n", rt.Host, cur, ver, channel)
	}
}

// capabilityPlan turns a route plus the manifest of the version being
// adopted into one converge plan, or refuses.
//
// Two rulings live here, and both are the difference between this feature
// working and this feature quietly doing the opposite of what it says:
//
//   - A manifest this pass could not obtain (m == nil: the binary is staged,
//     so ensureStaged succeeded, but neither the sibling nor the bucket
//     could answer for its manifest) REFUSES the version. caps.Plan's nil
//     Set means "leave the drop-ins alone", which is the right posture for a
//     pass that is merely surveying the box — and exactly the wrong one on a
//     version change, where it would point a new binary at the previous
//     version's unit shape and say nothing.
//
//   - A manifest that RESOLVED and declares NOTHING is not the same thing.
//     Manifest.Capabilities is a nil pointer for both a legacy manifest and
//     a fetch that never happened, so it is never passed through verbatim:
//     a resolved manifest always carries a non-nil Set, empty when it
//     declares nothing, which is what makes converge REMOVE stale drop-ins.
//     Set: m.Capabilities verbatim would mean a capability dropped from a
//     manifest is never removed — the exact opposite of spec §4d's removal
//     semantics.
//
// An unparseable grants column refuses too, rather than defaulting to no
// grants: "no grants" is a real instruction that revokes :25 and drops a
// mail server's listener, and a column this binary cannot read is not
// evidence that an operator asked for that.
//
// canRestart arms caps.Plan.DeferRestart: this converge runs BEFORE the
// symlink moves, so a daemon it started would be the OLD binary under the NEW
// unit shape, and the caller would then cycle it a second time. Deferring is
// only honest when somebody is actually going to do the restart — with no
// restart callback at all, converge keeps bringing the unit up itself.
func capabilityPlan(rt registry.Route, m *Manifest, canRestart bool) (caps.Plan, error) {
	if m == nil {
		return caps.Plan{}, fmt.Errorf("the manifest for this version could not be read or fetched, so its unit shape is unknown")
	}
	g, err := caps.ParseGrants(rt.Caps)
	if err != nil {
		return caps.Plan{}, fmt.Errorf("capability grants %q: %v", rt.Caps, err)
	}
	set := m.Capabilities
	if set == nil {
		set = &caps.Set{V: caps.Version}
	}
	return caps.Plan{
		Host:         rt.Host,
		App:          rt.App,
		Set:          set,
		Grants:       g,
		Hibernate:    rt.Hibernate,
		DeferRestart: canRestart,
	}, nil
}

// stageCheck is what ensureStagedChecked needs to judge a staged copy it is
// about to reuse (spec §4.2).
type stageCheck struct {
	beforeWrite func(string) error
	// refs reports the staged copies something on this box still refers to.
	// nil, or an error, means unknown, and unknown never permits a restage.
	refs func() (StagedPaths, error)
	// verified holds staged paths whose bytes already matched their canonical
	// release this pass, so N hosts moving onto one build hash it once, each
	// with the artifact sha it matched: a digest-bound reuse trusts the memory
	// only for the sha its own checked manifest names (stageFrom). nil
	// remembers nothing.
	verified map[string]string
}

// ensureStaged makes ver's build of app available in the version store,
// downloading and sha256-verifying it if needed. Idempotent across hosts in
// one pass. The manifest supplies the sha; the bytes are refused unless they
// hash to it — the channel pointer's signature plus this gate is the whole
// trust chain from operator to disk. host is used only to prefix this
// function's own log lines, matching every other line this file prints.
//
// It also keeps the staged manifest sibling (versions.go's StageManifest)
// current: a fresh download writes the sibling alongside the binary, and a
// binary that is ALREADY staged but missing its sibling gets one written
// without a re-download — the back-fill for every version staged before this
// sibling existed. For copies staged by this binary or later the sibling is
// the canonical manifest's exact bytes; only a copy an older binary staged
// carries a lossy re-marshal (see ReadStagedManifest). The bucket object stays
// canonical either way.
//
// The back-fill's own fetch failure is NOT fatal when the binary is already
// staged (2026-08-03 review, I2): this function's whole promise is "the
// binary is on disk", and a version already staged must not depend on the
// bucket answering for a sibling write that can be retried. Without this, a
// transient GetManifest failure would hard-fail adoption of a binary that
// needs no adoption at all — and on a canary route's 2-3-second fast loop,
// flood the journal every tick. adoptHost's backfillManifest (C1, same
// review) is what makes leaving this non-fatal safe: every subsequent pass
// gets another chance at the live version's sibling, so a transient failure
// here heals on its own instead of being permanently missed.
//
// It RETURNS the manifest, because adoptHost's capability converge builds its
// plan from it (spec §4d) and must never re-read the sibling to get one: the
// sibling write is best-effort, and a failed write reads back as (nil, nil) —
// "leave the drop-ins alone" — which on a version change would point a new
// binary at the previous version's unit shape without a word. The one path
// that has no manifest of its own to give (a staged binary whose manifest
// neither disk nor bucket would answer for) returns (nil, nil) DELIBERATELY,
// and adoptHost turns that into a refusal when — and only when — capabilities
// are armed for the route. Every other caller keeps today's contract exactly:
// an error means "the binary is not on disk", nil means it is.
//
// A version directory that is already staged is no longer trusted by name
// (zero-downtime deploys spec §4.2). A version name can be shipped again with
// different bytes once retention has pruned it, and the staged path is keyed
// by name alone. So a staged binary is hashed and compared with the canonical
// manifest's digest for this platform:
//   - a match reuses it, as before;
//   - a mismatch restages over it only when nothing on the box refers to it
//     (chk.refs: live links, Version, PrevVersion, the generations file,
//     and whatever WithPinSource adds); otherwise the version is refused with
//     ErrVersionReused, so a copy under another instance's build never
//     changes underneath it;
//   - a canonical manifest that cannot be read keeps today's answer: the copy
//     is used as staged, because a box cannot compare what it cannot read,
//     and the agent runs any attempt onto it in hold (CanonicalDemand).
//
// ensureStaged is ensureStagedChecked with no reference check. It has no
// registry in hand, so it cannot know what refers to a staged copy and never
// restages one: a copy that differs from its release is refused. It stays
// refused until a pass that adopts that version restages it, or the release
// is shipped under a new name.
func ensureStaged(ctx context.Context, s Store, optDir, acct, app, ver, host string) (*Manifest, error) {
	return ensureStagedChecked(ctx, s, optDir, acct, app, ver, host, stageCheck{})
}

// ensureStagedChecked is ensureStaged with the reuse check's references
// supplied by chk; see ensureStaged's doc for the whole contract.
func ensureStagedChecked(ctx context.Context, s Store, optDir, acct, app, ver, host string, chk stageCheck) (*Manifest, error) {
	path := VersionPath(optDir, acct, ver, app)
	_, statErr := os.Stat(path)
	staged := statErr == nil
	m, raw, err := GetManifestRaw(ctx, s, acct, app, ver)
	if err != nil {
		if !staged {
			return nil, fmt.Errorf("manifest for %s/%s: %w", app, ver, err)
		}
		// The binary IS staged, which is this function's whole promise, and
		// the canonical manifest cannot be read to compare it with. A sibling
		// already on disk is the manifest in hand, exactly as before this
		// check existed. Without one, a fetch failure is only worth a line;
		// backfillManifest retries on every pass.
		sib, serr := ReadStagedManifest(optDir, acct, ver, app)
		if serr == nil && sib != nil {
			return sib, nil
		}
		if serr != nil {
			fmt.Printf("  %s: staged manifest for %s: %v\n", host, ver, serr)
		}
		fmt.Printf("  %s: back-fill manifest for %s: %v\n", host, ver, err)
		return nil, nil
	}
	return stageFrom(ctx, s, optDir, acct, app, ver, host, chk, staged, m, raw, false)
}

// stageFrom is ensureStagedChecked from the point it holds the canonical
// manifest: m and raw, its exact bytes, read once by the caller. staged says
// whether a copy was on disk under the name. bound says m passed the
// authorizing pointer's digest (ensureStagedDigest), and changes two things
// only: a copy this pass already verified is reused only if it was verified
// against THIS artifact sha, never another manifest's under the same name;
// and the sibling is made exactly raw (syncSibling) rather than repaired only
// when it does not read. Unbound, it is ensureStagedChecked as zdd-4 wrote it.
func stageFrom(ctx context.Context, s Store, optDir, acct, app, ver, host string, chk stageCheck, staged bool, m *Manifest, raw []byte, bound bool) (*Manifest, error) {
	path := VersionPath(optDir, acct, ver, app)
	sibling := repairSibling
	if bound {
		sibling = syncSibling
	}
	art, hasBuild := m.Artifacts[adoptPlatform]
	if staged {
		if v := chk.verified[path]; v != "" && (!bound || hasBuild && strings.EqualFold(v, art.SHA256)) {
			sibling(optDir, acct, app, ver, host, raw, chk)
			return m, nil
		}
		sum, herr := FileSHA256(path)
		if herr != nil {
			return nil, fmt.Errorf("staged %s/%s: %w", app, ver, herr)
		}
		if hasBuild && strings.EqualFold(sum, art.SHA256) {
			if chk.verified != nil {
				chk.verified[path] = art.SHA256
			}
			sibling(optDir, acct, app, ver, host, raw, chk)
			return m, nil
		}
		// Different bytes under the same name. Overwrite only a copy nothing
		// refers to: overwriting a referenced one would change the bytes under
		// another instance's build (spec §4.2).
		if chk.refs == nil {
			return nil, fmt.Errorf("%s/%s: the staged copy differs from the release and what refers to it is unknown, so it was not restaged", app, ver)
		}
		pins, perr := chk.refs()
		// References another app could not complete (an unread build state
		// of one of its held instances) do not stop this one; this app's
		// do (Astra round 1 finding 2).
		if inc := (*IncompleteRefsError)(nil); errors.As(perr, &inc) && pins != nil && !inc.Covers(acct, app) {
			perr = nil
		}
		if perr != nil {
			return nil, fmt.Errorf("%s/%s: the staged copy differs from the release and what refers to it is unknown (%v), so it was not restaged", app, ver, perr)
		}
		// By the copy's real path as well as its name, so a symlinked opt dir
		// cannot make a pinned copy look unpinned (Astra r3 N4).
		referred, rerr := pins.Refers(path)
		if rerr != nil {
			return nil, fmt.Errorf("%s/%s: the staged copy differs from the release and what refers to it is unknown (%v), so it was not restaged", app, ver, rerr)
		}
		if referred {
			return nil, fmt.Errorf("%s/%s: %w", app, ver, ErrVersionReused)
		}
		fmt.Printf("  %s: staged %s differs from the release in the store and nothing on this box refers to it; restaging\n", host, ver)
	}
	if !hasBuild {
		// A box facing a release with no build for its platform: skip, never
		// install the wrong arch. The next release ships it and the box
		// catches up.
		return nil, fmt.Errorf("release %s has no %s build — skipping", ver, adoptPlatform)
	}
	// Streamed to disk and hashed on the way (#269): a build read whole
	// outgrew the stage pass's memory cap.
	if chk.beforeWrite != nil {
		if err := chk.beforeWrite(path); err != nil {
			return nil, err
		}
	}
	err := stageArtifactFrom(optDir, acct, ver, app, art.SHA256, func() (io.ReadCloser, error) {
		rc, _, err := OpenQualified(ctx, s, ArtifactKey(acct, app, ver, art.File))
		return rc, err
	})
	if err != nil {
		return nil, err
	}
	if chk.verified != nil {
		chk.verified[path] = art.SHA256
	}
	// The sibling is the CANONICAL bytes (spec §4.2), and a restage replaces
	// whatever sibling the stale copy had. Best-effort: a failure here is not
	// fatal to staging the binary, and backfillManifest gets another chance
	// at it on every later pass.
	if err := StageManifest(optDir, acct, ver, app, raw); err != nil {
		fmt.Printf("  %s: stage manifest sibling for %s: %v\n", host, ver, err)
	}
	return m, nil
}

// ensureStagedDigest is adoptHost's staging entry: ensureStagedChecked bound
// to the digest the authorizing pointer entry signs for ver (#253, spec
// §4.6). want "" is ensureStagedChecked exactly: a pointer from before
// digests, or one written with the marker off.
//
// With a digest, the canonical manifest must hash to want BEFORE anything is
// downloaded or restaged, and the manifest that passed is the one staging
// uses (stageFrom): it is never read a second time, so a swap between the
// check and the download cannot slip through. Everything after the check is
// zdd-4's reuse rule (spec §4.2), judged against that checked manifest: a
// staged copy whose bytes match it is reused, one that differs is restaged
// only when nothing on the box refers to it and refused otherwise. Unlike the
// no-digest path, a manifest that cannot be read or does not hash to want is
// a refusal even when a copy is staged: that copy cannot be shown to be the
// build the pointer signed.
func ensureStagedDigest(ctx context.Context, s Store, optDir, acct, app, ver, host, want string, chk stageCheck) (*Manifest, error) {
	if want == "" {
		return ensureStagedChecked(ctx, s, optDir, acct, app, ver, host, chk)
	}
	m, raw, err := CheckManifestDigest(ctx, s, acct, app, ver, want)
	if err != nil {
		return nil, fmt.Errorf("manifest for %s/%s: %w", app, ver, err)
	}
	_, statErr := os.Stat(VersionPath(optDir, acct, ver, app))
	return stageFrom(ctx, s, optDir, acct, app, ver, host, chk, statErr == nil, m, raw, true)
}

// syncSibling makes the sibling beside a digest-bound copy exactly raw, the
// manifest bytes that passed the pointer's digest (spec §4.2, §4.6): left
// alone when it already is, written otherwise, including over a sibling that
// reads but is some other manifest's bytes (an older binary's re-marshal, or
// a release since replaced under the name). Best-effort, like repairSibling.
func syncSibling(optDir, acct, app, ver, host string, raw []byte, checks ...stageCheck) {
	if cur, err := safefile.ReadFile(StagedManifestPath(optDir, acct, ver, app)); err == nil && bytes.Equal(cur, raw) {
		return
	}
	if err := stageGenerationSibling(optDir, acct, app, ver, raw, checks); err != nil {
		fmt.Printf("  %s: stage manifest sibling for %s: %v\n", host, ver, err)
	}
}

// repairSibling writes the canonical bytes beside a reused binary whose
// sibling is absent or does not read: the back-fill for versions staged
// before the sibling existed, and the repair for one a different binary
// wrote. A sibling that reads is left alone, as before. Best-effort.
func repairSibling(optDir, acct, app, ver, host string, raw []byte, checks ...stageCheck) {
	if sib, err := ReadStagedManifest(optDir, acct, ver, app); err == nil && sib != nil {
		return
	}
	if err := stageGenerationSibling(optDir, acct, app, ver, raw, checks); err != nil {
		fmt.Printf("  %s: stage manifest sibling for %s: %v\n", host, ver, err)
	}
}

// backfillManifest is adoptHost's per-pass repair for a LIVE version's
// staged sibling (2026-08-03 review, C1). ensureStaged's own back-fill only
// fires from the call inside adoptHost that sits behind the cur == ver
// short-circuit — so a version already live, once adopted, never calls
// ensureStaged again on its own account, and a sibling write that failed
// once would stay missing forever for exactly the version actually running.
// This gives every pass another shot at it: os.Stat first (the steady-state
// cost — one stat when the sibling is already there), fetch only when
// absent, and never return an error — an absent or unreadable sibling is
// the safe leave-alone state (ReadStagedManifest's nil-nil contract; the
// converge pass treats it as "unknown", never as "no capabilities"), so a
// repeated failure here is only ever worth a log line, host-prefixed like
// every other line adoptHost prints.
func backfillManifest(ctx context.Context, s Store, optDir, acct, app, ver, host string, checks ...stageCheck) {
	if _, err := os.Stat(StagedManifestPath(optDir, acct, ver, app)); err == nil {
		return
	}
	_, raw, err := GetManifestRaw(ctx, s, acct, app, ver)
	if err != nil {
		fmt.Printf("  %s: back-fill manifest for %s: %v\n", host, ver, err)
		return
	}
	if err := stageGenerationSibling(optDir, acct, app, ver, raw, checks); err != nil {
		fmt.Printf("  %s: stage manifest sibling for %s: %v\n", host, ver, err)
	}
}

// backfillManifestDigest is backfillManifest bound to the digest the
// authorizing pointer signs for ver (#253, spec §4.6). want "" is
// backfillManifest exactly. With one, the bucket manifest must hash to it
// (CheckManifestDigest) and the checked canonical bytes are what is written;
// a mismatch is a log line and no sibling, the same leave-alone state as any
// other failure here.
func backfillManifestDigest(ctx context.Context, s Store, optDir, acct, app, ver, host, want string, checks ...stageCheck) {
	if want == "" {
		backfillManifest(ctx, s, optDir, acct, app, ver, host, checks...)
		return
	}
	if _, err := os.Stat(StagedManifestPath(optDir, acct, ver, app)); err == nil {
		return
	}
	_, raw, err := CheckManifestDigest(ctx, s, acct, app, ver, want)
	if err != nil {
		fmt.Printf("  %s: back-fill manifest for %s: %v\n", host, ver, err)
		return
	}
	if err := stageGenerationSibling(optDir, acct, app, ver, raw, checks); err != nil {
		fmt.Printf("  %s: stage manifest sibling for %s: %v\n", host, ver, err)
	}
}

// settingsTarget runs Target for one exec instance (settings spec §4.3) and
// does what the decision needs before staging: on root's pass, write the
// consumed file (a resolution BEFORE staging); refuse what installs nothing.
// It returns the version to install, the decision, and stop: nothing is
// installed this pass. Every stop takes the host off the proof waiting list,
// as each of adoptHost's pre-admit returns does (plan 3 ruling 5).
//
// A route with default settings and no consumed file is today's adoption:
// the head, with bake, on the pointer's evidence, and no book, no Target and
// no consumed write. Only an unreadable failure book on a managed route
// changes that (the zero-downtime plans' refuseManaged posture). A consumed
// file another tenant of the hostname left (Consumed.ForeignTo) is read as
// none, and root's write replaces it (ruling R26).
func settingsTarget(ctx context.Context, optDir string, bake *FirstSeen, rt registry.Route, ptr *ChannelPointer, cur string, now time.Time,
	cfg passConfig, holds map[string]time.Duration, ignore bool) (string, TargetDecision, bool) {
	if cfg.rootUpdates != nil {
		return cfg.rootUpdates.target(ctx, rt, ptr, bake, now, cfg, holds, ignore)
	}
	stop := func(dec TargetDecision) (string, TargetDecision, bool) {
		cfg.proof.clear(rt.Host)
		return "", dec, true
	}
	acct, channel := RouteAccount(rt), routeChannel(rt)
	u, uerr := ParseUpdateSettingsLenient(rt.UpdateSettings)
	plain := uerr == nil && u.Default()
	managed := cfg.managed != nil && cfg.managed(rt)
	consumed, cerr := ReadConsumed(optDir, rt.Host)
	// A file root wrote for another tenant of this hostname is no record of
	// this route's (ruling R26): every rule below decides as if there were
	// none, and NextConsumed starts the route's record empty. A default route
	// still takes the Target path over one, so root's write replaces the old
	// tenant's file with this route's own (the decision is today's adoption
	// either way). An untagged file is read as the route's own, as before.
	foreign := consumed.ForeignTo(rt)
	if foreign {
		consumed = nil
	}
	if plain && consumed == nil && !foreign && (cerr == nil || !controlname.ValidHostShape(rt.Host)) {
		if ptr == nil {
			return "", TargetDecision{}, true // today's quiet skip
		}
		if managed && cfg.failedErr != nil {
			return stop(TargetDecision{})
		}
		return ptr.Version, TargetDecision{Version: ptr.Version, Reason: ReasonAdopt, Evidence: EvidencePointer, Bake: true, Detail: ptr.Version}, false
	}
	if cerr != nil {
		fmt.Printf("  %s: consumed file unreadable (%v): nothing installed\n", rt.Host, cerr)
		return stop(TargetDecision{})
	}
	// A column that does not parse even leniently is held where it is, never
	// read as auto (ruling R-SETTINGS-UNREADABLE): it may carry a hold this
	// binary cannot see. Root records reason unknown, so the wake check holds
	// too; nothing is installed and the live link stays.
	if uerr != nil {
		dec := TargetDecision{Reason: ReasonUnknown, Detail: uerr.Error()}
		if cfg.resolve {
			fmt.Printf("  %s: update settings unreadable (%v): held, nothing installed\n", rt.Host, uerr)
			in := TargetInput{Route: rt, Pointer: ptr, Consumed: consumed, Installed: cur}
			if err := publishConsumed(ctx, optDir, cfg.route, NextConsumed(in, dec)); err != nil {
				fmt.Printf("  %s: recording the update decision: %v\n", rt.Host, err)
			}
		}
		return stop(dec)
	}
	if cfg.failedErr != nil && (!plain || managed) {
		return stop(TargetDecision{}) // PassFilter said why, once
	}
	bs, berr := ParseBuildStateCopy(rt.BuildState)
	if berr != nil {
		if cfg.resolve {
			fmt.Printf("  %s: %v: read as unknown\n", rt.Host, berr)
		}
		bs = BuildStateCopy{}
	}
	in := TargetInput{Route: rt, Pointer: ptr, Settings: u, Build: bs, Consumed: consumed, Installed: cur}
	if ptr != nil {
		in.HeadHeld = Hold(bake, acct, rt.App, channel, ptr.Version, holds, ignore || ptr.FastTrack, now)
		in.HeadTag = cfg.tags[acct+"/"+rt.App][ptr.Version]
	}
	// The failure records outrank every rule for a route with settings
	// (§4.3). A route with default settings keeps today's adoption, which
	// never read the book.
	if !plain && cfg.failed != nil {
		in.Failed = func(v string) bool { return cfg.failed(acct, rt.App, v, rt.Host, v == rt.GoodVersion) }
	}
	dec := Target(in)
	if cfg.resolve {
		// Under the carry fence (ruling R-CARRY-FENCE): a route a move has
		// held since this pass listed it gets no record, and nothing is
		// installed for it on this pass.
		if err := publishConsumed(ctx, optDir, cfg.route, NextConsumed(in, dec)); err != nil {
			fmt.Printf("  %s: recording the update decision: %v; nothing installed\n", rt.Host, err)
			return stop(dec)
		}
	} else if dec.Resolve != nil && dec.Reason == ReasonRequest {
		// Only root resolves (§4.3 rule 2), and only root's write makes T
		// the request's: the next root pass does both. A decision that
		// carries a Resolve for another reason (a pin, a hold, a failure:
		// R-TARGET-FAILEDHEAD) does not depend on it, so it goes ahead.
		// The stop asks root for a pass through the stage relay (ruling
		// R-AGENT-RELAY), the way a stage this process cannot write does, so
		// a cold wake waits on root's resolution rather than an hour.
		if cfg.stageBlocked != nil && !cfg.hostWrites.Enabled {
			cfg.stageBlocked(rt.Host, dec.Version, ErrInstallRequestNeedsRoot)
		}
		return stop(dec)
	}
	if dec.Version == "" {
		if cfg.resolve && dec.Reason != "" {
			fmt.Printf("  %s: %s%s: nothing installed\n", rt.Host, dec.Reason, orDetail(dec.Detail))
		}
		return stop(dec)
	}
	return dec.Version, dec, false
}

// orDetail renders a decision's detail after its reason, or nothing.
func orDetail(d string) string {
	if d == "" {
		return ""
	}
	return " (" + d + ")"
}

// stageAuthorized stages a build settings spec §4.1 authorizes by its
// promotion record: the record's digest binds the manifest before anything
// is downloaded or replaced (ensureStagedDigest). This box's own staged copy
// is evidence on its own (§4.1's retained identity; Astra round 1 finding
// 7): when its bytes still hash to the artifact digest in the copy's
// sibling, the manifest it was verified against when it was staged, it is
// recovered whatever the remote side says: no record (promoted before
// records existed), a record or manifest that cannot be read now, or a
// staging the record path refused. Without such a copy the build is
// refused, and nothing is downloaded on anything weaker than the record's
// digest. A failure to read the record is never read as "no record".
// Nothing is backfilled.
func stageAuthorized(ctx context.Context, s Store, pub ed25519.PublicKey, optDir, acct, app, ver, host string, chk stageCheck) (*Manifest, error) {
	rec, err := ReadPromotionRecord(ctx, s, pub, acct, app, ver)
	var refused error
	switch {
	case err == nil && rec.ManifestDigest == "":
		refused = fmt.Errorf("%s: promotion record carries no manifest digest", ver)
	case err == nil:
		m, serr := ensureStagedDigest(ctx, s, optDir, acct, app, ver, host, rec.ManifestDigest, chk)
		if serr == nil {
			return m, nil
		}
		refused = serr
	case errors.Is(err, ErrNoPromotionRecord):
		refused = fmt.Errorf("%s: %s", ver, updatewords.Text("settings.evidence.no-record"))
	default:
		refused = fmt.Errorf("%s: promotion record: %w", ver, err)
	}
	if id, m := RetainedStaged(optDir, acct, app, ver); id != nil {
		fmt.Printf("  %s: %s recovered from this box's verified staged copy (%v)\n", host, ver, refused)
		return m, nil
	}
	return nil, refused
}