// The `carlos agent` verb — the single-box role: edge (TLS + host routing) +
// hibernation activator + hourly release adoption (plus a 2s fast-converge
// loop, fastConvergeLoop: serial poll + canary/unstaged sweep) + the fleet status
// page's 5-minute self-report composed in one process. The composition is
// new here, but each part is a port: the edge from Eleven's router.go, the
// activator from lchat activator.go, the adopt loop is release.Pass (or,
// for canaries, release.PassFilter) on the cadence of Eleven's updater
// timer, and the status loop is internal/status's
// CollectBox/WriteBox/BuildFleet/PublishFleet run on a timer with one
// publisher elected via internal/lease.
//
// Fleet status page config surface (every env var any part of this file
// reads for it, gathered in one place since no single function owns the
// whole contract):
//
//   - CARLOS_STATUS_BUCKET / CARLOS_STATUS_PREFIX / CARLOS_STATUS_REGION —
//     read by statusLoop. CARLOS_STATUS_BUCKET names the S3 bucket each
//     box writes its own status/boxes/<label>.json snapshot to and (when
//     it wins the lease) publishes status/fleet.json + the viewer assets
//     to. When CARLOS_STATUS_BUCKET is unset, it DEFAULTS to whatever
//     CARLOS_DEPLOYMENT_BUCKET is set to (resolveStatusBucket) — the edge
//     and console both serve the fleet status page out of the same
//     deployment bucket adoptLoop already publishes releases to, so the
//     common single-bucket deployment needs to set only one env var; an
//     explicit CARLOS_STATUS_BUCKET still always wins, for the deployments
//     that deliberately want a second, independent bucket. Both unset ->
//     the loop never starts, logging one line and returning — the same
//     "inert until configured" convention adoptLoop uses for
//     CARLOS_DEPLOYMENT_BUCKET/_DIR, and litestream's own env-gated
//     replication before that. CARLOS_STATUS_REGION feeds lease.Acquire
//     (the single-writer election for the publish half of each tick); it
//     may be empty (falls back to the AWS default chain). CARLOS_STATUS_PREFIX
//     does double duty: it namespaces that same lease key (path.Join(prefix,
//     "status-lease")) AND, since Task 5 of the customer-fleets plan,
//     rebases every key statusLoop's AND metricsLoop's store touches
//     (release.NewPrefixStore) — a fleet box set to
//     CARLOS_STATUS_PREFIX=fleets/<name>/ writes its own status snapshot at
//     fleets/<name>/status/boxes/<label>.json and its metrics/alerts under
//     fleets/<name>/metrics/boxes/... and fleets/<name>/alerts/boxes/...
//     instead of the shared paths every platform box also writes — so
//     neither its status heartbeat nor its metrics/alerts ever collide with
//     (or leak into) the platform's own fleet.json, metrics view, or alert
//     webhook. Empty (every platform box today) is a pure passthrough —
//     EXCEPT on a door-backed fleet box, where empty defaults to
//     "fleets/<fleet>/" because the door admits no other write scope
//     (resolveStatusPrefix). Log shipping is deliberately NOT rebased: see
//     logship.go. A
//     PUBLISHING box with a prefix set would relocate status/fleet.json
//     itself under that prefix too — moot today since nothing sets a prefix
//     on a publishing (non-fleet) box, and a fleet box never publishes at
//     all (CARLOS_FLEET's guard, below).
//   - CARLOS_FLEET — read by statusLoop (Task 5, customer-fleets plan). Set
//     (to the fleet's name — its value is otherwise unused here) marks this
//     agent as a fleet box: shouldSkipPublisherElection short-circuits
//     publish to false BEFORE lease.Acquire is ever called, so a fleet box
//     never contends for the cross-box publish lease and never runs
//     BuildFleet/PublishFleet — it still self-reports via statusTick's
//     unconditional WriteBox, just under its own CARLOS_STATUS_PREFIX.
//     HONEST LIMIT: the CARLOS_DEPLOYMENT_BUCKET fallback only ever
//     resolves to an S3 bucket. A deployment adopting from
//     CARLOS_DEPLOYMENT_DIR (a plain directory — tests, or a single-box
//     no-AWS deployment) has nothing for the fallback to resolve to, so
//     DIR-based deployments do not publish fleet status yet; give them
//     their own CARLOS_STATUS_BUCKET explicitly if they want the page.
//   - CARLOS_FLEET_BOX — this box's operator-chosen fleet_boxes.label (the
//     channel identity the edge dials, "<fleet>/<label>", NEVER the public
//     status page's hashed pseudonym). Optional: defaults to
//     CARLOS_TUNNEL_NAME with the "<fleet>/" prefix stripped, since a fleet
//     box's tunnel name already IS "<fleet>/<label>" (see
//     internal/host/tunnel_client.go). resolveFleetIdentity is the one
//     place both env vars are read and validated (registry.ValidTunnelName)
//     — an invalid result refuses to arm the wake-time/startup binding
//     publish (fleetbind.go) loudly, and degrades the heartbeat's label back
//     to the hashed pseudonym rather than disabling it outright.
//   - CARLOS_TUNNEL_URL / CARLOS_TUNNEL_TOKEN — the tunnel env
//     tunnel_client.go already dials with, read a SECOND time here (spec
//     2026-08-11-fleet-api-surface, F5) as the fleet store door's address
//     and credential: on a fleet box with NO bucket env set, openDeployStore
//     and the status/metrics store both become internal/fleetapi clients
//     speaking release.Store over the edge's /store/ routes as
//     "<fleet>/<label>", so the box needs no AWS credentials on the
//     platform's deployment bucket at all. A bucket still set WINS
//     everywhere (the migration fallback): every platform box, and any
//     grandfathered fleet box that still carries bucket credentials, keeps
//     direct S3 byte for byte. Missing tunnel env on a fleet box with no
//     bucket is a loud error, not a silent degrade — that box has no way to
//     reach a store at all.
//   - STATUS_BOX_REGION — an operator-set, coarse region word (e.g.
//     "EU · Ireland") carried into status.Probes.Region and published
//     verbatim on this box's snapshot; "" omits the field. Deliberately
//     free text chosen by the operator, never derived from cloud metadata —
//     the public commitment is a human-readable region, not a specific
//     datacenter.
//   - CARLOS_SYSTEM_CHANNEL — platform channel this box follows for
//     self-update, canary|stable (default stable).
//   - CARLOS_LAYOUT_DEFAULT — the instance layout this box converges to
//     when no layout intent is recorded, nested|flat (default nested;
//     isolated-exec design §8.6, "Nested by default"). Nested applies only
//     on a box whose isolation is ready; a recorded intent always wins.
//   - STATUS_HOST / STATUS_PUBLIC — NOT read anywhere in this file. They
//     gate whether ops/tofu fronts the bucket's published status/ prefix
//     with a public hostname; recorded here only so the fleet status
//     page's entire config surface is documented in one place rather than
//     split across this file and the tofu module.
//
// The status-lease above is one of TWO elections this process runs, and the
// only per-tick one. The other is the HOUSEKEEPER lease (housekeeper.go):
// the same internal/lease mechanism, held long-term, at the fixed
// deployment-bucket key control/housekeeper-lease/lock.json, electing the one
// box that performs this deployment's global duties — today the releases
// retention prune (retentionLoop), from Part 2 of the multi-region spec
// certificate issuance. It reads CARLOS_DEPLOYMENT_BUCKET (or
// CARLOS_DEPLOYMENT_DIR, which makes the box its own housekeeper) and the
// same CARLOS_FLEET guard, and adds no env var of its own.
//
// Domains config surface (read by domainsLoop / domainsTickConfig):
//
//   - PLATFORM_APPS_DOMAIN — the deployment's platform apps domain
//     (flagship: oncarlos.com). UNSET = the domains feature is inert: no
//     platform domains minted, no bucket manifests read, no health stamps —
//     zero behavior change, same litestream-convention gate as the loops
//     above.
//   - CARLOS_WILDCARD_DOMAINS — comma list of static wildcard-cert parent
//     domains for the edge (tito: go.tito.io); per-account parents under
//     PLATFORM_APPS_DOMAIN are derived automatically (edge wildcard.go)
//   - CARLOS_DNS_PROVIDER / CARLOS_DNSIMPLE_TOKEN — DNS-01 backend for
//     wildcard certs: unset/dnsimple needs the token (SSM → unit env),
//     route53 rides the box IAM role; nothing set = wildcards off,
//     per-host HTTP-01 only
//   - CARLOS_ACME_DELEGATION_ZONE — our own zone that delegated wildcard
//     parents' challenge TXTs publish into (spec §4); unset or an invalid
//     shape = delegation off, per-host and direct DNS-01 continue
//   - PLATFORM_EDGE_IPS — comma-separated edge IPs custom-domain DNS checks
//     compare against. Default: this box's public IPv4 from IMDS
//     (best-effort). It is no longer the WHOLE comparison set: each pass
//     unions it with the deployment's LIVE edges — every box's published
//     control/edges/<box>.json whose status snapshot is still fresh
//     (edgesfile.EdgeIPs) — because under latency steering a correctly-
//     pointed domain resolves to whichever region's edge is nearest, and
//     one box's boot-frozen list reads that as "elsewhere". The union only
//     ever WIDENS, so an unreadable bucket degrades to exactly this
//     variable. A pass that ends up knowing no addresses at all stands
//     down rather than guessing "elsewhere" about every host.
package host

import (
	"github.com/carlosframework/platform/internal/cliflag"

	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"math/rand/v2"
	"net"
	"net/http"
	"os"
	"os/exec"
	"os/signal"
	"path"
	"path/filepath"
	"runtime"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/agentsock"
	"github.com/carlosframework/platform/internal/analytics"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/apperr"
	"github.com/carlosframework/platform/internal/applog"
	"github.com/carlosframework/platform/internal/caps"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/controlsock"
	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/edgesfile"
	"github.com/carlosframework/platform/internal/fleetapi"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/instancefs"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rootparticipation"
	"github.com/carlosframework/platform/internal/rootstate"
	"github.com/carlosframework/platform/internal/schedule"
	"github.com/carlosframework/platform/internal/status"
	"github.com/carlosframework/platform/internal/version"
)

// dataDir resolves the fleet data directory (CARLOS_DATA_DIR, default /data).
func dataDir() string {
	if d := os.Getenv("CARLOS_DATA_DIR"); d != "" {
		return d
	}
	return "/data"
}

// RunAgent runs edge + activator + adoption in one process. Flags are exactly
// edge.Run's (-registry, -dev, -http, -https, ...): the args pass through to
// edge.RunWithActivator, which owns the serving loop and blocks until
// SIGTERM/SIGINT.
//
// With CARLOS_HOST_CLAIM and CARLOS_REGION_HA on, a box that is not yet
// enrolled with its guard armed runs region HA's joining mode instead, and
// RunAgent returns once it can host, so systemd starts it again as the
// claim agent (hostingGate, joining.go).
//
// Before either, a box its region's group launched resolves its group
// reference, whatever the switches say: the loops runAgent starts read
// the box's label once and keep it (agentEntry, joining.go).
func RunAgent(args []string) {
	agentEntry(os.Getenv, resolveGroupBoxReference, groupReferenceBackoff, log.Printf,
		func() bool { return hostingGate(args) },
		func() { runAgent(args, productionClaimFactory) })
}

