// Package activator is the hibernation state machine: instances hibernate.
// When the agent is given an S3 bucket (CARLOS_HIBERNATE_BUCKET) it stops
// relying on systemd to keep instance processes alive and instead runs them
// as its own children, on demand — "units are for the edge, instances are its
// children". A request for a host with no live process claims the per-host S3
// single-writer lease, restores the DB from S3 if the local copy is missing,
// spawns the instance, and proxies; after an idle window the instance is
// checkpointed back to S3, the lease released, and the process stopped. State
// parks in S3; processes exist only while someone's talking.
//
// Ported from lchat activator.go (Eleven's activator, issue #550 Part 4 v2),
// with the mesh (multi-box placement: gossip/HRW, peer proxying) dropped for
// single-box v1 — `// mesh seam:` comments mark every point where it slots
// back in. Membership of one: hash picks self, leases are uncontested except
// against a previous incarnation of ourselves.
//
// Sidecar co-spawn (LEGACY, but LIVE — Eleven runs on it today): a host with
// a sidecar staged (/data/<host>-sidecar.env) gets its `sidecar run` child
// spawned right after the instance turns healthy, and SIGTERMed first at
// sleep — the sidecar exists exactly while its instance is awake, so its
// lifecycle solves itself. Its own connections flow back through the edge
// (the box pins its hosted names to loopback), so the sidecar marks them
// (SidecarHeader) and the activator excludes them from the idle clock —
// otherwise the pair could never sleep, and an orphaned sidecar's reconnect
// loop would wake the instance right back (see ServeHTTP).
//
// Standalone sidecar members (NATIVE): a route with registry.Route.IsSidecar
// true is its own record, its own registry row, with no parent instance to
// hang its lifecycle from — WakeWorker is its request-less entry point, and
// ExecBackend.StartWorker (not StartSidecar) is what spawns it. This is the
// package's other sidecar mechanism, not a replacement for the co-spawn
// above: a host may carry one or the other but never both (the provisioning
// reconciler's mint-time check in internal/provision/sidecar_reconcile.go
// refuses to mint a native row for a host that already has a co-spawn env
// file staged). Retiring the co-spawn path is a follow-up, gated on Eleven's
// own migration onto native sidecar records.
//
// Opt-in is PER HOST (routes.hibernate): with the bucket env set, only
// flagged instances are activator-managed; the rest stay on their systemd
// units and are proxied straight. That's what makes a production canary safe —
// the env ships fleet-wide as a no-op, then flipping one host's flag migrates
// exactly one instance: its unit stops, the flag flips, and the first request
// wakes it here.
//
// The due-time index: scheduled server-side work fires on time even while its
// instance sleeps. At sleep the stop path asks the instance for its earliest
// pending delivery (GET /api/next-due over the socket, admin-gated) and
// records it in the registry (routes.next_due — persisted so it survives
// agent restarts); the sweeper wakes any sleeping host whose time has
// arrived, with no client request. Once up, the instance's own scheduler
// delivers, and the normal idle window puts it back to sleep — which
// re-records the NEXT due time, so the cycle maintains itself. Worst-case
// lateness ≈ one activator sweep interval (≤30s) + spawn + one scheduler tick.
package activator

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"path"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// SidecarHeader marks every request a co-spawned sidecar makes to its own
// instance through the edge. Marked traffic is activity-invisible: served
// while the instance is awake, refused (503) while it isn't, never a wake.
// The marker is deliberately spoofable-but-harmless: it only DOWNGRADES the
// sender's own requests (no activity credit; 503 while the instance sleeps).
// It cannot suppress other clients' activity stamps and cannot wake anything,
// so an outsider sending it gains nothing.
const SidecarHeader = "X-Carlos-Sidecar"

// Config is the agent-process opt-in, read from env. Nil (bucket unset) keeps
// proxy-only behavior byte-identical: instances run under systemd, the edge
// only proxies.
type Config struct {
	HostWrites hostwrite.Config
	Claims     *ClaimRuntime
	Bucket     string        // CARLOS_HIBERNATE_BUCKET (required to enable)
	Region     string        // CARLOS_HIBERNATE_REGION (optional; AWS default chain resolves "")
	Prefix     string        // CARLOS_HIBERNATE_PREFIX (optional key prefix inside the bucket)
	Idle       time.Duration // CARLOS_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s")
	// AgentSocket is the unix socket a worker child dials to reach back —
	// the direction that makes a sidecar possible at all: the agent listens,
	// children call in; the platform never dials a sidecar. Handed to every
	// worker as CARLOS_AGENT_SOCKET. Empty falls back to DefaultAgentSocket.
	AgentSocket string
}

// DefaultAgentSocket is where the agent listens for its worker children.
const DefaultAgentSocket = "/run/carlos/agent.sock"

// ConfigFromEnv reads the CARLOS_HIBERNATE_* env vars. A nil, nil return
// means the bucket is unset: hibernation disabled, don't construct a Service.
func ConfigFromEnv() (*Config, error) {
	bucket := strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_BUCKET"))
	if bucket == "" {
		return nil, nil
	}
	cfg := &Config{
		Bucket:      bucket,
		Region:      strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_REGION")),
		Prefix:      strings.Trim(strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_PREFIX")), "/"),
		Idle:        15 * time.Minute,
		AgentSocket: DefaultAgentSocket,
	}
	if v := strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_IDLE")); v != "" {
		d, err := time.ParseDuration(v)
		if err != nil || d <= 0 {
			return nil, fmt.Errorf("CARLOS_HIBERNATE_IDLE %q: want a positive duration like 15m or 90s", v)
		}
		cfg.Idle = d
	}
	return cfg, nil
}

// hibernatePrefix is the per-instance S3 key prefix: <Prefix>/<host>/ (the
// lease's lock.json is colocated under it by the lease package).
func (c Config) hibernatePrefix(host string) string {
	return HibernatePrefix(c.Prefix, host)
}

// HibernatePrefix is that layout as a free function, exported for the
// verification pass (internal/ops) that restore-drills these replicas. It is
// a function rather than a comment because a drill that guessed the prefix
// wrong would report "no replica" for a perfectly good backup — coverage
// theatre of exactly the kind the drill exists to end.
func HibernatePrefix(prefix, host string) string {
	return path.Join(prefix, host)
}

// mesh seam: Eleven also derived meshPrefix() here — <Prefix>/mesh/members,
// the membership heartbeats' home beside the per-instance prefixes (the
// bucket is the mesh's only shared substrate; instance prefixes are public
// FQDNs and the bare label "mesh" never is one, so they can't collide).

// --- backend seam ------------------------------------------------------------
//
// The state machine's side effects (S3 lease, restore, process spawn) sit
// behind these interfaces so the machine itself is unit-testable with fakes —
// wake coalescing, idle→sleep, and lease-lost stand-down need no AWS.

// Backend is the side-effect seam (Eleven's actBackend). ExecBackend is the
// production implementation for exec-child instances; UnitBackend drives
// carlos-app@ unit tenants; SplitBackend composes the two by the route's
// Backing. Every method takes the full route (Eleven passed bare host/dbPath
// strings) precisely so that composition can dispatch without a registry
// read of its own.
type Backend interface {
	// Acquire claims the per-host single-writer lease. Contention surfaces as
	// errLeaseHeld so the edge can answer 503 + Retry-After.
	Acquire(ctx context.Context, rt registry.Route) (Lease, error)
	// Restore materializes the local DB (rt.DBPath; the backend defaults it)
	// from S3 when it's absent (a no-op when the warm-cache copy is still on
	// disk). fresh reports that the local DB was absent AND no replica
	// existed — a first boot, never a restore. The activator runs the route's
	// Seed exactly then.
	Restore(ctx context.Context, rt registry.Route) (fresh bool, err error)
	// Seed runs the route's declared first-boot initialization (rt.Seed, a
	// JSON argv array) against the instance's own database, before anything
	// has spawned. Called by doWake only on a fresh boot with a seed
	// declared, so it must be safe to assume an empty database — and its
	// failure fails the wake, because a half-seeded instance must not serve.
	Seed(ctx context.Context, rt registry.Route) error
	// Start spawns the instance process and waits until its socket accepts. It
	// also returns the build it execed (the live/<host> symlink target read at
	// spawn), so the sweep can later spot a repoint under a still-running child
	// and converge it. Empty when unknown — the sweep treats that as "don't
	// cycle" (fail-safe).
	Start(ctx context.Context, rt registry.Route) (Proc, string, error)
	// NextDue asks a running instance for its earliest pending scheduled
	// delivery (unix seconds; 0 = nothing scheduled). Queried at sleep time,
	// while the socket still answers, to feed the due-time index.
	NextDue(ctx context.Context, rt registry.Route) (int64, error)
	// HasSidecar reports whether the route's host has a sidecar staged on
	// this box.
	HasSidecar(rt registry.Route) bool
	// StartSidecar spawns the host's sidecar. Fire-and-forget: a sidecar is a
	// pure outbound client with no socket, so there's nothing to health-check —
	// its own connect/backoff loop recovers from a slow start.
	StartSidecar(ctx context.Context, rt registry.Route) (Proc, error)
	// StartWorker spawns a child that binds NOTHING: no socket, no port. No
	// health probe is possible or wanted — readiness is the caller's settle
	// window, or the child's own call over the agent socket. rt.Args (a JSON
	// argv array) names what to run; empty or unparseable is refused rather
	// than exec'd bare. env is merged over the route's own env layers and
	// wins conflicts — it carries the per-wake sidecar token, which the
	// tenant's own env file must never be able to override.
	StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error)
}

// DueStore persists the due-time index (Eleven's actDueStore) — the registry
// in production (routes.next_due), so a sleeping host's timer survives agent
// restarts instead of dying with the process. *registry.Registry satisfies it
// directly.
type DueStore interface {
	SetNextDue(host string, due int64) error
	// ClearNextDueIf consumes a FIRED due time — see the registry's own doc
	// comment. Conditional on the row still holding the value that fired, so
	// a child that posted its next due time during this very wake keeps it.
	ClearNextDueIf(host string, due int64) error
	DueInstances(now int64) ([]registry.Route, error)
	// AlwaysOnWorkers lists the non-hibernating sidecar rows this box carries,
	// so the sweep can start one that appeared after the agent's boot pass —
	// which runs once, at arming, and never again.
	AlwaysOnWorkers() ([]registry.Route, error)
	// EarliestTimers is the hibernation guard's input: host -> soonest armed
	// timer. Read with DueInstances, before the sweep takes a.mu.
	EarliestTimers() (map[string]int64, error)
}

// RouteLookup is the registry read doWake performs before Acquire (design
// §6.8): the edge hands a wake the route it had cached at request time, which
// can be stale by the time the wake actually runs (a config edit landed in
// between); doWake re-reads the committed row so it spawns against
// whatever's really in the registry, not a snapshot. *registry.Registry
// satisfies this directly with its own Get.
type RouteLookup interface {
	Get(host string) (*registry.Route, error)
}

// leaseReleaseBudget bounds a wake bail-out's lease release: one small S3
// DELETE, the same five seconds lease.Replicator.Close gives its own.
const leaseReleaseBudget = 5 * time.Second

// releaseLease gives a failed wake's lease back under a budget of its own,
// never the wake's context. A fence or Shutdown cancels that context, and
// the real release (an S3 DeleteObject) fails at once on a dead one while
// spending the handle anyway, which leaves the lease held for its whole TTL
// and the host refusing wakes until then. WithoutCancel keeps the
// context's values and drops only its cancellation and deadline.
func releaseLease(ctx context.Context, lease Lease) error {
	rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), leaseReleaseBudget)
	defer cancel()
	return lease.Release(rctx)
}

// releaseRetrier is the OPTIONAL half of Lease that a failed Release
// leaves something to retry with (#292, Astra P3): lease.Held.Release
// spends its handle even when the DELETE fails, so the same conditional
// delete is re-issued through RetryRelease. execLease has it.
type releaseRetrier interface {
	RetryRelease(ctx context.Context) error
}

// owedRelease is a lease whose release failed: kept, and retried by
// finishPendingStops (the sweep, and Shutdown's polling) until it lands or
// Shutdown's budget ends. busy marks a retry in flight; lastLog paces the
// failure lines. Both under a.mu.
type owedRelease struct {
	host    string
	lease   releaseRetrier
	busy    bool
	lastLog time.Time
}

// giveBack is releaseLease for a lease the agent holds for host, with a
// failure kept as an obligation instead of forgotten (#292, Astra P3).
// It reports whether the release landed now.
func (a *Service) giveBack(ctx context.Context, host string, lease Lease) bool {
	err := releaseLease(ctx, lease)
	if err != nil {
		a.oweRelease(host, lease, err)
	}
	return err == nil
}

// oweRelease records a failed release of host's lease for retry. A lease
// that cannot retry is left to its TTL, as before. Never called with a.mu
// held.
func (a *Service) oweRelease(host string, lease Lease, err error) {
	r, ok := lease.(releaseRetrier)
	if !ok {
		log.Printf("activator: %s: giving the lease back failed: %v (the lease's TTL frees it)", host, err)
		return
	}
	log.Printf("activator: %s: giving the lease back failed: %v; retrying it", host, err)
	a.mu.Lock()
	if a.owedReleases == nil {
		a.owedReleases = make(map[*owedRelease]struct{})
	}
	a.owedReleases[&owedRelease{host: host, lease: r, lastLog: time.Now()}] = struct{}{}
	a.mu.Unlock()
}

// retryOwedReleasesLocked starts one retry of every owed release that has
// none in flight. The caller holds a.mu.
func (a *Service) retryOwedReleasesLocked() {
	for o := range a.owedReleases {
		if o.busy {
			continue
		}
		o.busy = true
		go a.retryOwedRelease(o)
	}
}

// retryOwedRelease is one conditional retry, under the release budget of
// its own. Landed, or no longer ours to release, the obligation goes.
func (a *Service) retryOwedRelease(o *owedRelease) {
	ctx, cancel := context.WithTimeout(context.Background(), leaseReleaseBudget)
	err := o.lease.RetryRelease(ctx)
	cancel()
	// A record past its own expiry is retired even though the delete still
	// fails (lease.ErrReleaseLapsed): the lease is free to any acquirer by
	// then, and a persistent refusal must not keep obligations forever.
	lapsed := errors.Is(err, lease.ErrReleaseLapsed)
	a.mu.Lock()
	o.busy = false
	logIt := err == nil || lapsed || time.Since(o.lastLog) >= 30*time.Second
	if err == nil || lapsed {
		delete(a.owedReleases, o)
	} else if logIt {
		o.lastLog = time.Now()
	}
	a.mu.Unlock()
	switch {
	case err == nil:
		log.Printf("activator: %s: lease given back on retry", o.host)
	case lapsed:
		log.Printf("activator: %s: lease release retired, never deleted: %v", o.host, err)
	case logIt:
		log.Printf("activator: %s: retrying the lease's release: %v", o.host, err)
	}
}

// leaseHealth is the OPTIONAL half of Lease: a held lease that can say
// whether it is still held (no failed renewal, not spent). execLease has it
// (lease.Held.Err). A hold's pinned wake asks it at its exec, before the
// lease has become a replica (final review F7).
type leaseHealth interface {
	Err() error
}

// leaseLost is l's liveness as an ownership refusal: errSwapLeaseLost
// (gateNoOwnership) when l says it is no longer held, nil when it is held
// or cannot say.
func leaseLost(l Lease) error {
	if h, ok := l.(leaseHealth); ok {
		if err := h.Err(); err != nil {
			return fmt.Errorf("%w: %v", errSwapLeaseLost, err)
		}
	}
	return nil
}

// Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's
// actLease).
type Lease interface {
	// Replicate trades the lease for running replication of the (now present)
	// local DB; the lease handle is spent on success. syncInterval is the
	// route's replication-class tick (replicationSyncInterval; 0 = stream).
	Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (Rep, error)
	// Release gives the lease up — the bail-out when a wake fails partway.
	Release(ctx context.Context) error
}

// Rep is running replication coupled to the lease (Eleven's actRep).
type Rep interface {
	// Err reports lost leadership (a lease renewal failed).
	Err() error
	// Close does the final checkpoint/sync and releases the lease.
	Close(ctx context.Context) error
}

// Proc is a running instance child process (Eleven's actProc).
type Proc interface {
	// Stop terminates gracefully (SIGTERM, bounded wait, then SIGKILL).
	Stop(ctx context.Context) error
	// Exited reports whether the process has died on its own.
	Exited() bool
}

// procCrasher is the OPTIONAL half of Proc: a child that can say how it
// exited. A hibernating worker's exit is ambiguous without it — design §7.2
// says a crash restarts, unqualified, while a clean exit from a bot that has
// finished its work is exactly what parking is for — and exit status is the
// one fact that tells them apart.
//
// Optional rather than a widening of Proc because it is only ever consulted
// for a worker, and ExecBackend (the only backend that spawns one;
// UnitBackend refuses StartWorker outright) is the only implementation that
// can answer. Kept additive so no other Proc — production or test — has to
// grow a method it cannot answer honestly.
type procCrasher interface {
	// Crashed reports a DEFINITELY non-zero exit. Only ever true for a child
	// that has actually exited.
	Crashed() bool
}

// procCrashed asks a Proc whether it crashed, defaulting to NO for a child
// that cannot say. That default is the safe direction: a crash misread as a
// finish parks a worker its schedule or an operator can still wake, while a
// finish misread as a crash restarts a bot that had nothing left to do —
// forever.
func procCrashed(p Proc) bool {
	c, ok := p.(procCrasher)
	return ok && c.Crashed()
}

// Replication classes — the registry's routes.replication vocabulary
// ("" | stream | batched | daily), mapped to litestream sync intervals by
// replicationSyncInterval.
//
// Aliases, not copies: internal/registry owns the vocabulary now that
// `carlos route --replication` writes the column against it (discussion
// #10), and two independent spellings of the same three strings is exactly
// how a setter and a reader drift into disagreeing about what `daily` means.
const (
	replicationStream  = registry.ReplicationStream  // change-driven ~1s sync — loss window seconds
	replicationBatched = registry.ReplicationBatched // sync at most every 5m — loss window minutes
	replicationDaily   = registry.ReplicationDaily   // sync at wake + every 24h — loss window up to a day
)

// replicationSyncInterval maps a class to the litestream replica sync
// interval it buys. 0 means "keep litestream's default" (1s, change-driven —
// idle ticks make zero S3 calls), which is what stream and the unset ""
// deliberately share so pre-ladder registries keep today's behavior exactly.
func replicationSyncInterval(class string) time.Duration {
	switch class {
	case replicationBatched:
		return 5 * time.Minute
	case replicationDaily:
		return 24 * time.Hour
	}
	return 0
}

// --- state machine -----------------------------------------------------------

// Per-host lifecycle. An absent entry is the fourth state: asleep.
//
//	asleep --request--------------------> waking   (one wake; concurrent
//	waking --wake ok--------------------> awake     requests coalesce on it)
//	waking --wake failed----------------> asleep   (error served; next request retries)
//	awake  --idle timeout / lease lost /
//	          child died----------------> stopping (final checkpoint, release lease)
//	stopping --stop done----------------> asleep   (local DB left as a warm cache)
//
// A request that lands on waking waits (bounded) for the wake; one that lands
// on stopping waits for the stop, then re-enters as a fresh wake.
type actState int

const (
	actWaking actState = iota
	actAwake
	actStopping
)

type actEntry struct {
	claimResult *wokenInstance
	// probe is set for an entry ProbeWake made (moves Plan 3 Task 5): the
	// claim path runs with the probe's policy, and the entry is held for
	// the move while its registry row is.
	probe *probeEntry
	// RecoveryProbe retains its separate recovery policy. recoveryPolicy is the
	// caller's probe policy, set at construction and never written again;
	// nil for every other entry, whose wake is main's. probeAdmitted is the
	// admission the probe started under, written by the wake before the
	// awake transition. probeHeld keeps requests, bounces and the idle clock
	// off the entry until the claim says the source opened on that session.
	recoveryPolicy *StartPolicy
	probeAdmitted  hostclaim.Observed
	probeHeld      bool
	probeReviewing bool
	probeObsolete  bool
	probeFence     uint64
	// bounceAfterWake is a Bounce that arrived while this INSTANCE was still
	// waking. The wake had already read its config and execed its build, so
	// the bounce's reason has not reached the child; wake() spends it the
	// moment the wake finishes, before any waiting request is admitted.
	bounceAfterWake bool
	host            string
	route           registry.Route
	state           actState

	ready       chan struct{} // waking: closed when the wake resolves; then check err
	err         error         // waking: the wake's outcome
	stopped     chan struct{} // stopping: closed when the stop completes
	wakeCancel  context.CancelFunc
	stopOutcome *stopObservation

	// stopWhy is why this entry is stopping (beginStopLocked), written under
	// a.mu; finishStop hands it to a run it interrupts (run spec §5.3).
	stopWhy string

	// cmd is the run attached to this instance (run spec §5.3), written
	// under a.mu. While it is set the idle sweep never stops the entry;
	// every other stop stops and settles its unit in finishStop, before
	// replication closes, and a unit not proven settled keeps the stop
	// pending with replication and the lease held (merge ruling M1).
	cmd *cmdHold
	// cmdLeft holds runs released before their unit was proven settled
	// (cmdRun.Release): their Stop failed or never came. finishStop settles
	// them as it does cmd, before replication closes. Written under a.mu.
	cmdLeft []*cmdHold

	sidecarUncertain bool

	// Activity tracking: an instance is idle only when nothing is in flight
	// AND the idle window has passed since the last request finished. In-flight
	// counting (not just timestamps) is what keeps an instance with one open
	// WebSocket awake — the proxy's ServeHTTP blocks for the tunnel's life.
	// (Sidecar-marked requests register in neither — see BeginSidecarRequest.)
	inflight   int
	lastActive time.Time

	rep  Rep
	proc Proc

	// Convergence: the `live/<host>` symlink target this child was spawned
	// against. A running process never re-reads the symlink, so once a ship
	// repoints it, an awake child keeps serving the old build until it's
	// cycled. spawnVersion lets the sweep notice the drift; staleSince is when
	// it first did, so a brief transient (a symlink moved then rolled straight
	// back) doesn't trigger a needless cycle. Both empty/zero until the wake
	// resolves.
	spawnVersion string
	staleSince   time.Time

	// gen is the generation this entry's process is (zero-downtime spec
	// §7.1): its number, socket, control socket and build. Set under a.mu at
	// the awake transition and never changed after it in this plan (plan 3's
	// switch replaces it). Zero for a worker and before the wake resolves.
	gen Gen

	// Sidecar co-spawn: the sidecar exists exactly while its instance is
	// awake. All sidecar fields are written under a.mu in the awake state only
	// (wake's attach, the sweeper's restart); a stop flips the state under the
	// same lock first, so stop() owns them uncontended from then on.
	sidecar         Proc          // the co-spawned `sidecar run` child; nil = none running
	sidecarWanted   bool          // the host had a sidecar staged at wake time
	sidecarStarted  time.Time     // when the current sidecar spawned (backoff reset)
	sidecarBackoff  time.Duration // last restart delay (doubles, capped)
	sidecarRetryAt  time.Time     // earliest next restart attempt
	sidecarStarting bool          // a restart's backend.StartSidecar call is in flight, off a.mu (design §6.7)
	// sidecarStartDone is closed by the in-flight restart goroutine when it
	// finishes (installed or discarded), non-nil exactly while
	// sidecarStarting is true. stop() waits on it, bounded by
	// sidecarStopTimeout, instead of polling sidecarStarting.
	sidecarStartDone chan struct{}
	// sidecarGen is the generation whose build the legacy sidecar runs;
	// written under a.mu.
	sidecarGen uint64

	// pendingStop marks a stop whose proc.Stop returned ErrStillRunning
	// (design §6.3): termination is unproven, so stop() returned without
	// releasing the lease or forgetting the entry, and the entry stays in
	// stopping. sweepOnce (and, during Shutdown, finishPendingStops) polls
	// proc.Exited() for every pendingStop entry and finishes the stop
	// (finishPendingStop) the moment it's proven.
	pendingStop bool
	// stopRetrying is true while a retried proc.Stop call for a still-running
	// pendingStop entry is in flight, so finishPendingStops does not launch a
	// second one on top of it. stopRetriedAt is when the last retry was
	// issued; finishPendingStops waits actStopTimeout between retries, the
	// same cadence proc.Stop itself is bounded by.
	stopRetrying  bool
	stopRetriedAt time.Time

	// ops is this entry's operation set (zero-downtime spec §7.4): every
	// generation other than the current one that is spawning,
	// health-checking or draining. Written under a.mu. Empty unless a swap
	// has run on this entry, and swaps need plan 1's generations switch.
	ops opSet
	// pin is a hold swap's chosen build (zero-downtime spec §7.8). The wake
	// that follows the hold's stop execs exactly this staged artifact rather
	// than the live link. It is set at construction (holdWakeLocked), before
	// the wake goroutine starts, and never written again, like the worker
	// fields. The one exception is a claims-on planned wake (plan 5,
	// doClaimWake), whose own wake goroutine sets it under a.mu before its
	// start policy is built; every reader is that goroutine or holds a.mu.
	pin *Gen
	// pinCheck is a hold's attempt checks, re-run under a.mu at the pinned
	// exec. Set at construction (holdWakeLocked), like pin.
	pinCheck func() error
	// pinExec is the pinned wake's execution evidence, set (under a.mu,
	// markPinExec) from its spawn's RESULT: a returned process, or an
	// unproven-start sentinel (final review F10). It is set before any
	// cleanup of the failed wake. Unset, nothing of the build ran, so the
	// hold is abandoned, never down (spec §6.4), and its refusal is no
	// member's event.
	pinExec bool
	// holdNext is a hold swap waiting for this entry's stop to finish.
	// finishStop installs its pinned wake. Written under a.mu.
	holdNext *holdSpec
	// pinFailed is the pinned build a claims-on start ran and saw fail
	// before it was ready, and pinFailReason why (zero-downtime spec §8.2).
	// startAdmitted writes them under a.mu, only for a failure classifyStep
	// calls the build's; plan 5 reports them. Nil when the pinned build
	// served, or never ran.
	pinFailed     *Gen
	pinFailReason string
	// settleCancel, on a hold canary's pinned entry, ends its settle
	// observation. beginStopLocked calls it while the process has not
	// exited, so an intentional stop is never read as a crash (spec §6.4).
	// Written under a.mu.
	settleCancel context.CancelFunc
	// genReaping is true while reapCurrentGen is running for this pending
	// stop, so finishPendingStops does not start a second one. Written
	// under a.mu.
	genReaping bool
	// sidecarJoin is a sidecar start a flag-off terminal stop stopped
	// waiting for (sidecarStartJoinWait) while it was still in flight: the
	// stop stays pending, with the lease, until it closes (Task 8, F6 plan
	// 5), and only then is the host cgroup proven empty. Written under a.mu.
	sidecarJoin chan struct{}
	// unprovenSidecars are sidecars of this stopping entry, generations off,
	// whose stop did not prove their exit: the co-stopped sidecar, or one a
	// restart brought back after the stop began (fix wave 1, item 1). The
	// lease is not released while one of them has not exited, whatever the
	// layout: a nested row has no host-cgroup proof to reach them. The sweep
	// re-stops them (reapCurrentGen). Written under a.mu.
	unprovenSidecars []Proc
	// launchLayout is the layout the wake launched this entry's child on
	// (wokenInstance.launchLayout), set at install. A flag-off terminal stop
	// decides from it whether the host cgroup must be proven empty
	// (stopHostProven, hostCgroupIsTheWakes). "" requires the proof. Written
	// under a.mu.
	launchLayout Layout

	// The rollout scheduler's view of this entry (zero-downtime plan 5).
	// plan is set by the wake goroutine before the spawn (doWake,
	// doClaimWake, checkPlan's replan) under a.mu, and read without it only
	// by that goroutine. firstWake is begin()'s cue to wait out the
	// first-wake budget; wakeStart (real time, set at the wake's start)
	// anchors B's health window. unexpectedExit marks a stop that began
	// because the child died on its own. firstWatch cancels a first wake's
	// observation when an intentional stop begins, and firstWatchCtx is
	// that observation's context, both registered in the awake transition
	// (armFirstWatchLocked) and read by watchFirstWake's goroutine. All but
	// plan are written under a.mu.
	plan           WakePlan
	firstWake      bool
	wakeStart      time.Time
	unexpectedExit bool
	firstWatch     context.CancelFunc
	firstWatchCtx  context.Context
	// firstWatchRelease releases the host's outcome fence the observation
	// took when it was armed (final review F3), once its terminal event has
	// been delivered. Under a.mu.
	firstWatchRelease func()

	// Standalone sidecar ("worker") entries. A worker is this entry's own
	// proc, not a passenger on someone else's: the co-spawn fields above stay
	// zero for it, and it never becomes a co-spawn parent either.
	//
	// worker, workerIdle, workerMaxRun and readyc are set once, at entry
	// construction, BEFORE the wake goroutine is launched, and never written
	// again — so wake/doWake read them off-lock safely (the `go` statement is
	// the happens-before edge). Everything else below is written under a.mu,
	// like every other post-wake field: workerToken and spawnedAt at the awake
	// transition and on a supervised restart, the rest by the agent socket.
	worker       bool          // this entry is a sidecar: no socket, no requests
	workerIdle   time.Duration // 0 = fall back to cfg.Idle
	workerMaxRun time.Duration // 0 = no ceiling
	readyc       chan struct{} // closed when the child reports ready (ends the settle window)
	readyNoted   bool          // guards readyc's one close

	// dueFired is the next_due value that TRIGGERED this wake, and only for a
	// worker woken by the due-time sweep (0 everywhere else). A fired timer is
	// one-shot: the wake consumes the schedule it acted on (consumeDueTime),
	// which is what stops a hibernating sidecar waking, idling, parking and
	// re-waking on every sweep forever. Set before the wake goroutine starts
	// and never written again, like the other construction-time worker fields.
	dueFired int64

	workerToken string    // the per-wake agent-socket token
	spawnedAt   time.Time // when this child spawned — the max_run and healthy-run clock
	// workerVersion is what the child last reported over the agent socket:
	// DISPLAY metadata only (spec §7.9). It can be empty or an app-chosen
	// string, so it never identifies the executable and is never written to
	// the registry, whose Version is root's install stamp.
	workerVersion string
	// workerServing is the build this worker's current child was exec'd
	// from, captured by the backend at the exec (spec §7.9): staged path and
	// sha256. Crash and max_run recycles respawn exactly this, never what
	// live/<host> names by then. Zero when the backend could not report it.
	// Written under a.mu at the awake transition and on every installed
	// respawn.
	workerServing release.StagedIdentity

	// Crash-restart pacing for an always-on worker, the co-spawn supervisor's
	// rule applied to a standalone member: doubling, capped, and reset by a
	// run that lasted sidecarHealthyRun.
	workerBackoff time.Duration
	workerRetryAt time.Time
	// workerRespawning is set while respawnWorkerLocked's goroutine owns this
	// entry's child: the sweeper skips the entry entirely so the moment
	// between the old child's stop and the new one's spawn — when the proc
	// reads as Exited — is not mistaken for a crash. respawnDone is closed
	// when that goroutine finishes, and is what stop() waits on: a stop must
	// not release the lease while a spawn against the same DB is in flight.
	workerRespawning bool
	respawnDone      chan struct{}

	// The socketless attempt's state (zero-downtime plan 5 Task 11,
	// ReplaceWorker). attemptBuilds names the builds an attempt in flight may
	// exec (its own and its fallback), so the generations file pins them
	// before anything execs them (Task 12); startWorker's pending generation
	// already publishes the exact build about to exec, so this is for the
	// fallback above all. retireTook is how long the last respawn took to
	// retire its old child: the "swapped" event's drain time for a worker
	// (spec §9.4).
	//
	// attemptGate is an attempt's §7.4 checks (swapChecksLocked with the
	// attempt's revision), and holdServing marks a ReplaceWorker fallback
	// whose publication waits for the scheduler's failure record (spec
	// §8.1). Both arm the NEXT respawnWorkerLocked only: it takes them when
	// it is called, so a later respawn (a restore, a fallback, a crash
	// restart) never inherits them, and a ReplaceWorker that stops waiting
	// cannot disarm a respawn still running. attemptVeto is that respawn's
	// answer when the gate refused, read and cleared by ReplaceWorker, and
	// attemptExec its execution evidence (final review F2): the respawn
	// started a process of the attempt's build, whether or not it was then
	// installed or shed. A supersession restores the replaced build only
	// when it is false.
	// heldServing is the publication holdServing kept back, delivered by
	// PublishHeldServing. workerInc is the worker child's incarnation,
	// bumped at every install (a wake, a respawn): an attempt counts only if
	// it moved, and a publication is delivered only for the latest.
	// parkedInc is the incarnation a clean-exit park stopped
	// (markCleanParkLocked), 0 for none: the only stop after which
	// ReplaceWorker counts an exited one-shot as a good build. All written
	// under a.mu.
	attemptBuilds []string
	retireTook    time.Duration
	attemptGate   func() error
	attemptVeto   error
	attemptExec   bool
	holdServing   bool
	heldServing   *heldServing
	workerInc     uint64
	parkedInc     uint64

	// shed is a respawn child, refused at its installation with generations
	// off, whose exit is not proven yet (final review R2-1). With
	// generations on the operation set holds such a child as a member. The
	// entry keeps it, and its build pinned, until the sweep proves it
	// exited (reapShedLocked): a terminal stop keeps the lease until then,
	// and no respawn execs beside it.
	shed *shedChild
}