func runAgent(args []string, buildClaims ClaimFactory) {
	agentCtx, cancelAgent := context.WithCancel(context.Background())
	defer cancelAgent()
	// Startup can stop before children drain; rollout keeps agentCtx until Close.
	startupCtx, cancelStartup := context.WithCancel(agentCtx)
	defer cancelStartup()
	claimCtx, err := claimEntryContext(agentCtx, os.Getenv, openDeployStore)
	if err != nil {
		log.Fatalf("agent: host claims: %v", err)
	}
	dd := dataDir()
	regPath := peekRegistryPath(args, dd)
	// A second handle on the same registry file the edge opens — the registry
	// is designed for concurrent handles (fresh-connection reads), and this
	// one feeds the activator's due-time index and the adopt loop.
	reg, err := registry.Open(regPath)
	if err != nil {
		if hint := registryOpenHint(regPath, err); hint != "" {
			log.Fatalf("agent: open registry %s: %v\nagent: %s", regPath, err, hint)
		}
		log.Fatalf("agent: open registry %s: %v", regPath, err)
	}

	// Constructed once here so the activator's ExecBackend (purges) and the
	// edge (reads + serves cached hits) share the very same store — see the
	// package doc comment above about caching applying before the
	// hibernation branch.
	cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)

	// Exec-child accounting: claim the cgroup subtree the unit file
	// delegates to us (Delegate= in carlos-edge.service). Nil when there is
	// no delegation — an old unit file, a dev run outside systemd, darwin —
	// in which case children spawn exactly as before and simply have no
	// per-instance readings.
	var cgm *cgroup.Manager
	if hostwrite.Enabled(claimEntryEnv(claimCtx)) {
		cgm = cgroup.SetupRetaining(log.Printf)
	} else {
		cgm = cgroup.Setup(log.Printf)
	}

	// Alert sinks and the operator paging hook (moves spec §13), built from
	// the sink env alone and armed here, before armMovesPass and before any
	// loop that runs a moves pass starts, so a stuck mark never finds paging
	// unarmed. They do not wait on the metrics store: the metrics loop is
	// handed this same instance, and a box whose store fails still pages.
	alerts := armAgentAlerts(agentCtx, os.Getenv, &processAlertCounters, newSNSClient)

	// Log shipping (`carlos logs`): one sink for every producer in this
	// process, armed only when there is somewhere to ship to — a nil sink
	// leaves every producer exactly as before (no tees, no pipes, no
	// events). The box identity is the one metrics publishes under.
	//
	// The gate is openStatusStore, not a bucket test: a cut-over fleet box has
	// no bucket at all and ships through the store door, and gating on the
	// bucket alone would leave `carlos logs` silently empty for exactly the
	// boxes an operator can't SSH into. Constructing the store here to throw
	// it away is deliberate and cheap (neither branch makes a call), and it
	// asks precisely the question logshipLoop will ask again for itself.
	var logSink *applog.Sink
	var errSink *apperr.Sink
	if _, _, err := openStatusStore(context.Background(), os.Getenv); err == nil {
		logSink = applog.NewSink(status.Label(readMachineID()), time.Now)
		errSink = apperr.NewSink(status.Label(readMachineID()), time.Now)
	}

	cfg, err := activator.ConfigFromEnv()
	if err != nil {
		log.Fatalf("agent: %v", err)
	}
	claimCtx = withClaimRuntimeResources(claimCtx, ClaimRuntimeResources{Mode: "agent", Registry: reg, Manager: cgm, DataDir: dd, OptDir: release.OptDir(), UnitsDir: "/etc/systemd/system"})
	// The moves park providers (Plan 5 Task 9a) register before the claim
	// assembly reads them, then the moves wake source right after them
	// (Plan 5 Task 9b): only with both switches on. The parks refuse by
	// name until the wake source registers, which opens the one gate
	// (movesWakeSourceRegistered), so the parks arm in the same start as
	// the wake owners. Task 5's expected refusals are quiet in the regional
	// worker's log and its anomalies page once (movesrefusals.go).
	movesRefusalLog := newMovesRefusals(metricsBoxLabel(os.Getenv, readMachineID()), PageOperator, log.Printf, time.Now)
	if parks, err := registerProductionMovesParks(claimCtx, claimEntryEnv(claimCtx), claimStore(claimCtx), *cfg, productionRosterRegion(claimStore(claimCtx), os.Getenv), movesRefusalLog); err != nil {
		log.Printf("agent: moves: the park providers are not registered: %v", err)
	} else if err := registerProductionMovesWakeSource(claimEntryEnv(claimCtx), parks, dd); err != nil {
		log.Printf("agent: moves: the wake source is not registered: %v", err)
	}
	// The C8 files manifest provider (moves Task 6) registers under the
	// claim switch alone: every claim-on serving wake establishes its files
	// manifest through it (claimfiles.go).
	if err := registerProductionClaimFiles(claimEntryEnv(claimCtx), dd, reg); err != nil {
		log.Printf("agent: host claims: the files manifest provider is not registered: %v", err)
	}
	if hostwrite.Enabled(claimEntryEnv(claimCtx)) {
		// Region HA (stage B Task 10): with CARLOS_REGION_HA on, the
		// relocation owner source is registered before the claim's owners
		// open. In claim mode the assembly is opened here, as the claim
		// factory would open it, so the relocation pass, or with region HA
		// off the completion pass (ruling 82), binds to the same one.
		if regionHAEnabled(claimEntryEnv(claimCtx)) {
			if err := registerRelocationSource(claimCtx, dd); err != nil {
				log.Fatalf("agent: region HA: %v", err)
			}
		}
		if claimCtx, err = ensureProductionClaimAssembly(claimCtx); err != nil {
			log.Fatalf("agent: host claims: %v", err)
		}
	}
	claims, err := newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
	if err != nil {
		log.Fatalf("agent: host claims: %v", err)
	}
	if err := configureWakeClaims(claims, cfg); err != nil {
		log.Fatalf("agent: host claims: %v", err)
	}
	// The build-state store (settings spec §4.2), built once after the claim
	// wiring so it locates claimed hosts through cfg.Claims. nil, nil on a box
	// with no replica bucket. The activator's BeforeExec hook comes from bs
	// (buildStateHook, below): a failure here is one operator line, the
	// agent keeps running, and every spawn that records its build is refused
	// until it is fixed: every exec-backed wake, worker, sidecar, swap and
	// claimed start (Tasks 5a-5d). A unit-backed row is started by systemd
	// on its live link and records nothing (activator identityFor is nil
	// for it), so it still starts.
	bs, bsErr := newBuildStates(cfg, reg, release.OptDir())
	// The status report says this box enforces update settings only while
	// the build-state store is up (settings spec §4.2; Astra I3 #1).
	noteBuildStatesInit(bsErr)
	if bsErr != nil {
		log.Printf("agent: build state store unavailable: %v. every exec-backed instance, worker and sidecar spawn is refused until it is fixed and the agent restarted (unit-backed rows still start)", bsErr)
	}

	// Fleet box identity (customer-fleets plan, Task 6 fix round — controller
	// ruling): resolved ONCE, here, before the activator is even constructed.
	// identErr is errNotFleetBox for the ordinary platform-box case
	// (CARLOS_FLEET unset — no log, nothing to arm) or names the specific
	// validation failure for a fleet box whose identity doesn't resolve to a
	// usable channel label, in which case the fleet machinery — OnWake and
	// the startup pass, both — stays disarmed (loud log) rather than run
	// with a label the edge could never dial. fleetStore is opened here too
	// (once) so it can be threaded into svc.OnWake BEFORE svc.Start() below —
	// assigning it after Start() would race the sweeper goroutine's own use
	// of the field.
	var fleetStore release.Store
	identity, identErr := resolveFleetIdentity(os.Getenv)
	switch {
	case errors.Is(identErr, errNotFleetBox):
		// Ordinary platform box; nothing to arm.
	case identErr != nil:
		log.Printf("agent: %v — refusing to arm fleet machinery", identErr)
	default:
		fs, err := openDeployStore(context.Background())
		if err != nil {
			log.Printf("agent: CARLOS_FLEET=%s set but deploy store unavailable: %v — fleet bindings disabled", identity.Fleet, err)
		} else {
			fleetStore = fs
		}
	}

	// The config guard (amadan #213), armed before the activator and every
	// reconcile loop that asks it: a box takes an instance's lease, or mints
	// its record, only if it can provide that instance's config. Armed
	// whether or not the activator is, because the mint gate is the reconcile
	// pass's. A box with no deployment store reads it as "nothing sealed
	// anywhere" and admits as before.
	guard := newConfigGuard(os.Getenv, openDeployStore)
	armConfigGuard(guard)
	guard.Announce()

	var act edge.Activator
	var svc *activator.Service
	started := make(chan struct{})
	var startupDone <-chan struct{}
	if cfg != nil {
		// Exec children and carlos-app@ unit tenants share one state machine;
		// the split backend dispatches per route on Backing. Note the arming
		// gate is still the hibernate bucket env even though unit tenants
		// never touch the bucket — one opt-in switch for the whole activator.
		eb := &activator.ExecBackend{
			Cfg:     *cfg,
			OptDir:  release.OptDir(),
			DataDir: dd,
			// The box's placement code, into every exec child's env
			// (issue #329) — same env the reconcile pass scopes by.
			Region:      strings.TrimSpace(os.Getenv("CARLOS_REGION")),
			Cache:       cache,
			Materialize: materializeHook(reg, dd),
			Cgroups:     cgm,
			// The memory tier for each spawned child's cgroup, read off the
			// staged manifest of the route's live version (memory-pressure
			// spec §2.3): 512M, or 1G for a family that declares children.
			Tier: memTier(stagedManifestOfLiveVersion(release.OptDir())),
			// The same read for one named version, for a generation that runs
			// a build other than the live link's (zero-downtime spec §7.3).
			TierOf: memTierOf(func(rt registry.Route, v string) (*release.Manifest, error) {
				return release.ReadStagedManifest(release.OptDir(), release.RouteAccount(rt), v, rt.App)
			}),
			// The scheduled-work verbs on every spawned instance's control
			// socket write to the SAME registry this agent's converger and
			// runner read (schedule spec §7.2): an app that asks for a
			// once-timer from inside itself is asking this box's timer table,
			// so the row is due-indexed by the next tick with nothing in
			// between. reg satisfies schedule.TimerStore directly.
			Timers: reg,
			// The control socket's read-only GET /updates answers from the
			// same registry copy (instance update settings spec §5.1).
			UpdatesReader: updatesReader(reg, release.OptDir()),
			// Asked inside Acquire, before the lease (amadan #213).
			ConfigGate: configGate(),
		}
		if claims.Writes.Enabled {
			eb.Timers = schedule.ClaimTimers{TimerStore: reg, Writes: claims.Writes, Context: agentCtx}
		}
		if logSink != nil { // a typed-nil in the interface would NOT be nil
			eb.Logs = logSink
		}
		// The verified home keys of this box's replica rows (L21 rulings 26
		// and 30), read again at every replica launch until O8 reads them
		// from each home's published env item, and the bucket's
		// configuration, read fresh before a mint (ruling 5) and at each
		// replica launch whose key is configured (ruling 27). The claim
		// path's backend gets the same pair (claimExecBackend).
		productionInstanceKeyReaders().wire(eb)
		// The root instance-control channel (design §5.3) and the box facts
		// a nested wake needs beyond the row (design §8.1, Task 6): the
		// journal check and the peer list both read this SAME registry
		// handle, so a nested row's classification here always agrees with
		// what the rest of this process already knows about the box.
		eb.Control = instancectl.NewClient(controlsock.Path)
		eb.LayoutFacts = activator.LayoutInputs{
			DataDir: dd,
			// A journal that cannot be read is not proof there isn't one: an
			// EACCES or EIO on /data/layout must fence the row rather than
			// wake it as if nothing were in flight, so only a definite
			// ErrNotExist counts as "no journal". It is the pass's own
			// predicate, an Lstat, so a symlink planted at the journal path
			// fences the host rather than being followed.
			Journal: func(host string) bool { return layout.JournalPresent(dd, host) },
			Peers:   reg.List,
		}
		// The layout pass is handed this same gate below, so it never makes
		// a nested tenant this backend would refuse to start. The gate's
		// first question is about this process itself: the probe of design
		// §8.5 runs under the instance templates' unit text and cannot see
		// the edge's, and RestrictSUIDSGID= on carlos-edge.service once made
		// every openat2 the agent issues answer ENOSYS (#126). A box in
		// that state wakes nothing nested and migrates nothing, and says
		// why, instead of failing on the first host's first open.
		selfCheck := func() error { return instancefs.SelfCheck(dd) }
		if err := selfCheck(); err != nil {
			log.Printf("agent: %v (nested wakes and the layout pass refuse until the unit is fixed)", err)
		}
		eb.NestedOK = nestedGate(selfCheck, instancefs.MountNoSymfollow, eb.Control.Ready)
		// CARLOS_SWAP=overlap turns on generations (plan 1), and with them
		// plan 3's Swap, before the Service exists, which reads the switch
		// once in New. Plan 2's per-generation containment goes on with
		// them: an overlap needs each generation in its own child. A box
		// that cannot contain generations refuses the flag here and stays
		// on today's paths (swapArming), so all three stay off.
		swapOn := swapArming(os.Getenv, eb)
		activator.SetGenerations(swapOn)
		eb.GenContainment = swapOn
		rolloutMaterialize = eb.Materialize
		svc = activator.New(*cfg, &activator.SplitBackend{
			Exec: eb,
			Unit: &activator.UnitBackend{OptDir: release.OptDir()},
		}, reg) // the registry satisfies activator.DueStore directly
		// doWake's own re-read of the committed row before it spawns
		// (design §6.8). The registry satisfies RouteLookup with its own
		// Get, exactly as DueStore above.
		svc.Routes = reg
		// Plans 2: an account's plan can shorten its entries' idle window. With
		// plans off (no marker, CARLOS_PLAN_LIMITS=off, or no record) the hook
		// answers 0 and every window is the box's own.
		svc.SetIdleFor(planIdleFor)
		// Settings spec §4.2: before any process runs a build against an
		// instance's database, the build state names it, under the lease. A
		// store that failed to start refuses every spawn (never "run
		// unrecorded"). Set before Start, like Routes.
		svc.BeforeExec = buildStateHook(bs, bsErr)
		// The held wake check (settings spec §4.3 rule 3): a held instance
		// never wakes on an installed build other than its target. The copy
		// is refreshed before every wake that could be held (wakeTarget), and
		// a target not installed is asked of root through the stage relay,
		// one throttled requester for every wake. As every stage relay, not
		// in host-writes mode (ruling R-AGENT-RELAY; release's OnStageBlocked
		// callers): there a held wake waits for root's own pass. A box with
		// no build-state store keeps today's wake.
		if bs != nil {
			svc.HeldTarget = func(h string) activator.HeldWake { return bs.wakeTarget(context.Background(), h) }
			if !cfg.HostWrites.Enabled {
				askStage := stageRequester(dataDir())
				// A wake waiting for root's decision on new settings names
				// no build (ruling R-HELD-PURE): one marker for a root pass,
				// nothing the relay would keep asking for.
				askPass := stageMarkerRequester(dataDir())
				svc.RequestStage = func(host, build string) {
					if build == "" {
						askPass(host, build, nil)
						return
					}
					askStage(host, build, nil)
				}
			}
		}
		// Every worker's captured serving build, published for root's pin
		// set (zero-downtime deploys spec §4.2, ruling R17), whatever
		// CARLOS_SWAP says. dd is dataDir(), the same data dir root's
		// RunAdopt and this agent's passes hand release.BakeDir, which is
		// where the pass reads release.GenerationsPath from. Set before
		// Start and before any wake.
		svc.GenerationsFile = release.GenerationsPath(dd)
		// A retiring generation's schedule writes are refused once a swap
		// has switched away from it, through each host's write fence, which
		// the switch takes (zero-downtime spec §7.10). Wired before Start,
		// like Routes. With swaps off nothing is ever retired or serialized,
		// so every write applies as before.
		eb.ScheduleFence = svc.ScheduleWrite
		if fleetStore != nil {
			// MUST be assigned before Start() below spawns the sweeper
			// goroutine, which reads svc.OnWake on every successful wake from
			// then on — setting it any later is a data race.
			svc.OnWake = fleetBindingHook(fleetStore, identity.Fleet, identity.Label)
		}
		if logSink != nil {
			svc.Logs = logSink
		}

		// The startup sweep (design §6.7): a leftover instance unit from a
		// previous agent process, one this box's registry no longer holds
		// a live wake for, is stopped before this process starts spawning
		// its own. Best-effort, like every other startup duty here: a
		// channel that never answers logs and moves on, and the box is no
		// worse off than before this existed (a leftover unit that keeps
		// running is exactly what an agent restart left before this pass).
		//
		// Then, on a legacy box, layout recovery before Start can wake
		// anything, and the instance-key backfill in the background after
		// it, which this does not wait for (the edge below must listen
		// within the update gate, whatever the box's size); on a claim
		// box, Start alone. The order and the branch are activatorStart's.
		svc.Startup = started
		startupDone = activatorStart{
			Configure: func(_ context.Context) {
				armAlways(svc, reg, dd) // flag on or off
				var rolloutSink activator.EventSink
				if logSink != nil { // a typed nil in the interface would not be nil
					rolloutSink = logSink
				}
				armRolloutAndRelay(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore)
				// The residence-move machinery (issue #331), armed with the
				// activator: the staging-fallback restore needs a same-region
				// mirror client, and the move worker needs the service's quiesce
				// answer. Both are best-effort — a mirror client that cannot
				// build logs and leaves the fallback dark, exactly like every
				// other optional duty here.
				if m, err := lease.NewS3Mirror(agentCtx, cfg.Region, cfg.Region); err != nil {
					log.Printf("agent: staging-fallback mirror client: %v (restore fallback disarmed)", err)
				} else {
					eb.Mirror = m
				}
				// Seeded instances (seeded-instances spec §3.4): the seed reader,
				// the placed recorder, the provision pass's hooks and the claim
				// flag the seed fallback refuses under. In Configure, like the
				// mirror, as it was before the startup moved to the background
				// (#281): before the sweep, before Start, and so before Ready
				// closes the startup latch. Every wake waits on that latch
				// (awaitStartup) and the reconcile loops start after it, so no
				// seeded row is woken, and no seeded record acted on, before
				// this has run. The import swap and rollback (spec §5.4, §5.6)
				// fence through this activator, delete the replica under
				// cfg.Prefix, and trust an unproven stop only after the startup
				// sweep below (seedSweptAtStart, set in Recover, which also
				// comes before the latch).
				wireSeedPlacement(eb, claims.Writes.Enabled, seedDataOpener(cfg.Bucket, cfg.Region),
					func() (release.Store, error) { return openDeployStore(context.Background()) },
					seedBox{Fence: svc, Prefix: cfg.Prefix, Swept: seedSweptAtStart.Load, NextFenceGen: fenceGenerations(reg)})
				armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, claims.Providers, rolloutSink, moveCarry(bs, bsErr))
				// carlos run's worker, armed with the moves pass as it was
				// before the startup moved to the background (#281). Its
				// passes ride the converge and instances loops, which the
				// duties start only once Ready has closed the startup
				// latch: no run is picked up before the startup sweep has
				// stopped a previous agent's leftover units, the cmd units
				// among them.
				armRunsPass(svc, reg, dd, rolloutSink)
				// The db-reads loop (db region forward spec §7.1): armed
				// only on a box that can serve; it takes nothing until a
				// console asks.
				armDBReads(*cfg)
			},
			Ready: func() {
				close(started)
				log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
			},
			Claims: claims.Writes.Enabled,
			Sweep: func(ctx context.Context) error {
				sw, err := eb.Control.Sweep(ctx)
				if err != nil {
					log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
				} else if len(sw.Units) > 0 {
					log.Printf("agent: swept %d instance unit(s) left from a previous agent: %v", len(sw.Units), sw.Units)
				}
				return err
			},
			Recover: func(ctx context.Context, swept bool) *layout.Pass {
				seedSweptAtStart.Store(swept)
				pass := newLayoutPass(reg, svc, dd, eb.NestedOK, swept)
				intent := startupRecoveryIntent(ctx, dd, openDeployStore, metricsBoxLabel(os.Getenv, readMachineID()))
				recoverLayoutAtStart(ctx, pass, intent)
				return pass
			},
			Routes:  reg.List,
			Route:   reg.Get,
			Keys:    eb,
			Slot:    svc.PrepareScheduledToken,
			Publish: armedLayoutPass.Store,
			Start:   svc.Start,
		}.Run(startupCtx)
		act = svc

		// The agent socket: the one channel a sidecar dials back over
		// (internal/agentsock, Task 7). Constructed only in this branch —
		// the same condition that constructs the activator — because with
		// hibernation off there are no workers on the box at all, and
		// nothing would ever call in. cfg.AgentSocket is never empty here:
		// activator.ConfigFromEnv defaults it to DefaultAgentSocket.
		//
		// The cancel is deferred right here, not called explicitly at the
		// shutdown site below: RunAgent's defers run LIFO at return, so this
		// one (registered first) fires AFTER the shutdown block's own
		// `defer cancel()` for svc.Shutdown's context — i.e. once the
		// children are already down — while staying the one place this
		// context's lifetime is decided, which is what satisfies go vet's
		// lostcancel check.
		agentSockCtx, agentSockCancel := context.WithCancel(context.Background())
		defer agentSockCancel()
		agentSrv := agentsock.New(svc)
		go serveAgentSocket(agentSockCtx, cfg.AgentSocket, agentSrv.Serve)
		log.Printf("agent: agent socket armed at %s", cfg.AgentSocket)

		// The boot pass (Task 8): a sidecar that does not hibernate must be
		// running whenever the agent is — an agent restart otherwise leaves
		// it down until a request or its due time happens to arrive, and an
		// always-on worker has no request path at all to wake it. Run off
		// the startup path (its own goroutine, not blocking RunAgent) since
		// WakeWorker can block for the settle window per host and this must
		// never delay the edge coming up; runWorkerBootPass itself wakes
		// its candidates concurrently, bounded, rather than serially.
		go func() {
			select {
			case <-started:
			case <-startupCtx.Done():
				return
			}
			routes, err := reg.List()
			if err != nil {
				log.Printf("agent: worker boot pass: list routes: %v", err)
				return
			}
			runWorkerBootPass(routes, svc.WakeWorker)
		}()
	}

	if svc == nil {
		close(started)
	}

	seedEnabled := claims.Writes.Enabled && os.Getenv("CARLOS_HOST_CLAIM_ORIGINS") == "1" && svc != nil
	safeClaimClose := true
	var shutdownCtx context.Context
	var shutdownCancel context.CancelFunc
	defer func() {
		if shutdownCancel != nil {
			defer shutdownCancel()
		}
		cancelAgent()
		if !safeClaimClose {
			return
		}
		base := shutdownCtx
		if !seedEnabled || base == nil {
			base = context.Background()
		}
		ctx, cancel := context.WithTimeout(base, 5*time.Second)
		defer cancel()
		if err := claims.close(ctx); err != nil {
			log.Printf("agent: regional claims shutdown: %v", err)
		}
	}()
	var stopHousekeeper func()
	if !claims.Writes.Enabled {
		// Certificate issuance must not wait for instance recovery.
		stopHousekeeper = startHousekeeper()
	}
	dutiesDone := make(chan struct{})
	runDuties := func() {
		defer close(dutiesDone)
		select {
		case <-started:
		case <-startupCtx.Done():
			return
		}
		if startupCtx.Err() != nil {
			return
		}

		// Fleet box binding publish, startup half (customer-fleets plan, Task 6):
		// a one-shot pass over every route already resident in this box's local
		// registry, covering an agent restart (activator.Service's in-memory
		// entries are gone, but an already-running child isn't) and routes that
		// never go through activator.wake at all. Non-authoritative
		// (publishResidentFleetBindings calls WriteFleetBinding with
		// takeover=false) — see fleetbind.go's doc comments for why a bare
		// "this box has this route resident" pass must never steal a binding a
		// wake-success write (lease-backed, authoritative) already holds for a
		// DIFFERENT box.
		if fleetStore != nil {
			if svc == nil {
				log.Printf("agent: CARLOS_FLEET=%s set but the hibernation activator is not armed — wake-time fleet bindings will never publish", identity.Fleet)
			}
			go publishResidentFleetBindings(fleetStore, identity.Fleet, identity.Label, reg)
		}

		// Scheduled work (schedule spec §6.2/§6.3), built ONCE here and shared by
		// all three legs below: the seed and the fast leg in fastConvergeLoop, the
		// hourly leg in adoptLoop, and the runner started here. One converger is a
		// correctness requirement, not tidiness — its etag memory is what makes a
		// declaration change cheap and a run signal exactly-once, and a second
		// instance elsewhere would replay or swallow one. The three legs run in
		// two different goroutines and do overlap (an hourly pass mid-flight when
		// a console bump fires the fast one); Converger.Pass serializes itself on
		// its own mutex, so no call site here needs one.
		//
		// The two halves arm on different conditions — armSchedule
		// (schedule_wiring.go) is that switch, and carries the why. In short:
		// declared schedules live in the bucket, so the converger needs the
		// deploy store; the RUNNER needs only the registry, because once-timers
		// are registered by instances on their control socket (mounted above,
		// Timers: reg, whenever the activator is armed) straight into this box's
		// rows, and a box that mounts the verb must also fire what it accepts.
		//
		// Guarded assignment, not a direct pass: logSink is a *applog.Sink, and
		// a nil one handed to an interface parameter is NOT a nil interface —
		// the same trap the activator's Logs assignment above avoids.
		var sink activator.EventSink
		if logSink != nil {
			sink = logSink
		}
		schedStore, schedErr := openDeployStore(agentCtx)
		sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd)
		// Move holds (moves spec §9.4): one gate for the runner, the converger
		// and a cut's drain, and the converger's freeze. Nothing with the
		// Phase 1 switch off.
		armScheduleHolds(sd, armedHoldView())
		if err := configureScheduleClaims(claims, sd); err != nil {
			log.Fatalf("agent: host claims: %v", err)
		}
		if sd != nil {
			sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
		}
		serialClaims := make(chan struct{}, 1)
		regionalReport := func(err error) { log.Printf("agent: regional claims: %v", err) }
		if hostwrite.Enabled(claimEntryEnv(claimCtx)) && moves.Enabled(os.Getenv) {
			regionalReport = movesRefusalLog.report
		}
		if err := claims.startRegional(agentCtx, serialClaims, regionalReport); err != nil {
			log.Fatalf("agent: regional claims: %v", err)
		}

		// The relocation pass (stage B Task 10): it says itself what it started.
		startRelocationPass(agentCtx, claimCtx, claims, reg, svc)
		// The change index (S3 change-index design §4): armed before the
		// loops that read through it, with its backstop running from here.
		// Unset CARLOS_CHANGE_INDEX, or a fleet box: nothing.
		if v := armChangeIndex(agentCtx, os.Getenv, openDeployStore); v != nil {
			go recordViewBackstopLoop(agentCtx, v)
		}
		go adoptLoop(reg, svc, sd, claims.Writes)
		go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)
		if logSink != nil {
			go logshipLoop(logSink, status.Label(readMachineID()))
		}
		if errSink != nil {
			go errshipLoop(errSink, status.Label(readMachineID()))
		}
		// The build-state report (settings spec rev 11 §4.4) names the rows
		// keepsBuildState selects, with host claims as the build-state store
		// sees them (cfg.HostWrites.Enabled, set by configureWakeClaims).
		buildStateClaims := cfg != nil && cfg.HostWrites.Enabled
		collectBuildStates := func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) }
		go statusLoop(reg, svc, collectBuildStates)
		// The memory early-warning (memory-pressure spec §2.4): per box, gated
		// by CARLOS_HOST_METRICS — one log line and nothing else where unset.
		go memwatchLoop()
		go domainsLoop(reg, peekAcmeCache(args), claims.Writes.Enabled)
		go instancesLoop(reg, claims.Writes, bs, claims.Providers.Provision)
		// The deployment-global election, armed BEFORE the loops it gates so a
		// single-box deployment is holding well before their first tick (a
		// minute away) — see housekeeper.go. Not a `go` of its own: it publishes
		// the process's housekeeper and spawns its loop itself. Its stand-down
		// runs in the post-drain block at the end of this function.
		if claims.Writes.Enabled {
			stopHousekeeper = startHousekeeper()
		}
		go retentionLoop(reg, collectBuildStates)
		go registrySnapshotLoop(reg)
		go edgesFileLoop(reg)
		// Region HA enrollment evidence (spec 1 §5.1): once, at start, and only
		// with CARLOS_REGION_HA on. See boxrecord.go.
		go boxRecordLoop()
		// The mesh key's own loop (multi-region backhaul spec, Part 3 Task 2) —
		// deliberately NOT a duty riding retentionLoop's tick: that loop returns
		// early, never arming at all, on a box missing CARLOS_RELEASE_PUBKEY,
		// and such a box still WINS the housekeeper election. A mint duty living
		// only inside retentionLoop would then silently never run fleet-wide.
		// See mintMeshKeyLoop's own comment (housekeeper.go).
		go mintMeshKeyLoop()
		// The DNS steering converger (multi-region ingress spec, Part 4) — its
		// own loop for mintMeshKeyLoop's exact reason, and dormant on every box
		// today: CARLOS_STEERING_ZONE_ID unset means one log line at arm and
		// nothing else, ever. See steering_loop.go.
		go steeringLoop()
		// The sweep reads the registry only through this closure — the stamped
		// bespoke units to sample beside the carlos- ones, re-read each tick.
		go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }, alerts)
		go configSyncLoop(reg, svc, dd, claims.Writes)
		go configGenLoop(reg, svc, dd, claims.Writes)
		go socketAdvisoryLoop(reg, svc)

	}
	if claims.Writes.Enabled {
		runDuties()
	} else {
		go runDuties()
	}

	// Seed handlers borrow this agent's resources and finish before they close.
	var seedCtx context.Context
	var seedCancel context.CancelFunc
	var seedServer *filesSeedServer
	if seedEnabled {
		seedCtx, seedCancel = signal.NotifyContext(claimCtx, syscall.SIGTERM, os.Interrupt)
		defer seedCancel()
		var seedFinish func()
		seedCtx, seedFinish = activator.WithCleanupBudget(seedCtx, 30*time.Second)
		defer seedFinish()
		stopSeed := context.AfterFunc(startupCtx, seedCancel)
		defer stopSeed()
		seedServer, err = startAgentFilesSeed(seedCtx, svc.Startup, svc)
		if err != nil {
			log.Printf("Files seed failed: %s.", err)
		}
	}

	// The same serving loop as `carlos edge`, with the activator injected.
	// Caching applies here too: cached hits still avoid the activator, which
	// is the entire point of checking the cache before the hibernation branch.
	// The ready hook fires synchronously before listening, so
	// tunnelLoop is handed off to its own goroutine rather than called here —
	// calling it directly would block the front door forever.
	edgeOpts := append([]edge.RunOption{
		// Certificate issuance is a deployment-global duty, so it runs where
		// the housekeeper runs (housekeeper.go). The gate goes DOWN into the
		// edge as a func — this package imports internal/edge and never the
		// reverse — and it is the GATE, not the bare flag: an ACME order runs
		// for seconds to minutes, and the context ends the instant this box
		// stops being the housekeeper. Armed above, so this closure is live
		// and correct by the time the edge reads it.
		edge.WithIssuerGate(housekeeperGate),
		// Whether this box contends in that election AT ALL — the cert
		// backfill's eligibility (#318). Mode is decided once from the
		// environment, so the closure is stable; see housekeeperContends
		// for the fleet-box exclusion it exists to state.
		edge.WithIssuerContender(housekeeperContends),
		// This box's own label, for the home map's own-box exclusion
		// (internal/edge/homemap.go). metricsBoxLabel is the SAME call
		// edgesFileLoop publishes under, deliberately: the map excludes our
		// hosts by matching the box field of the file we ourselves wrote, so
		// the two derivations must be one derivation. Without it the edge
		// would offer nothing (fail-closed), never dial itself.
		edge.WithBoxLabel(metricsBoxLabel(os.Getenv, readMachineID())),
		// The change index (#336): the cert mirror and the home map read
		// through the process's one watch. A loader, not the watch: on a box
		// without claim writes the index is armed on runDuties' goroutine,
		// which may not have run yet. Disarmed, it returns nil for good and
		// both keep today's full reads.
		edge.WithChangeWatch(armedChangeWatch.Load),
	},
		// The edge-local door (#175 P7), only when CARLOS_EDGE_DOOR names a
		// socket. It presents this box's public address, the one the edges
		// file publishes; with none, the edge keeps the door shut and says so.
		edgeDoorOptions(os.Getenv, readPublicIP)...)
	// A tunnel client takes the hub's GeoIP answer and mirrors nothing (geoip
	// spec §5.1); the same rule tunnelLoop arms on.
	if tunnelEnvComplete(os.Getenv) {
		edgeOpts = append(edgeOpts, edge.WithGeoTunnelClient())
	}
	edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
		go tunnelLoop(e)
		// The edge owns the health probers, so its state table is only
		// reachable once it exists — statusLoop (already running above) reads
		// it through this published source on its next tick. Method value, so
		// the loop always sees the CURRENT probe set: Reload replaces probers
		// behind this handle without the heartbeat needing to be re-wired.
		setUpstreamSource(e.UpstreamStates)
		// The GeoIP block for the heartbeat (geoip spec §5.5), the same way.
		geoIPSource.Store(e.GeoIPState)
		geoIPAlertSource.Store(e.GeoIPAlertCode)
		// The layout pass reloads routing through this as soon as a move
		// commits, instead of waiting for the 2s registry poll.
		reload := func() { _ = e.Reload() }
		layoutReload.Store(&reload)
		m := edge.NewUsageMeter()
		e.EnableUsage(m)
		// Same arming point as the meter: the edge observes an upstream 5xx
		// on the serving path and files it here (spec 2026-09-07 §3.2), and
		// errshipLoop above ships what this fills. Guarded, because a typed
		// nil inside the interface would NOT be nil to the edge.
		if errSink != nil {
			e.EnableErrors(errSink)
		}
		// The analytics meter (spec 2026-09-07 §3.3): the edge counts
		// pageviews per (account, app) while telemetryPass (fullPass and
		// instancesLoop) keeps its switch map current, and usageLoop
		// drains it into day objects on the same 5-minute tick as usage.
		am := analytics.NewMeter(dataDir(), status.Label(readMachineID()), time.Now)
		e.EnableAnalytics(am)
		setAnalyticsSink(e.SetAnalyticsOn)
		go usageLoop(m.Drain, am)
		// Edge request records (spec 2026-10-24-edge-request-protection-design):
		// sampled 404 and rule-match records plus per-minute counts, filed
		// into the log sink and shipped by logshipLoop. The same data dir as
		// the analytics meter, so the record salt is the analytics day salt.
		// The nil check is on the concrete recorder: a nil *Recorder passed
		// as the edge's interface would not be nil to it.
		if rec := newRequestRecorder(logSink, dataDir(), os.Getenv, time.Now); rec != nil {
			e.EnableRequestRecords(rec)
			go reqrecLoop(rec, time.NewTicker(time.Minute).C)
		}
	}, edgeOpts...)

	var seedErr error
	shutdownCtx, shutdownCancel, seedErr = joinAgentShutdown(seedCtx, seedCancel, seedServer, cancelStartup, startupDone, dutiesDone)
	if seedErr != nil {
		safeClaimClose = false
		log.Printf("Files seed failed: %s.", seedErr)
		os.Exit(1)
	}

	// The edge has drained; take the children down too (they're this process's
	// children — orphaning them would strand their leases until TTL expiry).
	if svc != nil {
		ctx := shutdownCtx
		svc.Shutdown(ctx)
		// Then the scheduler, under the same budget (spec §8.6, Task 11): an
		// attempt whose swap Shutdown just cancelled finishes its done(), its
		// outcome and the end of its intent on disk, and so does a wake event
		// under way, before this process exits. A planned restart mid-swap so
		// leaves nothing for the next agent to reconcile.
		if m := armedRollout(); m != nil {
			m.Close(ctx)
		}
		// The agent socket's own defer (above, at its construction) stops it
		// once this function returns — after Shutdown, per the LIFO ordering
		// explained there. That gap is real, not zero-width: svc.Shutdown
		// returning here and agentSockCancel actually firing are two
		// different points in the stack unwind, so the socket is technically
		// still accepting connections for the width of that unwind. It is
		// harmless: Shutdown has already stopped every entry and set
		// shuttingDown, so a call landing in the gap finds no live entry —
		// WorkerToken returns "" (authorize's unmatchable() then refuses it),
		// and NoteWorkerContact/NoteWorkerReady/SleepWorker are all no-ops on
		// a nil entry lookup. Nothing in the gap can reactivate a stopped
		// worker or authorize against a dead token.
	}
	// Same principle one level up: this process may hold the DEPLOYMENT's
	// housekeeper lease, and a restart that walked away from it would leave
	// control/housekeeper-lease/lock.json standing for up to its 15-minute
	// TTL — no pruner, and from Part 2 no cert issuer, across an ordinary
	// roll of a single-box deployment. Stopping cancels every duty in flight
	// and releases the lock, so the next box (or this one, a second later)
	// takes over immediately.
	if stopHousekeeper != nil {
		stopHousekeeper()
	}
}

// statusSocketProbe is the kernel-table probe the status loop hands
// wrapListeningProbe: listeningProbe itself with generations off (or no
// activator), and pointed at each managed route's current generation with
// them on (currentSocketProbe).
func statusSocketProbe(routes []registry.Route, svc *activator.Service) func(string) *bool {
	if svc == nil || !svc.GenerationsOn() {
		return listeningProbe
	}
	return currentSocketProbe(routes, svc.AppSocket, listeningProbe)
}

func configureWakeClaims(w *ClaimWiring, cfg *activator.Config) error {
	if w == nil {
		return hostwrite.ErrUnavailable
	}
	if !w.Writes.Enabled {
		return nil
	}
	if cfg == nil || w.Providers.Wake == nil || w.Writes.Selector == nil {
		return hostwrite.ErrUnavailable
	}
	cfg.HostWrites, cfg.Claims = w.Writes, w.Providers.Wake
	return nil
}

func configureScheduleClaims(w *ClaimWiring, sd *scheduleDeps) error {
	if w == nil {
		return hostwrite.ErrUnavailable
	}
	if !w.Writes.Enabled {
		return nil
	}
	if sd == nil || sd.runner == nil || sd.conv == nil || w.Writes.Selector == nil {
		return hostwrite.ErrUnavailable
	}
	sd.runner.HostWrites, sd.conv.HostWrites = w.Writes, w.Writes
	sd.runner.StartDelivery = w.Providers.Delivery
	return nil
}

// --- Task 8: the agent socket's retry/backoff, and the worker boot pass ----

// workersToStart is the boot pass's selection: a sidecar that does not
// hibernate must be running whenever the agent is. A hibernating sidecar is
// deliberately NOT started here — it waits for its due time, an operator
// command, or (later) a delivery signal, which is the whole point of the
// class. Status is checked the same way the due-time sweep checks it
// (activator.go's sweepOnce): only "ready" (or the legacy "" some pre-status
// rows carry) is eligible — a row still provisioning, or draining, must not
// be spawned.
func workersToStart(routes []registry.Route) []registry.Route {
	var out []registry.Route
	for _, rt := range routes {
		if rt.IsSidecar() && !rt.Hibernate && (rt.Status == "" || rt.Status == "ready") {
			out = append(out, rt)
		}
	}
	return out
}

// workerBootConcurrency bounds how many sidecars the boot pass wakes at
// once. WakeWorker blocks for the settle window (activator's sidecarSettle,
// 10s in production) whenever a child never calls /v1/ready — a slow or
// silent sidecar, or simply one that only survives the window rather than
// reporting early — so an unbounded fan-out could still spawn every child on
// the box in the same instant (a thundering herd against S3 leases and
// restore reads), while a serial pass would cost up to 10 seconds PER
// sidecar of agent startup. This bound trades both off: bounded fan-out, not
// one-at-a-time.
const workerBootConcurrency = 8

// runWorkerBootPass wakes every route workersToStart selects out of routes,
// concurrently (bounded by workerBootConcurrency) rather than one at a time
// — see workerBootConcurrency's doc comment for why serial is too slow and
// unbounded is too bursty. wake is svc.WakeWorker in production; a seam here
// so the concurrency shape is testable without a real activator.Service. A
// wake that errors is logged and does not stop the rest — one sidecar that
// can't come up must not strand every other one on the box at boot.
func runWorkerBootPass(routes []registry.Route, wake func(registry.Route) error) {
	toStart := workersToStart(routes)
	if len(toStart) == 0 {
		return
	}
	sem := make(chan struct{}, workerBootConcurrency)
	var wg sync.WaitGroup
	for _, rt := range toStart {
		rt := rt
		wg.Add(1)
		sem <- struct{}{}
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			if err := wake(rt); err != nil {
				log.Printf("agent: worker boot pass: wake %s: %v", rt.Host, err)
			}
		}()
	}
	wg.Wait()
}

// agentSocketRetryBase/Cap bound the agent socket's rebind backoff — vars,
// not consts, so a test can shrink them (sidecarSettle's precedent in
// internal/activator).
var (
	agentSocketRetryBase = time.Second
	agentSocketRetryCap  = 30 * time.Second
)

// serveAgentSocket runs serve (agentsock.Server.Serve in production) in a
// retry loop with doubling backoff, forever, until ctx is cancelled. serve
// is a seam so the retry/backoff shape is testable without a real unix
// socket.
//
// A failed bind must NEVER be fatal to the agent: agentsock.Serve
// deliberately REFUSES a socket another listener already owns (rather than
// unlinking it) to prevent two agents both believing they own the box's
// workers, and on an agent restart the old process's listener can take a
// few seconds to release — that refusal is expected and transient. An agent
// that exited because a socket was briefly busy would take down every route
// on the box, so this loop only ever logs and retries; it has no path that
// calls log.Fatal or returns an error to a caller who might.
//
// A clean return from serve (nil, once ctx is cancelled — agentsock.Serve's
// own contract) ends the loop immediately with no retry: that is a normal
// shutdown, not a failure.
func serveAgentSocket(ctx context.Context, path string, serve func(context.Context, string) error) {
	backoff := agentSocketRetryBase
	for {
		err := serve(ctx, path)
		if err == nil || ctx.Err() != nil {
			return
		}
		log.Printf("agent: agent socket %s: %v — retrying in %s", path, err, backoff)
		timer := time.NewTimer(backoff)
		select {
		case <-timer.C:
		case <-ctx.Done():
			timer.Stop()
			return
		}
		backoff *= 2
		if backoff > agentSocketRetryCap {
			backoff = agentSocketRetryCap
		}
	}
}

// nestedGate is what an isolated wake requires of the box (design §8.5,
// §8.6): this process able to make the openat2 call instancefs is built on
// (self; the edge unit's own hardening is what can take it away, #126),
// the three nosymfollow mounts a nested instance's tree depends on, and a
// control service that reports ready. The answer is recomputed on
// every wake (the mounts are cheap to read; /ready is one short request),
// so a box that becomes ready needs no agent restart and a box that stops
// being ready refuses the very next wake rather than the stale verdict a
// once-at-startup check would give.
func nestedGate(self func() error, mounts func(string) (bool, error), ready func(context.Context) error) func() error {
	return func() error {
		if err := self(); err != nil {
			return err
		}
		for _, m := range []string{"/data/instances", "/run/carlos/instances", "/run/carlos/logs"} {
			ok, err := mounts(m)
			if err != nil {
				return fmt.Errorf("%s: %v", m, err)
			}
			if !ok {
				return fmt.Errorf("%s is not a nosymfollow mount", m)
			}
		}
		ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineReady)
		defer cancel()
		return ready(ctx)
	}
}

// peekRegistryPath pre-parses the agent args for -registry (default
// <dataDir>/registry.db) without disturbing them — edge.RunWithActivator does
// the authoritative parse. The set mirrors edge.Run's flags so a full command
// line parses cleanly; an unknown flag just ends the peek early (the edge's
// own ExitOnError parse then reports it properly).
func peekRegistryPath(args []string, dd string) string {
	fs := flag.NewFlagSet("agent", flag.ContinueOnError)
	fs.SetOutput(io.Discard)
	regPath := fs.String("registry", filepath.Join(dd, "registry.db"), "")
	fs.String("acme-cache", "", "")
	fs.String("https", "", "")
	fs.String("http", "", "")
	fs.String("fallback", "", "")
	fs.Bool("dev", false, "")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)
	return *regPath
}

// materializeHook builds the ExecBackend's cold-wake staging hook, or nil
// when this box has no release store/pubkey — nil keeps today's behavior
// ("no live binary for <host>" wake failure), the safe degrade for a box
// whose routes are versioned by hand or not at all.
//
// The hook is the adopt pass scoped to one host (release.MaterializeHost),
// so a cold wake with nothing staged pays one slow request instead of 503ing
// until the hourly pass comes round (issue #77). Everything the pass refuses
// it still refuses here: a bake-held head is NOT materialized, and the wake
// fails with that reason rather than the head being installed early.
//
// Store and key are resolved ONCE, at agent start, exactly like adoptLoop's:
// a wake is on a request's critical path, and re-reading env plus re-opening
// an S3 client per cold wake would put that work there for nothing.
//
// OnStageBlocked is wired so a cold wake on a hardened box (this sandboxed
// process cannot write /opt/carlos) relays the stage it cannot perform to
// the root pass (carlos-stage.path) instead of just failing the wake. NO
// RestartHibernating here, deliberately: the host is cold, restart stays
// nil — the activator exec'ing the freshly-pointed binary IS the restart.
// The requester is built once with the hook, so its once-a-minute budget is
// the hook's own, same shape as the loops'.
func materializeHook(reg *registry.Registry, dd string) func(context.Context, string) error {
	store, err := openDeployStore(context.Background())
	if err != nil {
		log.Printf("agent: cold-wake materialize disabled: %v", err)
		return nil
	}
	pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		log.Printf("agent: cold-wake materialize disabled: %v", err)
		return nil
	}
	optDir := release.OptDir()
	stageReq := stageRequester(dd)
	return func(ctx context.Context, host string) error {
		// The process's one proof gate (proofask.go), so a cold wake cannot
		// adopt an unproven build of a listed app.
		return release.MaterializeHost(ctx, reg, store, pub, optDir, dd, host,
			release.OnStageBlocked(stageReq), release.WithProofGate(agentProofGate()),
			release.WithDigestRatchet(pointerRatchet(false, optDir, dd, os.Getenv)),
			updatesPassOption(dd, false))
	}
}

// adoptLoop runs a release-adoption pass roughly hourly (plus jitter, so a
// fleet never hits the store in the same second), forever. Missing deployment
// config disables the loop with one log line — an agent with no release store
// is a valid box (routes managed by hand, adopted by the timer unit, or not
// versioned at all).
// The activator is taken so the pass can cycle exec-backed instances the way
// only this process can (see makeRestartRoute); nil is fine — a box with no
// hibernation armed simply has no children to bounce.
//
// sd carries the scheduled-work converger; this loop is its HOURLY floor (the
// fast one rides the serial in fastConvergeLoop). sd.converge is a no-op when
// the box has no deploy store — the agent logged that once at arming.
func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config) {
	ctx := context.Background()
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: adopt loop disabled: %v", err)
		return
	}
	pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		log.Printf("agent: adopt loop disabled: %v", err)
		return
	}
	optDir := release.OptDir()
	restart := makeRestartRoute(reg, systemctlTryRestart, agentCheckedUnitRestart(systemdUnits{}, optDir, os.Getenv, nil), adoptBounceFn(svc))
	log.Printf("agent: adopt loop armed (hourly + jitter)")
	// First pass shortly after boot (Eleven's OnBootSec=5min, compressed —
	// the edge is already up by the time this fires), then hourly + jitter.
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = time.Hour + rand.N(10*time.Minute)
		// DO NOT add release.NagExpiredCanaries(...) to this call. This loop
		// runs hourly, but it is NOT "the hourly root adopt pass" the option's
		// own doc comment names — that's RunAdopt (carlos-adopt.service)
		// below, root-privileged and the one place config sync also runs. Two
		// hourly callers both nagging would double the log line every hour;
		// worse, if this loop's cadence ever changes, a nag wired here changes
		// frequency with it silently. One caller, one nag.
		//
		// DO NOT add release.WithCapabilities(...) to this call either, and
		// for a harder reason than the nag's: that option WRITES
		// /etc/systemd/system, and this loop runs inside carlos-edge.service,
		// which is ProtectSystem=strict with ReadWritePaths=/data /run/carlos.
		// It could not converge a drop-in if it tried; all it would produce is
		// a permission-denied line every hour. The capability converge belongs
		// to RunAdopt (carlos-adopt.service, root) alone — see below.
		//
		// OnStageBlocked, by contrast, is exactly right here: it fires only
		// when this sandboxed process cannot write the version store, and all
		// it does is ask the root pass to do the part this one cannot.
		//
		// RestartHibernating, by contrast again, belongs HERE and not in
		// RunAdopt: this process owns the exec children, so it is the only one
		// whose restart callback can reach a warm hibernator at all. Without
		// it the pass would split on the Hibernate FLAG and hand every exec
		// instance the pure repoint — the flag says "may sleep", and a busy
		// instance would go on serving the binary its symlink no longer names
		// until it happened to go idle. Safe because makeRestartRoute cannot
		// start a process on either of its paths; see release.RestartHibernating.
		//
		// WithStartUnit rides wherever the named-unit restart arm can fire:
		// a failed named-unit restart that this pass rolled back must not
		// leave the service down (release.WithStartUnit). It asks for no
		// authority the restart above did not already need — both are one
		// systemctl verb on the same unit — and it cannot wake anything: it
		// fires only after a try-restart RETURNED AN ERROR, which a stopped
		// unit's exit-0 no-op never does.
		//
		// DO NOT add the platform self-update here either: the system leg
		// lives in the fast loop and re-observes hourly on its own, so an
		// hourly call here would only double the pointer read.
		//
		// agentUnitsOption hands named units to root when this process is
		// not root (issue #315), which leaves WithStartUnit and
		// WithRollbackRestart to an agent run as root.
		opts := append([]release.PassOption{
			release.BakeDir(dataDir()), release.OnStageBlocked(stageRequester(dataDir())),
			release.RestartHibernating(restart),
			release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit)),
			release.WithRollbackRestart(agentRollbackUnitRestart(reg, systemdUnits{}, optDir, os.Getenv)),
			release.WithProofGate(agentProofGate()), release.WithHostWrites(writes),
			release.WithDigestRatchet(pointerRatchet(false, optDir, dataDir(), os.Getenv)),
			updatesPassOption(dataDir(), false)}, agentUnitsOption()...)
		if err := release.Pass(ctx, reg, store, pub, optDir, restart, opts...); err != nil {
			log.Printf("agent: adopt pass: %v", err)
		}
		// The schedule converge's hourly floor (schedule spec §6.2), AFTER the
		// release pass: an app whose schedules.json arrived with a new version
		// gets its rows against the version that is actually staged. Die-free
		// per app inside Pass — the joined error is one log line, never fatal.
		if err := sd.converge(ctx); err != nil {
			log.Printf("agent: schedule converge (hourly): %v", err)
		}
	}
}

// configSyncLoop delivers app config to exec-backed routes: hourly (plus
// a first pass ~1min after boot, like adoptLoop), materialize every
// exec-syncable route's <dataDir>/<host>.config.env from the bucket and
// bounce running instances whose config changed. Runs IN-PROCESS on
// purpose — /data is inside this unit's sandbox and the exec children
// are this process's own, so it can both write the files and restart
// the readers; the root adopt path keeps the unit-backed half
// (/etc/carlos/apps needs root). Inert until CARLOS_SECRETS_KEY is set,
// the same convention as every other optional loop.
func configSyncLoop(reg *registry.Registry, svc *activator.Service, dd string, writes hostwrite.Config) {
	raw := os.Getenv("CARLOS_SECRETS_KEY")
	if raw == "" {
		log.Printf("agent: config sync loop disabled: CARLOS_SECRETS_KEY not set")
		return
	}
	priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
	if err != nil {
		log.Printf("agent: config sync loop disabled: %v", err)
		return
	}
	ctx := context.Background()
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: config sync loop disabled: %v", err)
		return
	}
	log.Printf("agent: config sync loop armed (hourly + jitter)")
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = time.Hour + rand.N(10*time.Minute)
		if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
			log.Printf("agent: config sync: %v", err)
		}
	}
}

// configSyncMu serializes the two config loops' SyncExec passes against each
// other. They can genuinely overlap — hourly pass mid-flight when the 3s
// loop notices a fresh generation — and syncExecHost writes every route's
// file through a FIXED temp path (<host>.config.env.tmp, syncexec.go), so
// two interleaved passes could cross-rename each other's content: the loser
// lands STALE bytes after the winner's fresh ones, with the generation cache
// already advanced past the change, leaving nothing to re-fire until the
// next hourly pass. Serializing in-process closes that outright; the root
// adopt path is a different process writing a different namespace
// (/etc/carlos/apps) and is not this mutex's problem.
var configSyncMu sync.Mutex

// syncExecSerialized is the one way either config loop runs SyncExec — the
// mutex lives in a shared wrapper, not at the call sites, so a future third
// caller cannot forget it.
func syncExecSerialized(ctx context.Context, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error {
	return syncExecSerializedFiltered(ctx, nil, reg, store, priv, dd, svc, writes...)
}

// syncExecSerializedFiltered is syncExecSerialized over the routes keep
// accepts (nil: every route): the scoped pass's config leg (S3 change-index
// design §4.5). The filter is applied where SyncExec chooses its routes
// (appconfig.WithRouteFilter), so a route it leaves out costs no read.
func syncExecSerializedFiltered(ctx context.Context, keep func(registry.Route) bool, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error {
	configSyncMu.Lock()
	defer configSyncMu.Unlock()
	pending := map[string]string{}
	outcomes := map[string]bool{}
	opts := []appconfig.SyncOption{
		appconfig.SkipFenced(func(host string) bool { return layout.JournalPresent(dd, host) }),
		appconfig.WhenInstanceDirAbsent(func(host, dir string) { pending[host] = dir }),
		appconfig.WithHostOutcome(func(host string, done bool) { outcomes[host] = done }),
	}
	if keep != nil {
		opts = append(opts, appconfig.WithRouteFilter(keep))
	}
	for _, cfg := range writes {
		opts = append(opts, appconfig.WithHostWrite(cfg))
	}
	// CARLOS_SWAP=overlap: the scheduler restarts, paced, and acknowledges a
	// revision once it is active; refusing: no restart. configSyncMu above
	// is what makes ConfigDelivered serial per host.
	var err error
	if cb, ok := rolloutConfigCallbacks(ctx, store, configLegs{awake: managedAwake(svc, reg), restart: bounce(svc)}); ok {
		err = appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, dd, os.Getenv("PLATFORM_APPS_DOMAIN"), cb, opts...)
	} else {
		err = appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"), opts...)
	}
	// A filtered pass names only the routes it kept, so it is never
	// complete: a host it left out keeps its wait (pendingDirs.record).
	configDirsPending.record(pending, outcomes, err == nil && keep == nil)
	return err
}

// configDirsPending is what the last config pass could not write because a
// nested row's instance directory did not exist yet (harness Run 9, finding
// 6): host -> the directory. The activator makes that directory at the
// first wake, and before this the only pass that came back for the file was
// configSyncLoop's hourly one, so a nested instance created with settings
// woke and ran without them, its store binding included, for up to an hour.
// configGenLoop now looks at these directories on every tick and runs the
// pass again as soon as one is there.
var configDirsPending pendingDirs

type pendingDirs struct {
	mu   sync.Mutex
	dirs map[string]*pendingDir
}

// pendingDir is one host the config pass waits for: its instance directory,
// and, once the directory is there, how many passes since have left the host
// undone and how many ticks to let go by before the next retry.
type pendingDir struct {
	dir       string
	failures  int
	waitTicks int
}

// pendingRetryCap bounds the retry backoff: at most 16 generation-loop ticks
// (about a minute) between passes for a host whose directory is there but
// whose config a pass keeps leaving undone (a bundle that will not read).
const pendingRetryCap = 16

// record keeps what one pass found: absent, the hosts whose instance
// directory was not there; outcomes, every in-scope host the pass named,
// done or not (appconfig.WithHostOutcome). A host stays waited for until a
// pass reports it done (fix round 1, Astra's finding 3: a pass that left it
// undone on a read failure still returned nil), or until a complete pass no
// longer has it in scope (its row is gone or no longer takes a bundle). A
// pass that failed part way names fewer hosts than it might have, so it only
// adds. A host a pass leaves undone with its directory there backs off,
// doubling from the next tick up to pendingRetryCap ticks.
func (p *pendingDirs) record(absent map[string]string, outcomes map[string]bool, complete bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.dirs == nil {
		p.dirs = map[string]*pendingDir{}
	}
	for host, e := range p.dirs {
		if _, again := absent[host]; again {
			continue
		}
		done, named := outcomes[host]
		switch {
		case named && done, !named && complete:
			delete(p.dirs, host)
		case named:
			// The count saturates once the delay reaches the cap, so the
			// shift below never overflows however long a host keeps failing.
			if 1<<e.failures <= pendingRetryCap {
				e.failures++
			}
			e.waitTicks = min(1<<(e.failures-1), pendingRetryCap) - 1
		}
	}
	for host, dir := range absent {
		if e, ok := p.dirs[host]; ok {
			e.dir = dir
			continue
		}
		p.dirs[host] = &pendingDir{dir: dir}
	}
}

// appeared names a host whose instance directory a pass waited for, which
// exists now, and whose retry is due; ok is false when there is none. Each
// call is one tick of the backoff. Lstat only: the directory is the
// activator's, and nothing here follows or opens it.
func (p *pendingDirs) appeared() (host string, ok bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	for h, e := range p.dirs {
		if _, err := os.Lstat(e.dir); err != nil {
			continue
		}
		if e.waitTicks > 0 {
			e.waitTicks--
			continue
		}
		if !ok {
			host, ok = h, true
		}
	}
	return host, ok
}

// configAckFunc builds the callback config sync's two passes (the root
// pass's appconfig.Sync and this agent's own appconfig.SyncExec, called
// through syncExecSerialized above) invoke for a route they just converged
// (relay spec §5, internal/appconfig/configack.go): it stamps this box's
// label onto the ack — BoxLabel (metrics_loop.go) is the same identity
// every other per-box artifact this platform writes uses, so a config-ack
// and this box's heartbeat/metrics rows agree about which box they belong
// to — and writes it. A write failure is logged, never returned: an ack is
// best-effort telemetry for `carlos relay status`, not something config
// delivery itself depends on, and failing the whole sync pass over it would
// make a bucket blip on the ack object cost every OTHER route its config.
func configAckFunc(ctx context.Context, store release.Store, logPrefix string) func(appconfig.ConfigAck) {
	box := BoxLabel()
	return func(a appconfig.ConfigAck) {
		a.Box = box
		if err := appconfig.WriteConfigAck(ctx, store, a); err != nil {
			log.Printf("%s: config ack %s: %v", logPrefix, a.Host, err)
		}
	}
}

// configAckWriter is configAckFunc for the armed scheduler's ack writer
// (rollout.Manager.DeferAck): the same stamp and write, under the context
// the writer hands each write, which bounds it (Task 11, Astra R1 Flag
// triage T8). configAckFunc itself stays as it is, unbounded, for flag-off
// sync and the root pass (plan review #11).
func configAckWriter(store release.Store, logPrefix string) func(context.Context, appconfig.ConfigAck) {
	box := BoxLabel()
	return func(ctx context.Context, a appconfig.ConfigAck) {
		a.Box = box
		if err := appconfig.WriteConfigAck(ctx, store, a); err != nil {
			log.Printf("%s: config ack %s: %v", logPrefix, a.Host, err)
		}
	}
}

// reconcileMu serializes the two provisioning loops' Reconcile passes
// against each other, the same shape as configSyncMu above. They can
// genuinely overlap — instancesLoop's 5-minute pass mid-flight when a
// serial bump fires fastConvergeLoop's full pass set — and Reconcile's
// removal phase deletes any Provisioned, in-scope route that is absent from
// ITS OWN record snapshot, which it lists at the top of the pass
// (provision/reconcile.go): a route the newer pass minted between the older
// pass's apps/ listing and its registry listing looks exactly like a
// retired record to the older pass, which would delete it. The re-mint
// minutes later comes back with its capability grants revoked
// (registry.Delete purges grants — that is Delete's contract, not a bug).
// Serializing in-process closes the race outright; the root adopt path runs
// no Reconcile and is not this mutex's problem.
var reconcileMu sync.Mutex

// reconcileSerialized is the one way any agent loop runs the record-family
// reconcile passes — the mutex lives in a shared wrapper, not at the call
// sites, so a future third caller cannot forget it. Pinned by
// TestReconcileSerializedHoldsTheMutex.
//
// BOTH families run here, under ONE hold of the mutex: the instance pass and
// the sidecar pass. Each pass's removal phase is scoped to its own route Kind
// (provision.ReconcileSidecars' doc comment), so they cannot delete each
// other's rows — but they read and write the same registry, and the
// overlapping-pass hazard the mutex exists for (a route minted by a newer
// pass looking exactly like a retired record to an older one) applies within
// each family just as much. Running them as one serialized unit also means a
// caller cannot converge instances while leaving sidecars a tick behind.
//
// Errors are JOINED rather than short-circuited: a sidecar-side failure must
// not hide an instance-side one, and neither must stop the other running —
// each pass is independently safe to run and independently valuable.
func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig) error {
	return reconcileSerializedWith(ctx, s, reg, cfg, armedMovesWorker.Load())
}

// reconcileSerializedWith is reconcileSerialized with the moves worker whose
// offline hooks the pass takes (movesoffline.go): the armed worker in the
// agent, a box's own worker in a test. A nil worker, or one on a box that
// cannot take part in an offline move, adds nothing.
func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig, w *movesWorker) error {
	reconcileMu.Lock()
	defer reconcileMu.Unlock()
	// With the change index armed, the record walks below read the record
	// view's snapshot, refreshed here under reconcileMu so a pass's snapshot
	// is never older than the previous pass's (S3 change-index design §4.3).
	// Writes still go to s, and a key written through the snapshot store is
	// read from s afterwards. Disarmed, or when the view cannot vouch, rs is
	// s and the pass walks the bucket as before.
	rs, _, _ := viewStore(ctx, s)
	// A fresh row is minted at the layout this box's intent names
	// (ReconcileConfig.Layout): nested on a nested box, so a new tenant
	// never runs flat there. Set here, the one place both callers pass
	// through, because the intent is read from the store this pass has in
	// hand and scoped to the label this box publishes under.
	if cfg.Layout == nil {
		cfg.Layout = func(ctx context.Context, s release.Store) (string, error) {
			return mintLayoutIntentWith(ctx, s, reg.List)
		}
	}
	// Seeded records (seeded-instances spec §3.2): acted on only once the
	// activator has armed the data bucket (wireSeedPlacement).
	if cfg.Seeds == nil {
		cfg.Seeds = armedSeedHooks.Load()
	}
	// Move holds (moves spec §9.6): participant rows minted held and kept.
	// Unchanged with the Phase 1 switch off.
	cfg = withMovesHolds(cfg)
	// Offline move holds: the hold from the record alone, the keep from the
	// view, which is read here first if no listing has been read yet or the
	// last one failed, so keep never fails closed on a box whose listing
	// reads. That listing reads the bucket, never the record view: what a
	// move keeps is decided on live data (S3 change-index final review R1,
	// finding 1). Nothing on a box that cannot take part, and nothing where
	// Phase 1's hooks are set.
	if w != nil && w.cfg.OfflineCapable && w.cfg.Offline != nil {
		if !w.cfg.Offline.built() || w.cfg.Offline.failed() {
			refreshOfflineView(ctx, s, w.cfg.Offline)
		}
		cfg = withOfflineHolds(cfg, offlineHold(w.cfg), w.cfg.Offline)
	}
	// Runs ride this pass's record reads (run spec §5.2): the worker notes
	// every in-scope record with a run, and acts after the pass.
	cfg = withRunsHook(cfg)
	return errors.Join(
		provision.Reconcile(ctx, rs, reg, cfg),
		provision.ReconcileSidecars(ctx, rs, reg, cfg),
	)
}

// bounce builds the restart callback SyncExec calls for a route whose
// rendered config actually changed. Shared by BOTH config loops (hourly
// configSyncLoop and the fast configGenLoop) rather than duplicated per
// loop, unlike the adopt pair: those two differ from each other (filter,
// nag) where these two must not — the delivery must be identical whichever
// loop happened to notice a change first, and a divergence would mean an
// app converges one way at 3s and another way on the hour. A nil svc (no
// activator armed) still returns a working no-op: the files converge for
// future spawns, there is simply nothing running to restart.
func bounce(svc *activator.Service) func(string) error {
	return func(host string) error {
		if svc == nil {
			return nil
		}
		if svc.Bounce(host) {
			log.Printf("agent: config changed, bounced %s", host)
		}
		return nil
	}
}

// genAbsent marks an app whose generation object was READ and found
// missing, distinguishing it from an app never seen at all — the two are
// different in the only way this loop cares about: a never-seen app must
// record silently (first sight), while a known-absent app that suddenly
// has a generation object is an app whose config was just written for the
// first time, which must sync. A real etag can never collide with it: no
// store produces one containing a NUL byte (S3 etags are hex/quoted,
// DirStore's is hex).
const genAbsent = "\x00absent"

// genCache is everything configGenLoop remembers between ticks: one
// generation etag per config-carrying app, and the previous tick's
// route-shape digests. One object rather than two parameters because the
// two are remembered for exactly one reason — deciding what "changed"
// means — and a caller holding one without the other would silently lose
// half the loop's sensitivity.
type genCache struct {
	etags map[[2]string]string
	// exec/unit are appconfig.ConfigRouteFingerprints' two digests as of
	// the last pass that computed them; shapeSeen is false until the first
	// such pass. The bool is not redundant with an empty string: a box with
	// no config-carrying routes at all has a real, stable digest of the
	// empty set, and treating that as "never seen" would make its first
	// route look like a change on the tick after boot rather than on the
	// tick it appeared.
	exec, unit string
	shapeSeen  bool

	// gate skips the read of an app whose scope has not moved since its last
	// successful read, younger than the backstop (S3 change-index design §5).
	// nil, or a gate over a nil or unusable watch, reads every app every
	// tick, as before the index. now is the gate's clock (nil: time.Now).
	gate *release.ReadGate
	now  func() time.Time
}

func newGenCache() *genCache { return &genCache{etags: map[[2]string]string{}} }

// genPollPass is one tick of configGenLoop, split out for tests: read the
// generation etag of every app the box can converge — exec-config apps and
// unit-backed apps alike — plus the registry's own route shape, and report
// which HALVES moved since the cache was built.
//
// Semantics, each of which is a deliberate choice:
//
//   - First sight of an app (no cache entry) records silently and does NOT
//     sync. Boot must not re-bounce the world — every app would look
//     "changed" on the first tick after a restart. configSyncLoop's ~1min
//     first pass owns boot convergence.
//   - An unchanged etag does nothing.
//   - Absent that stays absent does nothing, forever: an app last written
//     by a pre-generation CLI has no generation object and never will
//     until someone writes its config again. That is legal, not a fault.
//   - Any movement syncs: the etag changed, or absent -> present.
//     present -> absent shouldn't happen (generation objects are only ever
//     created and overwritten, never deleted), but it's treated as a change
//     too — one extra SyncExec pass is cheap, and pretending we didn't see
//     it would be a lie about the only thing this function reports.
//   - ONE bool PER HALF for the whole pass, however many apps moved: the
//     caller runs one SyncExec (which resolves every route itself) and asks
//     for at most one root config pass (which sweeps every unit route
//     itself). Per-app syncs would multiply bucket reads to deliver exactly
//     the same files, and per-app root requests would ask the same pass to
//     run several times over.
//   - The halves are independent: an app whose routes are all exec-backed
//     sets execChanged alone, an all-unit app sets unitChanged alone, and an
//     app holding both shapes at once sets both — off the SAME single read,
//     because its whole config lives in one generation object however its
//     routes are backed.
//   - Apps that vanish from the route set lose their cache entries, so a
//     re-added app is a first sight again rather than a comparison against
//     an etag remembered from before it left (a route being re-added gets
//     fresh config from the spawn path anyway).
//   - The ROUTE SHAPE is watched too, not just the bucket. A route's
//     Environment binding lives in the box's own registry and moves no etag
//     anywhere, so without this the loop cannot see a rebinding at all and
//     delivery waits out configSyncLoop's hourly floor: issue #196, where a
//     live route served the wrong merge for ten minutes and the only symptom
//     was the app reporting a credential missing — which reads as a BAD
//     credential rather than an undelivered one. First sight of the shape is
//     silent for the same reason first sight of an app is. This covers BOTH
//     doors onto the column: `carlos route -environment` on the box, and the
//     provisioning reconciler converging Environment from an instance record
//     — which is the sanctioned door for a Provisioned row, and which the
//     issue's own suggested fix (bump the generation from `carlos route`)
//     would have missed entirely. It costs no bucket read: see
//     appconfig.ConfigRouteFingerprints.
//   - A read error that isn't "not found" leaves that app's cache entry
//     exactly as it was and keeps evaluating the rest: the errors come back
//     for the caller to aggregate into ONE log line per pass, never one per
//     app per tick. Clearing the entry instead would turn a recovered blip
//     into a bogus first sight or a bogus change.
//   - With the change index armed (cache.gate), an app already in the cache
//     is read only when its scope moved, its last read failed, or that read
//     is the backstop's age (S3 change-index design §5). A skipped app keeps
//     its entry and reports no movement.
//
// The cache is mutated in place — it lives across ticks in the loop, which
// is the whole point (there is nowhere else to remember what "changed"
// means).
func genPollPass(ctx context.Context, routes []registry.Route, store release.Store, cache *genCache) (execChanged, unitChanged bool, readErrs []error) {
	// The registry-side half (issue #196). Free — no bucket read at all —
	// and evaluated first so a rebinding converges on the same tick a var
	// change would, rather than one behind it.
	execFP, unitFP := appconfig.ConfigRouteFingerprints(routes)
	if cache.shapeSeen {
		execChanged = execChanged || execFP != cache.exec
		unitChanged = unitChanged || unitFP != cache.unit
	}
	cache.exec, cache.unit, cache.shapeSeen = execFP, unitFP, true

	execApps := appconfig.ExecConfigApps(routes)
	unitApps := appconfig.UnitConfigApps(routes)
	// The poll set is the UNION, read once per app: the two sets overlap
	// whenever an app has both route shapes, and its config is one object
	// either way. inExec/inUnit then say which half a movement belongs to.
	inExec := make(map[[2]string]bool, len(execApps))
	for _, k := range execApps {
		inExec[k] = true
	}
	inUnit := make(map[[2]string]bool, len(unitApps))
	for _, k := range unitApps {
		inUnit[k] = true
	}
	apps := make([][2]string, 0, len(execApps)+len(unitApps))
	apps = append(apps, execApps...)
	for _, k := range unitApps {
		if !inExec[k] {
			apps = append(apps, k)
		}
	}
	now := time.Now
	if cache.now != nil {
		now = cache.now
	}
	at := now()
	live := make(map[[2]string]bool, len(apps))
	for _, k := range apps {
		live[k] = true
		key := appconfig.GenKey(k[0], k[1])
		// The gate (spec §5): an app already in the cache is read again only
		// when its scope moved, its last read failed, or that read is the
		// backstop's age. An app not in the cache is always read: its first
		// sight is what a later change is compared against. The generation
		// is taken before the read, so a write landing during it stays due.
		scope, gated := release.ChangeScope(key)
		gen, due := cache.gate.Due(scope, at)
		if _, seen := cache.etags[k]; seen && gated && !due {
			continue
		}
		_, etag, err := store.Get(ctx, key)
		if gated {
			cache.gate.Mark(scope, gen, at, err == nil || errors.Is(err, os.ErrNotExist))
		}
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = genAbsent
		case err != nil:
			readErrs = append(readErrs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
			continue // cache entry untouched: this app is unknown this tick, not changed
		}
		prev, seen := cache.etags[k]
		cache.etags[k] = etag
		if seen && prev != etag {
			execChanged = execChanged || inExec[k]
			unitChanged = unitChanged || inUnit[k]
		}
	}
	for k := range cache.etags {
		if !live[k] {
			delete(cache.etags, k)
		}
	}
	return execChanged, unitChanged, readErrs
}

// configSyncMarker is the file configGenLoop touches to ask a root config
// pass to run now: carlos-adopt.path watches it (PathChanged) and starts
// carlos-config-sync.service (`carlos adopt -config-only`), which owns the
// unit half of config delivery (/etc/carlos/apps + try-restart) that this
// unprivileged process cannot — and only that half.
// It lives in the data dir, deliberately outside the *.config.env namespace
// reapConfigEnv sweeps — the reaper matches that suffix alone and must go
// on ignoring this.
const configSyncMarker = "config-sync-requested"

// requestRootSync touches the marker. The content is a timestamp only so an
// operator reading /data can see when the last request went out; systemd
// watches the path, not the bytes. Errors are the caller's to log and
// swallow: a request that cannot be written costs latency, not correctness,
// because carlos-adopt.timer's hourly pass is the floor underneath it.
func requestRootSync(dd string) error {
	return os.WriteFile(filepath.Join(dd, configSyncMarker),
		[]byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644)
}

// stageRequestMarker is the release half of configSyncMarker. The agent's
// adopt loops run inside carlos-edge.service (ProtectSystem=strict, no /opt
// write), so they can SEE a promotion within seconds and stage nothing; the
// only process that can write the version store is the root pass on
// carlos-adopt.timer, an hour away. Touching this file relays "there is a
// version here I cannot stage" to carlos-stage.path, exactly as the config
// half relays to carlos-adopt.path. Lives in the data dir beside its config
// twin, outside the *.config.env namespace reapConfigEnv sweeps.
const stageRequestMarker = "stage-requested"

// stageRequestEvery throttles those requests. Unlike the config marker —
// touched only when a generation etag MOVES, which is naturally rare — a
// blocked stage repeats on every tick of a loop that runs every 2-3 seconds,
// so an untouched throttle would ask systemd to start a root pass ~30 times a
// minute and each pass would re-block behind the one before it.
//
// Fifteen seconds, and a request inside the window is DEFERRED to its end,
// never dropped (stageThrottle). It was a minute and dropped, and a second
// release inside the minute then waited for whatever next moved the serial:
// 98 extra seconds on the flagship on 2026-10-06 (#231). A root pass takes
// about eight seconds, and the window has to outlast it: carlos-stage.path
// watches with PathChanged, and systemd does not act on a change made while
// carlos-stage.service is still running.
const stageRequestEvery = 15 * time.Second

// stageRequester returns a release.OnStageBlocked callback that touches the
// marker at most once per stageRequestEvery. The closure holds the clock, so
// each loop gets its own budget: the fast loop asking does not silence the
// hourly one, and neither can starve the other.
//
// Failures are logged and swallowed, like requestRootSync's: a request that
// cannot be written costs latency, never correctness, because
// carlos-adopt.timer remains the floor underneath it.
func stageRequester(dd string) func(host, ver string, err error) {
	ask := stageMarkerRequester(dd)
	return func(host, ver string, err error) {
		ask(host, ver, err)
		// The stage relay (stagerelay.go) learns the head here and not in
		// markerRequester, which mintRequester shares: a proof-mint request
		// is not a head that failed to stage.
		if r := armedStageRelay.Load(); r != nil {
			r.note(host, ver)
		}
	}
}