// shedChild is a respawn child that was exec'd and then refused at its
// installation, kept until its exit is proven (final review R2-1). Every
// field is written under a.mu. reaping marks a re-stop or proof in flight.
type shedChild struct {
	proc      Proc
	build     release.StagedIdentity
	reaping   bool
	retriedAt time.Time
}

// Sidecar restart pacing — Restart=always semantics without a supervisor: the
// sweeper redoes a dead sidecar with exponential backoff, capped, and a run
// that survived sidecarHealthyRun earns a fresh backoff (so one crash a day
// never escalates to the cap).
const (
	sidecarRestartBase = 2 * time.Second
	sidecarRestartMax  = time.Minute
	sidecarHealthyRun  = time.Minute
)

// sidecarStopTimeout serves two related bounds. First, its name: the
// sidecar's own stop at sleep time. `sidecar run` has no drain handler, so
// the SIGTERM is effectively immediate, and the SIGKILL fallback behind this
// timeout is insurance. Second (design §6.7), as sidecarStartJoinWait's
// default: how long a flag-off stop() waits on sidecarStartDone for an
// in-flight sidecar restart to land before carrying on with whatever
// e.sidecar currently is.
const sidecarStopTimeout = 5 * time.Second

// sidecarStartJoinWait is how long a flag-off terminal stop waits for an
// in-flight sidecar start before it carries on with the rest of the stop.
// It is a promptness bound only: a start still in flight then keeps the
// stop pending, with the lease, until it returns and has stopped what it
// started, and the host cgroup is proven empty after that (Task 8, F6
// plan 5). A var so tests can shrink it.
var sidecarStartJoinWait = sidecarStopTimeout

// sidecarSettle is the standalone worker's readiness window — what stands in
// for the health probe a child that binds nothing can never answer. A worker
// is up once it either reports ready over the agent socket (NoteWorkerReady)
// or survives this long without exiting; a child that exits INSIDE the window
// is a failed wake, released and surfaced exactly like a failed instance
// start. A var, not a const, so tests can shrink it — spawnHealthTimeout's
// precedent in this package (backend_exec.go).
var sidecarSettle = 10 * time.Second

// settlePoll derives the settle window's liveness poll: Proc reports death
// through Exited(), not a channel, so the window polls. Scaled off the window
// so a shrunken test window still gets several looks, and capped so the
// production window doesn't spin.
func settlePoll(window time.Duration) time.Duration {
	poll := window / 8
	if poll > 250*time.Millisecond {
		poll = 250 * time.Millisecond
	}
	if poll < time.Millisecond {
		poll = time.Millisecond
	}
	return poll
}

// maxWorkerStartsPerSweep bounds the sweep's always-on arm — how many
// stranded always-on sidecars it starts per tick. The same trade host's
// workerBootConcurrency makes: bounded fan-out, not one-at-a-time and not
// the whole box at once, because each wake claims a lease and restores a
// database. The remainder is started on the next sweep.
const maxWorkerStartsPerSweep = 8

// stopDiscardedSidecar stops a sidecar process a restart attempt returned
// but could not install. The wake it belonged to moved on (stopped, or
// replaced by a fresh wake of the same host) while the call was in flight.
// Bounded by sidecarStopTimeout so a wedged Stop can't leak the goroutine
// forever; fire-and-forget, since nothing is waiting on this beyond the
// process actually dying.
func stopDiscardedSidecar(sc Proc) {
	if sc == nil {
		return
	}
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
		defer cancel()
		sc.Stop(ctx) //nolint:errcheck
	}()
}

// stopReturnedSidecar stops a sidecar a flag-off restart brought back for
// an entry that is stopping (Task 8, F6 plan 5): synchronously, bounded by
// sidecarStopTimeout, so the start's join closes only after it. A sidecar
// whose stop did not prove its exit (sidecarStopUnproven) is not forgotten
// (fix wave 1, item 1): it is kept on the entry, and the stop keeps its
// lease until it has exited (sidecarsExited). The stop's host proof
// follows, and with a delegated cgroup that proof also reaches whatever
// this stop left. Never called with a.mu held.
func (a *Service) stopReturnedSidecar(e *actEntry, sc Proc) {
	if sc == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
	err := sc.Stop(ctx)
	cancel()
	if !sidecarStopUnproven(sc, err) {
		return
	}
	log.Printf("activator: %s: the sidecar a restart brought back for a stopping entry is not proven gone (%v); keeping the lease until it has exited", e.host, err)
	a.mu.Lock()
	e.unprovenSidecars = append(e.unprovenSidecars, sc)
	a.mu.Unlock()
}

// sidecarStopUnproven reports whether a flag-off sidecar stop left its
// termination unproven: the stop failed (a nested unit's ErrStillRunning,
// a signal that could not be sent) and the process has not exited. A stop
// that returned nil is the stop's own proof, as it is for the main
// process's flag-off stop: every real Proc returns nil only once its
// process is gone.
func sidecarStopUnproven(sc Proc, err error) bool {
	return err != nil && !sc.Exited()
}

// sidecarsExited reports whether every sidecar a flag-off stop kept
// unproven (actEntry.unprovenSidecars) has exited: the sidecar's own stop
// proof, which is the only one a nested row has. With restop it first
// stops again, bounded, each one still running (the sweep's retry,
// reapCurrentGen). Proven ones are forgotten. Never called with a.mu held.
func (a *Service) sidecarsExited(e *actEntry, restop bool) bool {
	a.mu.Lock()
	kept := slices.Clone(e.unprovenSidecars)
	a.mu.Unlock()
	if len(kept) == 0 {
		return true
	}
	var gone []Proc
	for _, sc := range kept {
		if restop && !sc.Exited() {
			ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
			if err := sc.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
				log.Printf("activator: %s: re-stop a sidecar not proven gone: %v", e.host, err)
			}
			cancel()
		}
		if sc.Exited() {
			gone = append(gone, sc)
		}
	}
	a.mu.Lock()
	e.unprovenSidecars = slices.DeleteFunc(e.unprovenSidecars, func(p Proc) bool { return slices.Contains(gone, p) })
	left := len(e.unprovenSidecars)
	a.mu.Unlock()
	if left > 0 {
		log.Printf("activator: stop %s: %d sidecar(s) it stopped have not exited; keeping the lease until they have", e.host, left)
	}
	return left == 0
}

// discardSidecar is how a sidecar start that lost its entry is shed. With
// generations off it is today's stopDiscardedSidecar, fire-and-forget. With
// them on it is synchronous and bounded, and the caller runs it before it
// closes the join (sidecarStartDone) a terminal stop waits on (spec §7.4).
// A sidecar whose termination it cannot prove stays tracked
// (trackStraySidecarLocked).
func (a *Service) discardSidecar(rt registry.Route, sc Proc) {
	if sc == nil {
		return
	}
	if !a.gens {
		stopDiscardedSidecar(sc)
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
	err := sc.Stop(ctx)
	cancel()
	if !a.sidecarTerminated(rt, sc) {
		log.Printf("activator: %s: a discarded SIDECAR is not proven gone (%v); keeping it tracked", rt.Host, err)
		a.mu.Lock()
		a.trackStraySidecarLocked(rt, sc)
		a.mu.Unlock()
	}
}

var errSidecarUnproven = errors.New("a legacy sidecar of this host is not proven gone")

// sidecarKiller is the backend's legacy-sidecar kill: whatever is left in
// the host's sidecar cgroup is killed, and it returns once that cgroup is
// empty.
type sidecarKiller interface {
	KillSidecar(ctx context.Context, rt registry.Route) error
}

// sidecarTerminated is THE termination proof for a legacy sidecar (spec
// §7.4, §7.5), as genTerminated is for a generation: its leader has exited
// AND its sidecar cgroup is proven empty. childProc.Stop signals only the
// leader, so its exit says nothing of a process it forked. A backend with
// no sidecar kill (a test double), or one that answers errNoGenContainment
// (the sidecar shares the host cgroup, flag off), has nothing more to
// prove. A nil process never ran. Never called with a.mu held. The cgroup
// is this host's legacy sidecar's alone: one entry per host, and the
// interlock and the join mean no second sidecar of the host runs while one
// is being proven.
func (a *Service) sidecarTerminated(rt registry.Route, sc Proc) bool {
	if sc == nil {
		return true
	}
	if !sc.Exited() {
		return false
	}
	k, ok := a.backend.(sidecarKiller)
	if !ok {
		return true
	}
	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	defer cancel()
	if err := k.KillSidecar(ctx, rt); err != nil && !errors.Is(err, errNoGenContainment) {
		log.Printf("activator: %s: the sidecar's leader exited but its cgroup is not proven empty: %v", rt.Host, err)
		return false
	}
	return true
}

// trackStraySidecarLocked keeps a legacy sidecar whose termination is
// unproven: it is never forgotten (spec §7.4). While one is tracked the host
// takes no new wake (errWakeUnproven), starts no sidecar (startSidecar), and
// no stop of it releases its lease (releasable); the sweep
// (reapStraySidecarsLocked) re-stops and re-proves it. The caller holds a.mu.
func (a *Service) trackStraySidecarLocked(rt registry.Route, sc Proc) {
	if a.straySidecars == nil {
		a.straySidecars = make(map[string][]*straySidecar)
	}
	a.straySidecars[rt.Host] = append(a.straySidecars[rt.Host], &straySidecar{proc: sc, rt: rt})
}

// reapStraySidecarsLocked re-stops each stray sidecar and re-proves it off
// the lock (the proof is a kernel or root call), forgetting it only once
// sidecarTerminated holds. The caller holds a.mu.
func (a *Service) reapStraySidecarsLocked() {
	for _, strays := range a.straySidecars {
		for _, s := range strays {
			if s.reaping {
				continue
			}
			s.reaping = true
			go func(s *straySidecar) {
				ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
				if err := s.proc.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
					log.Printf("activator: re-stop stray sidecar %s: %v", s.rt.Host, err)
				}
				cancel()
				proven := a.sidecarTerminated(s.rt, s.proc)
				a.mu.Lock()
				defer a.mu.Unlock()
				s.reaping = false
				if !proven {
					return
				}
				host := s.rt.Host
				keep := slices.DeleteFunc(a.straySidecars[host], func(x *straySidecar) bool { return x == s })
				if len(keep) == 0 {
					delete(a.straySidecars, host)
				} else {
					a.straySidecars[host] = keep
				}
			}(s)
		}
	}
}

// Build convergence: how long an awake child may serve a symlink-stale binary
// before the sweep gracefully cycles it, and how many to cycle per tick.
//   - ConvergeGrace comfortably exceeds one sweep interval (≤30s) and a
//     sidecar's ~32s reconnect, so a genuinely-persistent drift is required —
//     a symlink that moved then rolled straight back never trips it, and a
//     briefly-busy host gets a chance to idle-hibernate (which converges for
//     free) first.
//   - one per tick staggers a fleet-wide ship across sweeps instead of blipping
//     every host's connections at once.
//
// Convergence is a latency/observability fix, not urgent (the supply-chain bake
// window already holds releases), so a leisurely, gentle cadence is right.
const (
	ConvergeGrace             = 90 * time.Second
	maxConvergeCyclesPerSweep = 1
)

// ConvergeDecision is the pure state machine behind build convergence: given
// the symlink target read this tick (cur/readErr), the build the child execed
// (spawnVer), and when the drift was first seen (staleSince), it says whether
// to cycle now and returns the staleSince to carry forward. Split out so the
// grace/reset logic is unit-testable without a live activator. Three cases:
//   - read error: unknown — neither cycle nor clear the grace; retry next tick.
//   - converged / rolled-back / nothing to compare (cur=="" or spawnVer=="" or
//     cur==spawnVer): clear the grace clock, don't cycle.
//   - stale (symlink moved since spawn): start the grace clock on first sight,
//     cycle once the drift has persisted past grace.
func ConvergeDecision(cur string, readErr error, spawnVer string, staleSince, now time.Time, grace time.Duration) (cycle bool, nextStale time.Time) {
	switch {
	case readErr != nil:
		return false, staleSince
	case cur == "" || spawnVer == "" || cur == spawnVer:
		return false, time.Time{}
	default:
		if staleSince.IsZero() {
			staleSince = now
		}
		return now.Sub(staleSince) >= grace, staleSince
	}
}

// wakeWaitBound caps how long a request queues on someone else's wake or stop
// before giving up with 503 — a wedged wake must not pile up connections
// forever. A var so tests can shrink it.
var wakeWaitBound = 30 * time.Second

// wakeBoundExtended is a test seam, nil in production: awaitWake calls it
// right after a first wake's waiter extends its bound (spec §8.2), so a test
// can fail the wake inside the extension window.
var wakeBoundExtended func()

// errLeaseHeld: the per-host lease is held elsewhere. On one box this means a
// previous incarnation hasn't released (crashed agent; lease expires within
// ~lease.LeaseTTL) — the edge answers 503 with Retry-After.
type errLeaseHeld struct {
	holder     string
	retryAfter time.Duration
}

func (e *errLeaseHeld) Error() string {
	return fmt.Sprintf("instance lease held by %q (retry in %s)", e.holder, e.retryAfter)
}

// errConfigRefused: this box cannot provide the instance's sealed config
// (no key, the wrong key, or it could not tell), so it will not take the
// lease (amadan #213). The condition outlives one request and a keyed
// sibling may take the lease in the meantime, so the edge answers 503 with a
// longer Retry-After than a lease contention gets.
type errConfigRefused struct{ err error }

func (e *errConfigRefused) Error() string {
	return "instance config unavailable on this box: " + e.err.Error()
}
func (e *errConfigRefused) Unwrap() error { return e.err }

// configAdmitter is the optional Backend capability behind the config gate
// (amadan #213): AdmitConfig answers whether this box can provide rt's config.
// Acquire asks it before taking a lease; paths that keep a lease and start a
// child anyway (a worker's in-place respawn) ask it directly. A backend
// without it is not gated.
type configAdmitter interface {
	AdmitConfig(ctx context.Context, rt registry.Route) error
}

// seedPlacer is the optional half of a Backend that records a seed placed
// after the first upload (seeded-instances spec §3.4).
type seedPlacer interface {
	SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error
}

// configRefusedRetryAfter is the Retry-After (seconds) a refused wake sends.
const configRefusedRetryAfter = 30

var errWakeTimeout = errors.New("timed out waiting for instance wake")
var errStartupStopped = errors.New("activator stopped during startup")

// errShuttingDown refuses a request once Shutdown has begun (#292): nothing
// wakes, and nothing is recycled, after the agent has started going away.
var errShuttingDown = errors.New("activator: the agent is shutting down")

// Service is the activator (Eleven's activatorSvc). It structurally satisfies
// the edge package's Activator interface — the edge never imports this
// package: the dependency points activator→edge only (backend_exec.go takes
// edge.Cache and edge.ListenControl; serve calls edge.MarkAnswered), and the
// seam back is the interface the edge declares for itself.
type Service struct {
	// Startup gates admission while the agent recovers installations. Set before publication.
	Startup <-chan struct{}
	// startupSeenAt is when serve first found Startup closed (unix nanos,
	// 0 until then): requests in the window after it may carry a proxy the
	// edge chose before recovery moved their upstream.
	startupSeenAt atomic.Int64

	// idleFor is the plans idle hook (plans 2), set through SetIdleFor: given an
	// entry's route it names the plan's idle window, or 0 for no opinion. It
	// can only shorten a window. nil means today's behaviour.
	idleFor atomic.Pointer[func(registry.Route) time.Duration]

	cfg     Config
	backend Backend
	due     DueStore
	now     func() time.Time             // a seam for the unit tests' clock
	liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests

	// Routes, when non-nil, is read by doWake before Acquire (design §6.8)
	// to refresh the route against the registry's committed row rather than
	// spawning against the copy the edge cached at request time. Nil (every
	// test that doesn't wire it, and any caller that doesn't set it) keeps
	// today's behaviour: the wake runs on the route it was handed.
	Routes RouteLookup

	// Logs, when non-nil, receives member-visible platform events (wake
	// outcomes, restart bounces) for `carlos logs` — set before Start, like
	// the backend's own Logs field. Nil (every test, a box without the
	// shipping loop) emits nothing.
	Logs EventSink

	// mesh seam: Eleven kept the multi-box placement layer here (mesh *meshSvc):
	// requests for hosts whose HRW owner is another live box were proxied
	// there, and the due-time sweeper only woke hosts this box owned. Nil
	// meant a single box — hashing would pick self everywhere, so no mesh
	// codepath was touched at all. Single-box v1 drops the field; it re-enters
	// with the mesh port.

	// lifeCtx is cancelled by Shutdown: the seam that lets an in-flight worker
	// spawn be TORN DOWN rather than waited out. See Shutdown for why waiting
	// is the wrong trade there.
	lifeCtx  context.Context
	lifeStop context.CancelFunc

	mu      sync.Mutex
	entries map[string]*actEntry
	// shuttingDown is set by Shutdown before it stops anything, and read by
	// the supervised-respawn goroutines: the agent is going away, so nothing
	// new may be spawned against a DB whose lease is about to be released.
	// begin, WakeWorker, wake and the sweep read it too, with swaps on or
	// off (#292): once it is set nothing wakes, and an exited child is
	// Shutdown's to stop, never recycled as an unexpected exit.
	shuttingDown bool
	// shutdownCtx is Shutdown's budget, set with shuttingDown: a legacy
	// proof runs under the sooner of its own budget and this (#292).
	shutdownCtx context.Context
	// proofCancels are the legacy proofs running now (legacyRunGone), so
	// Shutdown can end them with its budget. Under a.mu.
	proofCancels map[*byte]context.CancelFunc

	// convergeBudget is how many build-convergence cycles this sweep may
	// still start: reset to maxConvergeCyclesPerSweep at the top of every
	// sweepOnce, and spent by the sweep's drift check and by a recycle that
	// converges (recycleWorkerLocked), so the two share one pace. Under a.mu.
	convergeBudget int

	// uncapturedSaid is the hosts whose uncaptured worker a recycle has
	// declined to respawn from the live link (recycleWorkerLocked), so the
	// agent says so once per host and not at every backoff or sweep. Under
	// a.mu.
	uncapturedSaid map[string]bool

	// liveExecGate is the backend's LiveExecGate as SetLiveExecGate set it:
	// nil on every box but one that declined CARLOS_SWAP=overlap and honours
	// failure records or hold-ran markers. The worker converge and recycle
	// ask it under a.mu before they stop anything (convergeWorkerLocked,
	// recycleWorkerLocked). liveGateSaid is the (host, build) a refusal there
	// was last logged for, so a drift the sweep sees every pass is said once.
	// Under a.mu.
	liveExecGate func(rt registry.Route, build string) error
	liveGateSaid map[string]string

	fences          map[string]uint64
	seedExclusions  map[string]*SeedExclusion
	seedDiagnostics map[string]*seedDiagnostic
	// fenceOwners names the owner of an owned fence (FenceAndStopAs), by
	// host. A fence installed any other way has no entry. Under a.mu.
	fenceOwners       map[string]string
	tokenPreparations map[string]chan struct{}
	stopHistory       map[string]*stopObservation
	// straySidecars are legacy sidecars whose termination is unproven, per
	// host (zero-downtime spec §7.4). Task 7 tracks into it; a stop keeps
	// its lease while its host has one (releasable). Under a.mu.
	straySidecars map[string][]*straySidecar
	// unprovenWakes are failed wakes whose generation is not proven stopped
	// (spec §7.4), per host, made lazily. Each keeps its lease and refuses
	// the host's next wake until the sweep proves it
	// (reapUnprovenWakesLocked). Under a.mu.
	unprovenWakes map[string]*unprovenWake
	// owedReleases are lease releases that failed and are retried until
	// they land (giveBack, #292). Under a.mu.
	owedReleases map[*owedRelease]struct{}
	// outcomeFences are planned-wake outcomes taken but not yet delivered
	// to OnGen, per host, made lazily (final review F3): a failed or
	// fallen-back first wake's, from before its entry goes until afterWake's
	// OnGen has returned, and a first wake's observation's, from its arming
	// until its terminal event has been delivered. A new wake of the host
	// waits for them (bounded) before it plans. Under a.mu.
	outcomeFences map[string]*outcomeFence

	sweepStop chan struct{}
	sweepDone chan struct{}
	// lastSweep is when the sweeper last finished a sweep, in unix
	// nanoseconds (0: never): the hosting loop's own beat, which the load
	// balancer's health answer reads (region HA spec 2 §6, LastSweep).
	lastSweep atomic.Int64

	// gens is the generations switch (SetGenerations) as read at New. Off:
	// every spawn binds the canonical socket and no retire hook fires.
	gens bool
	// cacheGens is host -> *atomic.Uint64, the newest generation that has
	// been current for that host (CacheGen). A sync.Map, never under a.mu:
	// the edge cache reads it while holding the cache's own lock.
	cacheGens sync.Map
	// retiredMu guards retiredHooks (OnGenRetired). A leaf lock: nothing is
	// called while it is held.
	retiredMu    sync.Mutex
	retiredHooks []func(host, socket string)

	// OnWake, when set, is called with the ROUTE after EVERY successful wake
	// — fleet boxes use it to publish a binding object to the deployment
	// bucket (customer-fleets plan, Task 6) naming this box as the current
	// holder of host's lease; nil (every non-fleet deployment) is a pure
	// no-op. Deliberately a callback rather than a fleet-aware field on
	// Config: the activator has no notion of "fleet" anywhere else in this
	// package, and threading FleetName/BoxLabel through Config would leak a
	// customer-fleets concept into a package that otherwise only knows
	// hosts, leases, and processes. wake() invokes it in its own goroutine
	// (see below) so a slow or failing publish can never delay a request
	// that's waiting on this same wake.
	//
	// The whole route rather than the bare host because the hook's own
	// refusal is kind-shaped: a binding must never be published for a
	// sidecar, and the wake path already HAS the route. Passing the host
	// alone forced the hook to re-read the registry to learn the kind, which
	// made a security-relevant refusal depend on an I/O call that can fail —
	// and fail OPEN, since a read error must not stop legitimate bindings.
	OnWake func(rt registry.Route)

	// Planner, when set, says which build each wake boots (zero-downtime
	// spec §8.2). The rollout manager sets it with CARLOS_SWAP=overlap; nil
	// keeps today's wake from the live link. It is called without a.mu held.
	Planner func(ctx context.Context, rt registry.Route) WakePlan
	// OnGen, when set, hears every planned wake's outcome. It is never
	// called with a.mu held: it reaches the scheduler's goodMu, which calls
	// back into Current (emitGen).
	OnGen func(GenEvent)
	// OnDrift, when set, replaces the sweep's convergence stop: with the flag
	// on, the scheduler owns every move onto a new build (spec §10). It runs
	// on its own goroutine, never under a.mu.
	OnDrift func(host string)

	// CoalesceRecycle, when set, is asked before a worker is recycled for a
	// crash or max_run (spec §7.9). True means an attempt for that host (a
	// queued release attempt admissible now, or one already running) will
	// make the one spawn itself, so the recycle is skipped. The sweep asks
	// again next time if the child is still due. Nil, the default and the
	// whole of plan 4, recycles every time. It is called with a.mu held: it
	// must answer from its own state, never block, and never call back into
	// Service. Set before Start, like OnWake. Plan 5's scheduler fills it.
	CoalesceRecycle func(host string) bool

	// BeforeExec records, under the instance's lease, that build is about
	// to run against host's database (settings spec §4.2), before every
	// spawn (recordBuild). The context names the lease's location
	// (BuildStateLocationFrom). A non-nil error refuses the spawn, and is
	// never a failure of the build. nil records nothing: a box with no
	// replica bucket. Set before Start, like OnWake.
	BeforeExec func(ctx context.Context, host, build string) error

	// HeldTarget, when set, is the agent's answer for a held instance's wake
	// (settings spec §4.3 rule 3): the build it must run, in root's rule
	// order, or Held false for an instance root follows the channel for. The
	// wake asks it before planning and before the lease (wakeCheck), never
	// for a hold swap's pinned wake or a moves probe. It runs off a.mu and
	// may read the bucket. nil checks nothing: today's wake. Set before
	// Start, like OnWake.
	HeldTarget func(host string) HeldWake
	// RequestStage, when set, asks root to install build for host (the
	// stage relay), for a held wake whose target is not installed. It must
	// not block: the wake waits for the install itself. Set before Start.
	RequestStage func(host, build string)

	// OnWorkerServing, when set, hears a worker's captured serving identity
	// each time a wake or a respawn installs a child (zero-downtime spec
	// §7.9), an empty version included. It runs with no lock held, through
	// publishWorkerServing, which delivers only the entry's latest
	// incarnation. The agent sets it whatever CARLOS_SWAP says, so a
	// flag-off box keeps `carlos sidecars ls`'s running version right
	// (registry ServingVersion). Set before Start, like OnWake.
	OnWorkerServing func(host string, id release.StagedIdentity)
	// servingMu orders OnWorkerServing deliveries: the incarnation check and
	// the delivery happen under it, so an older publication can never land
	// after a newer one. Taken BEFORE a.mu, never while holding it.
	servingMu sync.Mutex

	// GenerationsFile, when set, is where the activator publishes every
	// worker's captured serving build and every live or pending instance
	// generation's build for root's pin set (release.GenerationsPath of the
	// data dir root's passes use as BakeDir; see generations_file.go). A
	// staged spawn whose build cannot be written there first is refused.
	// Empty publishes nothing. Set before Start and before any wake, like
	// OnWake.
	GenerationsFile string

	// schedMu guards schedFences, each host's schedule write fence
	// (schedfence.go). It is never held with a.mu or a fence's own lock.
	schedMu     sync.Mutex
	schedFences map[string]*schedFence
	// genMu serialises the generations file's writers (persistGenerations).
	// It is taken BEFORE a.mu, never while holding it. genWritten (a write
	// has succeeded) and genDirty (the last write failed; the sweep retries)
	// are read and written under genMu only.
	genMu      sync.Mutex
	genWritten bool
	genDirty   bool
	// genPending holds the builds spawns are about to exec, published
	// before each exec (generations.go, ruling R19 item 3); genPendingSeq
	// names them. Both under a.mu.
	genPending    map[pendingGeneration]pendingBuild
	genPendingSeq uint64
	// joinWaiting, when set, is called by a terminal stop's joins as each
	// is about to wait: what is "operation set" (the barrier, joinOps, once
	// it has cancelled the attempts, on a non-empty set) or "sidecar start"
	// (stop() joining an in-flight supervised sidecar start). It is a test
	// seam: the positive signal that a stop has reached a join. Set before
	// the stop it observes; nil in production.
	joinWaiting func(host, what string)
	// wakeEnded, when set, is called as a wake goroutine (wake) returns,
	// after its last write (the generations file's republish). It is a test
	// seam: a test that removes the generations file's directory waits on
	// it first (fix wave 1, item 6). Set before the wake; nil in production.
	wakeEnded func(host string)
	// genHealth, when set, replaces a generation's health check. It is a
	// test seam. Nil is waitHealthyCtx on the generation's own socket (spec
	// §7.1: health uses the handle, never AppSocket).
	genHealth func(ctx context.Context, g Gen) error
	// claimSpawn, when set, replaces a claims-on successor's spawn. It is a
	// test seam: a claims-on spawn needs a delegated cgroup tree and an open
	// isolated launch. Nil in production (no production file assigns it,
	// TestClaimSpawnIsATestSeam), where every claims-on launch is closed on
	// this base and claimsSwapRefusal refuses the swap before its raise.
	claimSpawn func(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error)
}

// New wires the state machine to its side-effect backend and due-time store.
// In production due is the *registry.Registry itself.
func New(cfg Config, backend Backend, due DueStore) *Service {
	a := &Service{
		cfg:     cfg,
		backend: backend,
		due:     due,
		now:     time.Now,
		entries: make(map[string]*actEntry),
		gens:    GenerationsEnabled(),
	}
	// The service lifetime, cancelled by Shutdown. Only work that a shutdown
	// should ABANDON hangs off it — today that is exactly the supervised
	// worker respawn's exec. Everything a shutdown must still FINISH (the
	// final checkpoint, the lease release, the SIGTERM drain) keeps its own
	// Background context with its own timeout, so a shutdown can never cancel
	// the durability it exists to guarantee.
	a.lifeCtx, a.lifeStop = context.WithCancel(context.Background())
	// The convergence sweep's live/<host> reader. The exec backend knows the
	// symlink layout (OptDir is its field, not a package global — Eleven used a
	// global here); any other backend — the unit tests' fakes — leaves
	// convergence off ("" = unknown → never cycle) unless the test wires
	// a.liveVer itself.
	if lv, ok := backend.(interface {
		LiveVersion(host string) (string, error)
	}); ok {
		a.liveVer = lv.LiveVersion
	} else {
		a.liveVer = func(string) (string, error) { return "", nil }
	}
	return a
}

// startupFreshWindow bounds how long after startup completes a request's
// route is re-checked against the registry (serve). A proxy the edge chose
// before Startup closed reaches serve within milliseconds; the window is
// generous, and costs one registry read per hibernating request inside it.
const startupFreshWindow = 2 * time.Minute

// nearStartup reports whether this request is within startupFreshWindow of
// the first time serve found Startup closed. The first observation is at or
// after the real close, so a request racing the close is always inside it.
func (a *Service) nearStartup() bool {
	now := time.Now().UnixNano()
	at := a.startupSeenAt.Load()
	if at == 0 {
		if a.startupSeenAt.CompareAndSwap(0, now) {
			return true
		}
		at = a.startupSeenAt.Load()
	}
	return now-at < int64(startupFreshWindow)
}

func (a *Service) awaitStartup(ctx context.Context, deadline <-chan time.Time) error {
	if a.Startup == nil {
		return nil
	}
	select {
	case <-a.Startup:
		if a.lifeCtx.Err() != nil {
			return errStartupStopped
		}
		return ctx.Err()
	case <-ctx.Done():
		return ctx.Err()
	case <-a.lifeCtx.Done():
		return errStartupStopped
	case <-deadline:
		return errWakeTimeout
	}
}

// begin registers a request against rt's instance, waking it if asleep.
//
// stamp is what separates a visitor from the platform's own clock. Both take
// the in-flight count — a delivery in progress must never be cut off by a
// concurrent sweep — but only a visitor's release moves lastActive. A
// scheduled tick that stamped would reset the idle window on every fire, so
// any schedule tighter than that window would pin its instance awake forever
// for however little work the handler does.
//
// req is the member request being served, nil for a caller with none (a
// carlos run, beginRequest). Only the edge's request records read it: the
// request that starts a wake is marked (edge.MarkWoke) so its record can say
// it woke the app.
func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool, req *http.Request) (release func(), err error) {
	deadline := time.NewTimer(wakeWaitBound)
	defer deadline.Stop()
	arrival := time.Now()
	if err := a.awaitStartup(ctx, deadline.C); err != nil {
		return nil, err
	}
	extended := false
	for {
		a.mu.Lock()
		if a.fences[rt.Host] != 0 {
			a.mu.Unlock()
			return nil, ErrFenced
		}
		// Once Shutdown has begun, a request wakes nothing and recycles
		// nothing (#292), WakeWorker's rule. Checked on every pass of the
		// loop, not only at entry: a request queued on a stop or a wake
		// before the shutdown began would otherwise come round and wake the
		// host afresh, taking a lease Shutdown never sees. An exited child
		// found here is left for Shutdown's own stop, not read as the
		// build's failure.
		if a.shuttingDown {
			a.mu.Unlock()
			return nil, errShuttingDown
		}
		e := a.entries[rt.Host]
		if e == nil {
			// Asleep → kick off ONE wake; this request and any latecomers all
			// coalesce on e.ready rather than racing a second lease claim
			// (singleflight). The wake runs detached so even its initiator is
			// held to the same bounded wait as everyone queued behind it.
			//
			// FALL THROUGH to the switch holding the lock — the initiator
			// waits on the entry IT created. The old shape unlocked and
			// re-read the map, and a wake that FAILED inside that one-
			// preemption window had already deleted its entry: the initiator
			// found nothing and started a second wake — two lease acquires
			// and two restore attempts per request (#184, seen only as a
			// loaded-suite test flake because the window is that narrow).
			e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
			a.entries[rt.Host] = e
			go a.wake(e)
			edge.MarkWoke(req)
		}
		// A held recovery probe serves nothing until the source opens.
		if e.probeHeld {
			a.mu.Unlock()
			return nil, ErrFenced
		}

		switch e.state {
		case actAwake:
			// Stand down synchronously-detected failure: a dead child or a
			// lost lease found at request time turns into a stop, and the
			// request waits for it then re-wakes — recovery is the routine
			// wake path.
			if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
				why := StopExited
				if repErr != nil {
					log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
					why = StopLeaseLost
				} else {
					log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
					// Not an intentional stop: a first wake's observation
					// reads it as the build's failure (spec §6.4).
					e.unexpectedExit = true
				}
				a.beginStopLocked(e, why)
				a.mu.Unlock()
				continue
			}
			e.inflight++
			a.mu.Unlock()
			return a.releaseFunc(e, stamp), nil

		case actWaking:
			ready := e.ready
			a.mu.Unlock()
			if err := a.awaitWake(ctx, e, ready, deadline, arrival, &extended); err != nil {
				return nil, err
			}
			// A failed wake reports to every coalesced waiter (retrying
			// here would turn one bad wake into a storm); a successful one
			// loops to register against the now-awake entry.
			if e.err != nil {
				return nil, e.err
			}
			continue

		case actStopping:
			stopped := e.stopped
			a.mu.Unlock()
			select {
			case <-stopped:
				continue // entry is gone; loop wakes it fresh
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-deadline.C:
				return nil, errWakeTimeout
			}
		}
	}
}

// beginRequest is begin for member traffic: activity counts.
func (a *Service) beginRequest(ctx context.Context, rt registry.Route) (release func(), err error) {
	return a.begin(ctx, rt, true, nil)
}

// awaitWake is a waiter's wait on one wake: begin's and WakeWorker's. It
// returns nil once ready closes (the caller then reads e.err), ctx.Err(), or
// errWakeTimeout at the deadline. A first wake onto a new build gives its
// waiters arrival + firstWakeBudget (zero-downtime spec §8.2): enough for
// B's health window, B's kill and the good build's boot. Only that kind of
// wake (firstWake is set only by a planner, flag on), and once per caller:
// *extended carries across the caller's loop.
//
// After the extension it waits again on the SAME ready and never goes back
// to the entries map (round 2 Task 6). A wake that fails inside the
// extension closes ready and deletes its entry; a waiter that re-read the
// map found nothing and started a second wake, two lease acquires and two
// spawns for one caller (#184's shape). Called without a.mu.
func (a *Service) awaitWake(ctx context.Context, e *actEntry, ready <-chan struct{}, deadline *time.Timer, arrival time.Time, extended *bool) error {
	for {
		select {
		case <-ready:
			return nil
		case <-ctx.Done():
			return ctx.Err()
		case <-deadline.C:
			a.mu.Lock()
			fw := e.firstWake
			a.mu.Unlock()
			if !fw || *extended {
				return errWakeTimeout
			}
			*extended = true
			deadline.Reset(time.Until(arrival.Add(firstWakeBudget)))
			if wakeBoundExtended != nil {
				wakeBoundExtended()
			}
		}
	}
}

// errSidecarAsleep: a sidecar-marked request arrived for a host with no awake
// instance. Sidecar traffic never wakes anything — the activator itself spawns
// the sidecar as part of a wake, so a marked request for a sleeping host is
// either a stray reconnect racing a stop or an outsider spoofing the marker;
// both get a 503 and change nothing.
var errSidecarAsleep = errors.New("instance asleep; sidecar requests do not wake it")

// BeginSidecarRequest is beginRequest for sidecar-marked traffic: pass it
// through to an awake instance WITHOUT registering activity — no in-flight
// count, no lastActive stamp (the sidecar's own long-lived WebSocket must not
// hold its instance awake; that non-count is the whole reason the pair can
// sleep) — and refuse, rather than wake, when the instance is asleep or
// stopping.
func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) error {
	deadline := time.NewTimer(wakeWaitBound)
	defer deadline.Stop()
	if err := a.awaitStartup(ctx, deadline.C); err != nil {
		return err
	}
	for {
		a.mu.Lock()
		if a.fences[rt.Host] != 0 {
			a.mu.Unlock()
			return ErrFenced
		}
		e := a.entries[rt.Host]
		if e == nil || a.shuttingDown {
			// Shutting down: the stop is Shutdown's, and an exited child is
			// not recycled here (#292).
			a.mu.Unlock()
			return errSidecarAsleep
		}
		if e.probeHeld {
			a.mu.Unlock()
			return ErrFenced
		}
		switch e.state {
		case actAwake:
			// Same dead-child / lost-lease detection as a member request, but
			// the recovery stops at the stop: the loop then finds a stopping
			// (or absent) entry and turns this request away instead of
			// chaining into a re-wake.
			if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
				why := StopExited
				if repErr != nil {
					log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
					why = StopLeaseLost
				} else {
					log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
					e.unexpectedExit = true // the build's failure inside a first wake's observation (spec §6.4)
				}
				a.beginStopLocked(e, why)
				a.mu.Unlock()
				continue
			}
			a.mu.Unlock()
			return nil

		case actWaking:
			// Joining a wake someone else started is fine — the co-spawned
			// sidecar boots moments before its entry flips awake, so its first
			// dial can land in that sliver; joining can't initiate anything.
			// A failed wake just reads as asleep (the member path already
			// logged it).
			ready := e.ready
			a.mu.Unlock()
			select {
			case <-ready:
				if e.err != nil {
					return errSidecarAsleep
				}
				continue
			case <-ctx.Done():
				return ctx.Err()
			case <-deadline.C:
				return errWakeTimeout
			}

		default: // actStopping — the stop exists to shed exactly this connection
			a.mu.Unlock()
			return errSidecarAsleep
		}
	}
}

// mesh seam: Eleven exposed hasLiveEntry(host) here — the mesh's "already
// living here" check (waking and stopping both counted: a waking host is
// committed to this box, and a stopping one still holds the lease, so
// proxying it away would only trade a short wait for a loud lease race) —
// and awakeCount() for the fleet status page.

func (a *Service) releaseFunc(e *actEntry, stamp bool) func() {
	var once sync.Once
	return func() {
		once.Do(func() {
			a.mu.Lock()
			e.inflight--
			if stamp {
				e.lastActive = a.now()
			}
			a.mu.Unlock()
		})
	}
}

// consumeDueTime clears the due time that fired THIS wake, so a fired timer
// wakes its worker exactly once. A no-op for every other wake — an operator
// command, the boot pass, an instance's request wake — which is what leaves a
// schedule that has not yet come due completely untouched.
//
// Conditional on the row still holding the fired value (ClearNextDueIf): the
// child spawned by this very wake may already have called /v1/next-due during
// its settle window, and that newer schedule must survive the consumption of
// the older one.
func (a *Service) consumeDueTime(e *actEntry) {
	if !e.worker || e.dueFired <= 0 {
		return
	}
	if err := a.due.ClearNextDueIf(e.host, e.dueFired); err != nil {
		// Loud, but not a failed wake: the child is up and doing its work. The
		// cost of the miss is one extra wake on a later sweep, which is
		// idempotent — the opposite trade to failing a wake that succeeded.
		log.Printf("activator: consume fired due time for %s: %v — it may wake once more", e.host, err)
	}
}

// wake drives asleep→awake: lease, restore-if-missing, spawn, replicate,
// co-spawn the sidecar. It runs on the first requester's goroutine but under
// its own generous timeout, detached from that request's context — queued
// requests still need the wake to finish even if the requester who started it
// hangs up.
func (a *Service) wake(e *actEntry) {
	if a.wakeEnded != nil {
		defer a.wakeEnded(e.host)
	}
	parent := context.Background()
	if e.probe != nil && e.probe.parent != nil {
		// A probe's wake ends with its caller (ProbeWake).
		parent = e.probe.parent
	}
	ctx, cancel := context.WithTimeout(parent, 5*time.Minute)
	defer cancel()
	a.mu.Lock()
	e.wakeCancel = cancel
	// Real time, not a.now: it anchors a first wake's real health deadline
	// (spawnPlanned, doClaimWake).
	e.wakeStart = time.Now()
	// A wake that starts after Shutdown began is cancelled before its
	// lease, whatever CARLOS_SWAP says (#292): with swaps off it used to run
	// on, and take a lease the exiting agent never gave back.
	if a.fences[e.host] != 0 || a.shuttingDown {
		cancel()
	}
	a.mu.Unlock()
	wakeStart := a.now()
	w, err := a.doWake(ctx, e)
	if w != nil {
		w.wakeTook = a.now().Sub(wakeStart)
	}

	// A due-fired wake CONSUMES the schedule it acted on — here, on success,
	// and never at stop. The distinction is the whole fix: a worker that goes
	// to sleep BEFORE its timer fires must keep its schedule (that row is its
	// only way back up), while one whose timer has already fired must not be
	// handed the same fired time again on the next sweep. Ordered before the
	// awake transition below so nothing can observe this entry awake with the
	// spent schedule still standing. A FAILED wake deliberately consumes
	// nothing: the index is then the retry, exactly as the due sweep's own
	// comment describes.
	if err == nil {
		a.consumeDueTime(e)
	}

	// The member-facing record of this wake (spec §5): composed here, where
	// the route and the whole error are in hand — the journal line below the
	// backends already write stays exactly as it was.
	if err != nil {
		if genStartUnproven(err) {
			// The member's event reads as the fence (memberSafeWakeReason);
			// the generation detail is the operator's.
			log.Printf("activator: %s: wake failed and its generation is not proven stopped: %v", e.host, err)
		}
		if msg, ok := a.wakeFailureEvent(e, err); ok {
			emitEvent(a.Logs, e.route, "error", msg)
		}
	} else {
		emitEvent(a.Logs, e.route, "info", wakeEventMessage(w, e.worker))
		if steps := w.steps.String(); steps != "" {
			log.Printf("activator: %s: woke in %s: %s", e.host, w.wakeTook.Round(time.Millisecond), steps)
		}
	}

	var (
		servingID      release.StagedIdentity
		servingInc     uint64
		publishServing bool
		republish      bool
		releaseOutcome func()
	)
	a.mu.Lock()
	if a.outcomeOwedLocked(e, w, err) {
		// A failed or fallen-back first wake (final review F3; spec §8.1,
		// O7), or a hold first wake that ran and was ended with no verdict
		// (final review wave 2, N2): its failure or hold-ran marker is
		// recorded only by afterWake's OnGen, after this
		// critical section and after e.ready closes. The host's outcome
		// fence is taken here, before the entry goes and before anyone
		// waiting is released, so a new wake of the host does not plan
		// until that record exists. Released once OnGen has returned.
		releaseOutcome = a.holdOutcomeLocked(e.host)
	}
	if err != nil {
		// The failed wake's entry goes, and with it everything it published
		// (a worker's captured build, a plan's builds, a pinned build); a
		// generation retained as an unproven wake is listed from there.
		republish = e.worker || a.publishesLocked(e)
		if u := a.unprovenWakes[e.host]; u != nil {
			_, listed := a.genIdentity(u.gen)
			republish = republish || listed
		}
		e.err = err
		outcome := &stopObservation{uncertain: true}
		if w != nil && w.stopOutcome != nil {
			outcome = w.stopOutcome
		}
		a.rememberStopLocked(e.host, outcome)
		delete(a.entries, e.host) // next request retries from scratch
	} else {
		e.state = actAwake
		e.rep, e.proc = w.rep, w.proc
		e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted
		e.sidecarUncertain = w.sidecarUncertain
		e.sidecarGen = w.sidecarGen
		e.sidecarStarted = a.now()
		e.sidecarBackoff = 0
		e.sidecarRetryAt = time.Time{} // a wake-time spawn failure retries on the next sweep
		e.lastActive = a.now()
		// The max_run clock starts when the child does. Set for every entry
		// (it costs a field write) but read only for workers: an instance's
		// run length is bounded by its idle window and its members.
		e.spawnedAt = a.now()
		// The build this child actually execed, captured at Start() inside the
		// backend (not re-read here — a ship during the wake window would make a
		// re-read name the new target while the process runs the old binary, and
		// the sweep would never converge it). Empty = unknown → the sweep won't
		// cycle. Fresh wake, so it can't be stale yet.
		e.spawnVersion = w.spawnVersion
		e.workerServing = w.spawnIdentity
		if e.worker {
			// A new worker incarnation (spec §7.9): its identity is published
			// once the wake's outcome has been reported (below).
			e.workerInc++
			servingID, servingInc, publishServing = e.workerServing, e.workerInc, true
		}
		e.staleSince = time.Time{}
		// The generation is current from here (spec §7.1). A wake the
		// claim path ran (doClaimWake) spawned through its provider, not
		// startGen. With generations on and a runtime that honours
		// StartPolicy.Gen it carries the handle the provider bound (Task
		// 10); otherwise it gets the canonical handle and a number of its
		// own.
		e.gen = w.gen
		e.launchLayout = w.launchLayout
		if e.gen.N == 0 && !e.worker {
			ran := e.gen
			e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion)
			if ran.StagedPath != "" && ran.Build == w.spawnVersion {
				// A claimed start with no number names the file it ran
				// (startAdmitted): the handle keeps it, so a legacy
				// sidecar restart runs that file too (Task 5d).
				e.gen.StagedPath, e.gen.Digest, e.gen.Demand = ran.StagedPath, ran.Digest, ran.Demand
			}
		}
		if a.gens && e.worker && e.gen.N == 0 {
			// A worker woken some way that carried no number (doWake gives
			// every worker spawn its own): numbered here, with the build it
			// runs, so the barrier and the generations file see it. Worker:
			// its termination is proven on its host cgroup (Task 11).
			e.gen = Gen{N: nextGen(), Build: w.spawnVersion, Worker: true}
		}
		if e.gen.N != 0 {
			a.noteCurrent(e.host, e.gen.N)
		}
		// An instance's current generation is in the generations file from
		// here (spec §8.5). A worker's build was published at its install.
		republish = !e.worker && a.publishesLocked(e)
		if e.sidecarGen == 0 {
			// A wake that carried no generation number (the claim path) ran
			// its sidecar on the live link, which is that wake's own build.
			e.sidecarGen = e.gen.N
		}
		// Outside consumers of the canonical path reach the generation
		// that just became current (spec §7.1). A failed link costs them
		// the canonical path, not the wake: internal traffic never uses it.
		if a.gens && linksCanonical(e.route, e.gen) {
			if err := swapCanonicalLink(e.route.Socket, e.gen.Socket); err != nil {
				log.Printf("activator: %s: link %s to %s: %v (processes dialling the canonical path will not reach it)", e.host, e.route.Socket, e.gen.Socket, err)
			}
		}
		// A first wake's observation is registered here, after e.gen is the
		// new generation and before e.ready closes, in this critical
		// section (zero-downtime spec §6.4), not once the generation is
		// already reachable. The fence and bounce checks below come after
		// it, so a stop they begin cancels it.
		a.armFirstWatchLocked(e, w)
		if a.fences[e.host] != 0 || a.shuttingDown {
			why := StopFenced
			if a.fences[e.host] == 0 {
				why = StopShutdown
			}
			a.beginStopLocked(e, why)
		} else if e.bounceAfterWake {
			// A config delivery or `carlos restart` asked for while this wake
			// was under way (Bounce). Same spot and same reason as the fence
			// just above: the state flips to stopping before e.ready closes,
			// so a request waiting on the wake is never proxied to this child
			// — it waits out the stop and wakes a fresh one.
			e.bounceAfterWake = false
			a.bounceLocked(e)
		}
		if a.OnWake != nil && e.state == actAwake {
			// Fire-and-forget, off this goroutine: e.route is read once here,
			// before this goroutine can be reused for anything else — the
			// hook must never share e.host's own goroutine's fate (a wedged
			// bucket Put must not hold e.ready closed any longer than the
			// wake itself already takes).
			rt := e.route
			go a.OnWake(rt)
		}
	}
	if w != nil && w.pend != 0 {
		// The cold wake's pending build (step 2b): the generation installed
		// above, or the failed wake retained as unproven, now names it
		// itself, or nothing of it ran. Dropped in the same critical
		// section, so no snapshot sees neither.
		a.dropPendingGenerationLocked(w.pend)
		republish = true
	}
	a.mu.Unlock()
	close(e.ready)
	// A planned wake's outcome, for the scheduler: off a.mu, since OnGen
	// reaches its goodMu, which calls back into Current.
	a.afterWake(e, w, err)
	if releaseOutcome != nil {
		releaseOutcome()
	}
	if publishServing {
		// A worker's identity, after afterWake and never before it (spec
		// §8.1; P5 review finding 7): for a fallback wake, afterWake's
		// GenFellBack has the scheduler record B's failure synchronously,
		// so B is held before the fallback's identity is published, and a
		// publication that blocks cannot hold that record up.
		a.publishWorkerServing(e, servingID, servingInc)
	}
	if republish {
		// After a change, the write only logs: the file it failed to
		// replace pins at least as much, and the sweep retries it.
		_ = a.persistGenerations()
	}
}

// wakeEventMessage is the member-facing wake line, split out so the one thing
// an app can actually scrape has a test of its own.
//
// It carries the wake's timing because an instance cannot time its own wake:
// the activator execs the binary, the process starts, opens SQLite, runs its
// migrations and parses its templates, and only THEN can the app's own code
// observe anything. So a cold-start budget's most important number is the one
// the app structurally cannot produce, and every app on the platform meets
// the same wall (discussion #6).
//
// Two figures, because one would not settle the question the budget is
// really asking. `spawn` is Backend.Start alone — exec to the socket
// accepting, which is the app's OWN boot and none of the platform's work.
// The total around it is the lease, the restore, the seed and replication.
// An app reading "woke in 1.4s (spawn 1.2s)" knows its boot is the problem;
// reading "woke in 1.4s (spawn 200ms)" knows it is not, which is the
// distinction #6 says cannot be made at all today.
//
// Rounded to a millisecond: this is a log line a person reads and a script
// greps, and microseconds of a multi-hundred-millisecond wake are noise that
// makes both harder.
func wakeEventMessage(w *wokenInstance, worker bool) string {
	// A worker is not "woken" in the member's vocabulary — nothing dialed it,
	// and there was no request waiting on the other end of this. The timings
	// below still mean exactly what they mean for an instance, which is why
	// only the noun changes.
	msg := "instance woke"
	if worker {
		msg = "sidecar started"
	}
	if w == nil {
		return msg
	}
	if w.wakeTook > 0 {
		msg += " in " + w.wakeTook.Round(time.Millisecond).String()
		if w.spawnTook > 0 {
			msg += " (spawn " + w.spawnTook.Round(time.Millisecond).String() + ")"
		}
	}
	if w.spawnVersion != "" {
		msg += " (version " + w.spawnVersion + ")"
	}
	return msg
}