// stageMarkerRequester is stageRequester without the relay: the same marker,
// throttle and log lines, and nothing remembered. The moves worker uses it
// (movespass.go): its version is a move's held version, not a channel head,
// and a held stage neither moves the installed Version nor leaves a refusal,
// so the relay would have no way to know the request was answered.
func stageMarkerRequester(dd string) func(host, ver string, err error) {
	return markerRequester(dd, func(host, ver string, err error) {
		if err != nil {
			log.Printf("agent: cannot request root stage for %s %s: %v", host, ver, err)
			return
		}
		// One line for every cause the marker relays (Task 12a): a build
		// this process cannot stage, an install request only root
		// resolves (ErrInstallRequestNeedsRoot), a named unit only root
		// restarts (ErrUnitRestartNeedsRoot), and a held wake's target only
		// root installs. The throttle does not carry the cause.
		log.Printf("agent: %s needs a root pass for %s (a build only root can stage or install, a unit only root can restart, or an install request only root resolves); asked for one", host, ver)
	})
}

// mintRequester is the mint asker's (proofask.go) request: the same marker
// and the same throttle shape as stageRequester's, with its own budget and
// its own words. The asker logs why it asks; this logs only a failed write.
func mintRequester(dd string) func(host, ver string, err error) {
	return markerRequester(dd, func(host, ver string, err error) {
		if err != nil {
			log.Printf("agent: cannot request a root pass to mint a proof for %s %s: %v", host, ver, err)
		}
	})
}

// markerRequester touches the stage-request marker through a fresh
// stageThrottle and hands said each write's outcome.
func markerRequester(dd string, said func(host, ver string, err error)) func(host, ver string, err error) {
	t := &stageThrottle{every: stageRequestEvery, now: time.Now, after: func(d time.Duration, f func()) { time.AfterFunc(d, f) },
		write: func(host, ver string) {
			said(host, ver, os.WriteFile(filepath.Join(dd, stageRequestMarker),
				[]byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644))
		}}
	return func(host, ver string, _ error) { t.request(host, ver) }
}

// stageThrottle writes at most once per every, measured from the end of one
// write to the start of the next. While a write is scheduled (busy), every
// request rides it: a deferred write owns the window however late its timer
// fires. A request inside the window schedules one write for the window's
// end, so none waits longer than every after the write before it.
type stageThrottle struct {
	every time.Duration
	now   func() time.Time
	after func(time.Duration, func())
	write func(host, ver string)
	mu    sync.Mutex
	last  time.Time // when the last write finished
	busy  bool
}

func (t *stageThrottle) request(host, ver string) {
	t.mu.Lock()
	if t.busy {
		t.mu.Unlock()
		return
	}
	t.busy = true
	// The write runs under the lock: a request made once the marker has
	// changed may concern a pointer root has already read, so it waits for
	// the write to finish and schedules the next window, rather than
	// riding a write that cannot answer it.
	do := func() {
		t.mu.Lock()
		t.write(host, ver)
		t.busy, t.last = false, t.now()
		t.mu.Unlock()
	}
	if since := t.now().Sub(t.last); !t.last.IsZero() && since < t.every {
		t.mu.Unlock()
		t.after(t.every-since, do)
		return
	}
	t.mu.Unlock()
	do()
}

// configGenLoop is the fast-converge shape applied to issue #44: a config
// set should converge like a canary adopts — seconds, not the next hour.
// Every ~3s it reads one small object per config-carrying app (the
// generation object every config write touches) and, on movement, delivers
// the halves it can: ONE SyncExec pass for the whole tick for exec-backed
// instances, and one touch of the root-sync marker for unit-backed ones.
// Spurious movement costs a pass that finds nothing to write: SyncExec (and
// the root pass's appconfig.Sync) bounces only routes whose rendered file
// actually changed, so a redundant touch can never cost a restart.
//
// Gated on CARLOS_SECRETS_KEY exactly as configSyncLoop is, for a reason
// beyond convention: without the key SyncExec cannot render an app's
// environment at all, so this loop would poll the bucket every 3 seconds
// forever to signal no one. openDeployStore failing is the same story with
// nothing to read. The gate covers the unit half too, which is an accepted
// asymmetry rather than a considered one: a box whose agent lacks the key
// gets no fast path for its unit apps either, even though the root pass
// that would run holds its own key. In practice both units read the same
// /etc/carlos/host.env, so the case is theoretical — and the hourly timer
// floor stands regardless.
//
// Cardinality: unlike fastConvergeLoop's sweep, which is bounded to the
// deliberately rare canary/* routes plus still-unstaged ones, this watches
// EVERY config-carrying app on the box. That is accepted knowingly — one GET
// of the smallest object in the bucket
// per app per 3s, against the store fastConvergeLoop already hits every 2s. What
// keeps it bounded is that the pass reads generation objects only: no
// manifests, no pointers, no per-environment resolution. All of that
// happens inside the SyncExec a real change triggers.
//
// Deliberately NOT done here:
//
//   - No fast first pass. The loop sleeps BEFORE its first tick and the
//     first sight of every app is silent (see genPollPass) — boot
//     convergence belongs to configSyncLoop's ~1min first pass, and a box
//     coming up must not bounce every instance it has because it has never
//     seen their etags before.
//   - No unit-backed rendering or restarting. This process still writes
//     nothing under /etc and restarts no unit (see RunAdopt): a unit app's
//     movement only touches <dataDir>/config-sync-requested, which
//     carlos-adopt.path relays into a root carlos-config-sync.service run
//     (`carlos adopt -config-only`) that does both. The relay reaches ONLY
//     the config half — never release adoption, never the canary-TTL nag,
//     which would otherwise fire off an unrelated app's config write
//     instead of hourly. The relay can drop an event that lands while that
//     service is mid-run — accepted: the next config write re-touches the
//     marker, and the timer floor delivers within the hour regardless.
//   - No hourly floor replaced. configSyncLoop stays exactly as it was: it
//     catches apps written by older CLIs (no generation object at all),
//     ticks this loop missed, and the boot-retry gap of issue #27.
//
// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
// whole decision lives in genPollPass, which is tested.
func configGenLoop(reg *registry.Registry, svc *activator.Service, dd string, writes hostwrite.Config) {
	raw := os.Getenv("CARLOS_SECRETS_KEY")
	if raw == "" {
		log.Printf("agent: config generation loop disabled: CARLOS_SECRETS_KEY not set")
		return
	}
	priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
	if err != nil {
		log.Printf("agent: config generation loop disabled: %v", err)
		return
	}
	ctx := context.Background()
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: config generation loop disabled: %v", err)
		return
	}
	// One cache for the life of the process: what "changed" means is
	// exactly "differs from the last tick that could read it". Its gate is
	// built here, after RunAgent armed the change index.
	cache := newGenCache()
	cache.gate = newChangeGate()
	log.Printf("agent: config generation loop armed (3s + jitter, exec-config and unit-backed apps)")
	for {
		time.Sleep(3*time.Second + rand.N(time.Second))
		routes, err := reg.List()
		if err != nil {
			// The registry is a local file this process already holds open;
			// a failing List is a broken box, not a passing blip, so it is
			// worth a line even at this cadence (the same call is logged per
			// tick by statusLoop).
			log.Printf("agent: config generation poll: list routes: %v", err)
			continue
		}
		configGenTick(ctx, routes, reg, store, priv, dd, svc, writes, cache)
	}
}

// configGenTick is one tick of configGenLoop over the routes it listed, split
// out so a test can run the loop's own decision.
func configGenTick(ctx context.Context, routes []registry.Route, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes hostwrite.Config, cache *genCache) {
	execChanged, unitChanged, readErrs := genPollPass(ctx, routes, store, cache)
	if len(readErrs) > 0 {
		// ONE line per pass, never one per app: a bucket having a bad
		// minute would otherwise flood the journal at 20 lines a minute
		// per app. The count carries the scale, the first error carries
		// the cause; they are the same cause in practice.
		log.Printf("agent: config generation poll: %d apps unreadable: %v", len(readErrs), readErrs[0])
	}
	if unitChanged {
		// Requested first and never fatal: the write is instant while
		// syncExecSerialized can block on a serialized pass, so the root
		// run gets to start alongside it rather than behind it. A failed
		// marker write costs those apps the wait until
		// carlos-adopt.timer's next hourly pass — never correctness.
		if err := requestRootSync(dd); err != nil {
			log.Printf("agent: request root config sync: %v", err)
		} else {
			log.Printf("agent: config changed for unit-backed app(s), requested root config sync pass")
		}
	}
	// A nested row an earlier pass could not write for because its instance
	// directory was not there yet: the first wake has made it, so its
	// bundle goes in now (finding 6). The pass bounces the instance onto
	// its settings once the wake it is in has finished (Service.Bounce).
	if host, ok := configDirsPending.appeared(); ok && !execChanged {
		log.Printf("agent: %s: its instance directory is there now, writing its config", host)
		execChanged = true
	}
	if !execChanged {
		return
	}
	if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
		log.Printf("agent: config generation sync: %v", err)
	}
}

// Fast-converge cadence and clamps (carlos-deploy spec §3-§4). The tick is
// the serial poll's latency promise (~2s from console bump to pass); the
// clamp bounds what a burst of console writes can cost (one full pass set
// per 5s, with the last bump deferred, never dropped); the backoff bounds
// what a permanently-unstageable route can cost (one look per 5min at the
// cap, instead of one per tick forever).
const (
	fastTickEvery   = 2 * time.Second
	fastFullClamp   = 5 * time.Second
	fastBackoffBase = 2 * time.Second
	fastBackoffCap  = 5 * time.Minute

	// The restart-failure hold (noteRestartFailed) starts higher and caps
	// higher than the staging backoff above, because what it bounds is not
	// a retry's COST to this box but a service CYCLE on a tenant's process:
	// re-attempting a version whose restart just failed repoints, performs a
	// REAL restart of the running old service, fails, rolls back and starts
	// it again — retrying that 2 seconds later is near-certain to do it all
	// again. 30s → 15m doubling means a persistently broken head costs a
	// handful of honest cycles in its first half hour and then four an hour
	// from this loop, with the hourly floors' one-each on top — instead of
	// one every other second for as long as the broken head stands.
	fastRestartBackoffBase = 30 * time.Second
	fastRestartBackoffCap  = 15 * time.Minute
)

// fastDeps is one fastConvergeLoop's state and seams — a struct so the tick
// is testable without goroutines, env vars, or sleeps.
type fastDeps struct {
	reg          *registry.Registry
	store        release.Store
	pub          ed25519.PublicKey
	optDir       string
	restart      func(string) error
	passOpts     []release.PassOption      // fastPassOpts — every sweep pass carries all of them
	fullPass     func(ctx context.Context) // reconcile → adopt → config sync, or a test hook
	restartLeg   func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
	wakeLeg      func(ctx context.Context) // the sidecar wake check (design: sidecar members), or a test hook
	scheduleLeg  func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
	layoutLeg    func(ctx context.Context) // the layout pass (isolated-exec design §8.2), or a test hook; nil = not armed
	systemLeg    *systemLeg                // the platform self-update leg (spec §4); nil = not armed
	proof        *release.ProofGate        // D3's gate (agentProofGate); nil = no proof waits to keep
	mintLeg      func(ctx context.Context) // D3's mint asker (proofask.go); nil = not armed
	lastSerial   int64
	seeded       bool
	lastFull     time.Time
	notBefore    map[string]time.Time
	step         map[string]time.Duration
	restartHold  map[string]*restartFailure // hosts held out of the sweep after a failed restart; see noteRestartFailed
	serialBroken bool                       // last ReadSerial failed and was logged; reset on success
	serialNotify chan<- struct{}

	// The change index (CARLOS_CHANGE_INDEX=read, S3 change-index design
	// §4.5); watch nil leaves the tick exactly as before. The watch is shared
	// with the record view, so Poll's moved list is advisory: the tick diffs
	// Observed against indexActed, the generations it last acted on, and
	// moves that map on only once a pass has covered them.
	watch       *release.ChangeWatch
	scopedPass  func(ctx context.Context, moved []string) // passLegs.scoped, or a test hook
	fullDue     *atomic.Bool                              // the backstop's request for a full pass; nil = none
	indexActed  map[string]uint64                         // nil until the first good observation seeds it
	indexBroken bool                                      // last Poll failed and was logged; reset on success

	// canaryGate gates the sweep's canary pointer reads (spec §5): a canary
	// route stays a member of every tick, but its group's pointer is read
	// only when the app's scope moved, the last read failed or is the
	// backstop's age, or a route of the group is not live on the head that
	// read returned (canaryHeads). nil: every pointer is read, as before.
	canaryGate  *release.ReadGate
	canaryHeads map[[3]string]canaryRead // (account, app, channel) -> its group's last good read
}

// canaryRead is one canary group's last good pointer read: the head it
// returned, the app scope's generation captured before it, and when. The
// gate's mark is per app scope, so another channel of the same app can mark
// it; a group is settled only on its own read (final review R1, finding 6).
type canaryRead struct {
	head string
	gen  uint64
	at   time.Time
}

// restartFailure is one host's cross-pass memo of a version its restart
// refused: which version failed, the current hold step, and when the sweep
// may try that host again. Kept on the loop (fastDeps), never in
// release.Pass — the pass stays stateless; this loop is the caller whose
// cadence needs the bound.
type restartFailure struct {
	ver       string
	step      time.Duration
	notBefore time.Time
}

// fastPassOpts is the option set EVERY fast-converge pass carries — the
// serial-triggered full release.Pass and the sweep's release.PassFilter
// both, built in one place so neither call site can drift from the other.
//
// DO NOT add release.NagExpiredCanaries(...) to this set, ever. This
// is THE loop release.NagExpiredCanaries' doc comment warns about by
// name: every 2-3 seconds, canary/* always in the set. An expired canary sitting
// still would nag on every single tick — a journal-flooding poll storm
// out of what is supposed to be an hourly, easy-to-skim log line. The
// nag belongs to RunAdopt (carlos-adopt.service) alone.
//
// DO NOT add release.WithCapabilities(...) to this set, ever. Same
// sandbox as adoptLoop above (ProtectSystem=strict,
// ReadWritePaths=/data /run/carlos), and this one ticks every 2-3
// SECONDS: arming the converge here would write nothing and log a
// permission denial forever. The accepted consequence, ruled on in
// the 2026-08-03 review of the capability design: a freshly promoted
// canary runs its new binary against the unit shape the previous
// version left until the hourly root pass converges it — up to an
// hour. release.adoptHost prints one line per such repoint so the
// window is visible in the journal rather than silent.
//
// OnStageBlocked IS wired here, and this is the loop it matters most
// for: a canary is promoted to be picked up in seconds, and staging is
// the one step of that pickup this process cannot perform. Throttled
// inside stageRequester — this tick is 2-3 seconds, the request is at
// most one per stageRequestEvery (the requester is built ONCE per loop, so the budget
// is the loop's, not one tick's).
//
// RestartHibernating IS wired here too, and for the same reason as on
// adoptLoop: this process owns the exec children. It matters more on
// this loop than on any other — a promotion is bumped or fast-swept
// precisely to be picked up in seconds, and the routes this loop reaches
// are the warm ones somebody is watching. Pinned by
// TestFastSweepBouncesAWarmHibernatingCanary: dropping it would restore
// a dead-branch production bug without failing any other test.
//
// WithStartUnit IS wired here too (start — makeStartRoute in production,
// nil in tests that never reach a rollback): any pass whose restart
// callback carries the named-unit arm must also carry the start that
// undoes that arm's failed restart, or its rollback leaves the service
// down and the next pass stamps a never-run build
// (release.WithStartUnit). It asks for no authority the try-restart arm
// did not already need, and it cannot fire for a sleeper: only an
// ERRORED restart reaches it, and every arm that can meet a sleeping
// route swallows its failure.
//
// unitsOption hands named units to root when this process is not root
// (issue #315); see agentUnitsOption.
func fastPassOpts(dd string, stageReq func(host, ver string, err error), restart, start func(string) error) []release.PassOption {
	return append([]release.PassOption{
		release.BakeDir(dd),
		release.OnStageBlocked(stageReq),
		release.RestartHibernating(restart),
		release.WithStartUnit(start),
		release.WithDigestRatchet(pointerRatchet(false, release.OptDir(), dd, os.Getenv)),
		// Settings spec §4: the agent applies root's decisions, never
		// resolves. The book is opened per pass, not here: these options are
		// built once per agent run.
		updatesPassOption(dd, false),
	}, agentUnitsOption()...)
}

// agentRunsAsRoot is the agent's euid test, a seam for tests.
var agentRunsAsRoot = func() bool { return os.Geteuid() == 0 }

// agentUnitsOption is release.RootRestartsUnits for an agent that is not
// root (issue #315). carlos-edge.service runs as User=carlos, and systemd
// refuses that user a restart of any unit a route names, so a named-unit
// route goes to the root pass through the stage relay before anything moves.
// An agent run as root restarts the unit itself, as before.
func agentUnitsOption() []release.PassOption {
	if agentRunsAsRoot() {
		return nil
	}
	return []release.PassOption{release.RootRestartsUnits()}
}

// fastMembers computes the fast set for one tick — the routes with a reason
// to be looked at FASTER than the hourly floor:
//
//   - canary/* routes, always: a direct-bucket operator promote
//     (scripts/promote-app.sh, or `carlos promote` against the bucket)
//     never bumps the serial, so the sweep is the only thing keeping the
//     "canary adopts in seconds" promise (spec §4 correction).
//   - unstaged routes (no live symlink), once their backoff window has
//     passed: a route was just minted, or its stage failed; the sweep
//     picks it up as soon as there is something to stage.
//   - routes waiting for a dev proof (D3), each time the proof gate's
//     cached miss runs out: they have a live version, but their head is
//     held until its proof arrives.
//
// Computed BEFORE any store work so an empty set costs zero store reads —
// the idle box pays one serial GET per tick and nothing else. A registry
// list failure returns the empty set (a broken box, logged per tick by
// statusLoop's identical call — not worth a second line every 2s here).
func (d *fastDeps) fastMembers(now time.Time) []string {
	routes, err := d.reg.List()
	if err != nil {
		return nil
	}
	var members []string
	seen := map[string]bool{}
	for _, rt := range routes {
		// release.Adoptable, not a local copy: the sweep must count as
		// outstanding work exactly the routes a pass would consider —
		// anything else (platform aliases, non-ready routes, routes with no
		// app) would be swept forever, since the pass refuses it on shape.
		if !release.Adoptable(rt) || seen[rt.Host] {
			continue
		}
		// A host inside its restart-failure hold is outside the sweep
		// entirely — canary or unstaged. Sweeping it again at this cadence
		// would repoint, cycle the RUNNING old service through a restart
		// that is going to fail, roll back and start it again, every 2-3
		// seconds for as long as the broken head stands (before the
		// rollback-start fix that flap self-terminated after two ticks — by
		// stamping the lie instead). The hourly floors still retry held
		// hosts; a console promote's serial bump still converges them
		// immediately (fullPass is not membership-gated); and a restart that
		// SUCCEEDS clears the hold at once (observeRestart).
		if d.restartHeld(rt.Host, now) {
			continue
		}
		if isCanaryRoute(rt) {
			seen[rt.Host] = true
			members = append(members, rt.Host)
			continue
		}
		// D3 (spec §4.5, codex 14): a listed route waiting for a dev proof
		// stays in the sweep although it has a live version, whenever its
		// 30 s miss has run out, so it adopts soon after the proof arrives
		// rather than at the next serial bump or the hourly floor. A wait the
		// last sweep left standing (fastTick) is inside a backoff window, and
		// sits it out like an unstaged route.
		if d.proof.WaitingDue(rt.Host) {
			if nb, ok := d.notBefore[rt.Host]; ok && now.Before(nb) {
				continue
			}
			seen[rt.Host] = true
			members = append(members, rt.Host)
			continue
		}
		// A LiveVersion error reads as unstaged: the pass is the thing that
		// can repair a broken link, and the backoff bounds the retry cost.
		if v, err := release.LiveVersion(d.optDir, rt.Host); err == nil && v != "" {
			continue // staged: nothing outstanding at this cadence
		}
		if nb, ok := d.notBefore[rt.Host]; ok && now.Before(nb) {
			continue // still inside this host's backoff window
		}
		seen[rt.Host] = true
		members = append(members, rt.Host)
	}
	return members
}

// noteUnstaged records that host came through a sweep still unstaged:
// double its backoff (base 2s, capped at 5min) and push its next look out.
// Held routes cost almost nothing this way: the pass's own bake logic
// refuses them and the backoff keeps re-checks rare; the serial-triggered
// and hourly passes re-notice them when the window expires.
func (d *fastDeps) noteUnstaged(host string, now time.Time) {
	if d.step == nil {
		d.step = map[string]time.Duration{}
		d.notBefore = map[string]time.Time{}
	}
	s := d.step[host]
	if s == 0 {
		s = fastBackoffBase
	} else if s *= 2; s > fastBackoffCap {
		s = fastBackoffCap
	}
	d.step[host] = s
	d.notBefore[host] = now.Add(s)
}

// proofWaitRenewed reports whether host is on the proof gate's waiting list
// with a cached miss still running: the sweep just asked the gate about it
// and the gate held it again.
func (d *fastDeps) proofWaitRenewed(host string) bool {
	if d.proof.WaitingDue(host) {
		return false
	}
	for _, w := range d.proof.Waits() {
		if w.Host == host {
			return true
		}
	}
	return false
}

// backoffFor reports host's current backoff step (zero when none recorded).
func (d *fastDeps) backoffFor(host string) time.Duration { return d.step[host] }

// noteRestartFailed records that host's restart onto ver just failed, and
// holds the host out of the sweep for a growing window: same version failing
// again doubles the step (capped), a DIFFERENT version starts the hold fresh
// at the base — a new promotion is new work, not a continuation of the old
// failure. Keyed per host with the version recorded, rather than per
// (host, version) pairs accumulating: one live symlink per host means one
// version under attempt at a time, and the map stays bounded by the fleet.
func (d *fastDeps) noteRestartFailed(host, ver string, now time.Time) {
	if d.restartHold == nil {
		d.restartHold = map[string]*restartFailure{}
	}
	m := d.restartHold[host]
	if m == nil || m.ver != ver {
		m = &restartFailure{ver: ver, step: fastRestartBackoffBase}
	} else if m.step *= 2; m.step > fastRestartBackoffCap {
		m.step = fastRestartBackoffCap
	}
	m.notBefore = now.Add(m.step)
	d.restartHold[host] = m
}

// restartHeld reports whether host is inside its restart-failure hold.
func (d *fastDeps) restartHeld(host string, now time.Time) bool {
	m := d.restartHold[host]
	return m != nil && now.Before(m.notBefore)
}

// observeRestart wraps the restart dispatch so this loop SEES its outcomes —
// the seam the restart-failure hold feeds on. On an error the version under
// attempt is read from the live symlink (adoptHost repoints BEFORE it
// consults the callback, so at callback time the symlink names the version
// being tried; the read failing degrades the memo to host-keyed, which only
// makes the hold slightly coarser). On success the hold is cleared: the host
// is converged, or a recovery promote just landed, and either way there is
// nothing left to hold it out of the sweep for.
//
// The wrapper never SUPPRESSES a restart — suppression inside the callback
// would return nil to adoptHost, which would stamp SetVersions over a
// process nothing cycled: the exact lie this whole mechanism exists to
// prevent. The hold gates sweep MEMBERSHIP (fastMembers) instead, so a held
// host is simply not repointed at this cadence at all.
//
// Single-goroutine by construction, like every other fastDeps map: the
// callback runs inside PassFilter/Pass inside fastTick, on the loop's own
// goroutine.
func (d *fastDeps) observeRestart(restart func(string) error) func(string) error {
	return func(host string) error {
		err := restart(host)
		if err != nil {
			ver, _ := release.LiveVersion(d.optDir, host)
			d.noteRestartFailed(host, ver, time.Now())
		} else {
			delete(d.restartHold, host)
		}
		return err
	}
}

// convergeAdopted finishes the adoptions this process delegated or never saw.
//
// The agent is sandboxed (carlos-edge.service: ProtectSystem=strict, no /opt
// write), so on a real box a new version is staged, pointed live and stamped
// by a ROOT pass — carlos-stage.service on this loop's own request, or the
// hourly carlos-adopt.service. That pass cannot cycle an exec child
// (rootRestartRoute: the children are this process's), and once it has run,
// this process's own pass finds the host up to date and never reaches the
// restart callback RestartHibernating exists to hand it. The warm child was
// left on the old build until the activator sweep's ConvergeGrace ran out:
// 110-200s per deploy on the flagship, past `carlos deploy`'s own watch
// (discussion #231).
//
// So, for each warm instance the activator reports drifted, cycle it once the
// route is STAMPED with the version its symlink now names. The stamp is the
// adopting pass saying it is finished with the host (adoptHost writes it last,
// after its restart and any rollback); a drift the stamp does not yet agree
// with is an adoption still in flight, and this leg waits a tick for it rather
// than race it. execBounceable keeps the usual line between the two halves: a
// route whose process the root pass's systemctl arms can reach is never this
// leg's to cycle.
//
// At most limit instances are cycled per call, in host order. One pass can
// adopt every warm instance of an app at once — a root pass repoints them all
// inside a few seconds — and stopping them together would drain, checkpoint
// and later restore them together too: every host of the app blips at the
// same moment. The activator sweep spends the same care (one convergence
// cycle per sweep); this leg keeps it at its own cadence, so N instances
// converge over N ticks rather than in one. Only a cycle that actually
// started counts against the limit — a drifted host this leg must wait on
// costs the others nothing.
//
// Never wakes anything — ConvergeAdopted is a stop, and a sleeper execs the
// new build at its own next wake. drifted returning nothing is the ordinary
// tick and costs no registry read. Returns how many it cycled.
func convergeAdopted(drifted func() map[string]string, get func(host string) (*registry.Route, error), converge func(host, ver string) bool, limit int) int {
	stale := drifted()
	if len(stale) == 0 {
		return 0
	}
	hosts := make([]string, 0, len(stale))
	for host := range stale {
		hosts = append(hosts, host)
	}
	sort.Strings(hosts)
	n := 0
	for _, host := range hosts {
		if n >= limit {
			break
		}
		live := stale[host]
		rt, err := get(host)
		if err != nil || rt == nil || rt.Version != live || !execBounceable(*rt) {
			continue
		}
		if converge(host, live) {
			n++
			log.Printf("agent: %s was adopted onto %s by another pass — cycled its warm instance onto it", host, live)
		}
	}
	return n
}

// adoptedCyclesPerTick is convergeAdopted's limit per run of its loop: one
// warm instance every adoptedConvergeEvery. A single-instance app — the common
// case — converges on the first run after its adoption is stamped; an app with
// a dozen warm instances takes under half a minute, one blip at a time.
const adoptedCyclesPerTick = 1

// adoptedConvergeEvery is how often the converge leg looks for a finished
// adoption to act on.
const adoptedConvergeEvery = 2 * time.Second

// convergeLegFor builds the converge leg over the box's activator. nil for a
// box with none armed: there are no exec children to cycle. On a box that
// declined CARLOS_SWAP=overlap but honours failure records (heldHolds), a
// host whose installed build a record holds for it is passed by: cycling it
// would stop it into a wake its Planner refuses (final review F6).
func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() {
	if svc == nil {
		return nil
	}
	get := reg.Get
	if h := heldHolds(); h != nil {
		get = h.skipGet(reg.Get)
	}
	return func() {
		convergeAdopted(svc.Drifted, get, svc.ConvergeAdopted, adoptedCyclesPerTick)
	}
}

// runConvergeLoop runs the converge leg every `every` until stop closes (nil:
// for the life of the process).
//
// Its own loop, not a leg of fastTick, and the reason is measured: on the
// flagship's Ireland box a fast tick ran 25 to 50 seconds whenever the serial
// moved (a full reconcile and adopt pass over ~180 routes), and console
// activity across every tenant moves it about once a minute. Riding the tick,
// a stamped adoption waited ~48s for its cycle (discussion #231, 2026-10-05).
// The leg needs none of the tick's state: it reads the activator and the
// registry, both safe to read beside the tick, and its one write is a stop
// under the activator's own lock — where the tick's adopt bounce is
// version-aware (BounceAdopted), so the two cannot stop one child twice for
// one build.
func runConvergeLoop(leg func(), every time.Duration, stop <-chan struct{}) {
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-stop:
			return
		case <-t.C:
			leg()
		}
	}
}

// slowLegAfter is when logSlow speaks: a fast-loop leg that takes longer than
// this is worth a line, because the loop's promise is seconds.
const slowLegAfter = 5 * time.Second

// logSlow logs that what took longer than slowLegAfter, with detail. Meant
// for a defer: the loop's latency was invisible in the journal until someone
// reconstructed it from timestamps by hand (#231).
func logSlow(what string, start time.Time, detail string) {
	if d := time.Since(start); d > slowLegAfter {
		if detail != "" {
			detail = " (" + detail + ")"
		}
		log.Printf("agent: fast converge: %s took %s%s", what, d.Round(100*time.Millisecond), detail)
	}
}