// wokenInstance is a successful wake's yield. sidecar is nil when the host has
// none staged — or when the spawn failed, which is deliberately not an error:
// members > bots, so a host whose sidecar won't start still serves (the
// sweeper retries via sidecarWanted).
type wokenInstance struct {
	stopOutcome      *stopObservation
	sidecarUncertain bool
	rep              Rep
	proc             Proc
	sidecar          Proc
	sidecarWanted    bool
	sidecarGen       uint64                 // the generation whose build the legacy sidecar runs
	spawnVersion     string                 // live/<host> target read at exec — for build convergence
	spawnIdentity    release.StagedIdentity // the build exec'd, as the backend captured it (workers; spec §7.9)
	gen              Gen                    // the spawned generation's handle (instances only; zero for a worker)
	// genUnproven: the spawn failed and its error says the generation's
	// kill was not proven (genStartUnproven). cleanupFailedWake keeps the
	// lease until the sweep proves it, whatever a.gens says.
	genUnproven bool
	// legacyUncertain: the spawn was a start of the host's original run
	// unit that root may still carry out (legacyStartUncertain, #292).
	// cleanupFailedWake keeps the lease until the unit is proven gone.
	legacyUncertain bool
	// pend is the build a cold wake captured and published as pending at
	// step 2b (settleWakeIdentity; ruling R-COLDPUBLISH). wake() drops it in
	// the critical section that ends the wake: by then the entry's
	// generation (or a retained unproven wake's) names the build, or
	// nothing of it ran.
	pend pendingGeneration
	// launchLayout is the layout the row read immediately before this
	// wake's launch (final review round 1, finding 6, Astra): the failed
	// wake's cleanup decides from it, not from the layout at cleanup time,
	// whether the host cgroup must be proven empty (hostCgroupIsTheWakes).
	// "" when the backend cannot say (no Classifier), which requires the
	// proof.
	launchLayout Layout
	// spawnTook is how long Backend.Start took: exec to the socket accepting,
	// which is the instance's OWN boot and nothing of the platform's. wakeTook
	// is the whole wake around it, filled in by wake() rather than here
	// because the caller owns the clock that bounds it.
	//
	// They exist because an instance cannot time its own wake — almost all of
	// it happens before the app's code can observe anything — so the figure
	// an app most wants for a cold-start budget is the one it structurally
	// cannot produce (discussion #6). Reported on the wake event, where
	// `carlos logs` already carries the line.
	spawnTook time.Duration
	wakeTook  time.Duration
	// steps is the operator's breakdown of wakeTook, journal only.
	steps wakeSteps

	// planned is the generation a planned wake left serving; failed and
	// failReason name a first wake's build that failed; fellBack says the
	// good build replaced it (zero-downtime spec §8.2). untried says a
	// first wake's B was never exec'd because too little of its health
	// window was left (firstWakeMinVerdict), so the good build was booted
	// in its place as an ordinary wake.
	planned Gen
	// bootRan: the plan's own build (Boot) ran, or may have (a process came
	// back, or startMayHaveRun says its start is unproven or uncertain).
	// afterWake reads it for a hold first wake that ended with no verdict
	// (holdRanUnjudged; final review wave 2, N2).
	bootRan    bool
	failed     *release.StagedIdentity
	failReason string
	fellBack   bool
	untried    bool
}

// doWake runs the side effects of one wake. It takes the ENTRY, not the bare
// route, because two of its steps depend on what kind of child this is — the
// spawn call and the co-spawn — and those facts (e.worker, e.workerToken) are
// fixed at entry construction, before the wake goroutine starts, so reading
// them here without a.mu is safe by construction.
func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, error) {
	if a.cfg.HostWrites.Enabled {
		return a.doClaimWake(ctx, e)
	}
	rt := e.route
	host := rt.Host
	// Until the lease (step 1) this wake starts nothing, so a refusal before
	// it is known-no-start evidence, which leaves the host's stop history as
	// it was (seeded pre-land re-review, P2; census ruling 292fix).
	w := &wokenInstance{stopOutcome: &stopObservation{}}
	w.stopOutcome.notStarted()
	w.steps.begin(a.now())
	if ctx.Err() != nil || !a.wakeAllowed(host) {
		return w, ErrFenced
	}

	// 0. Refresh against the registry's committed row (design §6.8): the
	// edge cached rt at request time, and a config edit (provision, a path
	// change, a residence move) can land between that cache and this wake
	// actually running. Spawning against a stale DBPath/Socket would boot
	// the instance onto the wrong place on disk or bind the wrong socket, so
	// the wake re-reads the row it's actually about to spawn and carries
	// that forward. Every step below reads rt, not the edge's copy.
	if a.Routes != nil {
		fresh, err := a.Routes.Get(rt.Host)
		if err != nil {
			return w, fmt.Errorf("%s: route lookup: %w", rt.Host, err)
		}
		if fresh == nil {
			return w, fmt.Errorf("%s: no route", rt.Host)
		}
		if fresh.DBPath != rt.DBPath || fresh.Socket != rt.Socket {
			log.Printf("activator: %s: route refreshed: db %q -> %q socket %q -> %q", rt.Host, rt.DBPath, fresh.DBPath, rt.Socket, fresh.Socket)
		}
		// The refreshed row must still be the route this wake began for,
		// and the activator's to wake (final review F4, wakeidentity.go).
		if err := wakeRouteChanged(rt, *fresh, e.worker); err != nil {
			return w, err
		}
		rt = *fresh
		a.mu.Lock()
		if e := a.entries[rt.Host]; e != nil {
			e.route = rt
		}
		a.mu.Unlock()
	}

	// 0.5. Refuse a fenced or conflicting row before it ever reaches Acquire
	// or Restore (design §6.8). The in-Start/in-Seed refusal below still
	// exists as the second line of defense, but by the time either of those
	// runs the lease is already held and Restore has already written to
	// disk. For a fenced row that means restoring into a database a
	// migration journal says is being vacated, and for a conflict row it
	// means writing one host's database into another host's instance
	// directory. Checking here, on the row this wake just refreshed against,
	// closes both before either write happens.
	if c, ok := a.backend.(Classifier); ok {
		if layout, reason := c.Layout(rt); layout == LayoutFenced || layout == LayoutConflict {
			return w, fmt.Errorf("%s: refusing to wake: %s", rt.Host, reason)
		}
	}

	// 0.6. A failed wake whose generation is not proven stopped still holds
	// the host's lease, and its process may still run against the database
	// (zero-downtime spec §7.4). No new wake starts until the sweep proves
	// it (reapUnprovenWakesLocked). A legacy sidecar of the host whose
	// termination is unproven refuses it the same way
	// (reapStraySidecarsLocked).
	a.mu.Lock()
	_, unproven := a.unprovenWakes[host]
	unproven = unproven || len(a.straySidecars[host]) > 0
	a.mu.Unlock()
	if unproven {
		return w, fmt.Errorf("%s: refusing to wake: %w", host, errWakeUnproven)
	}

	// 0.65. The held wake check (settings spec §4.3 rule 3): a held
	// instance never wakes on an installed build other than its target
	// (wakeCheck). Before planning and before the lease, so a refusal or a
	// wait for root holds nothing. Not for a hold swap's pinned wake: the
	// swap chose e.pin from root's install. A route whose backend can exec a
	// named file (identityFor) may run a retained copy of the target
	// directly; any other waits for root. After a wait the row is re-read
	// as step 0 reads it, so the planner sees the build root just installed.
	var (
		held   string
		heldID *release.StagedIdentity
	)
	if e.pin == nil {
		var err error
		held, heldID, err = a.wakeCheck(ctx, host, identityFor(a.backend, rt) != nil)
		if err != nil {
			return w, fmt.Errorf("%s: %w", host, err)
		}
		if held != "" && heldID == nil && a.Routes != nil {
			fresh, err := a.Routes.Get(host)
			if err != nil {
				return w, fmt.Errorf("%s: route lookup: %w", host, err)
			}
			if fresh == nil {
				return w, fmt.Errorf("%s: no route", host)
			}
			// The wait for root can be long: the row read after it must
			// still be the route this wake began for, as at step 0 (final
			// review F4, wakeidentity.go).
			if err := wakeRouteChanged(rt, *fresh, e.worker); err != nil {
				return w, err
			}
			rt = *fresh
			a.mu.Lock()
			if e := a.entries[rt.Host]; e != nil {
				e.route = rt
			}
			a.mu.Unlock()
		}
	}

	// 0.7. Which build this wake boots (zero-downtime spec §8.2). A hold
	// swap's pinned wake (e.pin) already knows; everything else asks the
	// planner, off a.mu, before the lease: a refused plan takes no lease.
	var plan WakePlan
	if a.Planner != nil && e.pin == nil {
		// An earlier planned wake's outcome still on its way to the
		// scheduler is waited for, bounded (final review F3): planning
		// before its failure record exists would pick the failed build. A
		// wait that ends first refuses the wake before the planner and the
		// lease (final review wave 2, N1): nothing is planned or launched.
		if err := a.awaitOutcomes(ctx, host); err != nil {
			return w, fmt.Errorf("%s: refusing to wake: %w", host, err)
		}
		plan = a.Planner(ctx, rt)
		if plan.Refuse != "" {
			return w, fmt.Errorf("%s: refusing to wake: %s", host, plan.Refuse)
		}
		a.mu.Lock()
		e.plan, e.firstWake = plan, plan.FirstWake
		a.mu.Unlock()
	}
	// 0.8. A staged build, planned or pinned, is published before anything
	// execs it (spec §4.2, §8.5): the entry now names it (and a plan's
	// fallback), so the write puts both in root's pin set. A failed write
	// refuses the wake here, before the lease: nothing is taken, nothing
	// runs, and no build is blamed (errStagedBuildNotPublished).
	if plan.planned() || e.pin != nil {
		if err := a.publishBeforeExec(); err != nil {
			return w, fmt.Errorf("%s: refusing to wake: %w", host, err)
		}
	}

	// 1. The lease first: everything after this line is done as the prefix's
	// single writer. Contention is not retried here — on one box it means a
	// previous incarnation hasn't released; tell the client to come back after
	// the TTL instead of stalling its request.
	w.steps.mark(a.now(), "checks")
	*w.stopOutcome = stopObservation{}
	lease, err := a.backend.Acquire(ctx, rt)
	if err != nil {
		if lease != nil {
			w.stopOutcome.uncertain = true
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		}
		var held *errLeaseHeld
		if errors.As(err, &held) {
			log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)",
				host, held.holder, held.retryAfter)
		}
		return w, err
	}
	w.steps.mark(a.now(), "lease")

	if ctx.Err() != nil || !a.wakeAllowed(host) {
		w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		return w, ErrFenced
	}
	// 1.5. Flag off, the host cgroup is proven empty under the lease before
	// anything is restored or started (Task 8 fix round 1, finding 4). The
	// stop that released the lease proved it, but an agent restart forgets
	// a pending stop and its kill marks, so after the TTL a leftover can
	// still populate it. Unproven, the wake is refused and the lease is left
	// to its TTL, as an unproven seed leaves it: the leftover may still
	// write the database.
	if !a.gens {
		if err := a.wakeHostProven(ctx, e, rt); err != nil {
			log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
			w.stopOutcome.uncertain = true
			return w, fmt.Errorf("%s: refusing to wake: %w", host, err)
		}
	}
	// 2. Restore the DB from S3 when the local copy is gone — the restore must
	// happen *under* the lock (restoring without it could later shadow newer
	// S3 state written by the true holder). When it's still on disk it is the
	// warm cache: on one box, with the lease held, the local copy can only be
	// what we ourselves checkpointed at the last sleep. That holds only while
	// a region has one box. Claim ruling R6 ends it (region HA spec 1 H13):
	// once wakes are admitted by hostclaim's AdmitWriter, every admitted
	// writer purges its local copy and restores from the certified remote,
	// and no marker is ever read to skip that.
	fresh, err := a.backend.Restore(ctx, rt)
	if err != nil {
		if errors.Is(err, ErrSeedUnproven) {
			// An earlier seed's processes are still not proven gone (its
			// <db>.seed-unproven obligation, Task 8): they may still write
			// the database, so the lease is left to its TTL, as the seed's
			// own ErrSeedUnproven leaves it (step 3).
			log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
			w.stopOutcome.uncertain = true
		} else {
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		}
		return w, fmt.Errorf("restore %s: %w", host, err)
	}
	w.steps.mark(a.now(), "restore")
	if ctx.Err() != nil || !a.wakeAllowed(host) {
		w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		return w, ErrFenced
	}

	// 2b. One executable identity for this wake (settings spec §4.2, Astra
	// I1 #5): chosen now, recorded now, under the lease taken in step 1 and
	// before the seed and the spawn, and the only file this wake runs, its
	// seed included (settleWakeIdentity). A route whose backend cannot pin
	// an executable (identityFor nil: a unit-backed row, which systemd execs
	// from the live link itself) keeps today's calls and records nothing.
	// Any refusal here has run nothing: the lease goes back and the wake
	// fails, never blaming the build.
	//
	// A planned wake asks its selection gate first (checkPlan, ruling
	// R-REPLAN), so a replan lands before the identity is chosen and the
	// seed and the boot run one build. spawnPlanned asks again at the exec,
	// and startPinnedAt records again if that later replan changes the
	// build.
	if plan.planned() {
		next, err := a.checkPlan(ctx, e, rt, plan)
		if err != nil {
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
			return w, fmt.Errorf("start %s: %w", host, err)
		}
		plan = next
	}
	ib := identityFor(a.backend, rt)
	var id release.StagedIdentity
	if ib == nil && held != "" && !plan.planned() {
		// A held route whose backend cannot pin an executable (a unit row:
		// systemd execs the live link itself): nothing here can capture the
		// file, so the link is read once more just before the start. It
		// narrows a repoint after the wait; it cannot close it.
		if v := a.liveVersion(host); v != held {
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
			return w, fmt.Errorf("start %s: %w: %s is on %q, wants %s", host, errStaleLink, host, v, held)
		}
	}
	if ib != nil {
		var (
			idErr error
			wctx  context.Context
		)
		id, w.pend, wctx, idErr = a.settleWakeIdentity(ctx, ib, e, rt, held, heldID)
		if idErr != nil {
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
			return w, fmt.Errorf("start %s: %w", host, idErr)
		}
		if id.StagedPath == "" {
			// A live link outside the version store with no hook (ruling
			// R-EXEC-NOVERSION-2): today's live-link wake, nothing recorded.
			ib = nil
		} else {
			// The record is pending on wctx, written at the first exec
			// (amadan #272).
			ctx = wctx
		}
	}
	w.steps.mark(a.now(), "identity")

	// 3. First-boot seed, strictly between the restore and the spawn — so it
	// runs before Start's first health probe, and a brand-new instance is born
	// seeded rather than briefly serving an empty database. fresh is the
	// backend's one fact worth keeping: the local DB was absent AND no replica
	// existed, so a restore can never be mistaken for a first boot and a seed
	// can never run over recovered data. A seed that fails fails the WHOLE
	// wake (the Start-failure precedent: release the lease, surface the
	// error) — a half-seeded instance must not serve.
	//
	// The seed runs the wake's recorded identity (SeedExe) on every wake
	// kind, planned and pinned included.
	if fresh && rt.Seed != "" {
		w.stopOutcome.uncertain = true
		seed := func() error { return a.backend.Seed(ctx, rt) }
		if ib != nil {
			seed = func() error { return ib.SeedExe(ctx, rt, id) }
		}
		if err := seed(); err != nil {
			if errors.Is(err, ErrSeedUnproven) {
				// A SIGKILL survivor of the seed may still be writing the
				// database: releasing the lease here would be exactly the
				// write-after-release hazard the lease exists to prevent
				// (design §6.2), so it is left for its own TTL to expire,
				// the same way stop() leaves ErrStillRunning.
				log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
			} else {
				w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
			}
			return w, fmt.Errorf("seed %s: %w", host, err)
		}
		w.stopOutcome.uncertain = false
		w.steps.mark(a.now(), "seed")
	}

	// 4. Spawn before replication starts: a fresh instance creates its own DB
	// on boot, so waiting for the socket guarantees the file exists for
	// litestream to open. Writes that land in the gap are picked up by the
	// replica's first sync.
	//
	// Both doors are TIMED, because this is the half an app CANNOT see.
	// Backend.Start execs the binary and waits until its socket accepts, so it
	// contains the whole of the instance's own boot — opening SQLite, running
	// migrations, parsing templates — and none of the platform's lease,
	// restore or replication work. An app measuring its own TTFB starts the
	// clock after all of it (discussion #6). For a worker the figure means
	// less (StartWorker returns at exec, not at readiness — see below), but
	// it is recorded on the same terms rather than left absent, so the two
	// paths report through one field.
	//
	// A worker takes the other door: it binds nothing, so Start's health
	// probe has no socket to dial and would spend its whole timeout failing
	// the wake. StartWorker spawns and returns — readiness is the child's own
	// call over the agent socket, which is why the wake env below carries the
	// socket path and this wake's token.
	var (
		proc     Proc
		spawnVer string
		token    string
	)
	if ctx.Err() != nil || !a.wakeAllowed(host) {
		w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		return w, ErrFenced
	}
	w.stopOutcome.uncertain = true
	spawnStart := a.now()
	// The layout this launch uses, read on the row it launches, right
	// before it (finding 6): a layout that moves before a failed wake's
	// cleanup must not move where its descendants are looked for.
	if c, ok := a.backend.(Classifier); ok {
		w.launchLayout, _ = c.Layout(rt)
	}
	if e.worker {
		// Mint this wake's token here — under the lease, after the restore,
		// immediately before the spawn (design §7.2's order). A mint failure
		// is a wake failure like any other: release and report.
		token, err = newWorkerToken()
		if err != nil {
			// w goes back with the error, so wake() drops step 2b's
			// pending build (w.pend) and records this stop's outcome.
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
			return w, fmt.Errorf("start %s: %w", host, err)
		}
		// Published onto the entry NOW, not at the awake transition: the
		// child's first act is to call /v1/ready over the agent socket, which
		// authorizes against WorkerToken — and that call is what ENDS the
		// settle window. A token that only appeared once the entry was awake
		// would make the early-ready path unreachable: the socket would
		// refuse the very call it is waiting for, and every wake would pay the
		// full window. Publishing early gives nothing away: a wake that fails
		// deletes its entry, so the token still dies with it.
		a.setWorkerToken(e, token)
		// No registry write (spec §7.9): the row's Version is root's install
		// stamp, and the build this child runs is the identity the backend
		// captures at the exec, not anything the child later says.
		//
		// The generations file exists before any worker runs (ruling R17).
		a.publishGenerationsBeforeSpawn()
		var (
			ident release.StagedIdentity
			pend  pendingGeneration
		)
		env := workerEnv(a.agentSocket(), host, token)
		if plan.planned() {
			// A planned worker (zero-downtime spec §6.5): the named build,
			// behind planGate at each exec, with startWorker's publication
			// and the installation of the captured identity done inside
			// (PF8), and its readiness settled there too.
			proc, ident, err = a.spawnPlannedWorker(ctx, e, rt, plan, env, w, lease)
			if err == nil {
				w.planned = workerGenOf(ident)
			}
		} else {
			// startWorker publishes the build before its exec (ruling R19
			// item 3), and refuses the spawn when it cannot. With an
			// identity settled at step 2b it execs exactly that file
			// (onto): its hook publishes and gates it again, and skips the
			// record the wake already wrote. The second pending record of
			// the same build is harmless (the file lists each build once)
			// and is dropped as the wake's own is.
			var onto *release.StagedIdentity
			if ib != nil {
				onto = &id
			}
			proc, ident, pend, err = a.startWorker(ctx, rt, env, onto, nil)
		}
		spawnVer, w.spawnIdentity = ident.Version, ident
		if err == nil && !plan.planned() {
			// Installed while the entry is still waking, in place of the
			// pending build, in one critical section, so no snapshot sees
			// neither: the child is running these bytes from this moment,
			// settle window or not, so root must keep counting them as
			// referenced. The awake transition stores the same identity
			// again.
			a.mu.Lock()
			e.workerServing = ident
			a.dropPendingGenerationLocked(pend)
			a.mu.Unlock()
			_ = a.persistGenerations()
		}
		if a.gens {
			// The wake's generation, whether or not the worker becomes
			// ready (zero-downtime spec §7.4): a worker binds nothing, but
			// it is numbered and carries the build it execed, and
			// cleanupFailedWake proves its host cgroup before the lease
			// goes back (Worker, genChildEmpty).
			w.gen = Gen{N: nextGen(), Build: spawnVer, Worker: true}
		}
	} else if plan.planned() {
		// A planned instance wake (zero-downtime spec §8.2): the plan's
		// staged build, every exec behind planGate, with the first-wake
		// budget and fallback. w.gen carries the booted staged identity, so
		// the legacy sidecar below binds to it (startSidecar).
		var g Gen
		proc, spawnVer, g, err = a.spawnPlanned(ctx, e, rt, plan, w, lease)
		w.gen, w.planned = g, g
	} else if e.pin != nil { // swapsim:M3
		var g Gen
		proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin, lease)
		w.gen = g
	} else if ib != nil {
		// The identity step 2b recorded, as the next generation, exactly
		// where startGen would put it: the canonical socket (and, nested,
		// the original run unit) with the flag off, generation n's own
		// socket and unit with it on (startGenExe).
		var g Gen
		proc, g, err = a.startGenExe(ctx, ib, rt, id)
		spawnVer, w.gen = g.Build, g
	} else {
		var g Gen
		proc, g, err = a.startGen(ctx, rt)
		spawnVer, w.gen = g.Build, g
	}
	spawnTook := a.now().Sub(spawnStart)
	w.steps.mark(a.now(), "spawn")
	w.proc = proc
	if err != nil {
		// A start whose kill was unproven may have left its generation
		// running with no process handle (plan 2's startUnitAt). Its lease
		// is kept, process or not (P2-13).
		w.genUnproven = genStartUnproven(err) // swapsim:M9
		// A legacy run unit's start that root may still carry out (a start
		// job systemd queued, which cancelling the client does not cancel)
		// is stopped and proven gone before its lease goes (#292, P1).
		w.legacyUncertain = legacyStartMayRun(err)
		switch {
		case proc != nil || w.genUnproven || w.legacyUncertain:
			a.cleanupFailedWake(ctx, rt, w, lease)
		case launchNotStarted(err) && wakeStartedNothing(w):
			// A start its pre-exec gate refused ran nothing, and this wake
			// has started nothing else: its seed (step 3) has finished, and
			// its replica (step 5) and sidecar (step 6) come after this
			// spawn. Its stop is known, not uncertain (seeded pre-land
			// review, P2; census ruling 292fix).
			w.stopOutcome.notStarted()
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		case launchNotStarted(err):
			// Something of this wake is running after all: it goes through
			// the failed wake's cleanup and its proofs, not the shortcut.
			a.cleanupFailedWake(ctx, rt, w, lease)
		default:
			w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
		}
		return w, fmt.Errorf("start %s: %w", host, err)
	}

	// 4b. The settle window, workers only: the readiness check for a child
	// with nothing to probe. A child that dies inside it never becomes an
	// awake entry — the lease goes back and the wake reports, exactly as a
	// failed Start does, so a sidecar that crashes on boot doesn't sit in the
	// map holding a lease until someone notices.
	//
	// Through cleanupFailedWake, not a hand-rolled stop-and-release: that
	// helper is what records the stop's OUTCOME on w.stopOutcome, and a wake
	// that tore its own child down without recording it would leave the
	// stopHistory blind to exactly the case it exists for.
	// A planned worker's readiness already ran inside spawnPlannedWorker
	// (plannedWorkerReady, spec §6.5).
	if e.worker && !plan.planned() {
		if serr := a.awaitWorkerSettle(ctx, e, proc); serr != nil {
			a.cleanupFailedWake(ctx, rt, w, lease)
			return w, fmt.Errorf("start %s: %w", host, serr)
		}
		w.steps.mark(a.now(), "settle")
	}

	if ctx.Err() != nil || !a.wakeAllowed(host) {
		a.cleanupFailedWake(ctx, rt, w, lease)
		return w, ErrFenced
	}
	// 5. Trade the held lease for running replication, at the route's
	// replication-class sync interval (the per-plan durability ladder).
	// The DB path rides the route; when empty the backend defaults it to
	// <DataDir>/<host>.db internally (Eleven defaulted here via a package
	// global — base dirs are backend fields now).
	rep, err := lease.Replicate(ctx, rt.DBPath, replicationSyncInterval(rt.Replication))
	w.rep = rep
	if err != nil {
		a.cleanupFailedWake(ctx, rt, w, lease)
		return w, fmt.Errorf("replicate %s: %w", host, err)
	}

	// 5b. A seeded row's first upload, then its record (spec §3.4 step 5).
	// A failure here does not fail the wake: the instance is serving from
	// the placed file, and the next wake records it (finishPlacement).
	if p, ok := a.backend.(seedPlacer); ok {
		if err := p.SeedPlaced(ctx, rt, rep); err != nil {
			log.Printf("activator: %s: seed placement not recorded yet: %v", host, err)
		}
	}
	w.steps.mark(a.now(), "replicate")

	// 6. Co-spawn the sidecar, last — only once the instance is healthy and
	// replicating, so the sidecar always boots against a live instance (its
	// marked requests pass through the edge while we're awake). A spawn
	// failure is loud but never fails the wake: members come before bots, and
	// the sweeper retries with backoff.
	w.rep, w.spawnVersion, w.spawnTook = rep, spawnVer, spawnTook
	w.stopOutcome = nil
	// A worker is nobody's parent: it IS a sidecar, so it never co-spawns one.
	// Skipped outright rather than left to HasSidecar's answer — the staging
	// file is keyed by host, and a host that happened to have one staged must
	// not turn a standalone sidecar into a supervisor.
	if e.worker {
		return w, nil
	}
	if ctx.Err() != nil || !a.wakeAllowed(host) {
		return w, nil
	}
	if a.backend.HasSidecar(rt) {
		w.sidecarWanted = true
		w.sidecarGen = w.gen.N
		if sc, err := a.startSidecar(ctx, e, rt, w.gen, true); err != nil {
			w.sidecarUncertain = true
			w.sidecar = sc
			if sc == nil {
				log.Printf("activator: SIDECAR for %s failed to start: %v — serving without it (the sweep retries)", host, err)
			} else {
				log.Printf("activator: SIDECAR for %s failed to start: %v — retaining returned process", host, err)
			}
		} else {
			w.sidecar = sc
		}
		w.steps.mark(a.now(), "sidecar")
	}
	return w, nil
}

// actStopTimeout bounds a graceful instance stop (the instance's own SIGTERM
// drain is expected ~10s; leave headroom before the SIGKILL).
const actStopTimeout = 20 * time.Second

// beginStopLocked flips an awake entry to stopping and kicks the stop off; the
// caller holds a.mu. In-flight requests are not waited for (the caller's
// SIGTERM drain inside actStopTimeout is what covers them). Callers: the
// idle sweep (instance has been idle past cfg.Idle), a dead child, a lost
// lease (open requests on either of the last two are already doomed), and
// Bounce (config changed — the instance may be mid-request, but the drain
// still gives it actStopTimeout to finish before SIGKILL). Every caller says
// why (the Stop* reasons in cmd.go): a run the stop interrupts tells the
// member (run spec §5.3).
func (a *Service) beginStopLocked(e *actEntry, why string) {
	// A hold canary's settle ends with an intentional stop, before anything
	// is stopped (zero-downtime spec §6.4). A process that already exited
	// died of its build, and its stop must not hide that.
	if e.settleCancel != nil && (e.proc == nil || !e.proc.Exited()) {
		e.settleCancel()
	}
	// The same for a planned first wake's observation (spec §6.4): an
	// intentional stop ends it and records nothing. A stop that began
	// because the child died on its own (unexpectedExit), or of a process
	// that has already exited, leaves it to read the exit as the build's.
	if e.firstWatch != nil && !e.unexpectedExit && (e.proc == nil || !e.proc.Exited()) {
		e.firstWatch()
	}
	e.state = actStopping
	e.stopWhy = why
	if e.cmd != nil {
		e.cmd.cancelPrepare() // a stop interrupts a run's download
	}
	e.stopped = make(chan struct{})
	go a.stop(e)
}

// Owns reports whether host has a live entry in ANY state — waking,
// awake, or stopping. This is the external-socket advisory's predicate
// (issue #17): while an entry exists, whatever is serving the host's
// socket is this activator's own child — possibly still binding its
// listener, or draining SIGTERM — so any-state presence, not awake-only,
// is what keeps the advisory from calling the activator's own children
// strays.
func (a *Service) Owns(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.entries[host] != nil
}

// IdleWindow is cfg.Idle, for the box-side reporters that must tell a member
// whether their schedule is tight enough to hold the instance resident. Read
// only; the window is fixed at construction from CARLOS_HIBERNATE_IDLE.
func (a *Service) IdleWindow() time.Duration { return a.cfg.Idle }

// Bounce initiates a graceful stop of host's RUNNING instance so that
// its next wake — the next request, or its schedule — respawns it with
// a freshly composed environment. This is config delivery's restart
// half: the analogue of the unit path's try-restart, with the same
// caveat that in-flight requests are not waited for. It never starts
// anything: not-awake (waking, stopping, asleep, unknown) is a no-op,
// and the instance picks the new config up at its natural next wake.
//
// NOT kind-gated (Task 11 item 4, carried from Task 6's review) — e's
// entry has no branch on e.worker, so this stops an awake WORKER exactly
// as it stops an instance, and stop() itself is worker-aware (it skips
// the socket-only next-due probe, co-stops any attached sidecar first,
// etc.), so nothing here needs a kind check to behave correctly. The
// EXPLICIT decision this makes is what it deliberately does NOT do: start
// anything back up. For an instance that is exactly right — the next
// HTTP request is the wake half of the cycle, and Bounce owns only the
// stop half. A worker has no such request path; nothing dials one. A
// caller that wants a WORKER actually cycled — not just stopped — must
// pair this call with its own WakeWorker once Bounce returns true (see
// host/restart.go's cycleWorkers, the one caller that does): an
// always-on (Hibernate: false) worker left unpaired is stranded until the
// agent restarts (the boot pass that starts one runs once, at arming,
// never again), and a hibernating one stays silent until its schedule
// next comes due — neither of which "carlos restart" asked for. This is
// why SyncExec (appconfig/syncexec.go) never calls Bounce for a sidecar
// route at all: an automatic, unpaired bounce on every ordinary
// config-sync pass would carry exactly that risk with no operator command
// behind it.
func (a *Service) Bounce(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e != nil && e.probeHeld {
		// A held recovery probe runs the moves-held inputs, not delivered
		// config: the recovery owns its process until the source opens.
		return false
	}
	if e != nil && e.state == actWaking && !e.worker {
		// Not dropped: held for the end of the wake (bounceAfterWake). Still
		// reported as no stop, because none has happened — the callers that
		// count stops, or fall through to systemctl on false, see what they
		// always saw. Workers keep the old rule: their restart pairs a stop
		// with its own WakeWorker only on true (host/restart.go cycleWorkers),
		// and a stop held for later would park one with nothing to start it.
		e.bounceAfterWake = true
		return false
	}
	if e == nil || e.state != actAwake {
		return false
	}
	return a.bounceLocked(e)
}

// bounceLocked is Bounce's stop for an entry the caller has found awake,
// under a.mu.
func (a *Service) bounceLocked(e *actEntry) bool {
	// Both Bounce and BounceAdopted must preserve the admitted recovery.
	if e.probeHeld {
		return false
	}
	msg := "instance restarting (carlos restart)"
	if e.worker {
		msg = "sidecar restarting (carlos restart)"
	}
	emitEvent(a.Logs, e.route, "info", msg)
	a.beginStopLocked(e, StopRestart)
	return true
}

// startSidecar is every legacy co-spawned sidecar spawn: the wake tail,
// late attachment and the supervisor (sweepOnce), and a swap's replacement
// (zero-downtime spec §7.5). It runs its generation's recorded file by
// path whatever CARLOS_SWAP says (ruling R-EXEC-SIDECAR): a flag-off
// sidecar started from the live link could run a different build than its
// instance, and a nested one would rebind the host's shared app mount to
// the link. A generation that names no staged file (an instance woken
// before this binary, or the recovery probe's) has the live link resolved
// once, and that file run by path. On a declined box whose gate refuses
// that build, the instance's own build runs instead when the gate passes
// it (sidecarOwnBuild); otherwise the refusal stands. Each exec is
// recorded first (settings spec §4.2): the wake's co-spawn skips the build its wake already recorded, while a restart or
// a replacement writes it. A refused record starts nothing and is
// returned like any start failure, which never fails a wake. Only a route
// no backend can pin a file for (a unit row) keeps today's StartSidecar,
// live link and all, recording nothing.
//
// Immediately before the spawn it asks the gate's ownership check (spec
// §7.4): e must still be the host's, unfenced, with no shutdown, and (awake)
// its lease held; waking is the wake tail, whose lease is the wake's own.
// The check runs whatever the switch says (an accepted dormancy exception,
// recorded with the controller's rulings for Task 7): a host that does not
// own itself gets no co-spawned sidecar. While a stray sidecar of the host
// is tracked (its termination unproven) nothing starts beside it. The
// check is asked again after the record and before the exec, as
// startPinnedAt's gate is: a lease lost or a fence raised while the build
// state was written execs nothing.
//
// A resolved live link outside the version store with no BeforeExec set
// keeps today's live-link start, as the wake does (ruling
// R-EXEC-NOVERSION-2); with one set the record refuses it.
func (a *Service) startSidecar(ctx context.Context, e *actEntry, rt registry.Route, g Gen, waking bool) (Proc, error) {
	owned := func() error {
		a.mu.Lock()
		defer a.mu.Unlock()
		err := a.ownedLocked(e, waking)
		if err == nil && len(a.straySidecars[rt.Host]) > 0 {
			err = errSidecarUnproven
		}
		return err
	}
	if err := owned(); err != nil {
		return nil, err
	}
	s, ok := a.backend.(SidecarGenStarter)
	if !ok {
		return a.backend.StartSidecar(ctx, rt)
	}
	if g.StagedPath == "" {
		ib := identityFor(a.backend, rt)
		if ib == nil {
			return a.backend.StartSidecar(ctx, rt)
		}
		id, err := ib.LiveIdentity(ctx, rt)
		if err != nil {
			return nil, err
		}
		if id.Version == "" && a.BeforeExec == nil {
			return a.backend.StartSidecar(ctx, rt)
		}
		// The resolved file runs by path, past the backend's live-link
		// check (startSidecarChild), so a declined box's live exec gate is
		// asked here, before the record (askLiveExecGate). A refusal starts
		// nothing, as the backend's would, and the supervisor backs off.
		// A refused link whose build is not the instance's own falls back
		// to the instance's build (sidecarOwnBuild), so a sidecar is not
		// stranded beside an instance the gate already let run.
		if id.Version != "" {
			if err := a.askLiveExecGate(rt, id.Version); err != nil {
				own, ok := a.sidecarOwnBuild(rt, g.Build, id.Version)
				if !ok {
					return nil, err
				}
				id = own
			}
		}
		g.Build, g.StagedPath, g.Digest, g.Demand = id.Version, id.StagedPath, id.Digest, id.Demand
		// The resolved build's own tier (ruling R-TIER), never the live
		// tier read later.
		if gt, ok := a.backend.(interface {
			GenTier(rt registry.Route, build string) cgroup.Limit
		}); ok {
			g.Tier = gt.GenTier(rt, id.Version)
		}
	}
	if err := a.recordBuild(ctx, rt.Host, g.Build); err != nil {
		return nil, err
	}
	if err := owned(); err != nil {
		return nil, err
	}
	return s.StartSidecarGen(ctx, rt, g)
}

// sidecarOwnBuild is an unstaged sidecar's recovery when the declined gate
// refuses the build its live link names (readiness 2 Task 5): root
// repointed the link at a held build while the instance kept serving its
// own, and the supervisor would otherwise retry the held build forever. own
// is the generation's build (the instance's), refused the link's. When own
// is a different, named build, its staged copy is named and hashed through
// the backend (buildIdentifier) and the gate is asked about it too; both
// passing returns that identity, which startSidecar then records and runs
// by path as it would the link's. Anything else reports false, and the
// caller keeps the original refusal. A held build is never run: the gate
// has passed whatever this returns. The hash runs outside a.mu, as the
// link's own capture does.
func (a *Service) sidecarOwnBuild(rt registry.Route, own, refused string) (release.StagedIdentity, bool) {
	if own == "" || own == refused {
		return release.StagedIdentity{}, false
	}
	bi, ok := a.backend.(buildIdentifier)
	if !ok {
		return release.StagedIdentity{}, false
	}
	id, err := bi.BuildIdentity(rt, own)
	if err != nil {
		log.Printf("activator: %s: sidecar not started: this box will not start %s, the build its live link names, and the instance's own build %s cannot be named: %v", rt.Host, refused, own, err)
		return release.StagedIdentity{}, false
	}
	if id.Version != own {
		return release.StagedIdentity{}, false
	}
	if err := a.askLiveExecGate(rt, id.Version); err != nil {
		log.Printf("activator: %s: sidecar not started: this box will not start %s, the build its live link names, nor the instance's own build %s: %v", rt.Host, refused, own, err)
		return release.StagedIdentity{}, false
	}
	log.Printf("activator: %s: sidecar starting on the instance's own build %s (%s): this box will not start %s, the build its live link names", rt.Host, id.Version, id.StagedPath, refused)
	return id, true
}

// replaceSidecar is §7.5 step 8: a legacy co-spawned sidecar is restarted,
// stop-then-start, onto the generation just switched to. It claims the
// supervisor's interlock (sidecarStarting, sidecarStartDone) for the whole
// replacement. A supervised restart cannot race it, and stop() waits on it
// exactly as it waits on a supervised restart.
//
// A supervised start already in flight captured the generation it read
// when it began, which may be the one this swap just retired. Skipping the
// replacement because of it would leave the old build's sidecar beside the
// new server with nothing pending, so replaceSidecar joins it (its
// sidecarStartDone, bounded by the supervisor's DeadlineControl) and then
// replaces whatever it installed.
func (a *Service) replaceSidecar(e *actEntry) {
	a.mu.Lock()
	for e.sidecarStarting && a.entries[e.host] == e && e.state == actAwake {
		wait := e.sidecarStartDone
		a.mu.Unlock()
		if wait != nil {
			<-wait
		}
		a.mu.Lock()
	}
	if a.entries[e.host] != e || e.state != actAwake || !e.sidecarWanted || len(a.straySidecars[e.host]) > 0 {
		// A stray still unproven: the reaper proves it first, and the
		// supervisor then starts the current generation's sidecar.
		a.mu.Unlock()
		return
	}
	if e.sidecar != nil && !e.sidecar.Exited() && e.sidecarGen == e.gen.N {
		// Already on the current generation (the sweep's retry got here
		// first): nothing to replace.
		a.mu.Unlock()
		return
	}
	e.sidecarStarting = true
	done := make(chan struct{})
	e.sidecarStartDone = done
	old, rt, g := e.sidecar, e.route, e.gen
	a.mu.Unlock()
	defer close(done) // only after the old sidecar is proven gone, or retained
	if old != nil {
		ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
		if err := old.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
			log.Printf("activator: stop sidecar %s for its replacement: %v", e.host, err)
		}
		cancel()
		if !a.sidecarTerminated(rt, old) {
			// Unproven (spec §7.4, §7.5): the leader still runs, or something
			// it forked is left in its cgroup. It becomes a tracked stray, in
			// the same critical section that takes it off the entry, so no
			// supervisor sees the entry without a sidecar and nothing to wait
			// for. Nothing starts beside it, a stop keeps the lease
			// (releasable), and the reaper proves it.
			a.mu.Lock()
			if a.entries[e.host] == e {
				e.sidecar = nil
				e.sidecarStarting, e.sidecarStartDone, e.sidecarUncertain = false, nil, true
			}
			a.trackStraySidecarLocked(rt, old)
			a.mu.Unlock()
			log.Printf("activator: SIDECAR for %s is not proven gone for its replacement; tracking it (the sweep re-proves it)", e.host)
			return
		}
	}
	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	sc, err := a.startSidecar(ctx, e, rt, g, false)
	cancel()
	var discard Proc
	defer func() { a.discardSidecar(rt, discard) }() // after the unlock, before done closes
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.entries[e.host] != e {
		discard = sc
		return
	}
	e.sidecarStarting = false
	e.sidecarStartDone = nil
	if e.state != actAwake {
		// A terminal stop is joined on this replacement (done). It stops
		// e.sidecar after the join and proves it: install sc for it.
		e.sidecar, e.sidecarGen = sc, g.N
		return
	}
	e.sidecar, e.sidecarStarted, e.sidecarBackoff, e.sidecarGen = sc, a.now(), 0, g.N
	if err != nil {
		e.sidecarUncertain = true
		log.Printf("activator: SIDECAR for %s failed to start on generation %d: %v (the sweep retries)", e.host, g.N, err)
	}
}

// BounceAdopted is Bounce for an ADOPT pass's restart callback, with one
// thing Bounce cannot know taken into account: the callback fires after the
// pass has repointed live/<host>, so a child already exec'd from the build the
// symlink names has nothing to pick up, and stopping it buys a drain and a
// wake for no change of binary.
//
// That child exists whenever something else finished the adoption first. A
// root pass adopts, ConvergeAdopted cycles the instance, a request wakes it
// onto the new build — and a slower pass that read the old symlink before all
// of that reaches its own restart afterwards. Bounce would stop the
// replacement; this leaves it alone.
//
// Everything else is Bounce: same no-op on a host that is not awake, same
// event, same stop, and it applies to a worker exactly as Bounce does. An
// unknown spawn version or an unreadable symlink bounces — with nothing to
// compare, the safe reading of "adopted" is "the child may be stale".
//
// `carlos restart` and config delivery keep the unconditional Bounce: there
// the request IS the reason, whatever build is running.
func (a *Service) BounceAdopted(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || e.state != actAwake {
		return false
	}
	if e.spawnVersion != "" {
		if cur, err := a.liveVer(host); err == nil && cur == e.spawnVersion {
			return false
		}
	}
	return a.bounceLocked(e)
}

// Drifted lists the awake INSTANCES whose child was exec'd from a build
// live/<host> no longer names, each with the version the symlink names now.
// It is the cheap half of ConvergeAdopted below: a caller polling at seconds
// cadence asks this first and, on the ordinary tick where nothing has shipped,
// gets nothing back and reads no registry row at all. One readlink per awake
// instance, the same read the sweep makes.
//
// A read error, an unknown spawn version and an unknown symlink target are all
// "not drifted" — the sweep's own rule (ConvergeDecision): with nothing to
// compare there is nothing to act on. Workers are never listed; see
// ConvergeAdopted for why they stay with the sweep.
func (a *Service) Drifted() map[string]string {
	a.mu.Lock()
	defer a.mu.Unlock()
	var out map[string]string
	for _, e := range a.entries {
		if e.worker || e.state != actAwake || e.spawnVersion == "" || a.fences[e.host] != 0 {
			continue
		}
		cur, err := a.liveVer(e.host)
		if err != nil || cur == "" || cur == e.spawnVersion {
			continue
		}
		if out == nil {
			out = map[string]string{}
		}
		out[e.host] = cur
	}
	return out
}

// ConvergeAdopted cycles host's awake instance onto ver, a version an adopt
// pass has FINISHED adopting for it, without the sweep's ConvergeGrace. It
// reports whether it started a cycle.
//
// It exists because the adoption a real box performs is delegated. The agent
// runs sandboxed and cannot write the version store, so the root pass stages,
// repoints live/<host> and stamps the route — and cycles no exec child,
// because the children belong to this process. The agent's own pass then
// finds the host up to date and never reaches its restart callback, which
// left the sweep's grace as the only thing standing between a deploy and the
// build it shipped (discussion #231). This is the restart half of that
// adoption, performed by the one process that can.
//
// The grace is safe to skip HERE and nowhere else, because of what the caller
// vouches for: ver is the version the route is STAMPED with, and the stamp is
// written only after the adopting pass is done with the host — past its own
// restart and any rollback. The sweep cannot know that; it sees a symlink
// move and waits to learn whether it holds. ver must also still be what the
// symlink names, checked here under the lock: a stamp the symlink has moved
// on from is a newer adoption in flight, and its own stamp will bring the
// caller back.
//
// Like Bounce it never starts anything — an instance's next request is the
// wake half — and it is the same stop the sweep's convergence performs:
// SIGTERM drain, final checkpoint, lease release.
//
// Instances only. A worker's registry version is what its child REPORTS
// (cleared on every respawn), not what adoption stamped, so the agreement
// this rests on does not exist for one; the sweep's convergeWorkerLocked,
// behind the grace, remains a worker's path onto a new build.
func (a *Service) ConvergeAdopted(host, ver string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || e.worker || e.probeHeld || e.state != actAwake || a.fences[host] != 0 {
		return false
	}
	if ver == "" || e.spawnVersion == "" || e.spawnVersion == ver {
		return false
	}
	if cur, err := a.liveVer(host); err != nil || cur != ver {
		return false
	}
	log.Printf("activator: %s serving %s but %s is adopted — cycling onto the new build", host, e.spawnVersion, ver)
	a.beginStopLocked(e, StopRelease)
	return true
}

// --- standalone sidecars (workers) -------------------------------------------
//
// A worker is a member of this same state machine with the request path
// removed. Nothing routes to a sidecar — the edge refuses the row on kind, and
// ServeHTTP gates on `Kind == "instance"` — so no request can ever wake one:
// every wake arrives from the due-time sweep, an operator command, or the
// agent's boot pass, and its only sign of life is contact over the agent
// socket. beginRequest, BeginSidecarRequest and ServeHTTP are untouched.

// newWorkerEntry builds the waking entry for a standalone sidecar: the route's
// own ceilings (0 = platform default / no ceiling) and the channel its settle
// window waits on. The wake's token is NOT minted here — it is minted inside
// doWake, under the lease and immediately before the spawn, and published onto
// the entry there. Every field this does set is written BEFORE the entry is
// published to a.entries and the wake goroutine starts, so the wake path may
// read them without a.mu; nothing rewrites them afterwards.
func (a *Service) newWorkerEntry(rt registry.Route) *actEntry {
	return &actEntry{
		host:         rt.Host,
		route:        rt,
		state:        actWaking,
		ready:        make(chan struct{}),
		worker:       true,
		workerIdle:   time.Duration(rt.IdleSeconds) * time.Second,
		workerMaxRun: time.Duration(rt.MaxRunSeconds) * time.Second,
		readyc:       make(chan struct{}),
	}
}

// randRead is crypto/rand's entropy, behind a seam so a test can make minting
// fail. Nothing else in this package should read entropy.
var randRead = rand.Read

// newWorkerToken mints the per-wake agent-socket credential. A fresh one every
// wake is the point: a token from a previous incarnation of the child never
// matches, so an orphan that survived a stop cannot keep talking to the agent
// as though it were the live worker.
//
// It returns an error rather than panicking. The agent supervises every route
// on the box — one host that cannot mint a token must fail ITS wake (the lease
// released, like any other wake failure), never take the process, and every
// other tenant on the box, down with it. Deliberately no fallback source: an
// empty token would authorize nothing, but a predictable one would authorize
// everyone.
func newWorkerToken() (string, error) {
	var b [32]byte
	if _, err := randRead(b[:]); err != nil {
		return "", fmt.Errorf("mint worker token: %w", err)
	}
	return hex.EncodeToString(b[:]), nil
}

// errWorkerExitedEarly: the child died inside its settle window. That is a
// failed wake — a crash on boot (bad argv, missing config, panic at start) is
// exactly what the instance path's health probe would have caught.
var errWorkerExitedEarly = errors.New("worker exited during its settle window")