// fastTick is one fast-converge tick — two independent jobs (spec §3-§4):
//
//  1. Serial check: one ReadSerial. The FIRST successful read only seeds
//     lastSerial — a present object at its value (boot must not replay
//     history; adoptLoop's 1-minute first pass covers boot), an absent one
//     at zero (absence is not history, and the first-ever bump on a fresh
//     bucket must fire, not read as replay). Afterward any CHANGE
//     (not just increments — a reset must not wedge the box) runs the full
//     pass set, clamped to one per fastFullClamp: a bump inside the window
//     is deferred by NOT recording the acted serial, so the next tick past
//     the window re-fires. Read errors are tolerated silently (one log
//     line per state change, never per tick).
//     The restart leg (issue #113) rides that same move, AFTER the pass set:
//     a bump can carry a promote and a restart request at once, and cycling
//     processes before their new version and config land would cycle them onto
//     the old ones. It is a separate call rather than a fourth leg inside
//     fullPass because it is not release convergence — it restarts what is
//     already correct — and because it must still run on a box with no
//     CARLOS_SECRETS_KEY, which fullPass's config leg silently skips. The
//     sidecar wake leg (design: sidecar members) rides the same move
//     immediately after, for the identical reason: a bump can carry a
//     provisioning change and a wake request together, and it must run on
//     a box with no CARLOS_SECRETS_KEY too.
//     The schedule converge (schedule spec §6.2) is a further leg on that same
//     move, for both of the same reasons; see the call site. The layout
//     leg (isolated-exec design §8.2) follows it and only starts a pass in
//     the background; the system leg's observe comes last.
//  2. Fast sweep: fastMembers, and only when non-empty one PassFilter over
//     exactly those hosts — an empty sweep costs zero store reads. After
//     the sweep, members still unstaged back off, and so do members whose
//     proof wait the sweep left standing; other staged members clear their
//     backoff state.
func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
	// The index is observed BEFORE any pass this tick runs, so a pass covers
	// at least what was observed and indexActed can take it afterwards.
	obs, indexOK := d.observeIndex(ctx)
	fullRan := false
	doc, _, err := release.ReadSerial(ctx, d.store)
	switch {
	case err != nil:
		if !d.serialBroken {
			log.Printf("agent: fast converge serial read: %v", err)
			d.serialBroken = true
		}
	case !d.seeded:
		// The first successful read seeds — the ABSENT object included,
		// which ReadSerial reports as the zero doc. Absence is not history:
		// there is nothing to replay, and every deployment's bucket starts
		// without the object, so seeding zero is what lets the first-ever
		// console bump (0 → 1) fire instead of being swallowed as boot
		// noise. A PRESENT object seeds at its current value, and THAT is
		// the no-replay rule: the bumps behind it are history a booting box
		// must not act on (adoptLoop's 1-minute first pass covers boot).
		d.lastSerial, d.seeded = doc.Serial, true
		d.serialBroken = false
	default:
		d.serialBroken = false
		if doc.Serial != d.lastSerial && now.Sub(d.lastFull) >= fastFullClamp {
			d.runFull(ctx)
			fullRan = true
			if d.restartLeg != nil {
				d.restartLeg(ctx)
			}
			if d.wakeLeg != nil {
				d.wakeLeg(ctx)
			}
			// The schedule converge rides the same move, after the pass set for
			// the same reason the restart leg does: a bump can carry a promote
			// and a `carlos schedule set` at once, and converging schedules
			// before the version they belong to has staged would write rows
			// against the outgoing build. Also a separate leg rather than a
			// member of fullPass — it is not release convergence, and it must
			// still run on a box with no CARLOS_SECRETS_KEY.
			if d.scheduleLeg != nil {
				d.scheduleLeg(ctx)
			}
			// The layout pass converges to the fleet's recorded layout on
			// every move, after the pass set has minted and staged whatever
			// the move carried, so a freshly minted row is seen. It only
			// starts the pass: the pass runs on its own goroutine, because
			// it can wait on stops, and nothing above or below may wait on
			// it.
			if d.layoutLeg != nil {
				d.layoutLeg(ctx)
			}
			if d.serialNotify != nil {
				select {
				case d.serialNotify <- struct{}{}:
				default:
				}
			}
			// The platform self-update leg rides the move too, last: a
			// `carlos system update` is a console action like any other, and
			// this is what makes a promote noticed in seconds instead of at
			// the leg's own hourly floor. It only READS the pointer and sets
			// a deadline — the fetch and the request happen in the tick
			// below, after the spread, so a serial move costs one Get.
			if d.systemLeg != nil {
				d.systemLeg.observe(ctx)
			}
			d.lastSerial = doc.Serial
			d.lastFull = now
		}
	}

	// The backstop's full pass (spec §4.4), on this goroutine: requested
	// after each rebuild, run once, under the same clamp.
	if !fullRan && d.fullDue != nil && d.fullDue.Load() && now.Sub(d.lastFull) >= fastFullClamp {
		d.runFull(ctx)
		fullRan = true
		d.lastFull = now
	}
	// An index-only change runs the scoped pass (spec §4.5).
	d.indexTick(ctx, now, obs, indexOK, fullRan)

	// The self-update tick, outside the serial switch and before the sweep's
	// early return: it is what runs a spread deadline down and re-touches an
	// unanswered request, and neither has anything to do with routes. It
	// reads no store unless a deadline has actually come due.
	if d.systemLeg != nil {
		d.systemLeg.tick(ctx, now)
	}

	// D3's mint asker: on a proving box, ask for a root pass every 2 minutes
	// while a listed build has baked with no proof. Elsewhere it reads one
	// absent file and returns.
	if d.mintLeg != nil {
		d.mintLeg(ctx)
	}

	members := d.fastMembers(now)
	if len(members) == 0 {
		return // zero store reads: no PassFilter, no bake-clock write
	}
	defer logSlow("fast sweep", time.Now(), fmt.Sprintf("%d routes", len(members)))
	member := map[string]bool{}
	for _, h := range members {
		member[h] = true
	}
	opts := d.passOpts
	// The canary gate only on a tick whose own index observation succeeded:
	// a tick that could not read the index reads every pointer, as before
	// the index. (A failed poll also leaves the watch unknown, so the gate
	// would find every group due anyway; this keeps it off whatever another
	// caller's poll did in between.)
	if d.canaryGate != nil && indexOK {
		opts = append(slices.Clip(opts), release.WithPointerGate(d.canaryPointerGate(now)))
	}
	if err := release.PassFilter(ctx, d.reg, d.store, d.pub, d.optDir, d.restart,
		func(rt registry.Route) bool { return member[rt.Host] }, opts...); err != nil {
		log.Printf("agent: fast converge sweep: %v", err)
	}
	for _, h := range members {
		switch v, err := release.LiveVersion(d.optDir, h); {
		case err != nil || v == "":
			d.noteUnstaged(h, now)
			// An unstaged route whose wait this sweep renewed is waiting
			// for its proof, not failing to stage: spec §4.5 has it rejoin
			// the sweep each time its 30 s miss runs out, so its backoff
			// stops doubling there.
			if d.proofWaitRenewed(h) && d.step[h] > release.ProofMissTTL {
				d.step[h] = release.ProofMissTTL
				d.notBefore[h] = now.Add(release.ProofMissTTL)
			}
		case d.proof.WaitingDue(h):
			// The sweep neither renewed this host's proof wait (which would
			// cache a fresh 30 s miss) nor cleared it: adoptHost returned
			// before the gate, say on a channel with no pointer. Nothing will
			// change that by the next tick, so it backs off like an unstaged
			// route instead of being swept every 2 s (plan 3 ruling 5).
			d.noteUnstaged(h, now)
		default:
			delete(d.notBefore, h)
			delete(d.step, h)
		}
	}
}

// canaryPointerGate is the sweep's release.PointerGate at now. Only canary
// groups are gated; every other group is read as before. A canary group is
// skipped when its app scope has not moved since the group's own last good
// read (another channel's read of the same app does not count), that read is
// younger than the backstop, and every route of the group is live on the
// head it returned: then a read could only repeat what the box already
// runs. Anything unsettled (an unstaged route, a failed restart rolled back,
// a proof wait, a build the update settings chose over the head) keeps
// today's read on every tick.
func (d *fastDeps) canaryPointerGate(now time.Time) release.PointerGate {
	return func(acct, app, channel string, routes []registry.Route) (bool, func(*release.ChannelPointer, error)) {
		if !strings.HasPrefix(channel, "canary/") {
			return false, nil
		}
		scope, gated := release.ChangeScope(release.ChannelKey(acct, app, channel))
		if !gated {
			return false, nil
		}
		k := [3]string{acct, app, channel}
		gen, due := d.canaryGate.Due(scope, now)
		if !due && d.canarySettled(k, routes, gen, now) {
			return true, nil
		}
		return false, func(ptr *release.ChannelPointer, err error) {
			if d.canaryHeads == nil {
				d.canaryHeads = map[[3]string]canaryRead{}
			}
			if err != nil || ptr == nil {
				delete(d.canaryHeads, k)
				d.canaryGate.Mark(scope, gen, now, false)
				return
			}
			d.canaryHeads[k] = canaryRead{head: ptr.Version, gen: gen, at: now}
			d.canaryGate.Mark(scope, gen, now, true)
		}
	}
}

// canarySettled reports whether group k's own last good read is at gen (the
// generation Due returned) and younger than the gate's maximum age, and every
// route of the group is live on the head that read returned.
func (d *fastDeps) canarySettled(k [3]string, routes []registry.Route, gen uint64, now time.Time) bool {
	r, ok := d.canaryHeads[k]
	if !ok || r.head == "" || len(routes) == 0 || r.gen != gen || now.Sub(r.at) >= d.canaryGate.MaxAge() {
		return false
	}
	for _, rt := range routes {
		if v, err := release.LiveVersion(d.optDir, rt.Host); err != nil || v != r.head {
			return false
		}
	}
	return true
}

// runFull runs the full pass set's fullPass, taking any backstop request
// first: a full pass that starts after the backstop's rebuild satisfies it.
func (d *fastDeps) runFull(ctx context.Context) {
	if d.fullDue != nil {
		d.fullDue.Store(false)
	}
	d.fullPass(ctx)
}

// observeIndex polls the change index and returns its observed generations;
// ok is false when no watch is armed or the index is unusable (then the
// serial and the floors carry the tick, as before the index). A failure is
// logged once per state change, never per tick.
func (d *fastDeps) observeIndex(ctx context.Context) (map[string]uint64, bool) {
	if d.watch == nil {
		return nil, false
	}
	// Poll's moved list is advisory (shared watch): only Observed counts.
	// Bounded: the tick runs on a background context, and a hung index GET
	// would otherwise stall the tick and hold the watch's poll lock that the
	// edge's loops share. Running out is "unknown", like any other error.
	pctx, cancel := context.WithTimeout(ctx, release.ChangeIndexPollTimeout)
	_, err := d.watch.Poll(pctx)
	cancel()
	if err != nil {
		if !d.indexBroken {
			log.Printf("agent: fast converge change index: %v (scoped passes wait; the serial and the floors carry on)", err)
			d.indexBroken = true
		}
		return nil, false
	}
	obs, ok := d.watch.Observed()
	if ok && d.indexBroken {
		log.Printf("agent: fast converge change index readable again")
		d.indexBroken = false
	}
	return obs, ok
}

// indexTick acts on an index observation: the first one seeds indexActed
// without a pass (boot must not replay history; the backstop's first full
// pass and the loops' first passes cover boot), a full pass this tick
// covers everything observed, and otherwise app and account scopes that
// moved since indexActed run one scoped pass, clamped like the full pass:
// a change inside the clamp leaves indexActed as it was, so the next tick
// past the window acts on it. indexActed moves on only after a pass ran.
func (d *fastDeps) indexTick(ctx context.Context, now time.Time, obs map[string]uint64, ok, fullRan bool) {
	if !ok {
		return
	}
	if d.indexActed == nil || fullRan {
		d.indexActed = obs
		return
	}
	moved := indexScopes(d.indexActed, obs)
	if len(moved) == 0 {
		d.indexActed = obs // control/ scopes only: not this loop's business
		return
	}
	if now.Sub(d.lastFull) < fastFullClamp || d.scopedPass == nil {
		return
	}
	d.scopedPass(ctx, moved)
	d.indexActed = obs
	d.lastFull = now
}

// fastConvergeLoop replaces the canary-only fast adopt loop with the
// carlos-deploy fast set (spec §3-§4): a 2s serial poll that turns a console
// bump (promote, instance create, config write) into the ordinary verified
// pass set — reconcile → adopt → exec config sync — within seconds, plus a
// fast sweep of the routes with genuinely outstanding work (canary/* always;
// unstaged routes until they drain, with per-host backoff). Everything it
// triggers is the same signature-verified, bake-gated work the hourly loops
// run; the serial carries no authority, only earliness.
//
// Setup mirrors adoptLoop's: missing deployment config or pubkey disables
// the loop silently (adoptLoop already logged the reason). The config-sync
// leg degrades exactly as configSyncLoop does — no CARLOS_SECRETS_KEY, or a
// broken one, and the full pass set simply omits that leg (configSyncLoop
// already logged that too). Takes the activator for the same reason
// adoptLoop does: a fresh version on a warm exec instance must bounce onto
// its new binary, not wait for hibernation. Nil is fine.
//
// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
// whole decision lives in fastTick/fastMembers, which are tested.
func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {
	ctx := context.Background()
	store, err := openDeployStore(ctx)
	if err != nil {
		declinedHoldsStay("no deploy store")
		return // adoptLoop already logged this
	}
	pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		declinedHoldsStay("no release key")
		return // ditto
	}
	var priv *ecdh.PrivateKey
	if raw := os.Getenv("CARLOS_SECRETS_KEY"); raw != "" {
		if p, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY"); err == nil {
			priv = p
		}
	}
	dd := dataDir()
	optDir := release.OptDir()
	// d is built empty first so the restart dispatch can be wrapped in its
	// observer (observeRestart): a failed restart holds its host out of the
	// sweep instead of the sweep re-cycling the service every tick. The
	// remaining fields are filled in right below.
	d := &fastDeps{reg: reg, store: store, pub: pub, optDir: optDir, serialNotify: serialNotify}
	restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, agentCheckedUnitRestart(systemdUnits{}, optDir, os.Getenv, nil), adoptBounceFn(svc)))
	// Built once, used by both pass shapes below — see fastPassOpts for the
	// DO-NOT list and why the stage requester is per-loop, not per-tick.
	opts := append(fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit)), release.WithHostWrites(writes))
	// The rollback's restart (D1): checked like the forward one, and a replace
	// rather than a start, so a rejected build left up is cycled out.
	opts = append(opts, release.WithRollbackRestart(agentRollbackUnitRestart(reg, systemdUnits{}, optDir, os.Getenv)))
	// D3: the process's one proof gate rides every pass this loop runs, and
	// fastMembers reads its waiting list. The mint asker has its own
	// requester, so its 2-minute asks never spend the stage budget and its
	// journal line does not claim a stage.
	opts = append(opts, release.WithProofGate(agentProofGate()))
	d.proof = agentProofGate()
	d.mintLeg = (&mintAsker{reg: reg, store: store, sys: systemdUnits{}, optDir: optDir,
		policyPath: release.ProofPolicyPath, request: mintRequester(dd), now: time.Now}).tick
	// The restart check's memory (issue #113, narrowed requests added by
	// discussion #16), seeded ONCE here — two reads per app at arming (the
	// app-wide generation and the narrowed request log) and never again except
	// on a serial move, which is what keeps an idle box at exactly one serial
	// GET per tick. See restartSeed for why the baseline cannot wait for the
	// first request to establish itself.
	restartMem := newRestartMemory()
	// `carlos instances retry` (zero-downtime spec §8.4) reaches the rollout
	// manager through the host-retry log (a third read per app at arming,
	// and on a move, only where retryHost is set; hostRetryPass) or a legacy
	// entry in the narrowed-restart log, or, on a box that declined
	// the flag but honours its failure book, that book (Task 10); nil
	// otherwise. Each entry's grant is verified against the root key first
	// (spec §11).
	restartMem.pub = pub
	restartMem.retryHost = rolloutRetryHost()
	// The declined box's pointer pass (Task 10, ruling 194): `carlos promote
	// --retry` clears a record, a signed rollback a marker, read through
	// the same verified pointer read the armed scheduler uses, at most
	// every rollout.PointerReadEvery.
	if h := heldHolds(); h != nil {
		go h.pointerLoop(ctx, func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error) {
			return release.Channel(ctx, store, pub, acct, app, channel)
		})
	}
	restartSeed(ctx, reg, store, dd, restartMem)
	restartBounce := bounceFn(svc)
	restartWake := wakeFn(svc)
	// The sidecar wake leg's own memory, seeded the same way and for the
	// same reason: one silent read per served sidecar at arming so the
	// FIRST wake request after a boot is acted on rather than swallowed as
	// the baseline (sidecarWakeSeed's own doc comment).
	sidecarWakeMem := newSidecarWakeMemory(writes.Enabled)
	sidecarWakeSeed(ctx, reg, store, sidecarWakeMem)
	sidecarWake := wakeFn(svc)
	// The moves hold view, built before the converge below: that converge
	// runs before the first moves pass, and an unbuilt view freezes every
	// host (moves spec §9.4). Nothing with the Phase 1 switch off.
	refreshMovesHolds(ctx, store)
	// The offline keep view, likewise before any reconcile pass: once it is
	// read, keep answers false for every host with no offline move, so the
	// first pass removes what it always has (movesoffline.go).
	refreshOfflineView(ctx, store, armedOfflineView())
	// carlos run's boot pass (run spec §5.2, index item 2): reconcile's
	// first pass is a minute away and the CLI expires a run at 60 s.
	runRunsBootPass(ctx)
	// The schedule converge's own seed, on restartSeed's precedent and for the
	// same shape of reason: the leg below runs only on a serial MOVE, so
	// without one pass here a box that booted after the bump would carry no
	// timer rows (and report no status) until the hourly floor in adoptLoop —
	// up to an hour of a schedule silently not running. It also seeds the
	// converger's etag memory, which is what makes a run signal older than this
	// process read as history rather than as a fresh request to run.
	if err := sd.converge(ctx); err != nil {
		log.Printf("agent: schedule converge (start): %v", err)
	}
	d.restart = restart
	d.passOpts = opts
	// The scheduler's pass with CARLOS_SWAP=overlap armed, nothing while
	// managed starts are refused, today's convergeAdopted otherwise.
	if leg := fastConvergeLeg(svc, reg); leg != nil {
		go runConvergeLoop(leg, adoptedConvergeEvery, nil)
	}
	// The serial-triggered full pass set, in spec order: mint routes
	// (reconcile), then reconcile aliases, then stage them (adopt), then deliver
	// their config. Each error is one log line, never fatal — the hourly floors
	// stand underneath every leg. Not membership-gated by the restart hold:
	// a serial bump is an explicit console action, bounded by whoever keeps
	// making it — and it is what lets a recovery promote through the console
	// converge immediately even mid-hold.
	legs := &passLegs{store: store, reg: reg, pub: pub, optDir: optDir, restart: restart, opts: opts,
		priv: priv, dd: dd, svc: svc, writes: writes, bs: bs,
		cfg: func() provision.ReconcileConfig {
			return configuredClaimProvision(provisionTickConfig(os.Getenv), configure)
		}}
	d.fullPass = legs.full
	// The change index (S3 change-index design §4.5): an index-only change
	// runs the scoped pass, and the backstop's request runs the full one,
	// both on this loop's goroutine. Nil watch: the tick is as before.
	if watch := armedChangeWatch.Load(); watch != nil {
		d.watch = watch
		d.scopedPass = legs.scoped
		d.fullDue = &backstopFullDue
		// The sweep's canary pointer reads are gated on the same watch
		// (spec §5). Built here, after RunAgent armed the index.
		d.canaryGate = newChangeGate()
	}
	// The restart leg: this process's own children bounced here, everything
	// unit-shaped relayed to a root pass. Deliberately NOT inside fullPass
	// above — see fastTick's comment for both reasons.
	d.restartLeg = func(ctx context.Context) {
		routes, err := reg.List()
		if err != nil {
			log.Printf("agent: restart check: list routes: %v", err)
			return
		}
		restartRequestLeg(ctx, routes, store, dd, restartBounce, restartWake, restartMem)
	}
	// The sidecar wake leg: apps/<acct>/<app>/sidecar-wake/<name>.json
	// (release.SidecarWakeKey) compared and dispatched here, riding the
	// same serial move as the restart leg immediately above it — see
	// fastTick's comment for why both must run after fullPass and on a
	// box with no CARLOS_SECRETS_KEY.
	d.wakeLeg = func(ctx context.Context) {
		routes, err := reg.List()
		if err != nil {
			log.Printf("agent: sidecar wake check: list routes: %v", err)
			return
		}
		sidecarWakeLeg(ctx, routes, store, sidecarWakeMem, sidecarWake)
	}
	// The schedule leg: converge this box's timer rows from the apps'
	// declarations. Nil when the box has no scheduled work armed, which
	// fastTick tolerates.
	d.scheduleLeg = scheduleLegFor(sd)
	// The layout leg: this box's label is the one its status publishes
	// under, which is the label an operator scopes a layout request to.
	layoutBox := metricsBoxLabel(os.Getenv, readMachineID())
	layoutPasses := newLayoutLeg(ctx, func(ctx context.Context) bool { return runLayoutLeg(ctx, store, layoutBox) })
	d.layoutLeg = layoutPasses.kick
	// The platform self-update leg (spec §4): the sandboxed half, which only
	// ever fetches into /data and writes a request. It lives here rather than
	// in a loop of its own because the spread deadline needs a seconds
	// cadence to run down on, and this is the loop that already has one.
	d.systemLeg = newSystemLeg(store, pub, os.Getenv, dd, optDir)
	// One observe at arming, on restartSeed's precedent: the leg below only
	// re-observes on a serial move or at its hourly floor, so a box that
	// (re)starts under an already-promoted pointer would otherwise sit on the
	// old build for up to an hour with nobody to bump the serial. observe
	// handles "no pointer" itself, which is every deployment before its first
	// `carlos system update`.
	d.systemLeg.observe(ctx)
	log.Printf("agent: fast converge loop armed (2s + jitter: serial poll + canary/* + unstaged routes + system pointer)")
	for {
		time.Sleep(fastTickEvery + rand.N(time.Second))
		fastTick(ctx, d, time.Now())
	}
}

// isCanaryRoute reports whether rt follows a canary/* channel — the same
// two-segment convention bake.go's DefaultHolds wildcard matches (a bare
// "canary" with no slash is NOT covered, same as bake.go: it falls through
// to the unknown-channel default there, so it shouldn't fast-poll here
// either).
func isCanaryRoute(rt registry.Route) bool {
	return strings.HasPrefix(rt.Channel, "canary/")
}

// countExpiredCanaries counts routes (this box's own, platform aliases
// excluded — same convention as CollectBox) on a canary/* channel whose
// pointer has passed release.CanaryTTL, for the fleet status page's §5
// count (internal/status Probes.ExpiredCanaries). A canary route whose
// pointer can't be resolved at all — store unavailable, pubkey unusable, or
// the specific channel won't verify — nils the WHOLE count rather than
// undercounting: "we couldn't check" and "there are none" are different
// claims, and this function only ever makes the second one when it's sure.
// No canary routes at all is a confident, known 0 — it needs no store to
// answer "none" about routes that don't exist.
//
// Pointers are resolved at most once per distinct (account, app, channel)
// among the routes given (the local `cache` map below), mirroring adopt.go's
// Pass — this function is called once per 5-minute status tick
// (statusLoop), so N routes on this box following the same canary channel
// cost exactly ONE release.Channel() store read for that tick, not N
// (2026-08-03 review: confirmed and pinned by
// TestCountExpiredCanariesReadsEachChannelOnce). The cache is local to one
// call/one tick — nothing persists across ticks, which is fine at a
// 5-minute cadence.
func countExpiredCanaries(ctx context.Context, routes []registry.Route, store release.Store, pub ed25519.PublicKey, now time.Time) *int {
	type key struct{ acct, app, channel string }
	cache := map[key]*release.ChannelPointer{}
	n := 0
	for _, rt := range routes {
		if rt.Platform || !isCanaryRoute(rt) {
			continue
		}
		if store == nil || len(pub) != ed25519.PublicKeySize {
			return nil // a canary route exists but nothing here can resolve its pointer
		}
		k := key{release.RouteAccount(rt), rt.App, rt.Channel}
		ptr, cached := cache[k]
		if !cached {
			p, err := release.Channel(ctx, store, pub, k.acct, k.app, k.channel)
			if err != nil {
				return nil // this route's own pointer won't verify — unknown, not zero
			}
			ptr = p
			cache[k] = ptr
		}
		if release.CanaryExpired(ptr, now) {
			n++
		}
	}
	return &n
}

// peekAcmeCache pre-parses the agent args for -acme-cache the same way
// peekRegistryPath peeks -registry: the edge does the authoritative parse,
// this just lets the domains loop stamp cert expiry from the SAME cache dir
// the edge actually mints into when the operator moved it off the default.
// "" = flag not given; domainsTickConfig then derives the edge's own default
// ($CARLOS_DATA_DIR/acme).
func peekAcmeCache(args []string) string {
	fs := flag.NewFlagSet("agent", flag.ContinueOnError)
	fs.SetOutput(io.Discard)
	fs.String("registry", "", "")
	acme := fs.String("acme-cache", "", "")
	fs.String("https", "", "")
	fs.String("http", "", "")
	fs.String("fallback", "", "")
	fs.Bool("dev", false, "")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)
	return *acme
}

// domainsTickConfig builds one reconcile pass's Config from the environment
// — pure (getenv + the already-read public IP injected) so the whole env
// contract is testable without t.Setenv. PLATFORM_APPS_DOMAIN unset returns
// the zero Config, on which Reconcile is inert by contract.
//
// The SECOND return is the parsed PLATFORM_EDGE_IPS list (with the public-IP
// fallback applied). It is not a Config field: Config.EdgeIPs is now a func
// answering the deployment's LIVE edge set, which needs a deployment store —
// something this function must never touch if it is to stay pure and
// getenv-injectable. domainsLoop folds the two together once its store is
// open. A field only the agent's own wiring reads would be dead weight in
// every other caller and every test fixture, hence a return value.
//
// Resolve is armed for every configured deployment now, no longer gated on
// this box knowing an edge IP: a box with no PLATFORM_EDGE_IPS whose FOLD
// supplies edges must still stamp. The stand-down when nothing is knowable
// moved to where the set is actually known — domains.Reconcile's section 4,
// per pass — because "no edge addresses" is now a per-pass fact, not a
// boot-time one.
//
// The error return is ADVISORY, never fatal: it is non-nil only when
// CARLOS_RELEASE_PUBKEY was SET but would not parse, and it exists so the
// caller can log the real cause. release.ParsePublicKey says the same sentence
// for an empty value as for a typo'd one, so an unset key is reported as nil
// error and a nil ReleasePub — the two are different operator problems and the
// one log line about them must not name the wrong one. The Config is fully
// usable either way; only the auto-purge is off.
func domainsTickConfig(getenv func(string) string, publicIP string) (domains.Config, []string, error) {
	domain := strings.TrimSpace(getenv("PLATFORM_APPS_DOMAIN"))
	if domain == "" {
		return domains.Config{}, nil, nil
	}
	var ips []string
	for _, ip := range strings.Split(getenv("PLATFORM_EDGE_IPS"), ",") {
		if ip = strings.TrimSpace(ip); ip != "" {
			ips = append(ips, ip)
		}
	}
	if len(ips) == 0 && publicIP != "" {
		ips = []string{publicIP}
	}
	dd := getenv("CARLOS_DATA_DIR")
	if dd == "" {
		dd = "/data"
	}
	cfg := domains.Config{
		AppsDomain: domain,
		// The same default the edge's -acme-cache flag carries
		// (edge.RunWithActivator): $CARLOS_DATA_DIR/acme. domainsLoop
		// overrides this when the flag was given explicitly.
		AcmeDir: filepath.Join(dd, "acme"),
		Now:     time.Now,
		Resolve: resolveHost,
	}
	// EdgeIPs is deliberately left nil here — a Config out of this function
	// alone stamps nothing. domainsLoop installs the fold once it has a
	// store; see the doc comment.

	// The root release public key, same source and same parser as adoptLoop's.
	// Absence is TOLERATED and leaves ReleasePub nil, which switches the
	// auto-purge off: a box that cannot verify an app's production flag must
	// not destroy its objects. Everything else about the pass — including
	// hiding trashed apps, the reversible half — carries on regardless. A key
	// that is SET but unparseable is tolerated the same way, and RETURNED, so
	// the operator who typo'd it hears about the typo rather than being told
	// the variable is unset.
	if raw := strings.TrimSpace(getenv("CARLOS_RELEASE_PUBKEY")); raw != "" {
		pub, err := release.ParsePublicKey(raw, "CARLOS_RELEASE_PUBKEY")
		if err != nil {
			return cfg, ips, err
		}
		cfg.ReleasePub = pub
	}
	return cfg, ips, nil
}

// provisionTickConfig builds one instances reconcile pass's Config from the
// environment — pure (getenv-injected), mirroring domainsTickConfig's shape.
// Unlike domainsTickConfig, there is no inert zero-Config case: an unset
// CARLOS_REGION/CARLOS_POOLS is a LEGAL state, not a "feature not
// configured" one — see provision.Reconcile's doc comment — a box that
// hasn't been told its region/pools still serves only ""-scoped records,
// never someone else's residency promise. So this always returns a
// populated Config, and CARLOS_DATA_DIR is honored (default "/data")
// regardless of whether region/pools are set. CARLOS_INBOUND (design §8) is
// the same shape again: unset/"any" is today's behaviour exactly, and it is
// a declaration about what work THIS BOX will accept, independent of a
// record's own "no-inbound" placement request — see ReconcileConfig.Inbound.
func provisionTickConfig(getenv func(string) string) provision.ReconcileConfig {
	region := strings.TrimSpace(getenv("CARLOS_REGION"))
	var pools []string
	for _, p := range strings.Split(getenv("CARLOS_POOLS"), ",") {
		if p = strings.TrimSpace(p); p != "" {
			pools = append(pools, p)
		}
	}
	dd := getenv("CARLOS_DATA_DIR")
	if dd == "" {
		dd = "/data"
	}
	// CARLOS_INBOUND is the same apps-domain-style inert gate as
	// CARLOS_REGION/CARLOS_POOLS above: unset (or "any") is today's
	// behaviour exactly, a legal "no constraint" state, never a "feature not
	// configured" one. Unlike Region above it, this IS folded and validated
	// here (provision.NormalizeInbound) rather than left a pure passthrough:
	// review fix, Task 10 finding — an un-folded "None"/"NONE" or a typo'd
	// value compared exact-equal against InboundNone false, so a box meant
	// to be no-inbound would silently mint routed instances it cannot serve.
	// NormalizeInbound folds case/whitespace and maps any unrecognised value
	// to the restrictive reading (with its own log line), so
	// provision.Reconcile/ReconcileSidecars downstream see only the closed
	// vocabulary and their exact-equality comparisons stay correct.
	inbound := provision.NormalizeInbound(getenv("CARLOS_INBOUND"))
	cfg := provision.ReconcileConfig{
		Region:  region,
		Pools:   pools,
		DataDir: dd,
		Now:     time.Now,
		Inbound: inbound,
		// The armed config guard, asked before a mint (amadan #213). Nil
		// when none is armed (every test that does not arm one).
		ConfigGate: configGate(),
	}
	// The root release public key, same source and same parser as adoptLoop's
	// and domainsTickConfig's. It guards ONE decision in this pass: whether a
	// trash marker may retire a production-flagged app's routes (#340).
	//
	// Absence is tolerated and leaves ReleasePub nil, which retires trashed
	// routes UNGUARDED — deliberately the same end state domains.Config
	// documents and accepts for the same decision (§1c), because two passes
	// reaching opposite conclusions about one app is worse than either
	// answer. A key SET but unparseable lands in the same place, so it is
	// logged here rather than swallowed: it is a persistent misconfiguration,
	// and the operator who typo'd it must hear about the typo rather than
	// silently losing the guard.
	if raw := strings.TrimSpace(getenv("CARLOS_RELEASE_PUBKEY")); raw != "" {
		pub, err := release.ParsePublicKey(raw, "CARLOS_RELEASE_PUBKEY")
		if err != nil {
			log.Printf("agent: instances pass: %v — a trash marker will retire routes without checking the production flag", err)
		} else {
			cfg.ReleasePub = pub
		}
	}
	return cfg
}

// resolveHost is the production DNS seam: the default resolver with a hard
// 3s timeout, so one dead nameserver can't stall a whole reconcile pass.
func resolveHost(host string) ([]string, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	return net.DefaultResolver.LookupHost(ctx, host)
}

// imdsToken fetches an EC2 IMDSv2 session token (60s TTL) — the shared
// first step of every IMDS read in this file. Any failure (not on EC2,
// IMDS disabled, a transient metadata-endpoint hiccup) returns "", and
// every caller already treats a blank token as "no answer", never an
// error.
func imdsToken(client *http.Client) string {
	req, err := http.NewRequest(http.MethodPut, "http://169.254.169.254/latest/api/token", nil)
	if err != nil {
		return ""
	}
	req.Header.Set("X-aws-ec2-metadata-token-ttl-seconds", "60")
	resp, err := client.Do(req)
	if err != nil {
		return ""
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return ""
	}
	body, _ := io.ReadAll(resp.Body)
	return strings.TrimSpace(string(body))
}

// imdsLeaf reads one IMDS leaf under the given token, trimmed. The status
// code travels back with the body so a caller can tell "empty leaf" from
// "leaf did not answer 200" without a second round trip.
func imdsLeaf(client *http.Client, token, path string) (string, int) {
	req, err := http.NewRequest(http.MethodGet, "http://169.254.169.254"+path, nil)
	if err != nil {
		return "", 0
	}
	req.Header.Set("X-aws-ec2-metadata-token", token)
	resp, err := client.Do(req)
	if err != nil {
		return "", 0
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	return strings.TrimSpace(string(body)), resp.StatusCode
}

// readPublicIP best-effort reads this box's public IPv4 from EC2 IMDSv2
// (token PUT, then the public-ipv4 leaf). Any failure — not on EC2, IMDS
// disabled, no public address — returns "", and the caller treats that as
// "unknown", never as an error: a box without a discoverable public IP is a
// valid deployment whose DNS checks simply stand down.
func readPublicIP() string {
	client := &http.Client{Timeout: 2 * time.Second}
	token := imdsToken(client)
	if token == "" {
		return ""
	}
	ip, status := imdsLeaf(client, token, "/latest/meta-data/public-ipv4")
	if status != http.StatusOK {
		return ""
	}
	if net.ParseIP(ip) == nil {
		return ""
	}
	return ip
}

// edgeDoorOptions is the edge-local door's option (#175 P7): none unless
// CARLOS_EDGE_DOOR names a socket path, and then edge.WithEdgeDoor with the
// box's public address. publicAddr is a func so the address is read only for
// a box that asked for the door: off EC2, readPublicIP can block for its
// whole timeout. An empty address still passes through, and the edge logs
// that the door is off for want of one.
func edgeDoorOptions(getenv func(string) string, publicAddr func() string) []edge.RunOption {
	p := strings.TrimSpace(getenv("CARLOS_EDGE_DOOR"))
	if p == "" {
		return nil
	}
	return []edge.RunOption{edge.WithEdgeDoor(p, publicAddr())}
}

// ReadPublicIP is readPublicIP for callers outside this package — the relay
// advertises the box's Elastic IP as its relay address and learns it exactly
// as the edge does (relay spec §3, "Listeners"). Same contract: "" means
// unknown, never an error.
func ReadPublicIP() string { return readPublicIP() }

// readPlacementRegion best-effort reads this box's AWS region from EC2
// IMDSv2 (token PUT, then the placement/region leaf) — the same idiom and
// the same "" on any failure contract as readPublicIP: not on EC2, IMDS
// disabled, or a metadata-endpoint hiccup all read as "unknown" here, and
// the caller (edgesFileRegion) treats that identically to an unset
// CARLOS_AWS_REGION override — an edge with neither is simply unarmed for
// steering, not an error.
func readPlacementRegion() string {
	client := &http.Client{Timeout: 2 * time.Second}
	token := imdsToken(client)
	if token == "" {
		return ""
	}
	region, status := imdsLeaf(client, token, "/latest/meta-data/placement/region")
	if status != http.StatusOK {
		return ""
	}
	return region
}

// ReadPlacementRegion is readPlacementRegion for callers outside this
// package. The relay names its region in every heartbeat and checks it
// against each app's region list, and it must be the SAME string the edges
// file uses — hence the same probe, with the caller applying the same
// CARLOS_AWS_REGION override edgesFileRegion applies.
func ReadPlacementRegion() string { return readPlacementRegion() }

// domainsLoop runs the domains reconcile pass forever on the STATUS tick's
// cadence (5min + jitter), not the adopt loop's hour: the console's
// custom-domain guidance promises "Checks run every few minutes", and a DNS
// fix that takes up to an hour to turn green would read as broken. Thin and
// untested by convention (like adoptLoop/statusLoop) — everything with
// behavior lives in domainsTickConfig and domains.Reconcile. Missing config
// disables the loop with one log line, the same inert-until-configured
// shape as its siblings.
func domainsLoop(reg *registry.Registry, acmeCacheFlag string, hostClaim bool) {
	ctx := context.Background()
	// The inert gate is checked BEFORE readPublicIP: an unconfigured
	// deployment must make no IMDS call at all (a 2s link-local probe on
	// every agent start, on boxes that aren't even on EC2, for a feature
	// that is switched off). Inert means inert — no network, not just no
	// writes. domainsTickConfig re-checks the same variable; that's the
	// authoritative gate, this is the "don't pay for it first" one.
	if strings.TrimSpace(os.Getenv("PLATFORM_APPS_DOMAIN")) == "" {
		log.Printf("agent: domains loop disabled: PLATFORM_APPS_DOMAIN is not set")
		return
	}
	cfg, envIPs, keyErr := domainsTickConfig(os.Getenv, readPublicIP())
	cfg.HostClaim = hostClaim
	if cfg.AppsDomain == "" {
		log.Printf("agent: domains loop disabled: PLATFORM_APPS_DOMAIN is not set")
		return
	}
	if acmeCacheFlag != "" {
		cfg.AcmeDir = acmeCacheFlag
	}
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: domains loop disabled: %v", err)
		return
	}
	// The edge set is folded PER PASS, here rather than in
	// domainsTickConfig, because it needs the store this line above just
	// opened — and because a set read once at arm is the bug this replaces:
	// under latency steering a correctly-pointed customer domain resolves to
	// whichever region's edge is nearest, which a single box's boot-frozen
	// PLATFORM_EDGE_IPS reads as "elsewhere". envIPs stays in the union, so
	// a deployment whose bucket is unreadable degrades to exactly the old
	// behaviour and never below it. The bound matters: a hung store must
	// cost this pass 30s, not the loop forever.
	cfg.EdgeIPs = func() []string {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		return edgesfile.EdgeIPs(ctx, store, envIPs, time.Now())
	}
	log.Printf("agent: domains loop armed (5min + jitter), apps domain %s, configured edge IPs %v (the live set is folded from every box's edges file each pass) — health stamps every tick; claims reconcile only while this box holds the housekeeper lease", cfg.AppsDomain, envIPs)
	// One line, and it must name the REAL cause: a typo'd key and an absent
	// one are different problems with different fixes, and this is the only
	// place an operator learns that purging is not happening.
	if keyErr != nil {
		log.Printf("agent: domains loop auto-purge disabled: %v — trashed apps will be hidden but never destroyed", keyErr)
	} else if cfg.ReleasePub == nil {
		log.Printf("agent: domains loop auto-purge disabled: CARLOS_RELEASE_PUBKEY is not set — trashed apps will be hidden but never destroyed")
	}
	duty := domainsDuty(func(ctx context.Context) error {
		return domains.Reconcile(ctx, store, reg, cfg)
	})
	fleetBox := shouldSkipPublisherElection(os.Getenv)
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = 5*time.Minute + rand.N(time.Minute)
		// The deployment-global half, gated; then the per-box half on EVERY
		// platform box (#303): DNS and TLS health stamp this box's own
		// registry from this box's own view, no lease involved, ONCE per
		// tick — after the duties so a host section 3 just minted is
		// stamped in the same tick it appeared. Never on a fleet box: it
		// never contends for the lease for the same reason (housekeeperNever),
		// and it publishes no edges file, so it has no standing to judge the
		// deployment's DNS. (The old reason — that its EdgeIPs fall back to
		// its own address — no longer holds now that the set is folded from
		// every box's file; the exclusion stands on the publication rule
		// instead, which is the durable one.)
		runHousekeeperDuties(housekeeperGate, duty)
		if !fleetBox {
			if err := domains.StampHealth(ctx, reg, cfg, nil); err != nil {
				log.Printf("agent: domains health: %v", err)
			}
		}
	}
}

// domainsDuty wraps one domains reconcile pass as a housekeeper duty. The
// pass is deployment-wide state — materializing routes from claims, minting
// platform-domain and canary hosts, purging trashed apps — with one correct
// writer, and since multi-region (spec 2026-08-20) every platform box in a
// deployment carries PLATFORM_APPS_DOMAIN (the per-account wildcards and the
// steering probe need it), so an ungated loop would be N reconcilers racing
// over the same records. Gating on the lease is what makes the env safe to
// set fleet-wide. (The DNS/TLS health stamps are NOT in this pass: they are
// per-box, domains.StampHealth, run by domainsLoop on every box — #303.) Single-box
// deployments are unchanged: the one box always holds the lease. A box that
// never contends (a fleet box) never runs the pass — which is right, the
// pass is a platform-box job.
func domainsDuty(pass func(context.Context) error) housekeeperDuty {
	return housekeeperDuty{"domains reconcile", func(ctx context.Context) error {
		if pass == nil {
			return nil
		}
		return pass(ctx)
	}}
}

// instancesLoop runs the instance-provisioning reconcile pass forever on the
// same cadence as domainsLoop (5min + jitter, first tick after 1 minute) —
// turning bucket instance records into registry routes for THIS agent's
// region/pool scope (provision.Reconcile). Unlike domainsLoop there is NO
// apps-domain-style inert gate: an unset CARLOS_REGION/CARLOS_POOLS is a
// legal, still-armed state (see provisionTickConfig and
// provision.Reconcile's doc comment — a box with no region deliberately
// still serves ""-scoped records), so this loop arms whenever the deploy
// store opens, full stop. Missing deployment config disables the loop with
// one log line, the same "inert until configured" shape adoptLoop and
// domainsLoop use for their own missing config. Per-pass errors are logged;
// the loop lives on.
//
// Also runs ReconcileFleetBindings (customer-fleets plan, Task 6) on the same
// cadence, platform boxes only: turns every fleet's published
// fleets/*/bindings/*.json object into this platform box's own
// registry.Route.FleetBox — see fleetbind.go's doc comment for the
// fleet-placement/corrupt-object contract. Deliberately homed HERE rather
// than domainsLoop (an earlier pass rode domainsLoop's ticker, since it
// already had a store+reg in scope) — that coupled binding materialization
// to PLATFORM_APPS_DOMAIN being set, which has nothing to do with fleets;
// this loop has no such gate and shares the same store already opened above.
func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {
	ctx := context.Background()
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: instances loop disabled: %v", err)
		return
	}
	cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)
	backfill := newClaimBackfill(writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET"))
	log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools)
	// Platform-only: a fleet box (CARLOS_FLEET set) never reads other
	// fleets' bindings — it only ever WRITES its own (fleetbind.go's
	// fleetBindingHook/publishResidentFleetBindings, wired in RunAgent).
	// Reusing shouldSkipPublisherElection rather than re-testing the env var
	// here: both questions are exactly "is this box a fleet box", so one
	// predicate, already covered by TestShouldSkipPublisherElection, answers
	// both.
	isFleetBox := shouldSkipPublisherElection(os.Getenv)
	// An offline move's participants keep their reports current between the
	// passes below (finding 7); idle, it reads nothing.
	go offlineReassertLoop(ctx, store)
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = 5*time.Minute + rand.N(time.Minute)
		if err := backfill(ctx); err != nil {
			log.Printf("agent: claim backfill: %v", err)
		}
		if err := reconcileSerialized(ctx, store, reg, cfg); err != nil {
			log.Printf("agent: instances pass: %v", err)
		}
		// Every reconcile refreshes the build-state copies (settings spec
		// §4.2), this 5-minute one too: the fast leg above runs only on a
		// serial move.
		if bs != nil {
			bs.refreshAll(ctx)
		}
		telemetryPass(ctx, store, reg)
		plansPass(ctx, store, reg, os.Getenv)
		runMovesPass(ctx, store)
		runRunsPass(ctx)
		deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now())
		if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil {
			log.Printf("agent: instances alias pass: %v", err)
		}
		if !isFleetBox {
			if err := ReconcileFleetBindings(ctx, store, reg); err != nil {
				log.Printf("agent: fleet bindings reconcile: %v", err)
			}
		}
	}
}

// offlineReassertLoop runs the armed moves worker's offlineReassert every
// offlineReassertEvery plus up to 5 s of jitter, for the life of the agent:
// while this box takes part in an offline move, its source and sealer reports
// stay current for the doors whatever the serial does. Thin and untested by
// convention, like instancesLoop; the tick is offlineReassert.
func offlineReassertLoop(ctx context.Context, s release.Store) {
	for {
		time.Sleep(offlineReassertEvery + rand.N(5*time.Second))
		if w := armedMovesWorker.Load(); w != nil {
			w.offlineReassert(ctx, s)
		}
	}
}

// retentionLoop runs the release-retention pass forever on the slow
// reconcile cadence (5min + jitter, first tick after 1 minute) — never the
// 2s fast pass: pruning is not urgent, and a policy written now is enforced
// within minutes, which is the promise the CLI makes ("boxes prune within
// ~10 minutes"). Thin and untested by convention (like domainsLoop /
// instancesLoop); everything with behavior lives in retentionPass. Missing
// deployment config, a missing/unparseable release pubkey, or no box
// reports to read (retentionReports: the status store, or on a sole
// housekeeper with none this box's own build-state report) disables the
// loop with one log line — fail closed, domainsLoop's auto-purge shape: a
// box that cannot verify pointers, or cannot learn which builds the build
// states name (settings spec rev 11 §4.4), must not delete anything.
//
// Since the multi-region writer map (spec
// 2026-08-20-multi-region-ingress-design.md, Part 1) the pass itself is a
// HOUSEKEEPER DUTY: it runs only on the one box holding the deployment's
// housekeeper lease. releases/** is deployment-global state, and N boxes
// pruning it concurrently is the bug class the spec refuses to have. Every
// other box still arms this loop and still ticks — it just prunes nothing,
// silently, until the day it wins the lease (housekeeper.go). On a
// single-box deployment the holder is always this box, so the behaviour is
// unchanged.
func retentionLoop(reg *registry.Registry, buildStates func() (map[string]status.BuildStatePins, error)) {
	ctx := context.Background()
	// Platform boxes only (spec F6). Retention prunes the PLATFORM's release
	// objects — a customer's box scheduling deletes of them was already wrong
	// under direct S3, and through the store door it is refused outright (the
	// door's write/delete ACL scopes a fleet to fleets/<fleet>/**), so arming
	// here would buy nothing but an hourly loop of 404s. Reuses
	// shouldSkipPublisherElection because the question is exactly "is this a
	// fleet box", the same one instancesLoop asks — and it is deliberately the
	// bare CARLOS_FLEET test, not resolveFleetIdentity: a fleet box whose
	// LABEL is misconfigured is still a fleet box, and must still not prune.
	if shouldSkipPublisherElection(os.Getenv) {
		log.Printf("agent: retention loop is a platform-box job — skipped on a fleet box (spec F6)")
		return
	}
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: retention loop disabled: %v", err)
		return
	}
	raw := strings.TrimSpace(os.Getenv("CARLOS_RELEASE_PUBKEY"))
	if raw == "" {
		log.Printf("agent: retention loop disabled: CARLOS_RELEASE_PUBKEY is not set — retention policies will not be enforced by this box")
		return
	}
	pub, err := release.ParsePublicKey(raw, "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		log.Printf("agent: retention loop disabled: %v — retention policies will not be enforced by this box", err)
		return
	}
	// The box reports (settings spec rev 11 §4.4: the build-state builds
	// every box names) live in the status store, which is not always the
	// deployment store above (CARLOS_STATUS_BUCKET wins); a sole
	// housekeeper with none reads its own (retentionReports). Without them
	// no app can be pruned safely, so the loop does not arm.
	reports, err := retentionReports(ctx, os.Getenv, store, buildStates)
	if err != nil {
		log.Printf("agent: retention loop disabled: the box reports cannot be read (%v). retention policies will not be enforced by this box", err)
		return
	}
	if reports.local != nil {
		log.Printf("agent: retention: no status store; as the sole housekeeper this box reads its own build-state report")
	}
	log.Printf("agent: retention loop armed (5min + jitter) — prunes only while this box holds the housekeeper lease")
	// The pass runs under the HOLDING's context, not this loop's: a sweep
	// that is halfway through its deletes when renewal fails must stop, not
	// finish against a lock another box now owns (housekeeper.go).
	prune := housekeeperDuty{"retention pass", func(ctx context.Context) error {
		return retentionPass(ctx, store, reports, pub, reg)
	}}
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = 5*time.Minute + rand.N(time.Minute)
		runHousekeeperDuties(housekeeperGate, prune)
	}
}

// stampedUnitNames is the metrics sweep's one registry read: every distinct
// bespoke unit stamped on a route (Route.Unit, the #110 column — an
// operator's console.service, tito's titogo-canary@<tenant>.service), which
// the box's cgroup walk cannot discover by name convention.
//
// Called once per 60s tick from metricsLoop's probe, so it is deliberately a
// plain List + filter with no cache: the registry is a local SQLite file, one
// read a minute is nothing next to the pass loops already reading it, and a
// cache would be one more thing to invalidate when an operator stamps a unit.
//
// A read failure returns nil — this tick's stamped units are lost, the sweep
// and the agent are not. The names are returned sorted so the slice handed
// down each tick is stable (nothing below depends on the order; a stable one
// just makes a log or a test diff mean something).
func stampedUnitNames(reg *registry.Registry) []string {
	routes, err := reg.List()
	if err != nil {
		log.Printf("agent: metrics: read stamped units: %v", err)
		return nil
	}
	seen := make(map[string]bool, len(routes))
	var out []string
	for _, rt := range routes {
		if rt.Unit == "" || seen[rt.Unit] {
			continue
		}
		seen[rt.Unit] = true
		out = append(out, rt.Unit)
	}
	sort.Strings(out)
	return out
}

// resolveStatusBucket applies the status/deployment bucket defaulting
// documented in this file's package comment: an explicit CARLOS_STATUS_BUCKET
// always wins, and when it's unset the status loop falls back to whatever
// CARLOS_DEPLOYMENT_BUCKET is set to, so the common single-bucket deployment
// is one env var instead of two that must be kept in sync by hand. Returns
// "" when neither is set — the loop stays disabled, same as before this fix.
// Deliberately does NOT fall back to CARLOS_DEPLOYMENT_DIR: the status page
// has no directory-store serving path (the edge/console read status/ out of
// an S3 bucket, not a local directory), so a DIR-based deployment simply
// doesn't get this fallback — see the "HONEST LIMIT" note in the package
// comment above.
//
// Takes a getenv func rather than reading os.Getenv itself so the defaulting
// logic is testable without mutating process-wide environment variables —
// t.Setenv serializes tests within a package and the host package already
// has other env-reading tests; a plain map/func seam avoids that coupling
// entirely.
func resolveStatusBucket(getenv func(string) string) string {
	return ResolveStatusBucket(getenv)
}

// ResolveStatusBucket is resolveStatusBucket for callers outside this package
// — `carlos status -live` reads the boxes an agent wrote, and it must land on
// the SAME bucket the agent's status loop writes to, including the
// CARLOS_DEPLOYMENT_BUCKET fallback. Exported as a second name over one
// implementation rather than renamed at every internal call site, so the
// defaulting rule documented above stays in exactly one function.
func ResolveStatusBucket(getenv func(string) string) string {
	if b := getenv("CARLOS_STATUS_BUCKET"); b != "" {
		return b
	}
	return getenv("CARLOS_DEPLOYMENT_BUCKET")
}

// shouldSkipPublisherElection reports whether this tick must never contend
// for the cross-box publish lease at all — true exactly when CARLOS_FLEET is
// set (Task 5, customer-fleets plan): a fleet box self-reports its own Box
// snapshot every tick (statusTick's WriteBox call is unconditional) but must
// never win — or even attempt — the election that would have it merge and
// publish status/fleet.json, which is the platform's own boxes' shared feed.
// Checked BEFORE lease.Acquire is ever called, not after: a fleet box never
// makes the network call, not merely one that's discarded on return.
func shouldSkipPublisherElection(getenv func(string) string) bool {
	return getenv("CARLOS_FLEET") != ""
}

// upstreamSource holds the live edge's UpstreamStates method, published by
// RunAgent's edge-ready callback and read by statusLoop's probe set — the same
// shape tunnelStateVal uses for tunnel state (tunnel_client.go), and for the
// same reason: the status loop starts BEFORE the edge exists (RunAgent spawns
// the loops, then builds and serves the edge), so the two cannot be wired by
// passing a handle down.
//
// A func rather than the *edge.Edge itself, so what this package's tests
// inject is the same one value production sets — no fake Edge, no probe
// goroutine a test would have no way to stop.
var upstreamSource atomic.Value // func() map[string][]status.UpstreamState

// geoIPSource is the edge's GeoIP state for the heartbeat, published from
// the edge-ready callback like upstreamSource (geoip spec §5.5).
var geoIPSource atomic.Value // func() edge.GeoIPState

// geoIPAlertSource is the edge's geoip reason code for the box alert rule
// (geoip spec §10.4), published from the same ready callback.
var geoIPAlertSource atomic.Value // func(time.Time) (float64, bool)

// geoIPAlertReading is the rule's reading, or nil before the edge is ready
// (no data never fires and never clears).
func geoIPAlertReading(now time.Time) *float64 {
	fn, _ := geoIPAlertSource.Load().(func(time.Time) (float64, bool))
	if fn == nil {
		return nil
	}
	v, known := fn(now)
	if !known {
		return nil // no data never fires and never clears
	}
	return &v
}

// geoIPState reads it, or nil before the edge is ready.
func geoIPState() *status.GeoIPState {
	fn, _ := geoIPSource.Load().(func() edge.GeoIPState)
	if fn == nil {
		return nil
	}
	s := fn()
	return &status.GeoIPState{Mode: s.Mode, SHA256: s.SHA256, Type: s.Type, Error: s.Error}
}

// tunnelEnvComplete is tunnelLoop's own arming rule: all three set.
func tunnelEnvComplete(getenv func(string) string) bool {
	return getenv("CARLOS_TUNNEL_URL") != "" && getenv("CARLOS_TUNNEL_TOKEN") != "" && getenv("CARLOS_TUNNEL_NAME") != ""
}

// setUpstreamSource publishes (or, with nil, withdraws) the probe-state
// source. Called once, from the edge-ready callback.
func setUpstreamSource(fn func() map[string][]status.UpstreamState) {
	if fn == nil {
		upstreamSource.Store(func() map[string][]status.UpstreamState { return nil })
		return
	}
	upstreamSource.Store(fn)
}

// upstreamStates reads the current probe table for the heartbeat, or nil when
// no source has been published yet.
//
// nil is the honest answer for every box shape that has no edge probe table to
// report: `carlos edge` run without the agent, an agent whose ready callback
// hasn't fired yet, and the overwhelmingly common case of a box with no
// multi-upstream route at all. All three publish no upstreams key rather than
// an empty object (status.CollectBox's omitempty rule) — a box that isn't
// probing and a box running code that predates this field stay
// indistinguishable on the wire.
func upstreamStates() map[string][]status.UpstreamState {
	fn, _ := upstreamSource.Load().(func() map[string][]status.UpstreamState)
	if fn == nil {
		return nil
	}
	return fn()
}