// awaitWorkerSettle is the socketless child's health probe: it is up once it
// says so over the agent socket, or once it has survived sidecarSettle without
// exiting. Real time, not a.now — like waitHealthy, this bounds a real
// process's first seconds, and the test clock exists to move idle windows, not
// to model a boot.
func (a *Service) awaitWorkerSettle(ctx context.Context, e *actEntry, proc Proc) error {
	window := sidecarSettle
	if window <= 0 {
		return nil
	}
	deadline := time.NewTimer(window)
	defer deadline.Stop()
	tick := time.NewTicker(settlePoll(window))
	defer tick.Stop()
	for {
		select {
		case <-e.readyc:
			return nil // the child reported ready: up, early
		case <-deadline.C:
			if proc.Exited() {
				return errWorkerExitedEarly
			}
			return nil
		case <-tick.C:
			if proc.Exited() {
				return errWorkerExitedEarly
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// WakeWorker brings a sidecar up, or does nothing if it is already awake.
// It is the request-less twin of beginRequest: a sidecar has no socket, so no
// request can ever wake one — every wake arrives from the due-time sweep, an
// operator command, or the agent's boot pass.
//
// The route arrives whole and is the only input: the idle window and the run
// ceiling ride on the row, so no caller (and no sweep tick) needs a bucket
// read to decide a worker has outrun its ceiling.
//
// Idempotent: concurrent and repeated calls coalesce on the ONE wake, exactly
// as concurrent requests do — three calls produce one child.
func (a *Service) WakeWorker(rt registry.Route) error {
	deadline := time.NewTimer(wakeWaitBound)
	defer deadline.Stop()
	if err := a.awaitStartup(a.lifeCtx, deadline.C); err != nil {
		return err
	}
	arrival := time.Now()
	extended := false
	for {
		a.mu.Lock()
		// Refused once a shutdown has announced itself. Shutdown sets the flag
		// BEFORE it snapshots the entry map, so a wake dispatched after that
		// point — a sidecar-wake leg tick, a boot pass, an operator command
		// landing mid-drain — would insert a fresh entry that snapshot never
		// sees: a child holding the tenant's lease past the agent's own exit,
		// with nothing left to stop it. The respawn path checks the same flag
		// for the same reason.
		if a.shuttingDown {
			a.mu.Unlock()
			return fmt.Errorf("wake worker %s: the agent is shutting down", rt.Host)
		}
		e := a.entries[rt.Host]
		if e == nil {
			// Asleep → kick off ONE wake and FALL THROUGH to the switch still
			// holding the lock, waiting on the entry we just created. Unlocking
			// and re-reading a.entries[host] here is the #184 bug: a wake that
			// fails deletes its entry, so the initiator would find nothing and
			// start a second one.
			e = a.newWorkerEntry(rt)
			a.entries[rt.Host] = e
			go a.wake(e)
		} else if !e.worker {
			// A live entry for this host that is NOT a worker means the route
			// changed kind under a running instance (or two rows collided).
			// Refuse rather than adopt someone else's child: its proc, lease
			// and idle discipline are the instance path's.
			a.mu.Unlock()
			return fmt.Errorf("wake worker %s: an instance entry already holds this host", rt.Host)
		}

		switch e.state {
		case actAwake:
			// Same stand-down beginRequest makes: a lost lease or a dead child
			// found at wake time turns into a stop, and the loop then re-wakes
			// it. Without this, a boot pass or an operator command would report
			// success for a worker whose lease is gone — the one answer that
			// makes a broken sidecar look healthy. A respawn in flight is not a
			// dead child: its proc is mid-replacement, so leave it be.
			if !e.workerRespawning {
				if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
					why := StopExited
					if repErr != nil {
						log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
						why = StopLeaseLost
					} else {
						log.Printf("activator: worker %s exited — recycling", e.host)
						e.unexpectedExit = true // not an intentional stop (spec §6.4)
					}
					a.beginStopLocked(e, why)
					a.mu.Unlock()
					continue
				}
			}
			a.mu.Unlock()
			return nil // already up: a no-op, not a second child

		case actWaking:
			ready := e.ready
			a.mu.Unlock()
			// A worker's caller gets a first wake's extension as much as a
			// request does (Astra R3 flag triage T10): without it a
			// fallback that succeeds after the ordinary bound was reported
			// as a timeout.
			if err := a.awaitWake(context.Background(), e, ready, deadline, arrival, &extended); err != nil {
				return err
			}
			// A failed wake reports to every coalesced caller; retrying
			// here would turn one bad wake into a storm (beginRequest's
			// rule, for the same reason).
			return e.err

		case actStopping:
			stopped := e.stopped
			a.mu.Unlock()
			select {
			case <-stopped:
				continue // entry is gone; loop wakes it fresh
			case <-deadline.C:
				return errWakeTimeout
			}
		}
	}
}

// agentSocket is where a worker child dials back.
func (a *Service) agentSocket() string {
	if a.cfg.AgentSocket == "" {
		return DefaultAgentSocket
	}
	return a.cfg.AgentSocket
}

// workerEnv is what a spawn hands the child: where to call back, who it is,
// and this incarnation's credential. The backend applies it OVER the tenant's
// own env file so a tenant cannot forge its own token (StartWorker's
// contract). A free function of plain strings, not a method on the entry, so
// the two callers — the wake (off-lock, with a token still in a local) and the
// crash restart (under a.mu) — both compose it without touching entry state
// they don't own.
func workerEnv(socket, host, token string) map[string]string {
	name, _, _ := strings.Cut(host, ".") // the sidecar's declared name
	return map[string]string{
		"CARLOS_AGENT_SOCKET":  socket,
		"CARLOS_SIDECAR_TOKEN": token,
		"CARLOS_SIDECAR_HOST":  host,
		"CARLOS_SIDECAR_NAME":  name,
	}
}

// restartWorkerLocked replaces a worker's dead child — the caller holds a.mu
// and has already established that e is an awake worker whose proc has
// exited, and that the exit is one that must be restarted: ANY exit for an
// always-on worker (`hibernate: false` — the record exists ⇒ the process
// runs), or a non-zero exit for a hibernating one (a crash; a clean exit is
// that bot finishing its work, and parks instead).
//
// Restart=always without a supervisor, the co-spawn path's rule applied to a
// standalone member: doubling backoff from sidecarRestartBase to
// sidecarRestartMax, and a run that lasted sidecarHealthyRun earns a fresh
// one, so a crash a day never escalates to the cap. The first crash respawns
// immediately and ARMS the delay for the next one — a crash loop therefore
// paces itself instead of hot-looping the box. The pacing is decided here,
// under the lock; the exec itself is not (see respawnWorkerLocked).
//
// In place, not through a stop and re-wake: the lease and the replication are
// still ours (only the child died), and design §7.2 says an always-on worker
// runs with no lease churn. A hibernating worker restarted here keeps its
// lease for the same reason — with the consequence, recorded at the sweep's
// exited arm, that a crash LOOP holds that lease until an operator or the
// record intervenes.
func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) {
	if !e.spawnedAt.IsZero() && now.Sub(e.spawnedAt) >= sidecarHealthyRun {
		e.workerBackoff = 0
	}
	if now.Before(e.workerRetryAt) {
		return // still backing off; the child stays dead until it elapses
	}
	if e.workerBackoff == 0 {
		e.workerBackoff = sidecarRestartBase
	} else {
		e.workerBackoff = min(e.workerBackoff*2, sidecarRestartMax)
	}
	e.workerRetryAt = now.Add(e.workerBackoff)
	log.Printf("activator: worker %s exited — restarting (backoff now %s)", e.host, e.workerBackoff)
	// nil: this child is already dead, so there is nothing to retire first.
	a.recycleWorkerLocked(e, nil, "exited")
}

// convergeWorkerLocked replaces an always-on worker's child with one exec'd
// from the build live/<host> now points at — the caller holds a.mu and has
// already established the drift persisted past ConvergeGrace and that the
// per-sweep budget allows it. The old child is retired first: it is alive, and
// two incarnations of an outbound worker would double its work.
//
// nil: a converge execs what live/<host> names now. Until plan 5 queues it as
// a paced release attempt, this is the one path that moves a worker to a
// different installed build.
//
// build is the build live/<host> names, as the sweep just read it. On a box
// that declined CARLOS_SWAP=overlap and honours failure records or hold-ran
// markers, the live exec gate is asked about it first (Task 9): the old
// child is retired before the spawn, so a refusal at the spawn would leave
// the worker down. Refused, nothing is stopped, the worker keeps running its
// current build, and it reports false: no converge happened, and none of the
// sweep's budget is spent.
func (a *Service) convergeWorkerLocked(e *actEntry, build string) bool {
	if a.liveGateRefusesLocked(e, build, "it keeps running "+e.spawnVersion) {
		return false
	}
	log.Printf("activator: worker %s serving %s but live symlink is now %s — respawning onto the new build", e.host, e.spawnVersion, build)
	a.respawnWorkerLocked(e, e.proc, "converging onto a new build", nil)
	return true
}

// liveGateRefusesLocked asks the live exec gate (SetLiveExecGate) whether
// e may move onto build (a worker's converge or recycle, or the sweep's
// cycle of an instance or hibernating worker), the build live/<host> names,
// and reports a refusal, which it logs once per host and build. No gate
// (every box but a declined one that honours something) refuses nothing.
// The gate reads only memory and the row. The caller holds a.mu.
func (a *Service) liveGateRefusesLocked(e *actEntry, build, left string) bool {
	if a.liveExecGate == nil {
		return false
	}
	err := a.liveExecGate(e.route, build)
	if err == nil {
		delete(a.liveGateSaid, e.host)
		return false
	}
	if a.liveGateSaid[e.host] != build {
		if a.liveGateSaid == nil {
			a.liveGateSaid = map[string]string{}
		}
		a.liveGateSaid[e.host] = build
		log.Printf("activator: %s: live/%s names %s, which this box will not start (%v), so it does not move onto it: %s", e.host, e.host, build, err, left)
	}
	return true
}

// workerSpawnTimeout bounds a supervised respawn's exec. It is generous
// because ExecBackend.StartWorker may materialize the build from the bucket
// first (a dangling live/<host> after a retention prune), but it is BOUNDED:
// an unbounded spawn that never returns would leave workerRespawning set, and
// the sweeper skips a respawning entry — no stand-down, no restart, no idle,
// and nothing able to wake it. A wedged fetch must cost one entry a couple of
// minutes, not the entry forever.
const workerSpawnTimeout = 2 * time.Minute

// respawnJoinBound bounds a terminal stop's wait on a supervised respawn: the
// respawn goroutine's own worst case end to end (retiring the old child, the
// exec, and shedding a raced one). A var so tests can shrink it.
var respawnJoinBound = actStopTimeout + workerSpawnTimeout + sidecarStopTimeout

// respawnWorkerLocked replaces a worker's child on the SAME entry, keeping its
// lease and replication: only the process is new. The caller holds a.mu.
// stopOld is the child to retire first (nil when it is already dead).
//
// onto is the build to exec: the captured serving build for a recycle
// (recycleWorkerLocked), or nil for a converge, which execs what live/<host>
// names now. A recycle onto a staged copy whose bytes have changed is refused
// by the backend (release.ErrStagedCopyChanged) and lands in the failed-spawn
// arm below, which keeps the captured build: the worker stays down, and each
// crash restart asks for that build again and is refused again while its
// bytes differ (spec §4.2, ruling R17).
//
// Everything expensive happens OFF a.mu, on its own goroutine. That is not a
// stylistic preference: ExecBackend.StartWorker can materialize a build from
// the bucket, and the SIGTERM drain of the outgoing child is bounded but not
// instant — either one held across the global mutex would stall every request
// wake, Owns and Shutdown on the box. (The co-spawn supervisor above DOES exec
// under the lock; that is StartSidecar, which never fetches anything. The two
// are not interchangeable.)
//
// Three orderings carry the safety:
//   - Retire, THEN spawn. A sidecar is an outbound worker: two live
//     incarnations double its work, where a gap costs only latency.
//   - Ask again, under the lock, immediately BEFORE spawning. A stop that
//     landed while the old child was draining has already released the lease,
//     and a child spawned after that would be writing the tenant's DB with no
//     single-writer claim behind it. Shedding it afterwards is not enough —
//     the agent may be gone by then.
//   - Check once more before INSTALLING, and shed the child if the entry was
//     stopped or replaced meanwhile. A stop always wins.
//
// e.workerRespawning is the sweeper's interlock (mid-respawn the entry's proc
// reads as exited, which read as a crash would spawn a second child), and
// e.respawnDone is what stop() and Shutdown wait on so a stop never completes
// — nor the process exit — while a spawn is still in flight.
//
// A failed spawn deliberately leaves the dead child in place: the next sweep
// sees Exited and the crash-restart path takes over with its backoff. A failed
// converge degrades into supervision rather than into a hole.
func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, onto *release.StagedIdentity) {
	if a.cfg.HostWrites.Enabled {
		e.attemptVeto = ErrClaimIsolationUnavailable
		e.respawnDone = make(chan struct{})
		close(e.respawnDone)
		return
	}
	e.workerRespawning = true
	e.respawnDone = make(chan struct{})
	done, rt, host := e.respawnDone, e.route, e.host
	// A socketless attempt's checks and a fallback's held publication arm
	// this respawn only (ReplaceWorker sets them just before the call): taken
	// here, under the caller's a.mu, so no later respawn inherits them.
	attemptGate, hold := e.attemptGate, e.holdServing
	e.attemptGate, e.holdServing = nil, false
	// check is the ownership check every exec asks, whatever the flag (spec
	// §7.4), ANDed with an attempt's own checks when this respawn execs an
	// attempt's build (spec §6.3, §8.1): a newer release, or a halted
	// cohort, whose halt bumps the revision, that lands while the old child
	// drains or the build is prepared means the attempt's build must not
	// run. The caller holds a.mu.
	check := func() error {
		if err := a.ownedLocked(e, false); err != nil {
			return err
		}
		if attemptGate != nil {
			return attemptGate()
		}
		return nil
	}
	// With generations on, the respawn is a member of the operation set
	// until it has finished with every child it made: a stop's barrier joins
	// it, and cancelling it tears down the exec (zero-downtime plan 3,
	// Task 11). Off, m stays nil and spawnParent is just lifeCtx's child.
	spawnParent, cancelSpawn := context.WithCancel(a.lifeCtx)
	var m *genMember
	oldGen := e.gen // the retiring child's generation, for its proof
	if a.gens {
		m = &genMember{gen: Gen{N: nextGen(), Worker: true}, role: roleSpawning, cancel: cancelSpawn, done: make(chan struct{})}
		a.addMemberLocked(e, m)
	}
	go func() {
		// respawnDone closes LAST, once this goroutine is finished with every
		// child it created — including one it had to SHED. That ordering is
		// the invariant stop() leans on: when respawnDone is observed closed,
		// no child spawned by this respawn is still running, so the final
		// checkpoint and lease release below it cannot land while a live
		// process is still pointed at the same database. Closing it before the
		// shed (as this first did) only narrowed that race; it did not close
		// it.
		defer close(done)
		defer cancelSpawn()
		shed := false // a shed child went to dropMember, which leaves the set itself
		if m != nil {
			defer func() { // runs before close(done): defers are LIFO
				if !shed {
					a.leaveSet(e, m)
				}
			}()
		}
		var (
			proc  Proc
			ident release.StagedIdentity
			pend  pendingGeneration // the build published before the exec
			tok   string
			err   error
			// refused is the config gate's answer for this respawn (amadan
			// #213). The lease is still held, so Acquire's gate is not in
			// the way, and a respawn onto config this box cannot provide
			// would start a worker without its secrets while keeping the
			// lease from a sibling that could run it.
			refused error
		)
		// The old child's stop, timed on the wall clock: a socketless
		// attempt reports it as its drain (spec §9.4).
		var retireTook time.Duration
		if stopOld != nil {
			retireStart := time.Now()
			stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
			if serr := stopOld.Stop(stopCtx); serr != nil {
				log.Printf("activator: worker %s respawn: retiring the old child: %v", host, serr)
			}
			cancel()
			retireTook = time.Since(retireStart)
		}

		// The retired child must be proven gone, its host cgroup included,
		// before anything new execs beside its descendants (zero-downtime
		// spec §7.4, §7.9). Unproven: nothing spawns; the entry keeps its
		// lease, and the sweep's crash path retires it again through
		// genTerminated. stopOld may be nil (a crash restart, a restore):
		// the old child is gone, but what it forked lives in the host
		// cgroup, and genTerminated proves a Worker generation there with
		// or without a handle.
		retired := true
		if a.gens {
			oldGen.Worker = true
			retired = a.genTerminated(rt, oldGen, stopOld)
		} else {
			// Generations off, an earlier respawn's shed child that is not
			// proven gone may still run beside this exec (final review
			// R2-1): nothing execs until the sweep has proven it exited.
			a.mu.Lock()
			retired = e.shed == nil
			a.mu.Unlock()
		}

		// Still wanted? Asked HERE, before anything is spawned: past this
		// point a stop can only be raced, never prevented. The gate's
		// ownership check (spec §7.4): every exec asks it, whatever the
		// switch says. It covers the entry still being the host's, awake
		// and not shutting down, and adds the lease still held and no
		// fence. An attempt's own checks are ANDed in (check), and plan 3's
		// termination proof (retired) is kept beside both: nothing execs
		// beside an old child that is not proven gone.
		a.mu.Lock()
		ownErr := check()
		a.mu.Unlock()
		owned := ownErr == nil
		wanted := owned && retired

		if wanted {
			// Off the service lifetime, so Shutdown cancels this exec instead
			// of blocking on it. The timeout is the slow-materialize bound;
			// the parent is the "are we still here at all" one.
			ctx, cancel := context.WithTimeout(spawnParent, workerSpawnTimeout)
			// The ownership check again at the exec itself (final review
			// F1): the config admission, the materialization and the
			// build's publication all come between the check above and the
			// exec, and the lease can be lost, a fence set or the entry
			// stopped in that time. startWorker hands it to the backend as
			// the exec gate, asked after the publication and the cgroup's
			// preparation and before each start attempt (Astra R3 §5 item
			// 4). An attempt's checks ride on it (check): a supersession in
			// that time refuses the exec too.
			gate := func() error {
				if err := ctx.Err(); err != nil {
					return err
				}
				a.mu.Lock()
				defer a.mu.Unlock()
				return check()
			}
			if g, ok := a.backend.(configAdmitter); ok {
				refused = g.AdmitConfig(ctx, rt)
			}
			if refused == nil {
				// No registry write (spec §7.9): see doWake.
				if tok, err = newWorkerToken(); err == nil {
					proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok), onto, gate)
				}
			}
			cancel()
		}

		a.mu.Lock()
		e.workerRespawning = false
		// And once more before the child is installed: ownership lost after
		// the exec sheds the child (final review F1). ownedLocked covers what
		// the old liveness check did (the entry is the host's, awake, not
		// shutting down) and adds the lease and the fences; an attempt's
		// checks ride on it (check), so an attempt superseded during its exec
		// sheds its child rather than install it.
		if owned {
			ownErr = check()
		}
		live := ownErr == nil
		if attemptGate != nil {
			// The attempt's answer for ReplaceWorker: which gate refused, at
			// the first check, at the exec, or before the install. Nil when
			// nothing refused (an unproven old child, a failed exec, a
			// config refusal or an install), which ReplaceWorker reads from
			// the incarnation instead.
			switch {
			case ownErr != nil:
				e.attemptVeto = ownErr
			case refused == nil && verdictOf(err) != gateProceed:
				e.attemptVeto = err
			}
			// The execution evidence (final review F2): a process of the
			// attempt's build was started, installed below or shed, or the
			// start's error says it may have been (startMayHaveRun; final
			// review wave 2, N4). Without it, nothing of the build ran.
			e.attemptExec = startMayHaveRun(proc, err)
		}
		switch {
		case owned && !retired && live:
			a.mu.Unlock()
			log.Printf("activator: worker %s NOT respawned (%s): an earlier child's termination is not proven (its host cgroup or a shed respawn child may still hold a process, or the proof could not run), so nothing execs beside it; the sweep retries", host, why)
		case !wanted || !live:
			a.mu.Unlock()
			reason := ownErr
			if reason == nil {
				reason = errors.New("the old generation's termination is not proven")
			}
			log.Printf("activator: worker %s NOT respawned (%s): %v; abandoning it", host, why, reason)
			if proc != nil {
				// Spawned into a window that closed under us: shed it rather
				// than leave a child nothing owns. With generations on it is
				// the member's process: dropMember stops it and proves its
				// host cgroup empty, or keeps it in the set for the reaper.
				if m != nil {
					// The child's build moves to the member before its
					// pending record is dropped below: if dropMember cannot
					// prove the child gone, the member is retained, and the
					// generations file keeps naming its build until the
					// reaper proves it (final review F12).
					a.mu.Lock()
					m.proc, m.build = proc, ident
					a.mu.Unlock()
					shed = true
					a.dropMember(e, m)
				} else {
					// Generations off (final review R2-1): the entry takes
					// the child, and its build moves from the pending record
					// to it in the same critical section, BEFORE the Stop.
					// A terminal stop that gives up waiting on this respawn
					// still finds it, and keeps the lease; the sweep reaps it
					// once its exit is proven (reapShedLocked).
					s := &shedChild{proc: proc, build: ident}
					a.mu.Lock()
					e.shed, s.retriedAt = s, a.now()
					a.dropPendingGenerationLocked(pend)
					a.mu.Unlock()
					pend = 0
					shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
					serr := proc.Stop(shedCtx)
					shedCancel()
					a.mu.Lock()
					gone := proc.Exited()
					if gone && e.shed == s {
						e.shed = nil
					}
					a.mu.Unlock()
					if !gone {
						log.Printf("activator: worker %s: the shed respawn child's exit is not proven (%v); the entry keeps it and its build until the sweep proves it", host, serr)
					}
					_ = a.persistGenerations()
				}
			}
			if pend != 0 {
				// Its published build stays listed until the child is shed.
				a.mu.Lock()
				a.dropPendingGenerationLocked(pend)
				a.mu.Unlock()
				_ = a.persistGenerations()
			}
		case refused != nil:
			// Stand down: the old child is already gone, and nothing may
			// start in its place. A normal stop checkpoints and releases the
			// lease, so a keyed sibling can take it. Nothing re-wakes it here
			// until the box can provide the config (the config gate refuses
			// every wake before then).
			a.beginStopLocked(e, StopRestart)
			a.mu.Unlock()
			log.Printf("activator: worker %s NOT respawned (%s): %v — standing down and releasing its lease", host, why, refused)
		case verdictOf(err) != gateProceed:
			// The exec's own ownership check refused (startWorker's gate),
			// and ownership came back before the install check (a fence
			// lifted), or an attempt's checks refused there: nothing exec'd,
			// and this is no failure of the build. The dead child stays, and
			// the sweep's crash restart asks again (an attempt's restore is
			// ReplaceWorker's, from attemptVeto).
			a.mu.Unlock()
			log.Printf("activator: worker %s NOT respawned (%s): refused at its exec: %v", host, why, err)
		case err != nil:
			// The captured build is kept whatever the failure, a refused
			// recycle (release.ErrStagedCopyChanged) included: spec §4.2
			// refuses a recovery onto changed bytes, and that refusal must
			// outlast the retry (ruling R17), so the next crash restart asks
			// for the same captured build again. Only a converge, admitted
			// on its own terms, moves the worker onto another build.
			a.mu.Unlock()
			log.Printf("activator: WORKER %s failed to respawn (%s): %v — the sweep's crash restart takes it from here", host, why, err)
			emitEvent(a.Logs, rt, "error", "sidecar could not be restarted. retrying")
		default:
			e.proc = proc
			e.workerToken = tok
			e.spawnVersion = ident.Version
			e.workerServing = ident
			e.workerVersion = "" // the new child has not reported yet
			e.retireTook = retireTook
			// A new incarnation, published below (spec §7.9), unless this is
			// a ReplaceWorker fallback: its publication is held until the
			// scheduler has recorded the failure it falls back from (spec
			// §8.1), and PublishHeldServing delivers it.
			e.workerInc++
			inc := e.workerInc
			if hold {
				e.heldServing = &heldServing{id: ident, inc: inc}
			}
			if m != nil {
				e.gen = Gen{N: m.gen.N, Build: ident.Version, Worker: true}
			}
			// The captured build replaces the one published before the exec.
			a.dropPendingGenerationLocked(pend)
			// Only an exec from the live link starts a fresh drift clock. A
			// recycle onto the same captured build serves exactly what it
			// served before, so a drift it was serving keeps its age, and a
			// max_run shorter than ConvergeGrace cannot starve the converge
			// (ruling R13).
			if onto == nil {
				e.staleSince = time.Time{}
			}
			e.spawnedAt = a.now()
			e.lastActive = a.now()
			a.mu.Unlock()
			// The captured build may have changed (a converge): publish it.
			_ = a.persistGenerations()
			if !hold {
				a.publishWorkerServing(e, ident, inc)
			}
			log.Printf("activator: worker %s respawned (%s), now running %s", host, why, ident.Version)
			emitEvent(a.Logs, rt, "info", "sidecar restarted")
		}
	}()
}

// NoteWorkerContact stamps last-activity from an agent-socket call. It is a
// worker's only sign of life: nothing routes to a sidecar, so no request will
// ever stamp one, and without contact the idle clock runs to its window and
// the sweeper sleeps it.
func (a *Service) NoteWorkerContact(host string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil && e.worker {
		e.lastActive = a.now()
	}
}

// NoteWorkerReady records a worker's own "I am up": last-activity, the end
// of its settle window, and the build it SAYS it runs. That claim is display
// metadata only (zero-downtime deploys spec §7.9): it can be empty or an
// app-chosen string, so it never identifies the executable, never overwrites
// registry Version (root's install stamp) and never decides a recycle. The
// serving build is workerServing, captured at the exec.
//
// It acts on a WAKING entry as well as an awake one, which is the whole point
// of the settle window: a child that boots fast says so and its wake resolves
// immediately instead of waiting out sidecarSettle.
func (a *Service) NoteWorkerReady(host, version string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || !e.worker {
		return
	}
	e.lastActive = a.now()
	e.workerVersion = version
	if !e.readyNoted {
		e.readyNoted = true
		close(e.readyc)
	}
}

// SetWorkerNextDue records a sidecar's next scheduled work, straight to the
// due-time index — while the child is ALIVE and can say so. That is why the
// stop path never probes a worker: the answer is already recorded, and a probe
// of a process we are about to kill could only be worse.
func (a *Service) SetWorkerNextDue(host string, at int64) error {
	return a.due.SetNextDue(host, at)
}

// SleepWorker stops an awake HIBERNATING worker now — the /v1/idle call, "I
// have nothing to do". It is a stop REQUESTED immediately rather than an idle
// timestamp nudged backwards, so the ordinary stop path (park the DB, release
// the lease) does the work. Anything other than an awake worker is a no-op:
// this must never start something, and never touch an instance.
//
// An ALWAYS-ON worker's request is logged and REFUSED. Its contract is "the
// record exists, so the process runs" — and obeying would strand it: the stop
// deletes the entry, so not even the crash-restart sweep has anything left to
// see, and it stays down until the agent restarts. A sidecar that genuinely
// wants to stop exits its own process, which the sweeper then restarts (or
// its operator deletes the record, which retires it for good).
func (a *Service) SleepWorker(host string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || !e.worker || e.state != actAwake {
		return
	}
	if !e.route.Hibernate {
		log.Printf("activator: worker %s reported idle but is always-on (hibernate:false) — ignoring", e.host)
		return
	}
	emitEvent(a.Logs, e.route, "info", "sidecar reported idle — stopping")
	a.beginStopLocked(e, StopIdle)
}

// WorkerToken reports the live per-incarnation token for host, or "" — the
// agent socket's authorization check. A token from a previous wake, or from a
// previous incarnation of a restarted child, never matches.
//
// It answers for a WAKING entry as well as an awake one, because the child's
// first call — /v1/ready, which ends the settle window — necessarily arrives
// while the wake is still resolving. Refusing then would make the early-ready
// path unreachable and cost every wake the full window. Nothing else is
// loosened: a stopping entry, an asleep host, an instance and an unknown host
// all still get "", and a failed wake takes its token with it when the entry
// is deleted.
func (a *Service) WorkerToken(host string) string {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil && e.worker && (e.state == actAwake || e.state == actWaking) {
		return e.workerToken
	}
	return ""
}

// setWorkerToken publishes a wake's freshly minted credential onto the entry.
// Its own critical section because doWake runs off a.mu — every other write to
// an entry's fields is likewise made under the lock.
func (a *Service) setWorkerToken(e *actEntry, tok string) {
	a.mu.Lock()
	e.workerToken = tok
	a.mu.Unlock()
}

// stop drives awake→asleep: stop the child (its SIGTERM drain is the final
// write), checkpoint what's on disk to S3 and release the lease, then forget
// the entry. The local DB stays in place — it's the warm cache for the next
// wake; S3 is the durable truth under the lease.
func (a *Service) stop(e *actEntry) {
	outcome := &stopObservation{sidecarStop: e.sidecar == nil, sidecarExited: e.sidecar == nil, uncertain: e.sidecarUncertain}
	// The deliberate-stop marker for the member's timeline: hibernation,
	// convergence cycling, bounce, shutdown all pass through here, so an
	// "exited" line WITHOUT this line before it reads as what it is — a
	// crash.
	stopping := "instance stopping"
	if e.worker {
		stopping = "sidecar stopping"
	}
	emitEvent(a.Logs, e.route, "info", stopping)

	// Wait out a supervised respawn first. Everything below — the final
	// checkpoint and the LEASE RELEASE — assumes no other process is being
	// pointed at this DB; a respawn that finished after the release would run
	// a child with no single-writer claim behind it, on the one path whose
	// whole purpose is single-writer. The respawn's own pre-spawn check makes
	// this wait short (it aborts as soon as it sees this state flip); the
	// bound is insurance against a wedged exec, and instances never wait at
	// all (respawnDone is nil for them).
	//
	// Gated on the CHANNEL alone, deliberately, and never on
	// e.workerRespawning: that flag is cleared in the respawn's final critical
	// section, BEFORE it sheds a child it raced into existence, so a stop
	// goroutine scheduled after that section would read false and skip the
	// wait entirely — with up to sidecarStopTimeout of draining child left to
	// go. beginStopLocked flips the state under a.mu and then launches this
	// goroutine, so the two genuinely race for the lock with nothing ordering
	// them. The channel does not have that gap: it is created before the
	// respawn starts and closed only when the respawn is finished with every
	// child it made. A respawn that ended long ago costs a receive on an
	// already-closed channel.
	a.mu.Lock()
	waitRespawn := e.respawnDone
	a.mu.Unlock()
	if waitRespawn != nil {
		// The bound covers the respawn goroutine's OWN worst case end to end —
		// retiring the old child, the exec, and shedding a raced one — so this
		// wait can never give up on a respawn that is still making progress.
		bound := time.NewTimer(respawnJoinBound)
		select {
		case <-waitRespawn:
		case <-bound.C:
			log.Printf("activator: %s respawn did not finish before its stop; stopping the rest, the lease is kept until it has", e.host)
		}
		bound.Stop() // the common case closes in milliseconds; don't leave a 2m timer standing
	}

	// A sidecar restart (sweepOnce) runs its backend.StartSidecar call off
	// a.mu (design §6.7) and may still be in flight when a stop begins. Wait,
	// bounded by sidecarStopTimeout, for it to land: either it installs a
	// fresh e.sidecar in time and the co-stop below stops that one directly,
	// or it doesn't and the restart goroutine's own entry-identity check (it
	// re-reads a.entries[host] on the far side of the call) finds this entry
	// gone or stopping and stops whatever it started instead. Either way
	// nothing is left orphaned; this wait only decides which of the two
	// stops it. sidecarStartDone is closed by the restart goroutine when it
	// finishes, however it finishes, so this waits on that signal directly
	// rather than polling sidecarStarting on a timer.
	//
	// With generations off the wait is bounded (sidecarStartJoinWait), but
	// a start still in flight past it is not left behind (Task 8, F6 plan
	// 5): the rest of the stop runs, and then the stop stays pending, with
	// the lease, until the start returns and has stopped what it started
	// (the restart goroutine stops it before it closes sidecarStartDone).
	// Only then is the host cgroup proven empty and the lease released
	// (reapCurrentGen).
	a.mu.Lock()
	startDone, hook := e.sidecarStartDone, a.joinWaiting
	a.mu.Unlock()
	var sidecarJoin chan struct{}
	if startDone != nil {
		if hook != nil {
			hook(e.host, "sidecar start")
		}
		if a.gens {
			<-startDone // the start, and every process it left, is accounted for
		} else {
			bound := time.NewTimer(sidecarStartJoinWait)
			select {
			case <-startDone:
			case <-bound.C:
				// Recorded now, so every pending path below (a leader still
				// running, an unproven generation) waits for it too.
				sidecarJoin = startDone
				a.mu.Lock()
				e.sidecarJoin = startDone
				a.mu.Unlock()
			}
			bound.Stop()
		}
	}

	// The barrier (zero-downtime spec §7.4). Every generation in this
	// entry's operation set is cancelled and joined before the final
	// checkpoint and the lease release below: a spawning or health-checking
	// successor, and a draining retiring generation. With no swap ever run
	// on this entry the set is empty and this returns at once.
	joined := a.joinOps(e) // swapsim:M1

	// Co-stop, sidecar FIRST — before the next-due probe and the instance's
	// own SIGTERM. Two reasons: the sidecar must never outlive its instance
	// (an orphan's reconnect loop would be a standing wake source — the
	// ping-pong this feature exists to prevent — and its scheduled work would
	// keep firing into a host we're about to stop), and with the sidecar gone
	// first, the schedule the next-due probe reads can't shift under it.
	if e.sidecar != nil {
		scCtx, scCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
		err := e.sidecar.Stop(scCtx)
		outcome.sidecarStop, outcome.sidecarExited = err == nil, e.sidecar.Exited()
		if err != nil {
			log.Printf("activator: stop sidecar %s: %v", e.host, err)
		}
		scCancel()
		if !a.gens && sidecarStopUnproven(e.sidecar, err) {
			// Flag off, its exit is kept to be proven before the lease goes
			// (fix wave 1, item 1): a nested row has no host-cgroup proof to
			// reach it.
			a.mu.Lock()
			e.unprovenSidecars = append(e.unprovenSidecars, e.sidecar)
			a.mu.Unlock()
		}
		if a.gens {
			// A sidecar is gone only when its leader exited AND its cgroup is
			// empty (spec §7.4, §7.5). Unproven, it is tracked: the lease is
			// kept (releasable) and the sweep re-proves it.
			a.mu.Lock()
			rt := e.route
			a.mu.Unlock()
			if !a.sidecarTerminated(rt, e.sidecar) {
				a.mu.Lock()
				a.trackStraySidecarLocked(rt, e.sidecar)
				a.mu.Unlock()
			}
		}
	}

	// The due-time index: ask the instance for its earliest pending scheduled
	// delivery BEFORE stopping it — the socket dies with the process. (When the
	// stop was triggered by a dead child, the query just fails fast.) Any
	// failure records 0, i.e. "sleeps until a request": a missed probe must
	// degrade to pre-index behavior, never block or crash the sleep. A due time
	// already in the past is recorded as-is and the sweeper wakes the host on
	// its next tick — there is deliberately no minimum-sleep guard beyond
	// that. The idle sweep alone would make thrash impossible (an awake
	// instance only sleeps after a full idle window), but Bounce breaks that
	// bound on purpose: it can stop an instance shortly after it woke, so
	// consecutive wake/sleep cycles are bounded by config-change frequency,
	// not cfg.Idle. That's accepted — a bounce is rare relative to request
	// traffic, and each cycle is still a bounded SIGTERM drain, not a busy
	// spin.
	//
	// A WORKER skips this step entirely — both the probe and the write. It has
	// no socket to answer on, and its next-due already arrived over the agent
	// socket while it was alive (SetWorkerNextDue wrote it straight to the
	// registry), which is strictly better than asking a process we are about
	// to kill. Writing here would be actively wrong: the probe would fail, and
	// the failure path records 0 — erasing the schedule that is a sleeping
	// sidecar's ONLY way back up, since nothing can dial one.
	if !e.worker {
		dueCtx, cancelDue := context.WithTimeout(context.Background(), 5*time.Second)
		due, err := a.nextDue(dueCtx, e)
		cancelDue()
		if err != nil {
			log.Printf("activator: next-due probe %s: %v — recording none (it sleeps until a request)", e.host, err)
			due = 0
		}
		if err := a.due.SetNextDue(e.host, due); err != nil {
			log.Printf("activator: record next-due %s: %v", e.host, err)
		} else if due > 0 {
			log.Printf("activator: %s next scheduled delivery at %s — will wake for it", e.host, time.Unix(due, 0).UTC().Format(time.RFC3339))
		}
	}

	// Separate budgets: a SIGTERM drain that runs long must not eat the final
	// checkpoint's time — the S3 sync is the part that makes sleep durable.
	procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout)
	defer cancelProc()
	// Declared rather than inferred: the next-due probe above used to leave an
	// `err` in this scope, but it is inside the `if !e.worker` block now, so
	// this one is its own.
	var err error
	if e.proc != nil {
		err = e.proc.Stop(procCtx)
		if errors.Is(err, ErrStillRunning) {
			// The lease is never released over an unproven termination
			// (design §6.3, codex R2-5): stop() bails out here rather than
			// falling through to the checkpoint and lease release below,
			// leaving the entry in stopping with pendingStop set.
			// sweepOnce polls proc.Exited() for it and calls finishStop,
			// the rest of what this function would have done, the moment
			// termination is actually proven.
			log.Printf("activator: stop %s: %v; keeping the lease until termination is proven", e.host, err)
			// The call just above is this entry's first stop attempt, and it
			// was already bounded by actStopTimeout. keepPendingLocked
			// records it as stopRetriedAt, which is what makes
			// finishPendingStops' own cadence a true actStopTimeout apart
			// from THIS attempt, rather than firing a redundant retry the
			// moment pendingStop becomes visible to it.
			a.mu.Lock()
			a.keepPendingLocked(e, outcome)
			a.mu.Unlock()
			return
		}
		outcome.mainStop, outcome.mainExited = err == nil, e.proc.Exited()
	} else {
		err = ErrStopUnproven
	}
	if err != nil {
		log.Printf("activator: stop %s: %v", e.host, err)
	}
	if !joined {
		// A generation's termination is unproven, so the lease stays held,
		// as for ErrStillRunning above. finishPendingStops reaps the member
		// and then finishes this stop.
		log.Printf("activator: stop %s: a generation's termination is unproven; keeping the lease until it is", e.host)
		a.mu.Lock()
		a.keepPendingLocked(e, outcome)
		a.mu.Unlock()
		return
	}
	a.mu.Lock()
	settled := a.respawnSettledLocked(e)
	if !settled {
		// A respawn still running past the bounded wait above, or the
		// child it shed and could not prove gone (final review R2-1): the
		// lease stays held, as for ErrStillRunning, and finishPendingStops
		// finishes this stop once the respawn is done and its child proven
		// exited.
		a.keepPendingLocked(e, outcome)
	}
	a.mu.Unlock()
	if !settled {
		log.Printf("activator: stop %s: a respawn or the child it shed is not proven finished; keeping the lease until it is", e.host)
		return
	}
	if !a.gens && sidecarJoin != nil {
		// A sidecar start is still in flight past the bounded wait above
		// (Task 8, F6 plan 5): the stop stays pending with the lease until
		// it returns, and finishPendingStops proves the host cgroup then.
		log.Printf("activator: stop %s: a sidecar start is still in flight; keeping the lease until it returns", e.host)
		a.mu.Lock()
		a.keepPendingLocked(e, outcome)
		a.mu.Unlock()
		return
	}
	if !a.gens && !a.sidecarsExited(e, false) {
		// A sidecar this stop stopped, or one a restart brought back, has
		// not exited (fix wave 1, item 1): the stop stays pending with the
		// lease, and the sweep re-stops it (reapCurrentGen).
		a.mu.Lock()
		a.keepPendingLocked(e, outcome)
		a.mu.Unlock()
		return
	}
	if !a.gens && !a.stopHostProven(e) {
		// Flag off, the leader's exit says nothing of what it forked (Task
		// 8, F6): the host cgroup is proven empty first. Unproven, the stop
		// stays pending with the lease, as for ErrStillRunning, and the
		// sweep retries the proof (reapCurrentGen).
		log.Printf("activator: stop %s: its host cgroup is not proven empty; keeping the lease until it is", e.host)
		a.mu.Lock()
		a.keepPendingLocked(e, outcome)
		a.mu.Unlock()
		return
	}
	if a.gens && !a.releasable(e) { // swapsim:M8
		// THE retain-until-proven rule (spec §7.4): unproven, the stop stays
		// pending with the lease held, exactly as for ErrStillRunning, and
		// the sweep retries (reapCurrentGen). That covers a leader still
		// alive after a failed kill (a claims-on KillAndEmpty error is not
		// ErrStillRunning, so it reaches here), a child not proven empty, a
		// worker's host cgroup, and a stray sidecar.
		log.Printf("activator: stop %s: not every process of the entry is proven gone; keeping the lease until it is", e.host)
		a.mu.Lock()
		a.keepPendingLocked(e, outcome)
		a.mu.Unlock()
		return
	}
	a.finishStop(e, outcome)
}