// statusLoop runs the fleet status page's 5-minute tick, forever, once armed
// — see the package doc comment above for the full env contract. Missing
// CARLOS_STATUS_BUCKET disables the loop with one log line, the same
// "a box with no status store is still a valid deployment" shape adoptLoop
// uses for its own missing config.
//
// Each tick: gather fresh probes (small OS-reading funcs in probes.go, none
// of which this loop's own tests call — statusTick below takes a
// status.Probes value, so tests inject fixtures instead of touching /proc or
// a syscall), then contend for the cross-box status-lease. Acquire succeeding
// means this box is the one publisher for this tick: statusTick both writes
// this box's own snapshot AND merges+publishes the fleet-wide feed, and the
// lease is released again right after (held only for the publish, not across
// the 5-minute gap — the next tick, possibly on a different box, re-acquires
// fresh). A *lease.HeldError means another box already won this tick's
// election: completely normal contention, not a failure, so it's swallowed
// without a log line — every box hits this on every tick it doesn't win, and
// logging that every 5 minutes forever would bury the log in noise. Any
// OTHER acquire error (a real S3/credential problem) is logged, same as any
// other tick failure.
func statusLoop(reg *registry.Registry, svc *activator.Service, buildStates func() (map[string]status.BuildStatePins, error)) {
	ctx := context.Background()
	// bucket is read for the lease key alone (lease.Acquire below), which a
	// fleet box never reaches: shouldSkipPublisherElection short-circuits the
	// election before it, and a fleet box is the only shape that gets a store
	// with no bucket behind it (openStatusStore). Every other box resolves a
	// real bucket here exactly as before.
	bucket := resolveStatusBucket(os.Getenv)
	prefix := resolveStatusPrefix(os.Getenv)
	region := os.Getenv("CARLOS_STATUS_REGION")
	boxRegion := os.Getenv("STATUS_BOX_REGION")
	skipElection := shouldSkipPublisherElection(os.Getenv)

	// Fleet boxes heartbeat under the operator-chosen channel label
	// (fleet_boxes.label — the identity the edge dials via
	// edge.fleetChannelFor), NEVER the public status page's hashed
	// pseudonym (status.Label(machineID)): that anonymization exists so a
	// customer's machine id is never recoverable from the PUBLIC
	// status/fleet.json feed, but a fleet box's own snapshot never reaches
	// that feed at all (skipElection, above) — it only ever lands at
	// fleets/<fleet>/status/boxes/<label>.json, an owner-facing path no
	// public page reads, so the anonymization rationale simply doesn't
	// apply here. An invalid identity degrades to the hashed label rather
	// than disabling the heartbeat outright (RunAgent already refuses,
	// loudly, to arm the binding-publish machinery that actually needs the
	// label to be dial-able).
	fleetLabel := ""
	if identity, err := resolveFleetIdentity(os.Getenv); err == nil {
		fleetLabel = identity.Label
	} else if !errors.Is(err, errNotFleetBox) {
		log.Printf("agent: %v — heartbeat will publish under the hashed pseudonym instead", err)
	}

	s, target, err := openStatusStore(ctx, os.Getenv)
	if err != nil {
		if errors.Is(err, errNoStatusStore) {
			log.Printf("agent: status loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set (CARLOS_DEPLOYMENT_DIR deployments don't publish status yet — set CARLOS_STATUS_BUCKET explicitly)")
		} else {
			log.Printf("agent: status loop disabled: %v", err)
		}
		return
	}
	// store is what statusTick actually reads/writes through: identical to s
	// (a pure passthrough) unless CARLOS_STATUS_PREFIX is set, in which case
	// every key it touches is rebased under that prefix — see
	// release.NewPrefixStore and the package doc comment's
	// CARLOS_STATUS_PREFIX entry.
	store := release.NewPrefixStore(s, prefix)
	// The effective destination is worth a log line on its own: it's easy to
	// set CARLOS_DEPLOYMENT_BUCKET and CARLOS_STATUS_BUCKET to different
	// values by accident (or on purpose) and end up with the agent writing
	// status to a bucket the edge/console never look at — this line is the
	// one place an operator can confirm which bucket (or, on a bucketless
	// fleet box, which door identity) actually won.
	log.Printf("agent: status loop armed (%s + jitter), publishing to %s", status.SnapshotPeriod, target)

	// canaryStore/canaryPub resolve the release deployment config
	// countExpiredCanaries needs to answer §5's "how many canaries on this
	// box have expired" — the SAME env pair adoptLoop reads
	// (CARLOS_DEPLOYMENT_BUCKET/_DIR + CARLOS_RELEASE_PUBKEY), not
	// necessarily the status bucket above, which may deliberately be a
	// separate one. Either can come back unusable: no deployment store
	// configured, or no pubkey set. That is a normal, valid box shape (the
	// same "routes managed by hand" case adoptLoop's own doc comment
	// names) — it just means countExpiredCanaries can't answer and reports
	// nil, same as every other unmeasured probe in this loop. Resolved once
	// here, not re-read every tick, mirroring adoptLoop's own one-time
	// resolution.
	//
	// The error is logged rather than swallowed: "no deployment store" is a
	// valid box shape, but since the store can now also fail to build for a
	// REAL misconfiguration (a fleet box with no bucket and no tunnel env —
	// openFleetDoorStore's loud error), silence here would turn that into an
	// unexplained permanently-nil ExpiredCanaries probe.
	canaryStore, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: status loop: expired-canary probe unavailable: %v", err)
	}
	canaryPub, _ := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")

	// status.SnapshotPeriod IS this cadence — the const is named for the tick
	// below, and every reader that judges a box stale (the published boxes
	// table, the console's private host detail, `carlos status -live`) measures
	// against it, so changing it here must move them together rather than
	// leaving them judging a heartbeat that arrives on a different beat.
	// Jittered on the same shape as adoptLoop's delay: every box on the same
	// cadence would all hit the lease and the bucket in the same second
	// otherwise.
	delay := time.Minute
	for {
		select {
		case <-time.After(delay):
		case <-statusNudge:
			// A rollout transition (flag on only): publish within seconds,
			// at most every two, so a busy rollout cannot hammer the bucket.
			time.Sleep(2 * time.Second)
		}
		delay = status.SnapshotPeriod + rand.N(time.Minute)

		// Routes are listed here (not left to statusTick's own reg.List())
		// because the Listening probe needs to know which sockets are
		// systemd-socket-activated BEFORE it's built — see
		// wrapListeningProbe's doc comment. statusTick lists again for
		// itself since it stays store/lease-agnostic and self-contained;
		// one extra List() per 5-minute tick is not worth threading routes
		// through the pure function's signature for.
		routes, err := reg.List()
		if err != nil {
			log.Printf("agent: status tick: list routes: %v", err)
			continue
		}

		memTotal, memAvail := readMemBytes()
		p := status.Probes{
			MachineID:  readMachineID(),
			InstanceID: cachedInstanceID(),
			Region:     boxRegion,
			MemFrac:    readMemFrac(),
			DiskFrac:   readDiskFrac(),
			Listening:  wrapListeningProbe(unitBackedSockets(routes), statusSocketProbe(routes, svc)),
			// One systemctl call per 5-minute tick, asking the only question
			// that catches an enabled-but-never-armed adoption timer — the
			// flagship's silent no-adopt state, which nothing on any page
			// could see before this.
			AdoptRunning: adoptTimerActive(),
			// A second systemctl call per tick, asking the same question of
			// the config-sync trigger: an enabled-but-unstarted
			// carlos-adopt.path is a box whose unit-backed config silently
			// waits for the hourly timer (issue #44).
			ConfigPathRunning: configPathActive(),
			// A third, for the restart trigger: an enabled-but-unstarted
			// carlos-restart.path is a box where `carlos restart` answers
			// "requested" and nothing cycles until the hourly pass (issue #113).
			RestartPathRunning: restartPathActive(),
			// Whether this box's own alert rules have anywhere to deliver,
			// read from the env the metrics loop's sinks are built from
			// (discussion #192: an unset webhook was silent for weeks).
			Alerts: alertDeliveryProbe(os.Getenv, nil),
			// ExpiredCanaries answers spec §5 for this box's own routes; see
			// countExpiredCanaries and the canaryStore/canaryPub comment above.
			ExpiredCanaries: countExpiredCanaries(ctx, routes, canaryStore, canaryPub, time.Now()),
			// LagSecs stays nil: litestream's per-DB replication lag isn't
			// surfaced up to the agent process anywhere yet — the
			// replication machinery (internal/lease's Replicator) lives
			// inside the activator, one level down, and doesn't currently
			// export a "seconds since last successful sync" reading this
			// loop could poll. Publishing a guess (0? the box's own clock
			// skew?) would be worse than the honest null CollectBox already
			// knows how to carry — wiring this up is a seam for later, not
			// a gap to paper over now.
			MemTotalBytes: memTotal,
			MemAvailBytes: memAvail,
			Load1:         readLoad1(),
			DiskFreeBytes: readDiskFreeBytes(),
			Release:       version.Build,
			Platform:      runtime.GOOS + "-" + runtime.GOARCH,
			// The invalid-value line is newSystemLeg's to say, once at
			// arming; this tick publishes the channel the box actually
			// follows, five-minutely, and has nowhere to say it again.
			SystemChannel: boxSystemChannel(os.Getenv),
			System:        systemStateProbe(release.OptDir()),
			Tunnel:        tunnelState(),
			GeoIP:         geoIPState(),
			// The edge's probe table, read WHOLE every tick (never merged
			// with the last one), so a route this box has stopped probing
			// leaves no ghost behind in the next snapshot. nil until the
			// edge-ready callback publishes the source, and on any box that
			// probes nothing — both publish no key at all.
			Upstreams: upstreamStates(),
			// One stat (two with a WAL) per database route per tick, which
			// is what `carlos instances list` and `carlos routes` show as DB.
			DBBytes: dbFileBytes,
			// Layout is the armed pass's classification of every
			// instance-kind row, or, with no activator on this box, the
			// rows judged by path shape under Mode "unmanaged" — see
			// layoutStatusReport's own doc comment for why a box with no
			// pass still says something rather than nothing.
			Layout: layoutStatusReport(routes),
		}
		// The hosts this box is refusing because it cannot provide their
		// config (amadan #213): private Metrics only, nothing public.
		claimReport(&p)
		p.ConfigRefusals, p.ConfigRefusalsTotal = configRefusalProbe()
		p.ConfigGuard = configGuardProbe()
		// Builds the root pass rolled back (D1), as the metrics loop's
		// refusal watch last read them. Private Metrics only: hosts.
		p.ReleaseRefusals = releaseRefusalProbe()
		// The rollout scheduler's report (zero-downtime spec §8.4): nil
		// with CARLOS_SWAP off. Private Metrics only.
		p.Rollout = rolloutStatusProbe()
		// Hosts held for a dev proof (D3): private Metrics only.
		p.ProofWaits = proofWaitProbe()
		// The build-state builds of the instances this box hosts (settings
		// spec rev 11 §4.4), for the housekeeper's retention. A failed
		// collection is reported as such: the housekeeper prunes nothing
		// while it is fresh. Private Metrics only: hosts.
		if pins, err := buildStates(); err != nil {
			log.Printf("agent: status tick: build states: %v", err)
			p.BuildStatesErr = err.Error()
		} else {
			p.BuildStates = pins
		}
		// The update-settings protocol this box enforces, and root's last
		// decision per instance (settings spec §4.3). Private Metrics only.
		fillUpdateProbes(&p, reg, release.OptDir())

		// A fleet box (CARLOS_FLEET set) never gets here as a contender: it
		// self-reports only, never attempting the election at all — no
		// lease.Acquire call, no BuildFleet, no PublishFleet. See
		// shouldSkipPublisherElection and the package doc comment's
		// CARLOS_FLEET entry.
		var held *lease.Held
		publish := false
		if !skipElection {
			// path.Join (an S3 key, not a filesystem path — matches
			// activator.go's Config.hibernatePrefix convention) rather than a
			// bare "+" concat: guards against a trailing-slash
			// CARLOS_STATUS_PREFIX producing a doubled "//status-lease" key.
			var acqErr error
			held, acqErr = lease.Acquire(ctx, bucket, path.Join(prefix, "status-lease"), region)
			publish = acqErr == nil
			if acqErr != nil {
				var heldErr *lease.HeldError
				if !errors.As(acqErr, &heldErr) {
					log.Printf("agent: status tick: lease acquire: %v", acqErr)
				}
			}
		}

		// groupStatusTick is statusTick for every box but one its region's
		// group launched, which publishes only under its group reference
		// (census ruling region-ha-2b, boxreference.go).
		if err := groupStatusTick(ctx, store, reg, p, publish, fleetLabel, os.Getenv); err != nil {
			log.Printf("agent: status tick: %v", err)
		}

		if publish {
			if err := held.Release(ctx); err != nil {
				log.Printf("agent: status tick: lease release: %v", err)
			}
		}
	}
}

// statusTick is the fleet status page's pure per-tick work, extracted out of
// statusLoop so it needs no systemd, S3, or timers to test: given an
// already-built Store (DirStore in tests, the real S3Store in production)
// and an already-decided publish flag (statusLoop's lease result; a fixed
// true/false in tests), it does exactly what one tick does and nothing else.
// The box's own snapshot is written unconditionally — every box self-reports
// every tick regardless of who's publishing — and BuildFleet+PublishFleet
// only run when publish is true, which for a DirStore-backed dev/test caller
// with no lease in play at all just means "always true, no election needed"
// (a lone box is its own only publisher).
//
// labelOverride, when non-empty, replaces the box's hashed-pseudonym label
// (status.CollectBox derives one from p.MachineID by default) with the given
// value before WriteBox — statusLoop passes a fleet box's channel label
// (resolveFleetIdentity) here so its heartbeat lands at
// fleets/<fleet>/status/boxes/<label>.json under the SAME label the edge
// dials, not an opaque pseudonym only the public status page ever wanted.
// "" (every platform box) leaves CollectBox's own label untouched.
func statusTick(ctx context.Context, s release.Store, reg *registry.Registry, p status.Probes, publish bool, labelOverride string) error {
	routes, err := reg.List()
	if err != nil {
		return err
	}
	now := time.Now()

	// Aliases carries route_aliases' DNS/cert stamps (Task 2/5) up to the box
	// snapshot (issue #151) — read fresh alongside routes above rather than
	// folded into the caller's Probes in statusLoop, since reg is already in
	// scope here and a failed read must not sink the whole snapshot: logged
	// and left nil (CollectBox already treats nil the same as "no alias rows
	// exist"), same best-effort shape as every other probe in this loop.
	if aliases, err := reg.ListAliases(); err != nil {
		log.Printf("agent: status tick: list aliases: %v", err)
	} else {
		p.Aliases = aliases
	}

	box := status.CollectBox(routes, p, now)
	if labelOverride != "" {
		box.Label = labelOverride
	}
	if err := status.WriteBox(ctx, s, box); err != nil {
		return err
	}
	if !publish {
		return nil
	}

	// status.SnapshotPeriod is this loop's own tick cadence, named once in the
	// status package so the console's private host table and `carlos status
	// -live` measure staleness against the same period this publish does.
	fleet, err := status.BuildFleet(ctx, s, routes, now, status.SnapshotPeriod)
	if err != nil {
		return err
	}
	return status.PublishFleet(ctx, s, fleet)
}

// makeRestartRoute builds the adopt pass's restart callback, dispatched on
// what owns the route's process. THREE arms, in this order:
//
//  1. a route that NAMES its own unit (registry.Route.Unit — an operator-run
//     bespoke service, console.service) is try-restarted by that name, and
//     returns. This arm comes first because it is the only one that can be
//     right when it applies: such a route is addr-backed, so the activator
//     owns no child to Bounce and there is no carlos-app@<host> unit to
//     try-restart — both of the arms below would run and cycle nothing,
//     which is issue #110 (adoption stamped Version, and therefore
//     X-Carlos-Version, behind a console still serving the old build).
//  2. a carlos-app@<host> unit tenant is try-restarted by host (a no-op when
//     the unit isn't running — hibernators are never woken by this).
//  3. an exec child is Bounced through the activator, the same graceful
//     cycle config sync uses. systemctl cannot reach a child of THIS
//     process, which is why exec-backed warm instances used to keep serving
//     the old binary until they happened to hibernate: the symlink was
//     repointed under them and nothing cycled the process.
//
// Hosts the registry cannot answer for take arm 2: try-restart of a
// nonexistent unit is a no-op, whereas a bounce of a process we might not own
// is not.
//
// A bounce that cycled NOTHING falls through to try-restart rather than
// returning: an exec-backed route with an ACTIVE carlos-app@<host> unit is a
// real state — an unstamped unit tenant, the one adopt_advice.go warns about
// (issue #14) — and try-restart is what has always cycled it. The same
// fallthrough covers a box with no activator armed at all. When the route
// really is a sleeping exec child the cost is one try-restart of a stopped or
// absent unit, which starts nothing and at worst logs once: this callback
// only fires when the version actually moves (adoptHost early-returns on
// cur == ver), so it is one line per version change per host, never per tick.
//
// The named-unit arm takes NO part in that fallthrough, in either direction.
// It does not fall through (a route that names its unit has said where its
// process is; trying the other two afterwards could only cycle something
// else), and nothing falls through INTO it (the other arms are reached only
// when Unit is empty, i.e. when there is no name to try).
//
// No path can WAKE anything, which is what lets release.Pass hand this
// callback hibernating routes (RestartHibernating): Bounce returns false
// without starting anything when the host is not awake, and try-restart is
// defined as a no-op on a unit that is not running — including the named one,
// which is why it is try-restart and not restart (a console an operator has
// deliberately stopped stays stopped).
//
// ERRORS split by arm, and the split is the point:
//
// Arms 2 and 3 log and return nil, unchanged: a failed try-restart of a
// carlos-app@<host> unit must not poison the adopted version for the other
// hosts in the pass (release/adopt.go's rule), and it cannot be told apart
// from the ordinary case anyway — a sleeping exec child reaches arm 2 under
// RestartHibernating precisely because there is nothing there to cycle.
// Fire-and-forget is what makes that safe.
//
// Arm 1 RETURNS the failure, because for a named unit the ordinary case is
// not indistinguishable: try-restart of a unit that is loaded and stopped
// exits 0, so a non-zero exit means the name is wrong (exit 5, "not loaded")
// or the service would not come back. Both are "the new build is NOT
// running", and swallowing them would hand adoptHost a success it did not
// have: it would stamp Version, the edge would serve the new
// X-Carlos-Version, and `carlos deploy` would report live over the old
// process — issue #110 again, now with the stamp actively lying rather than
// merely unattended. Returned, adoptHost rolls the symlink back to the
// version that IS running, skips SetVersions, prints the failure, and the
// next pass retries; the header stays honest on the old version throughout.
func makeRestartRoute(reg *registry.Registry, tryRestart func(host string), tryRestartUnit func(rt registry.Route) error, bounce func(host string) bool) func(host string) error {
	return func(host string) error {
		rt, err := reg.Get(host)
		if err != nil {
			log.Printf("adopt: registry lookup %s: %v (restarting as a unit)", host, err)
		}
		if err == nil && rt != nil && rt.Unit != "" {
			return tryRestartUnit(*rt)
		}
		if err == nil && rt != nil && rt.Backing != registry.BackingUnit {
			if bounce != nil && bounce(host) {
				log.Printf("adopt: bounced exec instance %s onto its new version", host)
				return nil
			}
		}
		tryRestart(host)
		return nil
	}
}

// makeStartRoute builds the adopt pass's rollback-start callback
// (release.WithStartUnit): after a restart that FAILED has had its symlink
// rolled back, start the unit the route NAMES on the restored version —
// `systemctl start`, not try-restart, because the premise of that path is
// that the unit is down (a try-restart there is a no-op and the box stays
// serving nothing).
//
// ONE arm, deliberately, where makeRestartRoute has three: only a route that
// names its own unit has anything for this callback to start. The other two
// process owners never route here at all — their restart arms swallow their
// failures and return nil, so the rollback path this callback belongs to is
// unreachable for them — and a start of carlos-app@<host> or an activator
// child from here could WAKE a sleeper, which no adopt path may do. A route
// with an empty Unit, an unknown host, or a registry that will not answer
// therefore starts NOTHING and returns nil: for those shapes "nothing was
// stopped by the failed restart" is the only answer this callback can stand
// over. (The converged-daemon rollback is caps.Restore's business, not
// this callback's — see release.WithStartUnit.)
//
// This cannot wake anything, twice over. By ARGUMENT: it fires only after a
// try-restart of the named unit returned non-zero, and try-restart exits 0
// on a loaded-but-stopped unit — so a deliberately stopped service never
// produces the error that reaches this, and a unit whose name resolves to
// nothing gives systemctl start nothing to bring up either. And by
// STRUCTURE: a Hibernate-flagged route starts nothing here, full stop. That
// guard costs nothing real — provision refuses the addr-with-hibernate
// shape, so no legitimate named-unit route hibernates, and the only
// hibernating+Unit row is the misconfiguration restart.go already documents
// — and it turns "an adopt path cannot wake a sleeper" back into a property
// of the code's shape rather than of an argument about systemd exit codes.
func makeStartRoute(reg *registry.Registry, startUnit func(unit string) error) func(host string) error {
	return func(host string) error {
		rt, err := reg.Get(host)
		if err != nil || rt == nil || rt.Hibernate || rt.Unit == "" {
			return nil
		}
		return startUnit(rt.Unit)
	}
}

// systemctlTryRestart is makeRestartRoute's production unit path: cycle the
// route's carlos-app@ unit iff it's running, best-effort. A missing or
// stopped unit is not this pass's problem — try-restart is by definition a
// no-op on one, which is what makes makeRestartRoute's fallthrough safe:
// an exec route whose child is asleep returns false from bounce (nothing to
// cycle) and lands here under RestartHibernating, where try-restart finds no
// carlos-app@ unit for it and does nothing. Sleeping hosts DO reach this
// line; they simply have nothing here to restart, and pick the new version
// up on their next wake.
func systemctlTryRestart(host string) {
	out, err := exec.Command("systemctl", "try-restart", activator.UnitName(host)).CombinedOutput()
	if err != nil {
		log.Printf("adopt: try-restart carlos-app@%s: %v: %s (continuing)", host, err, strings.TrimSpace(string(out)))
	}
}

// systemctlTryRestartUnit is makeRestartRoute's NAMED-unit path: cycle the
// unit the route itself names, iff it's running. Same try-restart-not-restart
// reasoning as systemctlTryRestart above — a service an operator has stopped
// stays stopped, and try-restart on a loaded-but-stopped unit exits 0.
//
// It LOGS AND RETURNS, where its carlos-app@ sibling only logs, and the
// asymmetry is deliberate: this name came from an operator rather than from
// activator.UnitName, so a non-zero exit here means the stamp is wrong or the
// service will not come back — never the routine "nothing was running"
// systemctlTryRestart has to tolerate. The journal line is kept as well as
// the error because the error's own destination is a pass that prints one
// summary line per host; the exit status and systemctl's own words belong in
// the journal where an operator greps for the unit name.
//
// The unit reaches exec.Command as ONE argv element, so there is no shell for
// a name to escape into; the shape rail lives at the write (registry's
// ValidUnitName), where a typo can still be reported to the person who typed
// it.
func systemctlTryRestartUnit(unit string) error {
	out, err := exec.Command("systemctl", "try-restart", unit).CombinedOutput()
	if err != nil {
		log.Printf("adopt: try-restart %s: %v: %s", unit, err, strings.TrimSpace(string(out)))
		return fmt.Errorf("try-restart %s: %w: %s", unit, err, strings.TrimSpace(string(out)))
	}
	return nil
}

// unitOnly adapts a unit-name restart to makeRestartRoute's route-aware arm.
func unitOnly(f func(unit string) error) func(rt registry.Route) error {
	return func(rt registry.Route) error { return f(rt.Unit) }
}

// checkedUnitRestart is the named-unit arm with D1's settle check (spec
// §2.2-§2.3). The binding is verified first: the check judges a restart only
// when the unit's ExecStart runs THIS route's live link. Anything it cannot
// establish (show denied, wrong binding, unit stopped) falls back to today's
// plain try-restart and returns that command's own error.
//
// settled runs only when the settle check passed (never for a unit stopped
// during it, a failed settle or an unchecked fallback). The agent loops pass
// nil: the sandboxed agent cannot hash another unit's executable, and spec
// §4.5 has root record the healthy history.
func agentCheckedUnitRestart(sys unitSystem, optDir string, getenv func(string) string, settled func(rt registry.Route) error) func(rt registry.Route) error {
	return func(rt registry.Route) error {
		before, err := sys.Show(rt.Unit)
		if err != nil {
			log.Printf("adopt: %s: cannot read %s (%v); restart not checked", rt.Host, rt.Unit, err)
			return sys.TryRestart(rt.Unit)
		}
		link := filepath.Join(optDir, "live", rt.Host)
		if before.ExecPath != link {
			log.Printf("adopt: binding: %s runs %s, but the unit stamp is on %s; restart not checked", rt.Unit, before.ExecPath, rt.Host)
			return sys.TryRestart(rt.Unit)
		}
		if before.ActiveState != "active" {
			return sys.TryRestart(rt.Unit) // stopped by an operator: stays a no-op
		}
		want, err := filepath.EvalSymlinks(link)
		if err != nil {
			return sys.TryRestart(rt.Unit)
		}
		if err := sys.TryRestart(rt.Unit); err != nil {
			return err
		}
		timeout := unitSettleTimeout(getenv)
		err = settleUnit(sys, rt.Unit, before.InvocationID, settleOpts{
			Timeout: timeout, Poll: 500 * time.Millisecond, Hold: 5 * time.Second,
			WantDir: filepath.Dir(want), Ready: routeReady(rt, 2*time.Second),
			Now: time.Now, Sleep: time.Sleep})
		if errors.Is(err, errUnitStopped) {
			// An operator stopped it between the read above and now. Not a
			// verdict on the build, and a rollback restart would undo their
			// stop: same as a unit that was stopped before the restart.
			log.Printf("adopt: %s: %s was stopped during settle; not judged, not restarted", rt.Host, rt.Unit)
			return nil
		}
		if err == nil {
			log.Printf("adopt: %s: %s settled on %s", rt.Host, rt.Unit, filepath.Dir(want))
			// D3 (proofobserve.go): a build that passed the settle check is a
			// build this box has run well.
			if settled != nil {
				err = settled(rt)
			}
		}
		return err
	}
}

// rollbackUnitRestart is D1's rollback restart (release.WithRollbackRestart):
// the live link already names the restored build; replace whatever is
// running (a deadline failure can leave the rejected build up) and check
// the restored build comes up too.
func agentRollbackUnitRestart(reg *registry.Registry, sys unitSystem, optDir string, getenv func(string) string) func(host string) error {
	return func(host string) error {
		rt, err := reg.Get(host)
		if err != nil || rt == nil || rt.Unit == "" {
			return nil
		}
		before, _ := sys.Show(rt.Unit)
		if err := sys.ResetFailed(rt.Unit); err != nil {
			log.Printf("adopt: %s: reset-failed %s: %v (continuing)", host, rt.Unit, err)
		}
		if err := sys.Restart(rt.Unit); err != nil {
			return err
		}
		want, err := filepath.EvalSymlinks(filepath.Join(optDir, "live", host))
		if err != nil {
			return err
		}
		err = settleUnit(sys, rt.Unit, before.InvocationID, settleOpts{
			Timeout: unitSettleTimeout(getenv), Poll: 500 * time.Millisecond, Hold: 5 * time.Second,
			WantDir: filepath.Dir(want), Ready: routeReady(*rt, 2*time.Second),
			Now: time.Now, Sleep: time.Sleep})
		if errors.Is(err, errUnitStopped) {
			// The restored build is what the link names; a stop an operator
			// made while it settled is theirs to keep, not a failed recovery.
			log.Printf("adopt: %s: %s was stopped during the rollback's settle; not judged, not restarted", host, rt.Unit)
			return nil
		}
		if err != nil {
			return fmt.Errorf("rollback did not come up either: %w", err)
		}
		return nil
	}
}

// systemctlStartUnit is makeStartRoute's production path: bring the named
// unit up after a rolled-back restart left it down. START, not try-restart —
// try-restart is a no-op on a stopped unit, which on this path would leave
// the box with a correct symlink and no process, and hand the NEXT pass's
// try-restart an exit 0 to stamp a never-run build with. Idempotent where it
// matters: `systemctl start` of an already-running unit exits 0 and does
// nothing, so a service something else restored first is not cycled twice.
//
// LOGS AND RETURNS like systemctlTryRestartUnit, and for the same reason:
// this name came from an operator, its failure means the box is serving
// nothing, and the caller's one summary line per host must say so while the
// exit status and systemd's own words stay in the journal under the unit
// name an operator would grep for. Same no-shell shape too: the unit reaches
// exec.Command as ONE argv element.
func systemctlStartUnit(unit string) error {
	out, err := exec.Command("systemctl", "start", unit).CombinedOutput()
	if err != nil {
		log.Printf("adopt: start %s: %v: %s", unit, err, strings.TrimSpace(string(out)))
		return fmt.Errorf("start %s: %w: %s", unit, err, strings.TrimSpace(string(out)))
	}
	return nil
}

// rootRestartRoute is the root adopt pass's restart callback (RunAdopt, out of
// carlos-adopt.service): no bounce. That process has no activator and no
// children of its own — the exec instances belong to the agent, and another
// process's children are not its to stop. With bounce nil the dispatch's arm
// 3 is unreachable and every route it used to try-restart by host still is,
// byte-identically; what it gains is arm 1.
//
// Arm 1 has to be HERE, not only on the agent's in-process loops, and the
// reason is the sandbox. carlos-edge.service runs as User=carlos with
// ProtectSystem=strict and ReadWritePaths=/data /run/carlos, so the agent
// cannot write /opt/carlos: on a real box its pass reports the stage as
// blocked and relays to this verb (carlos-stage.path), which does the
// staging, the repoint, the restart and the version stamp. A restart arm
// wired only into the agent would therefore be a restart arm the consoles
// never reach.
//
// settled is checkedUnitRestart's hook for a named unit whose settle check
// passed (D3: recordSettled, which adds the build to the healthy history);
// nil records nothing.
// bounceFn adapts an activator to makeRestartRoute's bounce parameter,
// nil-safe: a box with no activator armed gets a nil callback, which
// makeRestartRoute reads as "nothing running to cycle" rather than falling
// back to a systemctl path that could not work for an exec route anyway.
func bounceFn(svc *activator.Service) func(host string) bool {
	if svc == nil {
		return nil
	}
	return svc.Bounce
}