// finishPendingStop finishes a stop that was left pending because proc.Stop
// returned ErrStillRunning (design §6.3): now that proc.Exited() has proven
// termination, spec §6.3 calls for retrying proc.Stop once more before
// declaring the main process closed. Spec §6.3 says "sweepOnce retries
// proc.Stop on its cadence until done closes", so outcome.mainStop reflects
// an actual observation rather than an assumption baked in at the
// ErrStillRunning return. A unitProc that has already exited takes its
// done branch and returns nil immediately; a childProc's Stop is a no-op
// on an already-exited process either way, so retrying costs nothing once
// the process is well and truly gone. mainExited is unconditionally true:
// the exit that made this entry eligible for this pass IS that proof,
// whatever this retried Stop call itself reports.
func (a *Service) finishPendingStop(e *actEntry, outcome *stopObservation) {
	ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
	err := e.proc.Stop(ctx)
	cancel()
	outcome.mainStop, outcome.mainExited = err == nil, true
	if err != nil {
		log.Printf("activator: stop %s: %v", e.host, err)
	}
	a.finishStop(e, outcome)
}

// finishPendingStops scans for entries left pending because proc.Stop
// returned ErrStillRunning (design §6.3, codex R2-5) and finishes every one
// whose proc has since exited, via finishPendingStop. Called from
// sweepOnce's own tick and, once more, from Shutdown. The sweeper is
// stopped by the time Shutdown runs, so without this second call a process
// that exits during shutdown would never get its final checkpoint. The
// caller holds a.mu.
func (a *Service) finishPendingStops() {
	// Failed wakes whose generation is not proven stopped hold their lease
	// until it is (zero-downtime spec §7.4).
	a.reapUnprovenWakesLocked()
	// Lease releases that failed are retried on the same tick (#292).
	a.retryOwedReleasesLocked()
	a.reapStraySidecarsLocked()
	for _, e := range a.entries {
		// Members first (zero-downtime spec §7.4): a generation whose stop
		// could not be proven is finished here once it exits.
		a.reapMembersLocked(e)
		// And a shed respawn child kept with generations off (final review
		// R2-1).
		a.reapShedLocked(e)
		// A parent restore that failed when the last member left keeps its
		// obligation and is retried here (final review F11).
		a.retryParentRestoreLocked(e)
		// A run whose stop was asked for and is not proven, pending stop
		// or not (re-review 6).
		a.retryCmdSettlesLocked(e)
		if !e.pendingStop || len(e.ops.members) > 0 || !a.respawnSettledLocked(e) {
			continue
		}
		if e.proc == nil || e.proc.Exited() {
			// Members gone and the main process (if any) exited: the stop
			// finishes once its termination proof holds (reapCurrentGen):
			// flag on, the retain-until-proven rule (releasable); flag off,
			// once a sidecar start it left in flight has returned, the host
			// cgroup proven empty (Task 8, F6). A process still alive falls
			// through to main's re-stop cadence below.
			if !e.genReaping && (a.gens || sidecarStartJoined(e.sidecarJoin)) {
				e.genReaping = true
				go a.reapCurrentGen(e)
			}
			continue
		}
		// Still running: spec §6.3 says sweepOnce retries proc.Stop on its
		// own cadence until done closes, and that applies here too, not only
		// once termination has already been proven some other way. Without
		// this, a stop whose first Control.Stop was itself lost (the socket
		// refused, a 502 mid-roll) never asks again, and the entry sits in
		// stopping holding the lease until the agent restarts.
		if e.stopRetrying || a.now().Sub(e.stopRetriedAt) < actStopTimeout {
			continue
		}
		e.stopRetrying = true
		e.stopRetriedAt = a.now()
		go a.retryPendingStop(e)
	}
}

// sidecarStartJoined reports whether a sidecar start a flag-off stop left
// in flight (actEntry.sidecarJoin) has returned: none was, or its done
// channel is closed.
func sidecarStartJoined(join chan struct{}) bool {
	if join == nil {
		return true
	}
	select {
	case <-join:
		return true
	default:
		return false
	}
}

// respawnDoneLocked reports whether e has no supervised respawn in flight.
// The caller holds a.mu.
func respawnDoneLocked(e *actEntry) bool {
	if e.respawnDone == nil {
		return true
	}
	select {
	case <-e.respawnDone:
		return true
	default:
		return false
	}
}

// respawnSettledLocked reports whether e's supervised respawns are done
// with every child they made: none is in flight, and no child one shed is
// kept unproven (final review R2-1). A terminal stop finishes, and gives
// up the lease, only then. The caller holds a.mu.
func (a *Service) respawnSettledLocked(e *actEntry) bool {
	return e.shed == nil && respawnDoneLocked(e)
}

// reapShedLocked finishes a shed respawn child kept with generations off
// (final review R2-1), once the respawn that shed it is done. An exited
// child is forgotten and its build unpinned. One still running is stopped
// again, bounded, every actStopTimeout, the cadence finishPendingStops
// keeps for the main process. The caller holds a.mu.
func (a *Service) reapShedLocked(e *actEntry) {
	s := e.shed
	if s == nil || s.reaping || !respawnDoneLocked(e) {
		return
	}
	if !s.proc.Exited() && a.now().Sub(s.retriedAt) < actStopTimeout {
		return
	}
	s.reaping, s.retriedAt = true, a.now()
	go func() {
		if !s.proc.Exited() {
			ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
			err := s.proc.Stop(ctx)
			cancel()
			if err != nil && !errors.Is(err, ErrStillRunning) {
				log.Printf("activator: %s: retry stop of the shed respawn child: %v", e.host, err)
			}
		}
		a.mu.Lock()
		s.reaping = false
		gone := s.proc.Exited()
		if gone && e.shed == s {
			e.shed = nil
		}
		a.mu.Unlock()
		if gone {
			log.Printf("activator: %s: the shed respawn child has exited", e.host)
			_ = a.persistGenerations()
		}
	}()
}

// retryPendingStop re-issues proc.Stop for a pendingStop entry that is
// still running, off a.mu like finishPendingStop's own call. Unlike that
// one, a return here proves nothing by itself. Done closing is still what
// proves termination, and the next finishPendingStops pass is what notices
// it and calls finishPendingStop. This only clears the in-flight guard so a
// later pass can try again.
func (a *Service) retryPendingStop(e *actEntry) {
	ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
	err := e.proc.Stop(ctx)
	cancel()
	if err != nil && !errors.Is(err, ErrStillRunning) {
		log.Printf("activator: retry stop %s: %v", e.host, err)
	}
	a.mu.Lock()
	e.stopRetrying = false
	a.mu.Unlock()
}

// finishStop is stop()'s (and finishPendingStop's) tail: the final
// checkpoint, lease release, and forgetting the entry. Split out (design
// §6.3, codex R2-5) so a pendingStop entry's eventual finish shares it with
// stop()'s own ordinary path. outcome carries forward whatever stop() had
// already observed (the sidecar co-stop, the due-time probe) before it hit
// ErrStillRunning and returned, plus whatever finishPendingStop's own
// retried proc.Stop just added.
func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
	// A run attached to this instance is stopped and settled before
	// replication closes (run spec §5.3, codex 1): everything it committed
	// is in the final checkpoint, and nothing writes the database after.
	// A run whose stop is not proven keeps the entry, replication, the
	// lease and its handle (merge ruling M1): the stop stays pending, as
	// for any process whose termination is unproven, and the sweep tries
	// the run's settle again (finishPendingStops). The check lives here, so
	// no stop path can skip it.
	if !a.settleCmds(e) {
		log.Printf("activator: stop %s: a run is not proven stopped; keeping replication and the lease until it is", e.host)
		a.mu.Lock()
		a.keepPendingLocked(e, outcome)
		a.mu.Unlock()
		return
	}
	// Close = final sync + lease release. After a lost lease the release part
	// fails harmlessly (someone else owns the lock); the sync still preserves
	// a local checkpoint.
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	repErr := ErrStopUnproven
	if e.rep != nil {
		repErr = e.rep.Close(ctx)
	}
	outcome.replicaClosed = repErr == nil
	if repErr != nil {
		log.Printf("activator: final checkpoint %s: %v", e.host, repErr)
	}
	a.mu.Lock()
	outcome.uncertain = !outcome.complete()
	e.stopOutcome = outcome
	a.rememberStopLocked(e.host, outcome)
	republish := e.worker || a.publishesLocked(e)
	delete(a.entries, e.host)
	retired := e.gen
	// The canonical link goes with the last generation (spec §7.1). Under
	// a.mu, the lock the awake transition creates links under, so a new
	// entry's link cannot be removed by this one's stop.
	if a.gens && linksCanonical(e.route, retired) {
		if err := removeCanonicalLink(e.route.Socket, retired.Socket); err != nil {
			log.Printf("activator: %s: remove link %s: %v", e.host, e.route.Socket, err)
		}
	}
	// A hold swap (zero-downtime spec §7.8): its pinned wake is installed in
	// the same critical section, before e.stopped closes, so a request
	// waiting on this stop finds the pinned entry, never an empty slot it
	// would fill with a wake of the live link.
	var next *actEntry
	if h := e.holdNext; h != nil {
		e.holdNext = nil
		var herr error
		next, herr = a.holdWakeLocked(e, h)
		h.woke <- holdWoke{next: next, err: herr}
	}
	a.mu.Unlock()
	if republish {
		// Its processes are gone, so its builds are no longer referenced by
		// them.
		_ = a.persistGenerations()
	}
	// The generation stopped being current when its entry went (spec §7.2:
	// its idle connections close now, not at the transport's idle timeout).
	a.genRetired(e.host, retired)
	close(e.stopped)
	if next != nil {
		go a.wake(next)
	}
	log.Printf("activator: %s hibernated", e.host)
}

// Start launches the background sweeper that puts idle instances to sleep and
// stands down lost leases/dead children between requests.
func (a *Service) Start() {
	// The generations file is written before anything spawns, whatever
	// CARLOS_SWAP says (the index contract): a file left by a previous
	// incarnation is stale, and one a crash cut short reads to root as an
	// incomplete pin set, so root would neither restage nor sweep until this
	// process changes it. The write is off a.mu, and a failure is logged and
	// marks the publication dirty for the sweep to retry; it never fails or
	// blocks Start.
	_ = a.persistGenerations()
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.shuttingDown {
		return
	}
	a.sweepStop = make(chan struct{})
	a.sweepDone = make(chan struct{})
	// Sweep well inside the idle window so a sleep happens promptly after it
	// elapses; clamp so second-granular test idles still sweep sanely.
	interval := a.cfg.Idle / 4
	if interval < 100*time.Millisecond {
		interval = 100 * time.Millisecond
	}
	if interval > 30*time.Second {
		interval = 30 * time.Second
	}
	go func() {
		defer close(a.sweepDone)
		if a.Startup != nil {
			select {
			case <-a.Startup:
			case <-a.sweepStop:
				return
			}
		}
		t := time.NewTicker(interval)
		defer t.Stop()
		for {
			select {
			case <-a.sweepStop:
				return
			case <-t.C:
				a.sweepOnce()
				a.lastSweep.Store(time.Now().UnixNano())
			}
		}
	}()
}

// LastSweep is when the sweeper last finished a sweep, or the zero time
// before its first. The sweep runs at least every 30 seconds, so a box
// whose last sweep is older than a minute has a hosting loop that is stuck
// (region HA spec 2 §6). A nil service has never swept.
func (a *Service) LastSweep() time.Time {
	if a == nil {
		return time.Time{}
	}
	n := a.lastSweep.Load()
	if n == 0 {
		return time.Time{}
	}
	return time.Unix(0, n)
}

// markCleanParkLocked records that a clean-exit park is stopping e's current
// worker incarnation (round 2 Task 6). ReplaceWorker's observation forgives
// an exited one-shot that is no longer awake (spec §6.5) only when this mark
// names the incarnation it watched: the park ran with no fence, the lease
// still held and a clean exit. A stop for a lost lease, a fence or a
// shutdown never marks, so the build is never recorded good on it. The
// conditions are checked here again so the mark never depends on the
// caller's order. The caller holds a.mu.
func (a *Service) markCleanParkLocked(e *actEntry) {
	if a.fences[e.host] != 0 || e.rep == nil || e.rep.Err() != nil || procCrashed(e.proc) {
		return
	}
	e.parkedInc = e.workerInc
}

// sweepOnce examines every awake instance once: idle ones hibernate; a lost
// lease or dead child stands down immediately regardless of activity. It then
// checks the due-time index and wakes any sleeping host whose scheduled
// delivery has come due.
func (a *Service) sweepOnce() {
	now := a.now()
	// The due query runs before the lock — it's a registry read, and nothing
	// under a.mu should wait on SQLite.
	dueRoutes, err := a.due.DueInstances(now.Unix())
	if err != nil {
		log.Printf("activator: due-time sweep: %v", err)
	}
	// The always-on arm's input, read on the same off-lock terms and for the
	// same reason: nothing under a.mu should wait on SQLite.
	alwaysOn, err := a.due.AlwaysOnWorkers()
	if err != nil {
		log.Printf("activator: always-on sidecar sweep: %v", err)
	}
	// The guard's input, read here for the same reason the due query is:
	// nothing under a.mu may wait on SQLite. A failure does NOT degrade to
	// today's behaviour: earliest ends up nil, so no host is held, and a
	// host with a schedule tighter than the idle window flips from one
	// resident process into a wake/sleep pair every tick — a 5m schedule
	// becomes ~288 checkpoint-and-restore traversals a day, precisely the
	// trade the guard exists to refuse. Fail open anyway: refusing to
	// hibernate anything until the query recovers would hold every
	// instance on the box awake, which is worse than losing the guard for
	// the instances that have one.
	earliest, err := a.due.EarliestTimers()
	if err != nil {
		log.Printf("activator: earliest-timer sweep: %v — hibernating without the schedule guard", err)
		earliest = nil
	}
	a.mu.Lock()
	if a.shuttingDown {
		// Shutdown has begun and is about to stop this sweeper: from here it
		// owns every entry (#292). A child found exited now is its stop's,
		// not an unexpected exit to recycle, and nothing is due-woken or
		// started. Shutdown runs finishPendingStops itself.
		a.mu.Unlock()
		return
	}
	// Finish stops that were left pending because proc.Stop returned
	// ErrStillRunning (design §6.3, codex R2-5): termination wasn't proven
	// at the time, so stop() kept the lease and returned without forgetting
	// the entry. Checked before the idle pass below for the same reason it's
	// a separate call at all: a pendingStop entry is in actStopping, not
	// actAwake, so the idle pass would never reach it.
	a.finishPendingStops()
	// Build-convergence cycles this tick may start (rate-limited), shared
	// with the recycles below that converge instead (recycleWorkerLocked).
	a.convergeBudget = maxConvergeCyclesPerSweep
	for _, e := range a.entries {
		// Recovery retains its cut fence. Supervise its process and lease
		// before the ordinary fenced-host skip; only the claim can open it.
		if e.probeHeld && e.state == actAwake {
			if a.retireRecoveryProbeLocked(e) {
				continue
			}
			if !e.probeReviewing {
				e.probeReviewing = true
				go a.reviewRecoveryHold(e)
			}
			continue
		}
		if a.fences[e.host] != 0 {
			continue
		}
		if e.state != actAwake {
			continue
		}
		// A converge respawn owns this entry's child until it finishes: its
		// old proc is stopped and its new one not yet installed, which every
		// check below would misread. One tick of patience, not a policy.
		if e.workerRespawning {
			continue
		}
		if err := e.rep.Err(); err != nil {
			log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err)
			a.beginStopLocked(e, StopLeaseLost)
			continue
		}
		if e.proc.Exited() {
			// Whatever follows, the child died on its own: a first wake's
			// observation reads it as the build's failure (spec §6.4).
			e.unexpectedExit = true
			// A worker's exit is read by STATUS, which is the only thing that
			// can tell "my work is done" from "I fell over":
			//
			//   - always-on (`hibernate: false`): the record exists ⇒ the
			//     process runs, so ANY exit is restarted in place — same
			//     supervision as the co-spawn path, and no lease churn,
			//     because only the child died (the lease and replication are
			//     still ours).
			//   - hibernating, clean exit (status 0): the bot finished. Park
			//     it — the stand-down below, exactly as for an instance.
			//   - hibernating, NON-ZERO exit: a crash, and design §7.2 says a
			//     crash restarts, unqualified. Without this arm a hibernating
			//     worker that fell over before setting a next-due slept
			//     forever, and the log called it "exited" as though it had
			//     finished.
			//
			// The restart carries the same doubling backoff, so a crash loop
			// PACES itself — it is not terminated. Nothing below this arm ever
			// runs for a crashing child: this branch continues unconditionally
			// while the proc reads as exited, and a successful respawn resets
			// both spawnedAt and lastActive, so neither max_run nor the idle
			// clock can fire between crashes. A hibernating sidecar that
			// crash-loops therefore keeps its entry and its lease at a 2s→1m
			// cadence until an operator parks it (a wake stands a broken
			// worker down and recycles it) or its record goes away. A
			// crash-COUNT ceiling that parks after N failures is the right
			// bound and is filed as a follow-up: it is a design decision about
			// when the platform gives up on a member's bot, not a detail of
			// this arm.
			if e.worker && (!e.route.Hibernate || procCrashed(e.proc)) {
				a.restartWorkerLocked(e, now)
				continue
			}
			if e.worker {
				log.Printf("activator: sidecar %s exited cleanly — its work is done; parking it", e.host)
				// Past this sweep's fence skip and lease check: the one
				// stop of an exited one-shot that a ReplaceWorker
				// observation may still count as a good build.
				a.markCleanParkLocked(e)
			} else {
				log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
			}
			a.beginStopLocked(e, StopExited)
			continue
		}
		// The idle predicate is shared verbatim; only its WINDOW can differ.
		// A worker declares its own on its row (Task 1) so this sweep never
		// needs a bucket read to decide one has sat idle long enough; zero —
		// every instance, and any sidecar that declared nothing — falls back
		// to the platform default, so an instance's behaviour is unchanged.
		// inflight is always 0 for a worker (nothing routes to it) and
		// lastActive is stamped by agent-socket contact instead of by a
		// released request, so the same expression means the right thing for
		// both.
		//
		// An ALWAYS-ON worker has no idle clock at all (design §7.2:
		// "hibernate: false … no idle clock, no lease churn"): idleness is
		// not a reason to stop something whose record says it must be
		// running. It does still have a run CEILING, but that ceiling
		// recycles its child rather than stopping the entry (see max_run
		// below) — a stopped always-on worker has no wake source of its own.
		// An instance is never in this arm: unflagged instances don't enter
		// the activator.
		hibernates := !e.worker || e.route.Hibernate
		// A held probe entry (ProbeWake) gets no traffic: the edge routes no
		// held row. It is not idle until the target's open marks it open
		// (OpenProbeEntry, Plan 3 Task 7).
		if e.probe != nil && !e.probe.managed && e.probe.phase < probeOpened {
			hibernates = false
		}
		idleWindow := a.cfg.Idle
		if e.workerIdle > 0 {
			idleWindow = e.workerIdle
		}
		if f := a.idleFor.Load(); f != nil {
			if d := (*f)(e.route); d > 0 && d < idleWindow {
				idleWindow = d
			}
		}
		if hibernates && e.inflight == 0 && e.cmd == nil && now.Sub(e.lastActive) >= idleWindow {
			// Don't sleep into an imminent tick. A wake/sleep pair is a full
			// checkpoint-and-restore round trip; spending one to be absent
			// for less time than we would have idled anyway is a straight
			// loss. The threshold is THIS entry's idle window rather than a
			// constant of its own: it is already the stated opinion about how
			// long residency is worth holding — the deployment's for an
			// instance, and the app's own manifest value for a worker, which
			// is exactly the figure a worker's schedule should be weighed
			// against.
			next, ok := earliest[e.host]
			heldForTick := ok && time.Unix(next, 0).Sub(now) <= idleWindow
			if !heldForTick {
				log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second))
				a.beginStopLocked(e, StopIdle)
				continue
			}
			// Fall through to the convergence check below: a host held awake
			// for its schedule never reaches a "next wake" to converge on, so
			// it is now one of the always-awake hosts that arm was written
			// for — the same class as a sidecar host that never sleeps.
		}
		// A worker that never reports idle still has a ceiling: a wedged
		// reconnect loop must not hold a lease and a box forever. Zero means
		// no ceiling — a sidecar that declared none is a legitimate shape, so
		// this must never invent one nobody asked for.
		//
		// It applies to an ALWAYS-ON worker too — the wedged reconnect loop
		// this guard exists for is exactly the shape a long-lived worker gets
		// into — but it is spent differently there: the child is RECYCLED on
		// the same entry, keeping the lease and replication, because stopping
		// an always-on worker leaves nothing to start it again. That is the
		// same move build convergence makes for one (convergeWorkerLocked),
		// and for the same reason.
		if e.worker && e.workerMaxRun > 0 && now.Sub(e.spawnedAt) >= e.workerMaxRun {
			if !e.route.Hibernate {
				log.Printf("activator: always-on sidecar %s hit its max_run of %s — recycling its child", e.host, e.workerMaxRun)
				a.recycleWorkerLocked(e, e.proc, "hit its max_run")
				continue
			}
			log.Printf("activator: %s hit its max_run of %s — stopping", e.host, e.workerMaxRun)
			a.beginStopLocked(e, StopIdle)
			continue
		}
		// Converge onto a shipped build. A running child never re-reads
		// live/<host>, so once a ship repoints the symlink this awake child is
		// serving a stale binary. Idle hosts converge for free on their next
		// wake; this cycle is for the always-awake ones — notably every
		// sidecar host, which never sleeps — that otherwise serve the old
		// build indefinitely. Gracefully cycle: beginStopLocked drains via
		// SIGTERM, takes the final checkpoint, releases the lease, and leaves
		// the host asleep; a member request (or the due-time index) then wakes
		// it onto the new binary — the same clean cycle `systemctl restart`
		// gives a unit-managed host. Unlike idle hibernation this can fire
		// with a request in flight; the stop still drains HTTP gracefully and
		// reconnecting clients coalesce onto the fresh wake, so a long upload
		// past the drain is cut exactly as a restart would cut it. A pinned
		// canary never trips this: the updater leaves its symlink alone, so
		// spawnVersion still matches.
		cur, readErr := a.liveVer(e.host)
		cycle, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, ConvergeGrace)
		e.staleSince = next
		if cycle && a.OnDrift != nil {
			// CARLOS_SWAP=overlap: the scheduler owns every move onto a new
			// build, paced (spec §10). The sweep only tells it to look, on
			// its own goroutine and so off a.mu (the scheduler calls back
			// into Current), and carries on with this entry's sidecar
			// supervision below. No converge budget is spent.
			host := e.host
			go a.OnDrift(host)
			cycle = false
		}
		if cycle && a.convergeBudget > 0 {
			// An always-on worker converges as one MOVE, not as a cycle: the
			// cycle's second half is "something wakes it back up", and a
			// worker with no request path and no schedule has nothing that
			// would. Stop-then-respawn on the same entry is that second half,
			// performed here. Both existing guards still apply — this only
			// runs once the drift has persisted past ConvergeGrace, and it
			// consumes the same one-per-sweep budget, so a bad build cannot
			// respawn the fleet at once.
			//
			// A converge the live exec gate refuses (a declined box that
			// honours a record or marker holding cur, Task 9) stops nothing
			// and spends no budget: the worker carries on to supervision
			// below, running its current build.
			//
			// So is the cycle of an instance or a hibernating worker (fix
			// round 1, ruling 2): its next wake would be refused, so stopping
			// it now turns a held build into an outage. A refusal keeps it
			// running on its current build.
			if e.worker && !e.route.Hibernate {
				if a.convergeWorkerLocked(e, cur) {
					a.convergeBudget--
					continue
				}
			} else if !a.liveGateRefusesLocked(e, cur, "it keeps running "+e.spawnVersion) {
				log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur)
				a.beginStopLocked(e, StopRelease)
				a.convergeBudget--
				continue
			}
		}
		// Sidecar supervision — Restart=always, sweep-driven, no supervisor
		// goroutine: a sidecar that died (or failed to spawn at wake) while
		// its instance stays awake is respawned with doubling, capped backoff;
		// a run past sidecarHealthyRun resets it. Spawned under a.mu
		// deliberately: an exec is quick, and the lock is what makes a restart
		// impossible to race with a stop (sidecar fields only change in the
		// awake state, and every stop flips the state under this same lock
		// first).
		//
		// A sidecar env staged MID-FLIGHT (attached to an awake child — the
		// opt-in path) is ADOPTED here rather than waiting for the next wake,
		// so opting in takes effect in seconds. Safe against double-running:
		// only hibernating hosts are activator children and their systemd
		// sidecar units stay disabled — the staged sidecar has exactly one
		// possible runner.
		//
		// Never for a worker: it is a sidecar itself, not a host that has one
		// (the same refusal doWake makes at wake time — restated here because
		// this adoption arm can fire on a long-awake entry that never went
		// through step 6 again).
		if e.worker {
			continue
		}
		if a.gens && e.state == actAwake && e.sidecarWanted && !e.sidecarStarting &&
			e.sidecar != nil && !e.sidecar.Exited() && e.sidecarGen != e.gen.N {
			// A legacy sidecar still on a retired generation's build: a
			// replacement whose stop of it was unproven (spec §7.5).
			go a.replaceSidecar(e)
		}
		if !e.sidecarWanted && a.backend.HasSidecar(e.route) {
			e.sidecarWanted = true
			log.Printf("activator: sidecar staged for awake %s — adopting", e.host)
		}
		if e.sidecarWanted && (e.sidecar == nil || e.sidecar.Exited()) && !now.Before(e.sidecarRetryAt) {
			if e.sidecar != nil && now.Sub(e.sidecarStarted) >= sidecarHealthyRun {
				e.sidecarBackoff = 0
			}
			if e.sidecarBackoff == 0 {
				e.sidecarBackoff = sidecarRestartBase
			} else {
				e.sidecarBackoff = min(e.sidecarBackoff*2, sidecarRestartMax)
			}
			e.sidecarRetryAt = now.Add(e.sidecarBackoff)
			// The restart call itself runs off a.mu (design §6.7, codex
			// R2-6): held under the lock, a stalled StartSidecar (a hung
			// exec, a wedged control socket) would block every other host's
			// wake and sweep behind it. sidecarStarting keeps a second
			// attempt from launching while one is already in flight.
			//
			// The far side re-checks by ENTRY IDENTITY (e0, the *actEntry
			// itself), not a generation counter: a per-entry counter that
			// starts at 0 and is bumped once when an entry becomes awake
			// reads 1 for every entry's first wake, so it can never tell
			// "this entry" apart from "a completely different entry, for
			// the same host, after a stop and rewake, that also just had
			// its first wake". A pointer
			// comparison against a.entries[host] has no such collision: a
			// fresh wake always allocates a brand new *actEntry (begin() and
			// the due-wake path both do `&actEntry{...}`), so identity alone
			// says whether this attempt's wake is still the current one.
			if !e.sidecarStarting {
				e.sidecarStarting = true
				done := make(chan struct{})
				e.sidecarStartDone = done
				e0 := e
				host := e.host
				route := e.route
				gen := e.gen
				prior := e.sidecar // exited, or nil
				backoff := e.sidecarBackoff
				go func() {
					defer close(done)
					if a.gens && prior != nil && !a.sidecarTerminated(route, prior) {
						// The sidecar being restarted exited, but something it
						// forked is left in its cgroup (spec §7.4, §7.5). It is
						// not forgotten: it becomes a tracked stray, and the
						// start below refuses beside it.
						a.mu.Lock()
						a.trackStraySidecarLocked(route, prior)
						a.mu.Unlock()
					}
					ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
					sc, err := a.startSidecar(ctx, e0, route, gen, false)
					cancel()
					var discard, stopNow Proc
					defer func() { // after the unlock, before done closes
						a.discardSidecar(route, discard)
						a.stopReturnedSidecar(e0, stopNow)
						// The join is given up only now, with the cleanup
						// finished (fix wave 1, item 1): a stop that reads
						// it while the sidecar this start brought back is
						// still being stopped must wait for that stop.
						a.mu.Lock()
						if e0.sidecarStartDone == done {
							e0.sidecarStarting, e0.sidecarStartDone = false, nil
						}
						a.mu.Unlock()
					}()
					a.mu.Lock()
					defer a.mu.Unlock()
					if a.entries[host] != e0 {
						// This host's map slot no longer holds e0 at all:
						// forgotten (stopped and deleted) or replaced by an
						// entirely different entry from a fresh wake. e0 is
						// unreachable from here on, so there's nothing left
						// on it to update; just stop whatever came back so
						// it doesn't outlive the wake it doesn't belong to,
						// or clobber a fresh wake's own sidecar.
						discard = sc
						return
					}
					// Still e0: this attempt owns its sidecarStarting flag
					// and join, and the deferred cleanup above clears them
					// once it has stopped whatever must be stopped,
					// regardless of what state e0 is in now: a stop may
					// have begun (flipping e0 to actStopping) while the
					// call was in flight. Cleared here, before the cleanup
					// ran, a stop reading the join in between would miss
					// the sidecar still being stopped (fix wave 1, item 1).
					if e0.state != actAwake {
						// Stopped mid-call, not yet forgotten: don't install
						// onto a wake that's on its way out.
						if a.gens {
							// stop() is joined on this goroutine (done). It
							// stops e.sidecar after that join, synchronously,
							// and records whether it exited: install sc so it
							// is that stop's to prove.
							e0.sidecar, e0.sidecarGen = sc, gen.N
							return
						}
						// Flag off, the stop either waited this start out or
						// is pending on it (e0.sidecarJoin, Task 8): sc is
						// stopped here, synchronously, before done closes,
						// so the stop's host proof runs after it.
						stopNow = sc
						return
					}
					if a.gens && e0.gen.N != gen.N {
						// A swap switched generations while this start was in
						// flight, so sc runs the retired generation's build
						// (spec §7.5). It is NOT discarded asynchronously
						// (stopDiscardedSidecar): the swap's replaceSidecar,
						// joined on this goroutine, would then start the new
						// sidecar while this one still ran, and a terminal
						// stop could pass the join over it (§7.4). It is
						// installed, so it stays tracked: replaceSidecar stops
						// it synchronously and starts the current generation's
						// only once it is proven gone, and a terminal stop
						// stops it like any sidecar.
						e0.sidecar, e0.sidecarStarted, e0.sidecarGen = sc, a.now(), gen.N
						if err != nil {
							e0.sidecarUncertain = true
						}
						return
					}
					if err != nil {
						e0.sidecarUncertain = true
						if sc == nil {
							log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", host, err, backoff)
						} else {
							log.Printf("activator: SIDECAR restart for %s failed: %v — retaining returned process", host, err)
						}
						e0.sidecar, e0.sidecarGen = sc, gen.N
						return
					}
					log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", host, backoff)
					e0.sidecar, e0.sidecarGen = sc, gen.N
					e0.sidecarStarted = a.now()
				}()
			}
		}
	}
	// Due-time wakes: a request-less wake through the same singleflight state
	// machine, so a client request arriving mid-wake just coalesces. Only hosts
	// with NO live entry are woken — an awake host is already serving its
	// schedule; a waking one is on its way; a stopping one will re-record its
	// next_due when the stop completes and be picked up on a later tick. The
	// registry's stale past value while the host is up is harmless for the same
	// reason. A failed due wake (e.g. a lease still held by a crashed
	// incarnation) removes the entry, and the index — only rewritten at the
	// next successful sleep — naturally retries it every tick until it works.
	for _, rt := range dueRoutes {
		// Only hibernating, ready hosts: the index exists to wake sleeping
		// instances, and a systemd-managed host is always on — spawning a
		// child for it would fight its unit over the socket. This also makes
		// a leftover next_due inert on a host flipped back to systemd (the
		// flip-off's release re-records one after the flag is already clear).
		// The registry's DueInstances is deliberately broader (no hibernate
		// filter), so the gate lives here, with the machine it protects.
		if !rt.Hibernate || rt.Status != "ready" {
			continue
		}
		if a.fences[rt.Host] != 0 {
			continue
		}
		if _, live := a.entries[rt.Host]; live {
			continue
		}
		// mesh seam: in the multi-box mesh only the HRW owner due-wakes a host —
		// otherwise every box sweeping a due entry would race a wake per tick
		// (the lease would let exactly one win, but loudly, forever). Eleven:
		// `if a.mesh != nil && !a.mesh.owns(rt.Host) { continue }`.
		if a.allowFilesDiagnosticLocked(a.seedDiagnostics[rt.Host], "due") {
			log.Printf("activator: %s has a scheduled delivery due — waking it", rt.Host)
		}
		// A sidecar row comes up as a WORKER (registry.DueInstances returns
		// both kinds, because the due-time index is a sidecar's only timer
		// wake source — nothing can dial one). Woken through the instance path
		// it would spawn via Start and spend the whole health-probe timeout
		// dialling a socket it never binds.
		var e *actEntry
		if rt.IsSidecar() {
			e = a.newWorkerEntry(rt)
			// The value that fired, carried into the wake so a SUCCESSFUL one
			// can consume exactly it (consumeDueTime) — a worker's stop writes
			// no next_due at all, so without this the row keeps its passed
			// value and this same sweep re-wakes it every tick.
			e.dueFired = rt.NextDue
		} else {
			e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
		}
		a.entries[rt.Host] = e
		go a.wake(e)
	}

	// The always-on arm: design §7.2 says the record exists ⇒ the process
	// runs, and until this existed that held only across agent RESTARTS. The
	// boot pass (host's runWorkerBootPass) runs once, at arming; the due loop
	// above gates on Hibernate; so an always-on sidecar created while the
	// agent was running had nothing that would ever start it — no request can
	// reach one, and it has no schedule. It also covers every other way an
	// always-on worker ends up with no live entry: a wake that failed, or a
	// stop with no re-wake behind it.
	//
	// Bounded per sweep for the reason host's workerBootConcurrency is
	// bounded: each wake claims an S3 lease and restores a database, so a box
	// just handed a fleet of sidecars starts them in batches. The remainder
	// comes up on the next sweep.
	started := 0
	for _, rt := range alwaysOn {
		if a.shuttingDown {
			break // the agent is going away; nothing new may claim a lease
		}
		if started >= maxWorkerStartsPerSweep {
			break
		}
		if _, live := a.entries[rt.Host]; live {
			continue // awake, waking, or draining a stop — none of those is stranded
		}
		log.Printf("activator: always-on sidecar %s is not running — starting it", rt.Host)
		e := a.newWorkerEntry(rt)
		a.entries[rt.Host] = e
		go a.wake(e)
		started++
	}
	a.mu.Unlock()
	// A generations file write that failed is retried here, off a.mu.
	a.retryGenerations()
}