// adoptBounceFn is bounceFn for an ADOPT pass's restart dispatch
// (makeRestartRoute on adoptLoop and fastConvergeLoop): the same bounce,
// except that a child already exec'd from the build live/<host> names is left
// running — see activator.BounceAdopted for the interleaving that produces
// one. `carlos restart` keeps bounceFn: a restart request cycles whatever is
// running.
func adoptBounceFn(svc *activator.Service) func(host string) bool {
	if svc == nil {
		return nil
	}
	if m := armedRollout(); m != nil {
		return m.Adopted
	}
	if rolloutUnavailable() != "" {
		return nil // no adopt bounce while managed starts are refused
	}
	if h := heldHolds(); h != nil {
		// Declined with failure records: a host a record holds keeps
		// running rather than stop into a refused wake (final review F6).
		return h.skipHost("adopt bounce", svc.BounceAdopted)
	}
	return svc.BounceAdopted
}

// wakeFn adapts an activator to cycleWorkers' wake parameter, nil-safe like
// bounceFn: a box with no activator armed gets a nil callback, which
// cycleWorkers reads as "nothing to wake" rather than panicking on a nil
// activator.
func wakeFn(svc *activator.Service) func(registry.Route) error {
	if svc == nil {
		return nil
	}
	return svc.WakeWorker
}

// openDeployStore opens the release store the box adopts from:
// CARLOS_DEPLOYMENT_BUCKET (S3), CARLOS_DEPLOYMENT_DIR (a directory — tests
// and single-box no-AWS deployments), or — on a fleet box with neither set —
// the edge's store door over the tunnel env (spec F5, openFleetDoorStore).
// Mirrors cli.go's openStore, but returns the error instead of dying: the
// agent degrades, the CLI exits.
func openDeployStore(ctx context.Context) (release.Store, error) {
	return openDeployStoreEnv(ctx, os.Getenv)
}

// OpenDeployStore is openDeployStore for callers outside this package. The
// relay reads its authority record and writes its heartbeat and metering to
// the same store the agent adopts from, chosen by the same rule, so a box
// that has moved onto the fleet door does not leave the relay behind on a
// bucket it can no longer reach.
func OpenDeployStore(ctx context.Context) (release.Store, error) { return openDeployStore(ctx) }

// openDeployStoreEnv is openDeployStore with the environment injected — the
// same getenv seam resolveStatusBucket already uses in this file, so the
// selection rule can be tested without mutating process-wide env.
//
// Order is the whole contract (spec F5): a bucket or dir still set WINS,
// before fleet identity is even consulted, so every platform box and every
// grandfathered fleet box that still has bucket credentials keeps exactly
// today's store, byte for byte. Only a box that resolves a fleet identity
// AND has no bucket/dir at all reaches the door — that is the cutover
// switch, per box, no flag day: drop the bucket var and the box is on the
// API; put it back and it is off.
func openDeployStoreEnv(ctx context.Context, getenv func(string) string) (release.Store, error) {
	if b := getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
		return release.NewDeploymentS3Store(ctx, b)
	}
	if d := getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
		ds := release.NewDirStore(d)
		ds.EnableChangeIndex()
		return ds, nil
	}
	identity, err := resolveFleetIdentity(getenv)
	switch {
	case err == nil:
		return openFleetDoorStore(getenv, identity)
	case !errors.Is(err, errNotFleetBox):
		// CARLOS_FLEET is set but the identity is unusable, and there is no
		// bucket to fall back to: this box has no store at all. Return the
		// identity error itself rather than the generic "no bucket" line —
		// the actual fix is the label, and the generic message would send an
		// operator hunting for a bucket var that is deliberately absent.
		return nil, err
	}
	return nil, errNoDeployStore
}

// openFleetDoorStore builds the fleetapi client a bucketless fleet box adopts
// through: the edge origin it already tunnels to (CARLOS_TUNNEL_URL), its
// channel identity ("<fleet>/<label>" — the same string tunnel_client.go
// registers with and edge.fleetChannelFor dials), and the tunnel bearer token.
//
// Missing tunnel env here is an ERROR, never a silent no-store: the caller
// established that this box is a fleet box with no bucket, so there is no
// other store it could possibly be meant to use, and a quiet nil would leave
// every loop in this file reporting "not configured" about a box that is in
// fact configured — just with one variable missing.
func openFleetDoorStore(getenv func(string) string, identity FleetIdentity) (release.Store, error) {
	base := strings.TrimSpace(getenv("CARLOS_TUNNEL_URL"))
	token := strings.TrimSpace(getenv("CARLOS_TUNNEL_TOKEN"))
	name := identity.Fleet + "/" + identity.Label
	if base == "" || token == "" {
		return nil, fmt.Errorf("fleet box %s has no deployment bucket and cannot reach the store door: both CARLOS_TUNNEL_URL and CARLOS_TUNNEL_TOKEN must be set", name)
	}
	return fleetapi.New(base, name, token)
}

// resolveStatusPrefix returns the prefix every key statusLoop and metricsLoop
// touch is rebased under (release.NewPrefixStore at both call sites).
//
// An explicit CARLOS_STATUS_PREFIX always wins, unchanged, on every box. What
// this adds is the DOOR-branch default: a fleet box publishing through the
// store door with an EMPTY prefix would otherwise write status/boxes/... and
// metrics/boxes/... at the shared platform root, and the door's write ACL
// admits only fleets/<own fleet>/** — every heartbeat and every flush would
// 404, forever, on a box that looks correctly configured. So on that branch
// alone the prefix defaults to "fleets/<fleet>/", which is exactly the layout
// the console's fleet page already reads.
//
// S3-path semantics are deliberately untouched: an empty prefix on a
// bucket-backed box is the platform box's shared root, which is correct
// there, and a bucket-backed fleet box keeps whatever the operator set (the
// customer-fleets rollout sets CARLOS_STATUS_PREFIX=fleets/<name>/ by hand).
func resolveStatusPrefix(getenv func(string) string) string {
	if p := getenv("CARLOS_STATUS_PREFIX"); p != "" {
		return p
	}
	if resolveStatusBucket(getenv) != "" {
		return ""
	}
	identity, err := resolveFleetIdentity(getenv)
	if err != nil {
		return ""
	}
	return "fleets/" + identity.Fleet + "/"
}

// errNoStatusStore is openStatusStore's "this deployment simply hasn't
// configured one" sentinel — distinct from a real construction failure, so
// each loop can keep its own long-standing disabled-log wording for the
// unconfigured case while sharing one selection rule.
var errNoStatusStore = errors.New("no status store configured")

// openStatusStore resolves the store statusLoop and metricsLoop publish
// through, plus a human phrase naming where that is for their armed log
// lines. Same F5 rule as openDeployStoreEnv, read through
// resolveStatusBucket: an explicit CARLOS_STATUS_BUCKET (else the
// CARLOS_DEPLOYMENT_BUCKET fallback) WINS and keeps direct S3; only a fleet
// box with neither set speaks the door.
//
// LEDGER NOTE (spec F5): the door serves the DEPLOYMENT store. A fleet box
// that sets CARLOS_STATUS_BUCKET to some OTHER bucket therefore keeps direct
// S3 for status/metrics while its releases may come through the door — that
// is the bucket-set-wins rule doing its job, not a bug: an operator who named
// a bucket explicitly gets that bucket.
//
// The CARLOS_STATUS_PREFIX rebase both loops apply (release.NewPrefixStore)
// is deliberately NOT applied here — it stays at each call site, unchanged,
// and wraps whichever store this returns.
func openStatusStore(ctx context.Context, getenv func(string) string) (release.Store, string, error) {
	if bucket := resolveStatusBucket(getenv); bucket != "" {
		// The fallback is the deployment bucket, so open it as one; status/ and
		// metrics/ are not change scopes and bump nothing.
		s, err := release.NewDeploymentS3Store(ctx, bucket)
		if err != nil {
			return nil, "", err
		}
		return s, "bucket " + bucket, nil
	}
	identity, err := resolveFleetIdentity(getenv)
	switch {
	case err == nil:
		s, derr := openFleetDoorStore(getenv, identity)
		if derr != nil {
			return nil, "", derr
		}
		return s, "the fleet store door as " + identity.Fleet + "/" + identity.Label, nil
	case !errors.Is(err, errNotFleetBox):
		return nil, "", err
	}
	return nil, "", errNoStatusStore
}

// RunAdopt is the `carlos adopt` verb: one immediate adoption pass — the
// carlos-adopt.timer unit's ExecStart, and an operator's "converge now".
// Runs as root (see carlos-adopt.service): this is the ONLY place the
// UNIT-BACKED half of config sync (appconfig.Sync) may run, because it
// writes /etc/carlos/apps and restarts units — the same filesystem-write
// shape release staging needs root for, and exactly what the sandboxed
// carlos-edge.service's in-process adopt loop (ProtectSystem=strict,
// ReadWritePaths=/data /run/carlos only) cannot do. The EXEC-BACKED half
// (appconfig.SyncExec) runs elsewhere, in the agent's in-process
// configSyncLoop: it needs neither root nor /etc access, because it only
// ever writes under /data and bounces its own children.
//
// -config-only runs the UNIT-BACKED CONFIG half ALONE: appconfig.Sync plus
// its restarts, no release.Pass, no canary nag, no backing advice. That is
// what carlos-config-sync.service execs, and it exists because the trigger
// behind it is a per-config-write path unit, not the hourly timer. Letting
// that trigger run a full adoption pass would mean one tenant's `carlos env
// set` could stage and cut over a DIFFERENT app's pending release within
// seconds of an unrelated write, and would fire the canary-TTL nag per
// config write instead of hourly — the "hourly root pass alone" split
// release/adopt.go and the nag comment below both depend on.
//
// -restart-only runs the RESTART-REQUEST half alone (issue #113): compare each
// app's restart generation against this box's marks and cycle the ones that
// moved, nothing else. That is what carlos-restart.service execs, and it is
// separated from -config-only for the same reason -config-only is separated
// from the full pass: the trigger behind it is a per-request path unit, and a
// request to cycle one app's processes must not stage a release, run the
// capability converge, or nag about canaries. It deliberately needs NO release
// key and NO secrets key — restarting verifies no signature and renders no
// config — so a box missing either still honours it.
func RunAdopt(args []string) { runAdopt(args, productionClaimFactory) }

func runAdopt(args []string, buildClaims ClaimFactory) {
	runAdoptWithEnsureSystemUnits(args, buildClaims, func() {
		ensureSystemUnits("/etc", filepath.Join(systemRootDir(release.OptDir()), systemLockFile), productionSysOps().systemctl, systemctlProbe, log.Printf)
	})
}

// runAdoptWithEnsureSystemUnits keeps the root-only unit effect injectable;
// the rest of adopt startup, including its fatal registry diagnostics, is real.
func runAdoptWithEnsureSystemUnits(args []string, buildClaims ClaimFactory, ensureUnits func(), sources ...adoptRootSource) {
	open := func() (rootstate.Store, error) { return rootstate.Open(rootReleasePrivate, true, false) }
	probe := rootparticipation.OpenForUpdate
	if len(sources) == 1 {
		open = sources[0].OpenRelease
		probe = sources[0].Census
	}
	err := withClaimRootRelease(context.Background(), hostwrite.Enabled(os.Getenv), open, func() error {
		return runAdoptUnderRootRelease(args, buildClaims, ensureUnits, probe, productionRootUpdates(dataDir(), probe))
	}, probe)
	if err != nil {
		log.Fatalf("adopt: %v", err)
	}
}

func runAdoptUnderRootRelease(args []string, buildClaims ClaimFactory, ensureUnits func(), census func() (rootparticipation.Ledger, bool, error), updateProviders ...release.RootUpdateOpener) error {
	reapStagingWrites(release.OptDir(), dataDir(), "/etc", log.Printf)
	if os.Geteuid() == 0 {
		if err := repairOffcloudCredentials("/etc/carlos/credentials"); err != nil {
			log.Printf("ALERT: adopt: offcloud credentials: %v", err)
		}
	}
	claimCtx, err := claimEntryContext(context.Background(), os.Getenv, openDeployStore)
	if err != nil {
		log.Fatalf("adopt: host claims: %v", err)
	}
	claimsEnabled := hostwrite.Enabled(claimEntryEnv(claimCtx))
	var claims *ClaimWiring
	if !claimsEnabled {
		// Keep the flag-off startup effects in their old order:
		// disabled wiring, flags, pass lock, unit repair, registry open.
		claims, err = newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
		if err != nil {
			log.Fatalf("adopt: host claims: %v", err)
		}
	}
	fs := flag.NewFlagSet("adopt", flag.ExitOnError)
	regPath := fs.String("registry", filepath.Join(dataDir(), "registry.db"), "registry database")
	configOnly := fs.Bool("config-only", false, "sync unit-backed config and restart changed routes only: no release adoption, no canary nag")
	stageOnly := fs.Bool("stage-only", false, "adopt releases without the canary nag: the agent-requested pass (carlos-stage.service)")
	restartOnly := fs.Bool("restart-only", false, "cycle the routes of apps whose restart generation moved, and nothing else: the agent-requested pass (carlos-restart.service)")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)
	var reg *registry.Registry
	if claimsEnabled {
		reg, err = openServiceRegistry(*regPath)
		if err != nil {
			if hint := registryOpenHint(*regPath, err); hint != "" {
				log.Fatalf("adopt: open registry %s: %v\nadopt: %s", *regPath, err, hint)
			}
			log.Fatalf("adopt: open registry %s: %v", *regPath, err)
		}
		defer reg.Close()
		claimCtx = withClaimRuntimeResources(claimCtx, ClaimRuntimeResources{Mode: "adopt", Registry: reg, DataDir: dataDir(), OptDir: release.OptDir(), UnitsDir: "/etc/systemd/system"})
		claims, err = newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
		if err != nil {
			log.Fatalf("adopt: host claims: %v", err)
		}
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), hostbox.RequestLimit)
		defer cancel()
		if err := claims.close(ctx); err != nil {
			log.Printf("adopt: host claims shutdown: %v", err)
		}
	}()

	// One exclusive lock for the whole pass, in every mode, because the four
	// units that exec this verb can start in the same second and the sweep
	// below is the first thing a pass does that a concurrent pass cannot
	// safely repeat (lockAdoptPass).
	//
	// A lock that cannot be taken is NOT fatal: it drops the sweep and runs
	// the pass exactly as this verb ran before the sweep existed. Every other
	// half is idempotent against a concurrent pass — that is why they were
	// unserialized for a year — so the honest response to "I cannot serialize"
	// is to skip the one half that needs it, not to stop converging the box.
	unlockPass, passLockErr := lockAdoptPass(dataDir())
	if passLockErr != nil {
		log.Printf("adopt: pass lock %s unavailable (%v) — continuing unserialized, staged-version sweep skipped", adoptPassLockPath(dataDir()), passLockErr)
	} else {
		defer unlockPass()
	}
	// Spec §7 belt 2: arm the self-update relay if the last hand roll did
	// not. Before the fatal gates below, and never fatal itself.
	if !*restartOnly {
		ensureUnits()
	}
	// Root keeps SQLite in the service-user child for every adoption mode,
	// including the early return for restart-only.
	if !claimsEnabled {
		reg, err = openServiceRegistry(*regPath)
		if err != nil {
			if hint := registryOpenHint(*regPath, err); hint != "" {
				log.Fatalf("adopt: open registry %s: %v\nadopt: %s", *regPath, err, hint)
			}
			log.Fatalf("adopt: open registry %s: %v", *regPath, err)
		}
		defer reg.Close()
	}
	// The moves protocol floor (moves spec §11): re-judge carlos.prev
	// against the armed floor and the held rows on every root pass, in every
	// mode, so a raised floor or a new hold reaches the units' choice within
	// the hour, or within seconds of a SetHold (its caller touches
	// stage-requested). Here, before the fatal gates below, so a pass that
	// dies later still judged. Never fatal; one stat on a box with no floor.
	restartChecks := productionRootRestartChecks(systemdUnits{}, release.OptDir())
	restartChecks.legacy = func() (bool, error) {
		if census == nil {
			return false, os.ErrPermission
		}
		ledger, required, err := census()
		if ledger.State != nil {
			ledger.State.Close()
		}
		return !required && err == nil, err
	}
	if err := recoverRootRestarts(claimCtx, systemdUnits{}, release.OptDir(), os.Getenv, restartChecks, claims.Writes); err != nil {
		log.Printf("adopt: %v", err)
	}
	if err := rootJudgePrev(reg); err != nil {
		log.Printf("adopt: moves protocol floor: %v", err)
	}

	// Step 0 (discussion #111): evict the staged versions nothing on this box
	// still names, BEFORE anything that can fail because the disk is full.
	//
	// The placement is the whole point of the fix, not a detail. Below this
	// line are three fatal gates — openDeployStore, the release pubkey parse,
	// and release.Pass, whose bake clock is written with CreateTemp under
	// /data and which log.Fatalf's the pass on error. On the box this is for,
	// with 1.2 MB free, a sweep placed after any of them never runs, and the
	// box has no way to free itself. It needs only the registry and /opt, so
	// it goes as early as having a registry allows. restartPass sits below
	// release.Pass's fatality for exactly this reason and says so.
	//
	// Hourly only: -stage-only is triggered per relayed release, up to once a
	// minute per loop, and walking the version store that often to free a
	// day's worth of shipping is work for no gain. -config-only and
	// -restart-only are narrow triggers that must do their one job.
	generationPins := productionRootGenerationPins()
	if passLockErr == nil && !*stageOnly && !*configOnly && !*restartOnly {
		sweepStagedVersionsPass(reg, generationPins)
	}

	ctx := context.Background()
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Fatalf("adopt: %v", err)
	}
	// One callback for both halves below, built once here now that it needs
	// the registry — it reads each route's `unit` column to decide what to
	// cycle. See rootRestartRoute for why the named-unit arm has to exist on
	// THIS pass.
	//
	// Only the RELEASE pass can actually reach that arm. appconfig.Sync skips
	// every route whose Backing is not BackingUnit before it restarts anything
	// (an exec-backed route's config is delivered by the agent's in-process
	// loop instead), and a unit-backed route is exactly what `carlos route
	// -unit` refuses to stamp — so config sync sees no unit-named route by
	// construction. It is handed the same callback anyway because there should
	// be ONE restart dispatch on this pass, not two that could drift.
	//
	// settled: D1's settle passing records the build in the healthy history
	// (D3, proofobserve.go). Root is the process that can hash the unit's
	// executable.
	observer := productionProofObserver(release.OptDir())
	observer.resolve = restartChecks.resolve
	settled := observer.recordSettled(release.RootHealthyDir)
	restart := rootRestartRouteWith(reg, systemdUnits{}, release.OptDir(), os.Getenv, settled, restartChecks)
	// The restart-request half, and the ONE place it dispatches from —
	// including for exec-backed routes, whose children belong to the agent:
	// rootRestartRoute's try-restart arm reaches a carlos-app@ unit, and an
	// exec child that this pass cannot cycle is cycled by nothing else, which
	// is a real limit of the -restart-only path recorded in restartPass.
	//
	// -restart-only returns here: no config sync, no capability converge, no
	// release.Pass, no nag, and — above all — no release public key, which is
	// parsed below and is fatal when unusable. A restart request must not fail
	// on a key it never needs.
	if *restartOnly {
		logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, func() error {
			routes, err := reg.List()
			if err != nil {
				log.Fatalf("adopt: list routes: %v", err)
			}
			restartPass(ctx, routes, store, restartMarksPath(dataDir()), restart)
			return nil
		}))
		return nil
	}
	// Parsed here, before config sync, exactly as the full pass always has
	// (a box with an unusable release key must die before it converges
	// anything), but SKIPPED entirely for -config-only: that mode never
	// verifies a signature, and making a config-sync trigger fatal on a
	// release-key problem would fail every config write on a box whose
	// releases were already going nowhere.
	var pub ed25519.PublicKey
	if !*configOnly {
		if pub, err = release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY"); err != nil {
			log.Fatalf("adopt: %v", err)
		}
	}
	// Config sync is optional — inert until CARLOS_SECRETS_KEY is set, the
	// same convention every other optional loop in this package follows. It
	// runs before release adoption so a newly adopted route's very first
	// start already sees its env file, and a pass carrying both a release
	// and a config change never starts the new binary against stale config.
	if raw := os.Getenv("CARLOS_SECRETS_KEY"); raw != "" {
		secretsPriv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
		if err != nil {
			log.Printf("adopt: config sync disabled: %v", err)
		} else if err := appconfig.Sync(ctx, reg, store, secretsPriv, appconfig.DefaultEnvDir, restart, configAckFunc(ctx, store, "adopt"), claims.Writes); err != nil {
			log.Printf("adopt: config sync: %v", err)
		}
	}
	if *configOnly {
		// Everything below is release adoption and its reporting — the half
		// carlos-config-sync.service must never reach, however often a
		// config write triggers it. The restart half is below for the same
		// reason: a config write is not a restart request, and running the
		// comparison here would only move it earlier by minutes at the cost of
		// firing on a trigger that has nothing to do with it.
		return nil
	}
	// The restart-request half again (issue #113), here as the FLOOR under
	// carlos-restart.path: that relay can drop an event that lands while the
	// service is mid-run, exactly as the config one can, and a dropped restart
	// with no floor would be a request that never happens at all.
	//
	// BEFORE release.Pass deliberately. That call is fatal on error, so a box
	// with an unstageable release would otherwise never reach this line — and
	// a wedged instance is most worth cycling precisely on a box where
	// something else is already wrong.
	logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, func() error {
		if routes, err := reg.List(); err != nil {
			log.Printf("adopt: list routes for the restart pass: %v", err)
		} else {
			restartPass(ctx, routes, store, restartMarksPath(dataDir()), restart)
		}
		return nil
	}))

	// Step 2 (spec §4d): the grant-drift capability converge, BETWEEN config
	// sync and release adoption. Before release.Pass because it repairs the
	// shape of what is ALREADY live — a revoked grant, a deleted drop-in, a
	// socket unit nobody enabled — and a route that release.Pass then moves
	// to a new version converges again inside adoptHost, which is a no-op
	// when nothing about the shape changed: same bytes, no write, no reload,
	// no restart.
	//
	// The zero caps.Ops is the production one (real /etc/systemd/system, real
	// systemctl, real useradd), for the same reason WithCapabilities below is
	// only ever set here: this verb is root out of carlos-adopt.service, and
	// the in-process loops in this file cannot write unit configuration at
	// all.
	//
	// Never fatal. A failure here is the registry not listing; adoption of
	// releases is a separate job and must still run.
	if err := capabilityPass(reg, release.OptDir(), productionClaimCapsOps(caps.Ops{}, restartChecks.allowLegacy)); err != nil {
		log.Printf("adopt: capability pass: %v", err)
	}

	// NagExpiredCanaries(os.Stdout) is set HERE ONLY (2026-08-03 review) — the
	// canary-TTL nag (spec §5) belongs to the hourly root pass
	// (carlos-adopt.service, this verb's ExecStart), the same split PR #9's
	// config sync above uses. adoptLoop and fastConvergeLoop in this same
	// package call release.Pass/PassFilter without the nag option, so an
	// expired canary nags once an hour from here, never
	// every 2-3 seconds from the in-process fast loop.
	//
	// WithCapabilities(caps.Ops{}) is likewise set HERE ONLY, and the zero Ops
	// is the production one (real /etc/systemd/system, real systemctl, real
	// useradd). This verb runs as root out of carlos-adopt.service; the
	// in-process loops in this same file run inside carlos-edge.service's
	// ProtectSystem=strict sandbox and cannot write unit configuration at all,
	// which is why adoptLoop and fastPassOpts each carry a comment saying
	// never to add this one.
	//
	// -stage-only drops the nag and keeps everything else. It is what
	// carlos-stage.service runs when the agent relays a version it could not
	// stage: that request arrives at most once per stageRequestEvery per loop, and the nag
	// is an HOURLY visibility feature — firing it per request would flood the
	// journal with the same expired-canary lines, which is precisely what the
	// option's own doc comment says never to do. The capability converge and
	// config sync stay: both are jobs only this root pass can do, and a
	// newly staged version wants its unit shape and env file converged with
	// it, not an hour later.
	//
	// WithStartUnit is wired on THIS pass above all: on a hardened box the
	// sandboxed agent's stage is blocked and relayed here, so this is the
	// pass that actually try-restarts a console's named unit — and therefore
	// the pass whose failed restart, rolled back, must start that unit again
	// on the restored version rather than leave a correct symlink over no
	// process (release.WithStartUnit; without it the NEXT pass's try-restart
	// no-ops on the stopped unit, exits 0, and stamps a build that has never
	// run). -stage-only keeps it for the same reason it keeps the converge:
	// a relayed stage is exactly the request that performs such a restart.
	pr := proofRoot{
		obs: observer, reg: reg, store: store, bakeDir: dataDir(),
		healthyDir: release.RootHealthyDir, policyPath: release.ProofPolicyPath,
		readKey: readProvingKey, hourly: !*stageOnly, logf: log.Printf,
	}
	history, err := configureRootProof(&pr, restartChecks, func() (*release.RootHistoryPass, error) {
		return openRootHistory(store, release.OptDir())
	})
	if err != nil {
		log.Printf("adopt: %v", err)
	} else if history != nil {
		defer history.History.Store.Close()
		linkRootHistory(history, restartChecks, release.OptDir(), os.Getenv)
	}
	recovery := productionNamedUnitRecovery(store, pub, release.OptDir(), claims.Writes)
	passOpts := []release.PassOption{release.WithRootGenerationPins(generationPins), release.WithCapabilities(productionClaimCapsOps(caps.Ops{}, restartChecks.allowLegacy)), release.BakeDir(dataDir()), release.WithRestartAttemptSettled(restartChecks.finish),
		release.WithRunningVersion(restartChecks.runningVersion),
		release.WithSelectedRestarts(
			rootSelectedRestart(reg, systemdUnits{}, release.OptDir(), os.Getenv, settled, restartChecks, false),
			rootSelectedRestart(reg, systemdUnits{}, release.OptDir(), os.Getenv, nil, restartChecks, true)),
		release.WithStartUnit(rootStartRoute(reg, release.OptDir(), restartChecks)), release.WithHostWrites(claims.Writes),
		release.WithDigestRatchet(pointerRatchet(true, release.OptDir(), dataDir(), os.Getenv)),
		release.WithRollbackRestart(rollbackUnitRestart(reg, systemdUnits{}, release.OptDir(), os.Getenv, restartChecks)),
		// D3's gate: a fresh gate per root pass, so the policy file is read
		// at the start of this pass (BeginPass) like every other. Healthy
		// history is root-owned; diagnostic refusals still use BakeDir.
		release.WithProofGate(release.NewRootProofGate(release.ProofPolicyPath, dataDir())),
		// Settings spec §4.3: root alone resolves install requests and
		// writes /opt/carlos/consumed/<host>, before it stages.
		updatesPassOption(dataDir(), true, updateProviders...)}
	passOpts, err = rootAdoptionOptions(passOpts, restartChecks, history, func() release.PassOption { return recovery.passOption(reg, restartChecks, os.Getenv) })
	if err != nil {
		return err
	}
	if !*stageOnly {
		passOpts = append(passOpts, release.NagExpiredCanaries(os.Stdout))
	}
	// D3 (proofroot.go) wraps the pass. Before it, record every named-unit
	// route's live build that has run well as healthy, so the build live when
	// a requirement is switched on is in the history before this pass can
	// adopt a proven head over it (final review H4). After it, mint on a
	// proving box and, on the hourly pass, say when a requirement is off;
	// a route the pass changed is looked at again first.
	passErr := pr.around(ctx, func() error {
		return release.Pass(ctx, reg, store, pub, release.OptDir(), restart, passOpts...)
	})
	if passErr != nil {
		log.Printf("adopt: %v", passErr)
	}

	// A move's held stage (moves design §7.3, T7): the sandboxed agent
	// cannot write /opt, so a move's target leaves a request under
	// /data/moves/held-stage and touches stage-requested. Served here, after
	// release adoption, on -stage-only and on the hourly pass (its floor).
	// Reads nothing and writes nothing unless a request exists, and none
	// exist while CARLOS_MOVES_PHASE1 is off.
	heldStagePass(ctx, dataDir(), release.OptDir(), pub, pointerRatchet(true, release.OptDir(), dataDir(), os.Getenv), reg, store, BoxLabel())

	// Step 4: the advisories. The two capability advisories that fall out of
	// READING manifests (declared-but-ungranted, and capabilities on an
	// exec-backed route) are printed by capabilityPass above, where the
	// manifests are already in hand; this one needs the box's firewall and
	// its passwd database instead, so it lives here with the other
	// systemd-probing advisory.
	if routes, err := reg.List(); err == nil {
		adviseBackingMismatches(routes)
		adviseHostRoleGateMissing(routes, lookupUID, nftGateUIDs)
	}
	return passErr
}