// RetireExcept stops (final checkpoint + lease release) every live entry whose
// host is not in keep — the reload-time reconcile behind flipping a host's
// hibernate flag off (and, as a bonus, route removal while awake). The flip
// clears the route's flag and reloads the edge; the map swap stops routing
// requests through the activator, and this puts the child down and gives the
// S3 lease back so the host's systemd unit can bind the socket. It reuses the
// ordinary stop path — the narrowest hook that exists — so the release still
// records next_due and checkpoints to S3 like any sleep. In-flight requests on
// a retired child are cut off as a deploy restart would cut them — the flip is
// a deliberate operator action.
func (a *Service) RetireExcept(keep map[string]bool) {
	// A probe's entry (ProbeWake) is held for its move while the registry
	// row is: the edge manages ready rows only, so it is never in keep.
	seen, held := a.heldProbes(keep)
	if retireBetweenLocks != nil {
		retireBetweenLocks()
	}
	a.mu.Lock()
	var waking []*actEntry
	for _, e := range a.entries {
		if e.probe != nil {
			snap, covered := seen[e]
			switch {
			case covered && e.probe.seq != snap.seq:
				// The probe changed since this pass's snapshot (an open, or a
				// newer pass took it over): the registry read is older than
				// the change and cannot judge it. Kept for this reload.
				continue
			case !e.probe.managed && keep[e.host] && e.probe.phase >= probeOpening:
				// The edge manages the opened probe's ready row: from here it
				// is an ordinary serving entry (Plan 3 Task 7). The metadata
				// stays (ruling 2).
				e.probe.managed = true
				e.probe.seq++
				continue
			case !e.probe.managed && (!covered || held[e]):
				continue
			}
		}
		if keep[e.host] {
			continue
		}
		// Workers are outside this reconcile's world entirely. `keep` is the
		// EDGE's managed set, built by edge.Reload — which refuses a sidecar
		// row on kind as the first act of its loop, so a worker's host can
		// never appear in it. Without this guard every edge reload (any route
		// change at all) would stop every running standalone sidecar. A
		// sidecar's removal authority belongs to the sidecar reconcile pass,
		// not to the edge's route map.
		// A held recovery belongs to its claim operation, not the edge's
		// serving set. Explicit fences and claim/session review still stop it.
		if e.worker || e.probeHeld {
			continue
		}
		switch e.state {
		case actAwake:
			log.Printf("activator: %s is no longer hibernate-managed — releasing", e.host)
			a.beginStopLocked(e, StopRetired)
		case actWaking:
			// A wake can't be stopped mid-flight; finish the retire once it
			// resolves (below, off-lock).
			waking = append(waking, e)
		case actStopping:
			// Already on its way out.
		}
	}
	a.mu.Unlock()
	for _, e := range waking {
		go func(e *actEntry) {
			<-e.ready
			a.mu.Lock()
			defer a.mu.Unlock()
			// Only if THIS entry is still the live one and its wake succeeded —
			// a failed wake already removed itself, and a fresh entry for the
			// same host means the route was re-flagged in the meantime.
			if a.entries[e.host] == e && e.state == actAwake {
				log.Printf("activator: %s is no longer hibernate-managed — releasing (wake finished first)", e.host)
				a.beginStopLocked(e, StopRetired)
			}
		}(e)
	}
}

// Quiesced answers whether a host currently has NO live activator entry —
// no child running, none waking, none mid-stop. The residence-move cut
// (internal/host's move worker) gates its final mirror round on this: the
// route's Status change retires the entry through RetireExcept, and only a
// quiet host's replica prefix is byte-final.
func (a *Service) Quiesced(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	_, live := a.entries[host]
	return !live
}

// Shutdown checkpoints and stops every child — the agent going down takes its
// instances with it (they're its children; leaving them orphaned would strand
// the leases until TTL expiry and leave the sockets bound against the next
// incarnation's spawns). Blocks until all stops complete or ctx expires.
func (a *Service) Shutdown(ctx context.Context) {
	// mesh seam — drain step 1, order-pinned: Eleven left the mesh membership
	// FIRST — the heartbeat is deleted before any checkpoint or lease release
	// below, so peers stop proxying here immediately and their wakes succeed
	// the moment each lease releases. Deploys, box failure drills, and
	// scale-down all ride this same path. (`a.mesh.stop(ctx)` went here.)
	// Announce the shutdown BEFORE anything stops: a supervised respawn in
	// flight checks this flag immediately before it spawns, so it abandons
	// rather than starting a child this shutdown would then have to chase.
	// (stop() waits out any respawn that got past that check, so the waits
	// below cover the rest.)
	a.mu.Lock()
	a.shuttingDown = true
	// Proofs that start or run from here end with this budget too
	// (legacyRunGone, #292): none may outlive Shutdown.
	a.shutdownCtx = ctx
	sweepStop, sweepDone := a.sweepStop, a.sweepDone
	a.mu.Unlock()
	// …and those already running end with it too (Astra, round 3).
	context.AfterFunc(ctx, a.cancelProofs)
	// …and cancel the service lifetime, which TEARS DOWN any worker spawn
	// already past that check rather than letting the stop below wait it out.
	// The wait is right for a Bounce or an idle sleep — no deadline, and
	// letting the exec finish is tidier — but a shutdown has a deadline it
	// does not control: systemd's TimeoutStopSec is 90s by default, and
	// workerSpawnTimeout alone is 2 minutes. An agent that blocks on one
	// host's bucket materialize gets SIGKILLed mid-shutdown, which strands
	// leases and sockets for EVERY host on the box — far worse than
	// abandoning one respawn. The cancelled respawn aborts and closes its
	// respawnDone, so stop()'s wait falls through immediately.
	if a.lifeStop != nil {
		a.lifeStop()
	}
	if sweepStop != nil {
		close(sweepStop)
		<-sweepDone
	}
	a.mu.Lock()
	// The sweeper is stopped as of the line above, so nothing else will ever
	// poll a pendingStop entry (design §6.3) again on its own. Run the same
	// pass once here, and again on the ticker below while waiting, so a
	// process that finally exits during shutdown still gets its final
	// checkpoint instead of being abandoned mid-drain.
	a.finishPendingStops()
	var waits []chan struct{}
	var waking []*actEntry
	for _, e := range a.entries {
		switch e.state {
		case actAwake:
			a.beginStopLocked(e, StopShutdown)
			waits = append(waits, e.stopped)
		case actStopping:
			waits = append(waits, e.stopped)
		case actWaking:
			// Shutdown covers waking entries too (zero-downtime spec §7.4),
			// whatever CARLOS_SWAP says. The wake is cancelled, and once it
			// resolves, wake() itself stops the entry (through the barrier
			// with swaps on), or the failed wake has given its lease back
			// under a budget of its own (releaseLease).
			//
			// Swaps off used to leave a waking entry to resolve on its own,
			// with the lease's TTL to bound the damage. That is #292: an
			// edge restart stops every carlos-instance@ unit before the
			// agent's SIGTERM, the agent recycles one and a request wakes it
			// again, and that wake's start is a systemctl start job queued
			// behind the edge restart. The agent exited with the wake still
			// holding its lease, and the next agent refused the host for the
			// lease's whole 15 minutes.
			if e.wakeCancel != nil {
				e.wakeCancel()
			}
			waking = append(waking, e) // swapsim:M6
		}
	}
	a.mu.Unlock()
	// One ticker for every wait below, the wake joins included (#292,
	// Astra P2): a wake that ignores its cancellation can hold a join until
	// the deadline, and a stop left pending meanwhile must still be
	// finished, and its lease given back, as soon as its process is proven
	// gone, not once every wake has joined.
	poll := time.NewTicker(100 * time.Millisecond)
	defer poll.Stop()
	pollPending := func() {
		a.mu.Lock()
		a.finishPendingStops()
		a.mu.Unlock()
	}
	for _, e := range waking {
		for joined := false; !joined; {
			select {
			case <-e.ready:
				joined = true
			case <-poll.C:
				pollPending()
			case <-ctx.Done():
				a.logUnresolvedAtExit()
				return
			}
		}
		a.mu.Lock()
		if a.entries[e.host] == e && e.state == actStopping {
			waits = append(waits, e.stopped)
		}
		a.mu.Unlock()
	}
	for _, w := range waits {
		for done := false; !done; {
			select {
			case <-w:
				done = true
			case <-poll.C:
				pollPending()
			case <-ctx.Done():
				a.logUnresolvedAtExit()
				return
			}
		}
	}
	// Retained failed wakes (zero-downtime spec §7.4, P2-13) have no entry
	// for the loop above to wait on, so they get the same polling here: the
	// reaper (reapUnprovenWakesLocked, inside finishPendingStops) retries
	// each one's stop and proof and gives its lease back once it holds. A
	// wake still unproven when the budget runs out keeps its lease; the
	// lease's TTL frees it.
	//
	// With generations off the map fills too, so flag-off Shutdown waits
	// here as well, bounded by ctx: from GenContainment's sentinels
	// (genStartUnproven), and, on any box, from a failed wake whose leader
	// survived its stop (cleanupFailedWake's leaderLives, the fail-closed
	// exception to dormancy from the task 2 fix). That flag-off retention
	// proves only the leader's exit, not its descendants (final review F6,
	// deferred). With neither, the map is empty and this returns at once.
	//
	// Lease releases that failed (#292, Astra P3) are retried here the same
	// way, inside the same budget (retryOwedReleasesLocked, also inside
	// finishPendingStops), and named at exit if they are still owed.
	for {
		a.mu.Lock()
		left := len(a.unprovenWakes) + len(a.owedReleases)
		if left > 0 {
			a.finishPendingStops()
		}
		a.mu.Unlock()
		if left == 0 {
			return
		}
		select {
		case <-poll.C:
		case <-ctx.Done():
			a.logUnresolvedAtExit()
			return
		}
	}
}

// logUnresolvedAtExit names what Shutdown's budget ran out on: failed
// wakes not proven stopped, whose leases are kept, and lease releases that
// failed and were not resolved by a retry. The TTL frees either.
func (a *Service) logUnresolvedAtExit() {
	a.mu.Lock()
	var held, owed []string
	for host, u := range a.unprovenWakes {
		held = append(held, fmt.Sprintf("%s (generation %d)", host, u.gen.N))
	}
	for o := range a.owedReleases {
		owed = append(owed, o.host)
	}
	a.mu.Unlock()
	if len(held) > 0 {
		slices.Sort(held)
		log.Printf("activator: shutdown: out of time with %d failed wake(s) not proven stopped, their leases still held: %s", len(held), strings.Join(held, ", "))
	}
	if len(owed) > 0 {
		slices.Sort(owed)
		log.Printf("activator: shutdown: out of time with %d lease release(s) unresolved, held until their TTL: %s", len(owed), strings.Join(owed, ", "))
	}
}

// ServeHTTP answers one proxied request for a hibernating instance: wake (or
// join a wake), proxy, stamp activity. Lease contention and wake timeouts are
// 503s — "sleeping hosts are slow to wake", never a 5xx cascade of retries.
// It reports whether it handled the request (wrote the response or proxied
// it); false means "not activator-managed — proxy straight", so the edge can
// call it unconditionally.
func (a *Service) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
	return a.serve(w, r, rt, proxy, true)
}

// ServeScheduled is ServeHTTP for a platform-delivered scheduled tick: same
// wake, same in-flight hold, same error mapping — no activity credit.
//
// A dedicated method rather than a wire marker like SidecarHeader, and the
// difference is the delivery path, not taste. The sidecar dials its
// instance's public URL, so its traffic arrives through the edge and the
// request is the only place a signal could ride. A tick never leaves the
// process (schedule_wiring.go: "the tick never leaves the box"), so the call
// site IS the signal, and there is no header for an outsider to send.
//
// A tick never wakes a managed route this box has stopped serving (a status
// other than ready: the residence-move cut's "moving", instance-moves design
// G12). The due-wake sweep holds the same gate. The edge never hands the
// activator such a row, but the runner fires whatever row the registry
// returns, and a tick that woke a host mid-cut would restore it and let it
// write behind the move's final mirror round. The move worker's fence closes
// the race in-process; this gate is the half that survives an agent restart.
func (a *Service) ServeScheduled(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
	if rt.Kind == "instance" && rt.Hibernate && rt.Status != "" && rt.Status != "ready" {
		maintenanceResponse(w, r)
		return true
	}
	return a.serve(w, r, rt, proxy, false)
}

func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler, stamp bool) bool {
	if rt.Kind == "instance" && rt.Hibernate && a.Startup != nil {
		deadline := time.NewTimer(wakeWaitBound)
		err := a.awaitStartup(r.Context(), deadline.C)
		deadline.Stop()
		if err != nil {
			return a.serveWakeError(w, r, rt, err)
		}
		// The edge may have chosen rt and its proxy before startup recovery
		// ran, whether this request waited here or arrived just as Startup
		// closed. If recovery moved this host's upstream (a layout journal
		// finished), that proxy points at the old socket: answer as a wake
		// still in progress, and the retry is routed afresh. Only requests
		// near startup are checked; after the window the edge's own
		// selection is current.
		if a.nearStartup() && a.Routes != nil {
			if fresh, ferr := a.Routes.Get(rt.Host); ferr == nil && fresh != nil &&
				(fresh.Socket != rt.Socket || fresh.Addr != rt.Addr) {
				return a.serveWakeError(w, r, rt, errWakeTimeout)
			}
		}
	}

	a.mu.Lock()
	fenced := a.fences[rt.Host] != 0
	a.mu.Unlock()
	if fenced {
		maintenanceResponse(w, r)
		return true
	}
	// Per-host gating: only a hibernate-flagged instance is activator-managed;
	// unflagged instances and sibling services are the edge's to proxy
	// straight — the deploy-safety property that lets the bucket env ship
	// fleet-wide as a no-op before any host is flagged.
	if rt.Kind != "instance" || !rt.Hibernate {
		return false
	}

	// Sidecar-marked traffic (SidecarHeader — the sidecar stamps every REST
	// call and its WebSocket dial) is activity-invisible: proxied while the
	// instance is awake, refused while it isn't, never a wake. The co-spawned
	// sidecar dials its instance's public URL through this very edge (the box
	// pins its hosted names to loopback), so without this its keepalives
	// would read as member activity — the instance could never idle — and
	// after a sleep its reconnect loop would be the wake source: ping-pong
	// forever. Spoofing the marker is harmless by construction: it only
	// DOWNGRADES the sender's own requests (no activity credit; 503 while
	// asleep). It cannot suppress other clients' activity stamps and cannot
	// wake a sleeping host, so an outsider sending it gains nothing.
	if r.Header.Get(SidecarHeader) != "" {
		// The marker is an edge-side signal only (activity exclusion, above);
		// strip it so nothing downstream can ever grow a meaning an outsider
		// could spoof.
		r.Header.Del(SidecarHeader)
		if err := a.BeginSidecarRequest(r.Context(), rt); err != nil {
			if errors.Is(err, ErrFenced) {
				maintenanceResponse(w, r)
				return true
			}
			if errors.Is(err, context.Canceled) {
				return true // the sidecar hung up while queued on a wake
			}
			w.Header().Set("Retry-After", "5")
			// Ours, not the app's: the edge must not file this 503 as an
			// edge error (edge/answered.go; spec 2026-09-07 §3.2).
			edge.MarkAnswered(r)
			http.Error(w, "instance is asleep; sidecar requests do not wake it", http.StatusServiceUnavailable)
			return true
		}
		proxy.ServeHTTP(w, r)
		return true
	}

	// mesh seam: Eleven proxied a member request for a host with no live
	// local entry and an HRW owner on another live box to that peer here (one
	// hop max, guarded by a forwarded marker); a host with ANY live local
	// entry stayed local — placement converges at the next sleep boundary,
	// since its state is already in S3. Single-box v1 always wakes locally.

	release, err := a.begin(r.Context(), rt, stamp, r)
	if err != nil {
		return a.serveWakeError(w, r, rt, err)
	}
	defer release()
	proxy.ServeHTTP(w, r)
	return true
}

func (a *Service) serveWakeError(w http.ResponseWriter, r *http.Request, rt registry.Route, err error) bool {
	var held *errLeaseHeld
	var refused *errConfigRefused
	switch {
	case errors.Is(err, ErrFenced):
		maintenanceResponse(w, r)
	case errors.As(err, &refused):
		// Quiet by design: the config guard logs once per host and
		// publishes the refusal in the box's status snapshot. A line per
		// request here would bury that one.
		w.Header().Set("Retry-After", strconv.Itoa(configRefusedRetryAfter))
		edge.MarkAnswered(r)
		http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
	case errors.As(err, &held):
		retry := int(held.retryAfter/time.Second) + 1
		w.Header().Set("Retry-After", strconv.Itoa(retry))
		edge.MarkAnswered(r)
		http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
	case errors.Is(err, errWakeTimeout):
		w.Header().Set("Retry-After", "5")
		edge.MarkAnswered(r)
		http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
	case errors.Is(err, context.Canceled):
		// The client hung up while queued; nothing to write.
	default:
		// The wake failure is already the member's platform log error
		// (this log line, shipped by logsink.go); marking it keeps the
		// edge from filing the same failure a second time.
		if a.allowFilesDiagnostic(rt.Host, err, "http") {
			log.Printf("activator: wake %s: %v", rt.Host, err)
		}
		edge.MarkAnswered(r)
		http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
	}
	return true
}

// SetIdleFor installs the plans idle hook (plans 2). A plan can only shorten
// an entry's idle window: the sweep takes the smaller of the entry's own
// window and the hook's answer, and ignores 0. Always-on workers have no idle
// clock, so the hook never reaches them. nil removes the hook.
func (a *Service) SetIdleFor(f func(registry.Route) time.Duration) {
	if f == nil {
		a.idleFor.Store(nil)
		return
	}
	a.idleFor.Store(&f)
}