// Ported from lchat activator.go (the s3Backend half) and the exec/env/health
// helpers it leaned on from lchat control.go and versions.go.
package activator

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"log"
	"maps"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/carlosframework/platform/internal/applog"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/instancekey"
	"github.com/carlosframework/platform/internal/instanceupdates"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/safefile"
	"github.com/carlosframework/platform/internal/schedule"
	"github.com/carlosframework/platform/internal/seeds"
	"github.com/carlosframework/platform/internal/unixdial"
)

// spawnHealthTimeout bounds how long a spawned instance may take to bind its
// socket and answer. A var so tests can shrink the failure path.
var spawnHealthTimeout = 30 * time.Second

// ErrSeedUnproven is seedUnit's answer when a seed's cgroup never reports
// empty (design §6.2): only a SIGKILL survivor reaches this, and while its
// cgroup stays populated it may still be writing the database, so the
// caller must treat this the way stop() treats ErrStillRunning and leave
// the lease for its own TTL to expire rather than release it over a write
// that might still be happening.
var ErrSeedUnproven = errors.New("activator: seed unproven; processes remain")

// seedUnprovenSuffix names a flat seed's unproven-cleanup obligation, kept
// beside the database (<db>.seed-unproven): a seed failed, its processes
// were not proven gone, and so its database was kept. While the file is
// there no wake treats the database as warm (Restore). It does not end in
// ".db", so the box-wide litestream glob never picks it up.
const seedUnprovenSuffix = ".seed-unproven"

// writeSeedObligationFile is the obligation's durable write: a var so a
// test can make it fail (a full disk).
var writeSeedObligationFile = safefile.WriteFile

// quarantineSuffix names where a restore that failed its integrity check is
// parked. It deliberately does NOT end in ".db": the box-wide litestream
// generator globs <data>/*.db, and a quarantined corpse must never be picked
// up for replication.
const quarantineSuffix = ".corrupt"

// ExecBackend is the production Backend: leases + restore via the lease
// package's litestream primitives, processes via exec of the same versioned
// binary a systemd unit would run — `<OptDir>/live/<host>` with the
// `<DataDir>/<host>.env` EnvironmentFile.
type ExecBackend struct {
	Cfg Config
	// OptDir is the on-box install root: the version store and the
	// per-instance "live" symlinks live under it (/opt/carlos in production;
	// a struct field, never hardcoded, so tests point it at a temp dir).
	OptDir string
	// DataDir holds each instance's DB and env files (<host>.db, <host>.env,
	// <host>-sidecar.env) — /data in production.
	DataDir string
	// Materialize, when non-nil, stages a binary and points the host's live
	// symlink at it via an external mechanism instead of the backend doing
	// filesystem work itself — Eleven needed this because its router ran
	// sandboxed (ProtectSystem=strict) and couldn't write to /opt itself.
	// With it nil, a missing live symlink is an error: Eleven fell back to
	// staging its OWN binary (instance and control binary were one program),
	// but a carlos app is a separate binary from the release store — there is
	// nothing sensible to invent here, and the adopt pass owns the symlinks.
	Materialize func(ctx context.Context, host string) error
	// Mirror, when non-nil, arms Restore's staging fallback (issue #331):
	// a host with no replica under the ordinary instances/ prefix may have
	// one under moves/<host>/ — a residence move's staged copy, pushed by
	// the SOURCE box. Restore then copies it into place (own bucket, own
	// IAM), restores as normal, and deletes the staging copy once CheckDB
	// vouches for what it restored. Nil = the fallback does not exist,
	// byte-identical to the pre-move behaviour.
	Mirror lease.MirrorStore
	// Phase1, set by the agent from moves.Enabled, retires the staging
	// fallback above: Restore never promotes a moves/<host>/ tree, because
	// under Phase 1 the staged copy is the target's to read through the
	// landing, and a restore that promoted it would be a second, older
	// source of truth (moves design §5.3, M15). Off, the legacy promotion
	// is exactly as it was until Plan 6 Task 8 removes it.
	Phase1 bool
	// Region is the box's CARLOS placement code (CARLOS_REGION), handed to
	// every exec child as its own CARLOS_REGION (issue #329, titogo's
	// residence design: the app reads where it lives — e.g. to clear a move
	// freeze after landing). Agent-owned in BOTH directions, exactly like
	// the control-channel vars below: set from what the agent knows, and
	// STRIPPED when the agent knows nothing — a tenant's config env must not
	// be able to tell an app it runs somewhere it does not.
	Region string
	// ReplicaHomeKeys answers, for a replica row's host, the SHA-256 of its
	// home's effective CARLOS_INSTANCE_KEY (instance-key plan rulings 26 and
	// 30). O8 will supply it from the home's published env item; until then
	// the agent reads the box's list (/etc/carlos/replica-home-keys), which
	// only the pre-roll check writes, again at every call. Nil, or false for
	// a host, means the home key cannot be verified, and a replica launch
	// that would be handed a key is refused (replicaKeyGate).
	ReplicaHomeKeys func(host string) (sha256 string, ok bool)
	// ConfiguredKey reads rt's configuration from the bucket now and
	// returns the SHA-256 of the CARLOS_INSTANCE_KEY it renders, "" for
	// none (appconfig.ConfiguredKeyDigest). Two readers ask it. The mint
	// asks before it mints, so a key the configuration sets is never
	// shadowed by one minted while that configuration was still on its
	// way to the box (ensureInstanceKey; ruling 5): a key the bucket sets
	// and the box has not delivered defers the launch, and nil is a box
	// with no configuration source, where nothing is on its way. The
	// replica gate asks at every launch whose key is configured, and
	// refuses a replica whose configured key is not that digest: a
	// configuration change is in flight (ruling 27); there nil refuses
	// every configured key.
	ConfiguredKey func(ctx context.Context, rt registry.Route) (sha256 string, err error)
	// Cache, when non-nil, adds the purge verb to each spawned instance's
	// control socket so it can invalidate its own cached responses right
	// after a write. Nil disables that ONE verb, not the socket: the control
	// channel is bound for every exec child either way (§7.2), because the
	// scheduled-work verbs need it on boxes with no edge cache too.
	Cache *edge.Cache
	// Cgroups, when non-nil, places every spawned child — instance, sidecar,
	// seed — in a per-host cgroup under the agent's delegated subtree so the
	// metrics sampler can account it (exec-child accounting spec). Nil — no
	// delegation: old unit file, tests, darwin — spawns exactly as before.
	Cgroups *cgroup.Manager
	// Tier, when non-nil, picks the memory cap for a route's carlos-exec@
	// cgroup (memory-pressure spec §2.3). The box wiring reads the staged
	// manifest of the live version; tests leave it nil and get the default
	// tier. Instance, seed and sidecar share the cgroup, so they share the
	// tier — a seed's first-boot cost is that instance's work.
	Tier func(registry.Route) cgroup.Limit
	// GenContainment selects the generation layout (zero-downtime spec
	// §7.3, §7.6): each instance spawn gets its own child cgroup g<n> (or
	// the carlos-instance-gen@<id>:<n> unit for a nested host) under a
	// parent that holds no processes. False, the zero value, is today's
	// layout byte for byte: one cgroup per host and carlos-instance@. Set
	// only by plan 5's CARLOS_SWAP wiring.
	GenContainment bool
	// TierOf is one named version's tier, read from that version's staged
	// manifest (zero-downtime spec §7.3). A generation's limits are its own
	// artifact's, never whatever the live link names at the moment. Nil
	// falls back to Tier (tests, and every flag-off spawn).
	TierOf func(rt registry.Route, version string) cgroup.Limit
	// Logs, when non-nil, gets each instance and seed child's stdout/stderr
	// teed in line-wise for `carlos logs` (spec §5) — journald keeps every
	// byte it got before. Nil (tests, a box without the shipping loop)
	// spawns exactly as before: fd passthrough, no pipes.
	Logs EventSink
	// Timers, when non-nil, mounts the scheduled-work verbs on each spawned
	// instance's control socket so an app can set and clear its own one-shot
	// timers (spec 2026-08-23-scheduled-work-design.md §7.2). The box wiring
	// passes the registry. Nil — tests, and any box whose agent predates the
	// feature — binds the control socket without those verbs, so an app that
	// asks gets 404 rather than a socket that isn't there.
	Timers schedule.TimerStore
	// UpdatesReader answers the control socket's read-only GET /updates
	// (instance update settings spec §5.1); nil mounts nothing.
	UpdatesReader func(host string) (instanceupdates.Report, error)
	// LayoutFacts supplies the box facts the classification of §8.1 needs
	// beyond the row: journals and peers. Zero value: never fenced, never
	// conflict. Named apart from the Layout method (the Classifier
	// interface's Layout(rt) (Layout, string)) so a struct field and a
	// method can coexist on the same type.
	LayoutFacts LayoutInputs
	// Control is the root instance-control channel a nested wake talks to
	// (design §5.3). Nil refuses every nested wake with "instance control
	// channel not configured" rather than dialing nothing. A flat or
	// foreign host never needs it, and a nil Control does not touch them.
	Control Control
	// UnitState answers `systemctl show` for the observer (design §6.5).
	// Nil: systemctlUnitState, the real systemctl binary.
	UnitState UnitState
	// CgroupEvents reads a unit's cgroup.events for the observer and for a
	// seed's cgroup-empty wait. Nil: cgroupEventsUnder("/sys/fs/cgroup"),
	// or CgroupRoot when set.
	CgroupEvents CgroupEvents
	// CgroupRoot overrides where cgroup v2 is mounted, for the default
	// CgroupEvents reader. Nil (empty): /sys/fs/cgroup.
	CgroupRoot string
	// NestedOK gates a nested wake on the box facts a unit test can't see:
	// the isolation mounts, the control service's own /ready. Nil: no
	// gate (every unit test and any box that predates the check). Wired in
	// production by Task 6.
	NestedOK func() error
	// Seeds reads a seed's snapshot from this region's data bucket
	// (seeded-instances spec §3.4). Nil: no seed can be placed here, and a
	// seeded row's wake fails rather than starting empty.
	Seeds SeedObjects
	// ClaimArmed is CARLOS_HOST_CLAIM on this box. The seed fallback is the
	// plain path only; with claims armed it refuses (spec §3.4, #250).
	ClaimArmed bool
	// OnSeedPlaced records a seed placed (Task 4): the record's CAS to
	// placed, then the snapshot's delete. Nil: not recorded here.
	OnSeedPlaced func(ctx context.Context, account, app, host, id string) error
	// ConfigGate answers whether this box can provide rt's config (amadan
	// #213): nil means yes, an error means refuse. Acquire asks it BEFORE
	// lease.Acquire, so a box that would start the instance without its
	// secrets never holds the lease, and a keyed sibling takes it instead.
	// The gate owns its own policy, logging and operator-visible record of
	// the refusal; this backend only honours the answer. Nil: no gate (every
	// unit test, and a box with no deploy store to read config from).
	ConfigGate func(ctx context.Context, rt registry.Route) error
	// LiveExecGate answers whether rt may start build, the build a start
	// from the live link is about to run: nil means yes, an error refuses
	// the start and comes back as it is. Only a box that declined
	// CARLOS_SWAP=overlap and honours failure records or hold-ran markers
	// from an earlier armed run sets it (internal/host, through
	// Service.SetLiveExecGate, before Start), so the plan-time refusal is
	// repeated at the real launch with the build actually being started
	// (Astra R1 #6, the R2/R3 declined residuals). It is asked by a flat
	// instance's or sidecar's live-link spawn (which then execs the staged
	// path it resolved, not the link), a nested instance's or sidecar's
	// live-link start and a seed from the live link (each pins the link
	// first, declinedLivePin, and runs the file it pinned), and a worker's
	// spawn from the live link. Nil, on every other box: each of
	// those spawns is exactly as it was. It reads only memory, the row it
	// is handed and the live link, and writes nothing: the activator asks
	// it under its own lock too (convergeWorkerLocked).
	LiveExecGate func(rt registry.Route, build string) error
	// ObservePoll is how often the exit observer (unitproc.go) and a
	// seed's cgroup-empty wait re-check unit state. 0: one second. A field
	// so tests can shrink the wait without a package-global var shared
	// across every test in the suite.
	ObservePoll time.Duration
	// RunDirOverride is test-only: when set, layoutInputs() classifies
	// against it instead of provision.SocketDir and pathsFor resolves a
	// nested run dir and log socket beneath it, so a test can exercise the
	// nested layout without writing to /run/carlos.
	RunDirOverride string
	// ScheduleFence runs a schedule write made on the control socket of
	// host's generation gen (the number the listener was bound for, 0 for
	// a spawn that is not a generation's), unless the activator has
	// switched away from that generation, and reports whether it ran
	// (zero-downtime spec §7.10, final review F8). Wired to
	// Service.ScheduleWrite by the agent. Nil (tests, a box before this
	// wiring) refuses nothing.
	ScheduleFence func(host string, gen uint64, write func()) bool
	// layoutFn, when set, replaces Classify in StartCmd's layout check; a
	// unit test's way to stand a temp directory in for a nested host.
	layoutFn func(rt registry.Route) (Layout, string)
	// Strict, when non-nil and true for a host, makes every spawn of that
	// host strict (moves design §5.3): the child starts only inside a group
	// prepareStrict prepared, which is drained (KillAndEmpty) and closed
	// when the child exits, and a spawn that cannot be made strict is
	// refused with ErrStrictRequired. The host package sets it from the hold
	// view: a host whose record has a move in flight, with the switch on
	// (Task 14). Nil, or false for a host, is today's spawn exactly.
	Strict func(host string) bool
	// prepareStrict is the strict path's seam. Nil: Cgroups.PrepareStrict,
	// and a nil Cgroups under Strict is ErrStrictRequired. Tests replace
	// this seam, never the Cgroups field.
	prepareStrict func(host string, limit cgroup.Limit) (strictProc, error)
	// killGen is the spawn path's own generation kill (startChildAt's
	// failed health wait). Nil: KillGen. Tests replace this seam, never the
	// Cgroups field.
	killGen func(ctx context.Context, rt registry.Route, n uint64) error
	// cgroupKill is seedDescendantsGone's kill and proof: child "" is the
	// host cgroup (KillHost), any other name one of its children. Nil: the
	// cgroup manager's own. Tests replace this seam, never the Cgroups
	// field.
	cgroupKill func(ctx context.Context, rt registry.Route, child string) error
	// openHostCleanup is EmptyHost's open of the host cgroup. Nil:
	// Cgroups.OpenCleanup. Tests replace this seam, never the Cgroups
	// field.
	openHostCleanup func(host string) (*cgroup.Cleanup, error)
	// strictPreflight is NextSpawnStrict's seam for PrepareStrict's runtime
	// preflight. Nil: Cgroups.StrictPreflight, the very function
	// PrepareStrict runs, so S2 and the spawn cannot disagree.
	strictPreflight func() error
	// strictGates holds one hostGate per host that has spawned with Strict
	// set. strictMu guards only the map's lookup and insert, never a start
	// or any filesystem work, so a stalled start blocks only its own host.
	strictMu    sync.Mutex
	strictGates map[string]*hostGate
	// strictHook, when set, is called at named points of the strict
	// cleanup path. Test-only: it makes interleavings deterministic.
	strictHook func(point string)
	// launchHook, when set, is called by a staged spawn once its
	// preparation is done, immediately before its exec gate (runExecGate):
	// flat, inside the cgroup manager's before-start hook, so once per
	// start attempt (a PID-move retry asks again); nested, before root's
	// start call. Test-only: it lands a lease loss, a fence or a
	// supersession after the preparation and before the launch (final
	// review F1).
	launchHook func(host string)
}

// hostGate serialises everything that touches one host's group once Strict
// is wired (moves design §5.3): the ownership check and an ordinary start,
// a strict spawn taking ownership, the exit path's directory removal, a
// retry's removal, and S2's read. owner, and owner.pending, are read and
// written only with mu held. While owner is set, from a strict spawn until
// its cleanup succeeds (retries included), no other spawn of the host may
// use the group and nothing but the owner may remove it.
type hostGate struct {
	mu    sync.Mutex
	owner *strictOwner
}

// gate returns host's gate, creating it.
func (b *ExecBackend) gate(host string) *hostGate {
	b.strictMu.Lock()
	defer b.strictMu.Unlock()
	if b.strictGates == nil {
		b.strictGates = map[string]*hostGate{}
	}
	g := b.strictGates[host]
	if g == nil {
		g = &hostGate{}
		b.strictGates[host] = g
	}
	return g
}

// lookupGate returns host's gate, or nil if it never had one.
func (b *ExecBackend) lookupGate(host string) *hostGate {
	b.strictMu.Lock()
	defer b.strictMu.Unlock()
	return b.strictGates[host]
}

func (b *ExecBackend) hook(point string) {
	if b.strictHook != nil {
		b.strictHook(point)
	}
}

// ErrStrictGroupInUse is a spawn refused because a strict child of the same
// host is alive in the host's group: an ordinary child placed beside it
// would be killed by its exit drain.
var ErrStrictGroupInUse = errors.New("activator: a strict child of this host is still running in its group")

// ErrStrictCleanupPending is a spawn refused because the host's last strict
// group has not been cleaned up: its retained handle still owns the group,
// and a retry's drain would kill anything placed in it.
var ErrStrictCleanupPending = errors.New("activator: this host's last strict group is still being cleaned up")

// ErrStrictRequired is a spawn refused because its host is in a move and
// the child could not be started under strict containment (moves design
// §5.3): no delegated cgroup subtree, a populated or unverifiable group, or
// a strict start that failed. Nothing was spawned.
var ErrStrictRequired = errors.New("activator: a host in a move spawns only under strict containment")

// strictProc is what a strict spawn holds for the life of its child.
// *cgroup.Prepared satisfies it on every build (strict_linux.go,
// strict_other.go).
type strictProc interface {
	Start(*exec.Cmd) error
	KillAndEmpty(context.Context) error
	Close() error
}

var _ strictProc = (*cgroup.Prepared)(nil)

// strictDrainTimeout bounds each KillAndEmpty a strict group's cleanup runs.
var strictDrainTimeout = 30 * time.Second

// strictCleanupBackoff is the bounded retry schedule for a strict group
// whose cleanup failed on exit: one more attempt after each wait. When it
// runs out the handle is still kept, and the host's strict spawns stay
// refused until the agent restarts: a group that never drained may still
// hold a process nothing accounts for.
var strictCleanupBackoff = []time.Duration{time.Second, 5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute, 10 * time.Minute}

// execLogLineMax caps one teed log line (spec §5): longer is truncated with
// a marker, never split into phantom lines.
const execLogLineMax = 8 << 10

// execDrainGrace bounds how long a dead child's held-open output pipes can
// delay its Wait. Teeing turns the child's stdout/stderr into PIPES (with a
// plain *os.File the fd passed straight through), and os/exec's Wait then
// waits for the read side to drain — which a grandchild holding the write
// end keeps open forever. Same discipline, same reason as seedDrainGrace;
// without it, an exited instance would look alive to the activator for as
// long as some stray grandchild lived, and Stop would hang on p.done.
var execDrainGrace = 5 * time.Second

// teeStd wires cmd's stdout/stderr: straight to the agent's own streams
// (journald under systemd) exactly as before, PLUS — only when b.Logs is
// set — a line-splitting tee into the log sink tagged as rt's app output.
func (b *ExecBackend) teeStd(cmd *exec.Cmd, rt registry.Route) {
	if b.Logs == nil {
		cmd.Stdout = os.Stdout
		cmd.Stderr = os.Stderr
		return
	}
	acct, app := release.RouteAccount(rt), rt.App
	emit := func(line string) { b.Logs.App(acct, app, rt.Host, line) }
	cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit))
	cmd.Stderr = io.MultiWriter(os.Stderr, applog.NewLineWriter(execLogLineMax, emit))
	cmd.WaitDelay = execDrainGrace
}

var _ Backend = (*ExecBackend)(nil)

// tier is Tier with the nil default applied.
func (b *ExecBackend) tier(rt registry.Route) cgroup.Limit {
	if b.Tier == nil {
		return cgroup.TierDefault
	}
	return b.Tier(rt)
}

// spawnInCgroup is the ONE seam all three child shapes — seed, instance,
// sidecar — start through: the route's cgroup, the route's tier. A spawn
// site that bypassed it would compile and silently ship uncapped children;
// keeping the tier plumbing here makes that a diff a reviewer can see and
// TestSpawnInCgroupAppliesTheTier can pin.
//
// Under a move (Strict) it is also the one place a spawn is made strict, so
// no child shape can start a host's process outside the strict group. The
// returned settle runs after the child has been reaped: for a strict spawn
// it drains and closes the group, for a plain one it does nothing.
//
// A strict handle owns the host's group from the spawn until its cleanup
// succeeds (hostGate.owner). While it does, every other spawn of the host
// is refused, an ordinary one included: the check and an ordinary spawn's
// start happen under the host's gate, so no strict group can be prepared
// for the host in between. With Strict nil (a box with no moves wiring)
// none of this runs and no lock is taken: the spawn is main's exactly.
func (b *ExecBackend) spawnInCgroup(rt registry.Route, build func() *exec.Cmd) (*exec.Cmd, func(), error) {
	return b.spawnInCgroupGated(rt, cgroup.Limit{}, build, nil)
}

// spawnInCgroupGated is spawnInCgroup with a before-start hook (final review
// F1), asked immediately before each start attempt: the cgroup manager's
// (StartGated), or, strict, after the strict group is prepared. A refusal
// starts nothing and is cleaned up as a failed start is. A nil before is
// spawnInCgroup exactly.
//
// tier is the artifact's own (a recorded or staged build's, ruling
// R-TIER); zero is b.tier(rt), the live link's, which is every caller but
// a staged spawn.
func (b *ExecBackend) spawnInCgroupGated(rt registry.Route, tier cgroup.Limit, build func() *exec.Cmd, before func() error) (*exec.Cmd, func(), error) {
	if b.Cfg.HostWrites.Enabled {
		return nil, nil, ErrClaimIsolationUnavailable
	}
	if tier == (cgroup.Limit{}) {
		tier = b.tier(rt)
	}
	if b.Strict == nil {
		cmd, err := b.Cgroups.StartGated(rt.Host, tier, build, before)
		return cmd, func() {}, err
	}
	strict := b.Strict(rt.Host)
	g := b.gate(rt.Host)
	g.mu.Lock()
	if o := g.owner; o != nil {
		pending := o.pending
		g.mu.Unlock()
		cause := ErrStrictGroupInUse
		if pending {
			cause = ErrStrictCleanupPending
		}
		if strict {
			return nil, nil, fmt.Errorf("%s: %w: %w", rt.Host, ErrStrictRequired, cause)
		}
		return nil, nil, fmt.Errorf("%s: %w", rt.Host, cause)
	}
	if !strict {
		cmd, err := b.Cgroups.StartGated(rt.Host, tier, build, before)
		g.mu.Unlock()
		return cmd, func() {}, err
	}
	o := &strictOwner{}
	g.owner = o
	g.mu.Unlock()
	sp, err := b.prepareStrictProc(rt.Host, tier)
	if err != nil {
		b.disown(rt.Host, o, false)
		return nil, nil, fmt.Errorf("%s: %w: %v", rt.Host, ErrStrictRequired, err)
	}
	o.sp = sp
	cmd := build()
	if before != nil {
		if err := before(); err != nil {
			// Nothing started: the strict group goes as after a failed start.
			b.settleStrict(rt.Host, o)
			return nil, nil, err
		}
	}
	if err := sp.Start(cmd); err != nil {
		b.settleStrict(rt.Host, o)
		return nil, nil, fmt.Errorf("%s: %w: %v", rt.Host, ErrStrictRequired, err)
	}
	return cmd, func() { b.settleStrict(rt.Host, o) }, nil
}

// prepareStrictProc is prepareStrict with its default applied.
func (b *ExecBackend) prepareStrictProc(host string, limit cgroup.Limit) (strictProc, error) {
	if b.prepareStrict != nil {
		return b.prepareStrict(host, limit)
	}
	if b.Cgroups == nil {
		return nil, errors.New("no delegated cgroup subtree")
	}
	p, err := b.Cgroups.PrepareStrict(host, limit)
	if err != nil {
		return nil, err
	}
	return p, nil
}

// strictOwner is the strict handle that owns a host's group. pending
// (guarded by the host gate's mu) is set once its first cleanup attempt
// failed and retries own the group. drained records a KillAndEmpty that
// already succeeded: the handle refuses a second one, so a retry after a
// failed Close only closes. sp and drained are touched only by the one
// goroutine running the owner's cleanup at a time.
type strictOwner struct {
	sp      strictProc
	pending bool
	drained bool
}

// attempt is one cleanup attempt: drain (unless already drained), then
// close. It runs outside the host gate (a drain may take strictDrainTimeout);
// the owner it belongs to keeps every other user of the group out meanwhile.
func (o *strictOwner) attempt(host string) error {
	if !o.drained {
		ctx, cancel := context.WithTimeout(context.Background(), strictDrainTimeout)
		err := o.sp.KillAndEmpty(ctx)
		cancel()
		if err != nil {
			log.Printf("activator: %s: strict group did not drain: %v", host, err)
			return err
		}
		o.drained = true
	}
	if err := o.sp.Close(); err != nil {
		log.Printf("activator: %s: strict group not closed: %v", host, err)
		return err
	}
	return nil
}

// settleStrict drains a strict child's group and closes the handle. On
// success the owner lets go and the exit path's cleanupHost removes the
// group as before. On a failure the owner stays, marked pending, retried on
// strictCleanupBackoff: until a retry succeeds every spawn of the host is
// refused and only the owner may remove the group directory, because the
// handle's identity is that directory and removing it would strand the
// handle for ever.
func (b *ExecBackend) settleStrict(host string, o *strictOwner) {
	if o.attempt(host) == nil {
		b.disown(host, o, false)
		return
	}
	b.hook("cleanup-failed")
	g := b.gate(host)
	g.mu.Lock()
	o.pending = true
	g.mu.Unlock()
	go b.retryStrictCleanup(host, o, append([]time.Duration(nil), strictCleanupBackoff...))
}

// disown ends o's ownership of host's group. removeGroup removes the group
// directory first, under the host gate, so no spawn can recreate it in
// between: the retry owner's last act.
func (b *ExecBackend) disown(host string, o *strictOwner, removeGroup bool) {
	g := b.gate(host)
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.owner != o {
		return
	}
	if removeGroup {
		b.Cgroups.CleanupHost(host)
	}
	g.owner = nil
}

// cleanupHost is the exit path's CleanupHost. With Strict nil it is main's
// call, with no lock. Otherwise it runs under the host gate and removes
// nothing while any strict owner, live or pending, holds the group: the
// owner removes it once its handle is closed.
func (b *ExecBackend) cleanupHost(host string) {
	if b.Strict == nil {
		b.Cgroups.CleanupHost(host)
		return
	}
	g := b.gate(host)
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.owner != nil {
		return
	}
	b.hook("exit-cleanup-checked")
	b.Cgroups.CleanupHost(host)
}

// retryStrictCleanup retries o's cleanup once after each wait in schedule,
// a copy taken when the cleanup was queued, and removes the group directory
// when one succeeds.
func (b *ExecBackend) retryStrictCleanup(host string, o *strictOwner, schedule []time.Duration) {
	for i, wait := range schedule {
		time.Sleep(wait)
		if o.attempt(host) == nil {
			b.disown(host, o, true)
			log.Printf("activator: %s: strict group cleaned up on retry %d", host, i+1)
			return
		}
	}
	log.Printf("activator: %s: strict group still not cleaned up after %d retries; spawns of this host stay refused until the agent restarts", host, len(schedule))
}

// strictCleanupPending reports, under the host gate, whether host's group
// is owned by a strict handle whose cleanup is being retried.
func (b *ExecBackend) strictCleanupPending(host string) bool {
	g := b.lookupGate(host)
	if g == nil {
		return false
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.owner != nil && g.owner.pending
}

// NextSpawnStrict is the source's S2 input (moves design §7.3, §5.3): will
// host's next spawn be strict? OK only when Strict names the host, the box
// has a delegated cgroup subtree, PrepareStrict's own runtime preflight
// passes (the agent placed in that subtree or its supervise child, on a
// real cgroup2 mount, on a build that supports it), and no earlier strict
// group of the host awaits cleanup. The per-host group state PrepareStrict
// also judges (a populated group) is the idle stop's to clear.
func (b *ExecBackend) NextSpawnStrict(host string) moves.Check {
	switch {
	case b.Strict == nil || !b.Strict(host):
		return moves.Check{Reason: "S2: the next spawn of this host is not strict"}
	case b.Cgroups.Root() == "":
		return moves.Check{Reason: "S2: this box has no delegated cgroup subtree, so it cannot spawn strictly"}
	case b.strictCleanupPending(host):
		return moves.Check{Reason: "S2: this host's last strict group has not been cleaned up yet"}
	}
	preflight := b.strictPreflight
	if preflight == nil {
		preflight = b.Cgroups.StrictPreflight
	}
	if err := preflight(); err != nil {
		if errors.Is(err, cgroup.ErrStrictRefused) {
			return moves.Check{Reason: "S2: the agent is not placed in its delegated cgroup subtree, so it cannot spawn strictly"}
		}
		return moves.Check{Reason: "S2: this box cannot spawn strictly"}
	}
	return moves.Check{OK: true}
}

func (b *ExecBackend) liveDir() string             { return filepath.Join(b.OptDir, "live") }
func (b *ExecBackend) liveLink(host string) string { return filepath.Join(b.liveDir(), host) }

// LiveVersion reports the version host's live symlink currently points at, or
// "" (no error) if the host hasn't been put under version management yet. It
// doubles as the Service's convergence-sweep seam — New picks it up via an
// optional interface.
//
// Delegated to release.LiveVersion rather than reading the parent directory,
// because the version store gained an account level: a migrated tenant's link
// is <opt>/versions/<ver>/<acct>/<app>, and "the parent dir" there is the
// account SQID. ConvergeDecision compares this against the route's version,
// so returning a sqid would convince it that every migrated tenant was
// converged forever — a tenant that silently stops taking deploys. One
// definition of "which segment is the version", in one place.
func (b *ExecBackend) LiveVersion(host string) (string, error) {
	return release.LiveVersion(b.OptDir, host)
}

// Acquire claims the per-host single-writer lease. A contested lease is
// re-surfaced as errLeaseHeld (never a litestream type — the lease package's
// *HeldError is this package's only view of it) so the edge can 503 with a
// Retry-After that tracks the holder's expiry.
func (b *ExecBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
	host := rt.Host
	// Admission before the lease: the instance's config has to be
	// providable from THIS box, or it must not be the one that starts it.
	// Here, not in doWake, because this is the only place an instance or
	// sidecar lease is taken; every wake source reaches it.
	if err := b.AdmitConfig(ctx, rt); err != nil {
		return nil, err
	}
	held, err := lease.Acquire(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
	if err != nil {
		var he *lease.HeldError
		if errors.As(err, &he) {
			retry := time.Until(he.ExpiresAt)
			if retry < time.Second {
				retry = time.Second
			}
			return nil, &errLeaseHeld{holder: he.Owner, retryAfter: retry}
		}
		return nil, err
	}
	return &execLease{held: held, defaultDB: b.pathsFor(rt).dbPath}, nil
}

// AdmitConfig asks the config gate whether this box can provide rt's config:
// nil means yes (or no gate), an *errConfigRefused means no. Acquire calls it
// before the lease; the activator calls it again before an in-place worker
// respawn, which keeps its lease and so never passes through Acquire.
func (b *ExecBackend) AdmitConfig(ctx context.Context, rt registry.Route) error {
	if b.ConfigGate == nil {
		return nil
	}
	if err := b.ConfigGate(ctx, rt); err != nil {
		return &errConfigRefused{err: err}
	}
	return nil
}

// execLease adapts *lease.Held to the Lease seam.
type execLease struct {
	held      *lease.Held
	defaultDB string // used when the route carries no DBPath
}

func (l *execLease) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (Rep, error) {
	if dbPath == "" {
		dbPath = l.defaultDB
	}
	rep, err := l.held.Replicate(ctx, dbPath, syncInterval)
	if err != nil {
		return nil, err
	}
	return rep, nil
}

// The Rep execLease.Replicate hands back is lease.Held's *lease.Replicator,
// and SeedPlaced records a seed placed only through its SyncAndWait. Pinned
// here so a change to either fails the build, not every seeded wake.
var (
	_ func(*lease.Held, context.Context, string, time.Duration) (*lease.Replicator, error) = (*lease.Held).Replicate
	_ interface{ SyncAndWait(context.Context) error }                                      = (*lease.Replicator)(nil)
)

func (l *execLease) Release(ctx context.Context) error { return l.held.Release(ctx) }

// RetryRelease is releaseRetrier: the held lease's conditional retry of a
// release that failed (#292).
func (l *execLease) RetryRelease(ctx context.Context) error { return l.held.RetryRelease(ctx) }

// Err is the held lease's liveness (leaseHealth): a failed renewal, a
// spent handle or a poisoned one.
func (l *execLease) Err() error { return l.held.Err() }

// Restore puts the row's database in place for a wake. A row whose seed is
// placeable (ready, swapped or placed) or committed goes through the seed
// fallback (seedplace.go), which never reports fresh; every other row
// restores plainly. A seed column that does not parse fails the wake:
// reading it as "no seed" would boot a seeded instance empty.
func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
	src, err := seeds.ParseRowSeed(rt.SeedSource)
	if err != nil {
		return false, err
	}
	if seedRestores(src) {
		return b.seededRestore(ctx, rt, src)
	}
	return b.restorePlain(ctx, rt)
}

// restorePlain is Restore for a row with no seed to place: the warm cache,
// then the staging fallback, then the host's own replica, and fresh only
// when there is none of the three.
func (b *ExecBackend) restorePlain(ctx context.Context, rt registry.Route) (bool, error) {
	host, dbPath := rt.Host, b.pathsFor(rt).dbPath
	// A failed seed whose processes were not proven gone left its database
	// and an obligation (Task 8): it is answered before the database can
	// count as a warm cache, and before the marker check below can purge it.
	if err := b.settleSeedObligation(ctx, rt, dbPath); err != nil {
		return false, err
	}
	if err := reconcileSeedMarker(rt, dbPath); err != nil {
		return false, err
	}
	if _, err := os.Stat(dbPath); err == nil {
		return false, nil // warm cache — see doWake
	} else if !os.IsNotExist(err) {
		return false, err
	}
	// The DB is gone, so any leftover litestream sidecar state (meta dirs,
	// WAL/SHM) describes a database that no longer exists — litestream's own
	// recovery guidance is to delete it. Clear it so the restored copy starts
	// a clean lineage. Two meta dirs: litestream's default suffix ("-litestream",
	// its MetaDirSuffix — spelled out because this package never imports
	// litestream) as any box-wide backup replicator would use, and the lease
	// package's activator-scoped ".activator-<db>-litestream".
	PurgeLocalDBArtifacts(dbPath, false)

	// The staging fallback (issue #331): when a move staged a replica
	// here, promote it (server-side, own bucket) and restore from the
	// promoted copy. Staging wins when present (promoteStagedReplica,
	// review round 1): a staged copy exists only while an inbound move is
	// landing, so it is newer than any lineage the own prefix kept from an
	// earlier stay on this box (the legacy cut does not delete the own
	// prefix it leaves behind), and the promotion's mirror replaces it.
	var promoted map[string]lease.MirrorObject
	if b.Mirror != nil && !b.Phase1 {
		var err error
		if promoted, err = b.promoteStagedReplica(ctx, host); err != nil {
			// HARD failure, same rule as restoreReplica's own error paths
			// (review round 1): "continuing without it" here would fall
			// through to a FRESH boot + seed, and once that empty database
			// replicates, the staged data is shadowed forever. A failed
			// wake retries; a seeded ghost does not.
			return false, fmt.Errorf("staging fallback for %s: %w", host, err)
		}
	}

	restored, err := restoreReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
	if err != nil {
		// "We couldn't tell" is never reported as "there was nothing there":
		// fresh stays false on every error path, so a bucket outage can't
		// route a live instance into the seed.
		return false, err
	}
	if restored {
		// Prove the restored file before anything serves from it. A replica
		// can come back damaged — a torn page, a truncated object, a lineage
		// that never had a snapshot — and the alternatives to failing here are
		// both worse than a down route: serving a corrupt database, or (if
		// this returned fresh) running the route's seed over recoverable data.
		if err := lease.CheckDB(dbPath); err != nil {
			q := dbPath + quarantineSuffix
			_ = os.Remove(q) // keep exactly one: the newest failure is the evidence
			if mvErr := os.Rename(dbPath, q); mvErr != nil {
				// Quarantine failed, so the live path may still hold the bad
				// file — remove it rather than leave a warm cache that the
				// next wake would boot with no check at all.
				_ = os.Remove(dbPath)
				log.Printf("activator: %s restored CORRUPT (%v) and could not be quarantined (%v) — removed", host, err, mvErr)
			} else {
				log.Printf("activator: %s restored CORRUPT from s3://%s/%s: %v — quarantined at %s, instance NOT started",
					host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), err, q)
			}
			for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
				_ = os.Remove(p)
			}
			// The next wake re-attempts the restore rather than refusing on
			// sight of the quarantined file: a replaced or repaired replica
			// must recover with no operator action. The price is a repeated
			// restore for as long as the replica stays broken, which is the
			// right way round — a permanently corrupt replica is an alarm
			// someone is answering, not a steady state to optimise for.
			return false, fmt.Errorf("restored replica for %s is corrupt: %w", host, err)
		}
		log.Printf("activator: restored %s from s3://%s/%s", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
		// A clean restore of a copy promoted THIS call is the moment the
		// staging has served its purpose; a corrupt one (above)
		// deliberately LEFT it in place as evidence, and staging that was
		// never consulted (an abort's leftovers under a live own lineage)
		// is not this restore's to touch.
		if len(promoted) > 0 {
			b.dropStagedReplica(ctx, promoted)
		}
	} else {
		// Never replicated: a brand-new instance boots fresh and creates its
		// own DB — the register-only create path (provisioning as a metadata
		// write) landing on its first wake. This — DB absent AND no replica —
		// is the ONLY fresh boot, and the one moment the route's seed runs.
		log.Printf("activator: no replica for %s in s3://%s/%s — booting fresh", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
	}
	return !restored, nil
}

// restoreReplica is lease.Restore behind a package var so the fresh-detection
// matrix (replica present / absent / unreachable) is testable without S3.
var restoreReplica = lease.Restore

// Seed runs the route's first-boot initialization, once, before the instance
// has ever been spawned (doWake calls it only when Restore reported fresh).
//
// THE CONTRACT, FOR APP AUTHORS: the platform invokes your binary with your
// manifest's `seed` argv followed by `--db <path>` — you create your schema
// and starter content in that database and exit 0. (`titogo seed` is the shape
// this was named for.) It runs under the same merged env layers the instance
// itself gets, so the credentials and config in your env file are there; it
// does NOT get the purge-channel vars, which describe a serving process's
// cache and mean nothing to a one-shot init — and it does not get them even
// if your own env file sets them, those two names being the agent's to
// author. A non-zero exit fails the wake:
// a half-seeded instance must not serve.
// Seed dispatches by layout, the same way Start does: a nested row seeds
// through its own launch file and unit, a fenced or conflicting row
// refuses, and a flat or foreign row seeds through today's exec child.
func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
	p := b.pathsFor(rt)
	switch p.layout {
	case LayoutNested:
		return b.seedUnit(ctx, rt, nil, cgroup.Limit{})
	case LayoutFenced, LayoutConflict:
		return fmt.Errorf("%s: refusing to seed: %s", rt.Host, p.reason)
	}
	return b.seedChild(ctx, rt, b.liveLink(rt.Host), cgroup.Limit{})
}

// SeedExe is Seed on the identity a wake recorded (settings spec §4.2):
// id.StagedPath is checked against id.Digest first, and a copy that
// changed seeds nothing (never the live link instead). A flat or foreign
// row execs that file by path. A nested row asks root to run the seed unit
// on it, which root validates and binds in place of the live link. The
// seed runs at id's own tier (GenTier, ruling R-TIER), never the tier of
// whatever the live link names by then.
func (b *ExecBackend) SeedExe(ctx context.Context, rt registry.Route, id release.StagedIdentity) error {
	if err := release.VerifyStaged(id); err != nil {
		return fmt.Errorf("%s: not seeded with %s: %w", rt.Host, id.Version, err)
	}
	tier := b.GenTier(rt, id.Version)
	p := b.pathsFor(rt)
	switch p.layout {
	case LayoutNested:
		return b.seedUnit(ctx, rt, stagedExeOf(id), tier)
	case LayoutFenced, LayoutConflict:
		return fmt.Errorf("%s: refusing to seed: %s", rt.Host, p.reason)
	}
	return b.seedChild(ctx, rt, id.StagedPath, tier)
}

// seedArgv parses rt.Seed's JSON argv array and appends the --db flag, the
// composition both spawn paths need identically (design §6.2). An empty
// declared seed (an empty array, len 0) returns a nil argv, which both
// callers treat as "nothing to run" (doWake only calls Seed for a
// non-empty rt.Seed; this stays defensive rather than assuming that).
func seedArgv(rt registry.Route, dbPath string) ([]string, error) {
	var seedArgs []string
	if err := json.Unmarshal([]byte(rt.Seed), &seedArgs); err != nil {
		// Never a silent skip: a mangled seed column must fail the wake
		// loudly, not boot an empty instance that looks healthy.
		return nil, fmt.Errorf("parse seed argv for %s (%q): %w", rt.Host, rt.Seed, err)
	}
	if len(seedArgs) == 0 {
		return nil, nil
	}
	return append(seedArgs, "--db", dbPath), nil
}

// removeSeedDB deletes a seed's own half-built database and its WAL/SHM
// siblings, proven out of reach of its processes: the cleanup both spawn
// paths and the next wake's discharge run when, and only when, the
// database is the seed's own work (seedCreatedTheDB). Shared so the three
// cannot drift on which files "the database" means. A file already gone
// is fine; any other failure is returned, and the directory is synced so
// the removals are durable before the caller removes the obligation.
func removeSeedDB(dbPath string) error {
	for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
		if err := os.Remove(p); err != nil && !errors.Is(err, fs.ErrNotExist) {
			return fmt.Errorf("remove the seed's %s: %w", p, err)
		}
	}
	if err := syncDir(filepath.Dir(dbPath)); err != nil {
		return fmt.Errorf("sync %s after removing the seed's database: %w", filepath.Dir(dbPath), err)
	}
	return nil
}

// seedChild is Seed's flat body. exe is the file the seed execs: the live
// link (Seed), which is materialized first if it is missing, or a staged
// path SeedExe verified, which is exec'd by path whatever the link names.
// With LiveExecGate set the live link is pinned instead (declinedLivePin)
// and the file it named is exec'd. tier is that file's own; zero is the
// live tier (b.tier).
func (b *ExecBackend) seedChild(ctx context.Context, rt registry.Route, exe string, tier cgroup.Limit) error {
	host := rt.Host
	paths := b.pathsFor(rt)
	argv, err := seedArgv(rt, paths.dbPath)
	if err != nil {
		return err
	}
	if argv == nil {
		return nil // defensive: doWake only calls us for a non-empty seed
	}
	if exe == b.liveLink(host) {
		if b.LiveExecGate != nil {
			// A declined box seeds the build its gate passed, by path, at
			// that build's tier (declinedLivePin), never the link. Before the
			// database stat below: a refusal runs nothing and so can never
			// count as a failed seed whose database is removed.
			id, err := b.declinedLivePin(ctx, rt)
			if err != nil {
				return err
			}
			exe, tier = id.StagedPath, b.GenTier(rt, id.Version)
		} else if err := b.ensureLiveLink(ctx, host); err != nil {
			return err
		}
	}
	// Mint before the env is read, exactly as Start does — a seed is the FIRST
	// thing a fresh instance runs, so this is usually where the token comes
	// into existence, and instanceEnv's contract is that a seed sees the same
	// environment the instance will.
	if err := EnsureAdminToken(paths.envFile); err != nil {
		return fmt.Errorf("admin token for %s: %w", host, err)
	}
	// The instance key, before the env is read for the same reason: a seed
	// is usually the first process of a fresh instance (L21).
	if _, err := b.ensureInstanceKeyFor(ctx, rt, paths); err != nil {
		return fmt.Errorf("instance key for %s: %w", host, err)
	}
	fileEnv, keyLayer, err := b.instanceEnvLayer(rt)
	if err != nil {
		return err
	}
	// A seed has no control socket, so a replica's seed is gated as a
	// launch without one (ruling 26), on the layer this same read found.
	if err := b.replicaKeyGate(rt, fileEnv, keyLayer, false); err != nil {
		return err
	}
	dbPath := paths.dbPath

	// Whether the database existed BEFORE the seed ran decides whether a
	// failed seed may clean up after itself (below). On the wake path it is
	// always absent — Restore reported fresh.
	_, statErr := os.Stat(dbPath)
	seedOwnsDB := seedCreatedTheDB(statErr)

	ctx, cancel := context.WithTimeout(ctx, seedTimeout)
	defer cancel()
	var stderr capBuffer
	log.Printf("activator: seeding %s: %s %s", host, exe, strings.Join(argv, " "))
	// The seed shares the instance's carlos-exec@<host> cgroup: first-boot
	// cost is that instance's work. No CleanupHost here — the instance spawn
	// reuses the dir moments later, and its exit path (or the next agent
	// start's sweep) removes it.
	child, _, perr := b.placeFor(spawnSeed, 0)
	if perr != nil {
		return perr
	}
	// The obligation goes on disk, durably, before the seed's first
	// instruction (Task 8 fix round 1): from here until its processes are
	// proven gone, nothing may treat the database as warm, whatever happens
	// to the seed, its disk or this agent. A write that fails refuses the
	// seed with nothing run.
	if err := recordSeedObligation(host, dbPath, seedObligation{ownsDB: seedOwnsDB, layout: LayoutFlat}); err != nil {
		return err
	}
	cmd, settle, err := b.spawnInCgroupAtTier(rt, child, false, tier, func() *exec.Cmd {
		cmd := exec.CommandContext(ctx, exe, argv...)
		// The control-channel vars are agent-owned in BOTH directions here too
		// (Start's discipline, see withoutVars), and here they are always
		// STRIPPED: the seed gets no control listener of its own, so fileEnv is
		// tenant-supplied and without this a tenant's own env file could hand
		// its seed a CARLOS_CONTROL_SOCKET the agent never opened and a
		// CARLOS_PURGE for a channel that does not exist — and the seed has no
		// serving cache to purge, nor any business setting timers, in any case.
		// This is what makes the contract above true rather than merely
		// intended.
		cmd.Env = childEnv(seedVars(fileEnv, host))
		// Like the instance child, the seed logs through the agent's own streams
		// (journald under systemd) — and its stderr is ALSO teed into a bounded
		// buffer, because the wake error this returns is the whole diagnosis an
		// operator gets for a host that refuses to come up. With Logs set the
		// seed's output ships to `carlos logs` too: it is app output, and on a
		// failed first wake it is the diagnosis the MEMBER gets.
		cmd.Stdout = os.Stdout
		cmd.Stderr = io.MultiWriter(os.Stderr, &stderr)
		if b.Logs != nil {
			acct, app := release.RouteAccount(rt), rt.App
			emit := func(line string) { b.Logs.App(acct, app, host, line) }
			cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit))
			cmd.Stderr = io.MultiWriter(os.Stderr, &stderr, applog.NewLineWriter(execLogLineMax, emit))
		}
		// WaitDelay is what makes seedTimeout real. Because Stderr is a plain
		// io.Writer (the tee), os/exec pipes it and Wait waits for that pipe to
		// close — and a seed that spawned its own grandchild holds the write end
		// open long after the timeout killed the seed itself. Without this, a
		// wedged seed would hold the lease until the wake's own 5m budget expired.
		cmd.WaitDelay = seedDrainGrace
		return cmd
	}, pendingRecordHook(ctx)) // the wake's record, at the exec (amadan #272)
	spawned := err == nil
	if spawned {
		err = cmd.Wait()
		settle()
	}
	// WaitDelay reports on the SUCCESS path too: a seed that exits 0 after
	// backgrounding something (a helper, a stray `&`) leaves the inherited
	// stderr pipe open, and Wait then returns ErrWaitDelay INSTEAD of nil
	// even though the process exited 0. Treating that as a failure would
	// fail the wake AND delete the database the seed had just correctly
	// built — leaving the host fresh again next wake, i.e. permanently
	// un-wakeable. The exit status is the verdict; the leaked pipe is a
	// log line.
	exitedZero := spawned && (err == nil || (errors.Is(err, exec.ErrWaitDelay) && cmd.ProcessState != nil && cmd.ProcessState.Success()))
	var failure error
	if !exitedZero {
		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
			err = fmt.Errorf("timed out after %s and was killed: %w", seedTimeout, err)
		}
		failure = fmt.Errorf("seeding %s: %w", host, err)
		if tail := strings.TrimSpace(stderr.String()); tail != "" {
			failure = fmt.Errorf("seeding %s: %w: %s", host, err, tail)
		}
	}
	// What the seed left is proven gone before its database is touched
	// (Task 8). CommandContext and Wait speak for the leader only, and a
	// process it forked keeps writing the database, a removed one included.
	// A failed seed's cgroup is killed and proven empty: flag on its role
	// child, flag off carlos-exec@<host>, which holds nothing else yet (a
	// fresh host, before the spawn, under the wake's lease). A successful
	// seed's role child is too (flag on): a helper that outlived the seed
	// would otherwise run on beside the instance, out of reach of any
	// generation kill. Flag off, a successful seed's helper stays in the
	// host cgroup, where the terminal stop's proof (stopHostProven) finds
	// it. A spawn that failed ran nothing.
	if spawned && (failure != nil || child != "") {
		pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
		perr := b.seedDescendantsGone(pctx, rt, child)
		pcancel()
		if perr != nil {
			// The database is kept, as seedUnitFailed keeps it: removing it
			// under a live writer is the hazard itself. The obligation
			// written ahead stays, so every later wake refuses it as warm
			// until the proof holds (Restore), and the caller leaves the
			// lease to its TTL (ErrSeedUnproven).
			if failure == nil {
				failure = fmt.Errorf("seed for %s: %w: processes remain after a successful seed: %v", host, ErrSeedUnproven, perr)
			} else {
				failure = fmt.Errorf("%w: %w (the seed's processes remain after stopping it; its database was left in place: %v)", failure, ErrSeedUnproven, perr)
			}
			log.Printf("activator: %s: the seed's processes are not proven gone; its database is kept and %s refuses it as warm until they are", host, dbPath+seedUnprovenSuffix)
			return failure
		}
		b.Cgroups.RemoveChild(host, child) // the seed's own child only; no-op for "" (flag off)
	}
	// From here the seed's processes are proven gone (or a spawn that
	// failed ran nothing, or a flag-off success's helper is the terminal
	// stop's to prove), so the obligation is discharged. A failure takes
	// the half-built database with it first. Failing the wake is only half
	// the promise: the file a crashed seed left behind would look like a
	// warm cache to the NEXT wake (DB present, so not fresh, so no seed) and
	// the instance would boot on it, which is the half-seeded instance this
	// whole path exists to prevent. Only a database that wasn't there when
	// we started is ours to remove, and under the lease, with the seed's
	// processes proven gone above, nothing else can have written it. A
	// discharge that fails keeps the obligation and fails the wake while
	// nothing serves; the next wake proves and removes again. A successful
	// flag-off seed's failed discharge first proves its processes gone, since
	// no instance will (ErrSeedUnproven otherwise).
	if derr := dischargeSeedObligation(dbPath, failure != nil && seedOwnsDB); derr != nil {
		unprovedSuccess := spawned && failure == nil && child == "" // the proof above was skipped
		if failure == nil {
			failure = fmt.Errorf("seed for %s: %w", host, derr)
		} else {
			failure = fmt.Errorf("%w; and %w", failure, derr)
		}
		if unprovedSuccess {
			// A successful flag-off seed has no proof of its own: its helper
			// is left for the instance's terminal stop to prove. With the
			// discharge failed the wake fails, so no instance will run to
			// prove it, and releasing the lease over it would be the hazard
			// itself (fix wave 1, item 2): the proof is tried now. Unproven,
			// the seed is ErrSeedUnproven and the lease is left to its TTL.
			pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
			perr := b.seedDescendantsGone(pctx, rt, child)
			pcancel()
			if perr != nil {
				log.Printf("activator: %s: the seed's obligation could not be discharged and its processes are not proven gone; %s stays", host, dbPath+seedUnprovenSuffix)
				return fmt.Errorf("%w: %w: the seed's processes are not proven gone: %v", failure, ErrSeedUnproven, perr)
			}
		}
		return failure
	}
	if failure == nil {
		if err != nil {
			log.Printf("activator: seeded %s (it exited 0 but left a child holding its output pipe open for more than %s)", host, seedDrainGrace)
		} else {
			log.Printf("activator: seeded %s", host)
		}
		return nil
	}
	return failure
}

// seedDescendantsGone kills what a flat seed left and proves it gone: flag
// on, its own role child (child); flag off, carlos-exec@<host> (""), which
// holds nothing else yet: a fresh host, before the spawn, under the wake's
// lease. An absent cgroup is empty. A box with no delegated cgroup
// (cgroup.ErrNoDelegation) ran the seed unaccounted, with nothing that
// could reach a descendant, so the leader's exit is the only proof there
// is (ledger 127): it logs and answers nil. So is a kernel with no
// cgroup.kill (cgroup.ErrNoKill). Any other failure is returned, and the
// caller keeps the database (ErrSeedUnproven).
func (b *ExecBackend) seedDescendantsGone(ctx context.Context, rt registry.Route, child string) error {
	var err error
	switch {
	case b.cgroupKill != nil:
		err = b.cgroupKill(ctx, rt, child)
	case child == "":
		err = b.KillHost(ctx, rt)
	default:
		err = b.killChild(ctx, rt.Host, child)
	}
	if errors.Is(err, cgroup.ErrNoDelegation) {
		log.Printf("activator: %s: the seed ran with no delegated cgroup; its leader's exit is the only proof there is", rt.Host)
		return nil
	}
	if errors.Is(err, cgroup.ErrNoKill) {
		// A kernel before 5.14 can kill nothing in a cgroup (EmptyHost's
		// rule): read as unproven, an obligation could never be discharged
		// on such a box and the row would stay down for good.
		log.Printf("activator: %s: this kernel has no cgroup.kill; the seed's leader's exit is the only proof there is", rt.Host)
		return nil
	}
	return err
}

// seedObligation is what <db>.seed-unproven records, written ahead of a
// seed (Task 8 fix round 1): before the seed's first instruction runs, so
// a seed that dies of a full disk, or an agent that dies mid-seed, cannot
// leave a partial database with nothing to refuse it. It is removed only
// once the seed's processes are proven gone (dischargeSeedObligation).
//
// ownsDB is seedCreatedTheDB at the seed's start: only then is the
// database the obligation's to remove. layout is where the seed ran:
// LayoutNested is root's seed unit in the host slice, not
// carlos-exec@<host>, so its proof is that unit's (nestedSeedGone);
// LayoutFlat is an exec child; "" is an obligation that does not say.
type seedObligation struct {
	ownsDB bool
	layout Layout
}

// The obligation's body: three lines, the first a version. Anything a
// reader does not recognise reads as the cautious answer: a database that
// was not the seed's (kept), and the layout the row reads now.
const (
	seedObligationHeader = "carlos seed obligation v1"
	seedObligationOwned  = "db=owned-by-seed"
	seedObligationKept   = "db=there-before-seed"
	seedObligationFlat   = "seed=flat"
	seedObligationNested = "seed=nested"
)

func (o seedObligation) body() []byte {
	db, seed := seedObligationKept, seedObligationFlat
	if o.ownsDB {
		db = seedObligationOwned
	}
	if o.layout == LayoutNested {
		seed = seedObligationNested
	}
	return []byte(seedObligationHeader + "\n" + db + "\n" + seed + "\n")
}

// recordSeedObligation writes the obligation ahead of a seed, durably
// (safefile: file fsync, rename, dir fsync). One that is already there is
// an earlier seed not yet answered (Restore answers it before any seed of
// a wake): it is never overwritten, since that could lose what it says the
// earlier seed owned, and the seed is refused with ErrSeedUnproven. A write
// that fails refuses the seed as an ordinary error: nothing ran.
func recordSeedObligation(host, dbPath string, o seedObligation) error {
	ob := dbPath + seedUnprovenSuffix
	if _, err := os.Lstat(ob); err == nil {
		return fmt.Errorf("%s: refusing to seed: %w: an earlier seed's obligation %s is not answered", host, ErrSeedUnproven, ob)
	} else if !errors.Is(err, fs.ErrNotExist) {
		return fmt.Errorf("%s: refusing to seed: read the seed obligation %s: %w", host, ob, err)
	}
	if err := writeSeedObligationFile(ob, o.body(), 0o600); err != nil {
		return fmt.Errorf("%s: refusing to seed: record its obligation %s: %w", host, ob, err)
	}
	return nil
}

// readSeedObligation reads dbPath's obligation. present is false only on a
// definite "not there"; any other failure to look is an error, never "no
// obligation".
//
// The name sits beside the database, on ground the instance itself can
// write, so it is opened as the env files are (openOwnEnvNoFollow): a
// symlink or a FIFO planted there is refused in the open itself, never
// followed or waited on. Such an entry was not written by a seed, so it
// reads as an obligation that does not say (the zero value): it is proven
// and removed like any other, and the database beside it is kept, which
// neither hangs the wake nor holds the row down.
func readSeedObligation(dbPath string) (o seedObligation, present bool, err error) {
	f, err := openOwnEnvNoFollow(dbPath+seedUnprovenSuffix, os.O_RDONLY, 0)
	if errors.Is(err, fs.ErrNotExist) {
		return seedObligation{}, false, nil
	}
	if errors.Is(err, errOwnEnvSymlink) || errors.Is(err, errOwnEnvNotRegular) {
		return seedObligation{}, true, nil
	}
	if err != nil {
		return seedObligation{}, true, err
	}
	raw, err := io.ReadAll(io.LimitReader(f, 4<<10))
	f.Close()
	if err != nil {
		return seedObligation{}, true, err
	}
	lines := strings.Split(string(raw), "\n")
	if len(lines) >= 3 && lines[0] == seedObligationHeader {
		o.ownsDB = lines[1] == seedObligationOwned
		switch lines[2] {
		case seedObligationNested:
			o.layout = LayoutNested
		case seedObligationFlat:
			o.layout = LayoutFlat
		}
	}
	return o, true, nil
}

// dischargeSeedObligation runs once a seed's processes are proven gone:
// the database files go first when removeDB (removeSeedDB, every error
// checked and the directory synced), and only then the obligation, synced
// too, so a crash between the two leaves the obligation, never a database
// without it. Any failure keeps the obligation, and the next wake proves
// and removes again.
func dischargeSeedObligation(dbPath string, removeDB bool) error {
	if removeDB {
		if err := removeSeedDB(dbPath); err != nil {
			return err
		}
	}
	ob := dbPath + seedUnprovenSuffix
	if err := os.Remove(ob); err != nil && !errors.Is(err, fs.ErrNotExist) {
		return fmt.Errorf("remove the seed obligation %s: %w", ob, err)
	}
	if err := syncDir(filepath.Dir(dbPath)); err != nil {
		return fmt.Errorf("sync %s after removing the seed obligation: %w", filepath.Dir(dbPath), err)
	}
	return nil
}

// settleSeedObligation answers an earlier seed's obligation before anything
// treats dbPath as warm. With none there it does nothing. Otherwise the
// seed's processes are proven gone first (seedObligationProof). A proof
// that fails refuses the wake with ErrSeedUnproven and removes nothing.
// One that holds discharges the obligation: a database the seed owned goes
// with it, so the wake goes on as a first boot and the seed runs again; a
// database that was there before the seed stays, and is warm. The
// obligation is a file, so it survives an agent restart.
func (b *ExecBackend) settleSeedObligation(ctx context.Context, rt registry.Route, dbPath string) error {
	host := rt.Host
	o, present, err := readSeedObligation(dbPath)
	if !present {
		return nil
	}
	if err != nil {
		// "Couldn't tell" is never "no obligation", and never "proven
		// gone" either (fix wave 1, item 2): ErrSeedUnproven leaves the
		// lease to its TTL. A nested wake runs no host proof before this,
		// so nothing else says the seed's processes are gone.
		return fmt.Errorf("%s: %w: read the seed obligation %s: %w", host, ErrSeedUnproven, dbPath+seedUnprovenSuffix, err)
	}
	if err := b.seedObligationProof(ctx, rt, o); err != nil {
		return fmt.Errorf("%s: %w: an earlier seed's processes are not proven gone, so its database is not used: %v", host, ErrSeedUnproven, err)
	}
	if err := dischargeSeedObligation(dbPath, o.ownsDB); err != nil {
		return fmt.Errorf("%s: after an earlier seed proven gone: %w", host, err)
	}
	if o.ownsDB {
		log.Printf("activator: %s: the earlier seed's processes are proven gone; its database is removed and the seed runs again", host)
	} else {
		log.Printf("activator: %s: the earlier seed's processes are proven gone; the database that was there before it is kept", host)
	}
	return nil
}

// seedObligationProof proves an obligation's seed gone, by where it ran. A
// nested seed ran as root's seed unit in the host slice: that unit is
// stopped and its cgroup proven empty (nestedSeedGone). A flat seed ran in
// carlos-exec@<host>, flag off in it and flag on in its role child, so the
// host cgroup is proven, whatever CARLOS_SWAP says now: it holds the role
// child too, so it covers the seed whichever way it ran. An obligation that
// does not say (an unknown body) is proven by the layout the row reads now.
//
// Killing there is safe: this runs under the wake's lease, before anything
// of this wake has run, and no other entry of the host is live (an unproven
// failed wake or stray sidecar refuses the wake before the lease).
func (b *ExecBackend) seedObligationProof(ctx context.Context, rt registry.Route, o seedObligation) error {
	layout := o.layout
	if layout == "" {
		layout = b.pathsFor(rt).layout
	}
	if layout == LayoutNested {
		return b.nestedSeedGone(ctx, rt)
	}
	pctx, cancel := context.WithTimeout(ctx, actStopTimeout)
	defer cancel()
	return b.seedDescendantsGone(pctx, rt, "")
}

// nestedSeedGone stops root's seed unit for rt and proves its cgroup empty
// (waitSeedCgroupEmpty), each bounded and detached from ctx's own
// cancellation, which may be why the seed failed. A failed stop is logged:
// the empty cgroup is the proof.
func (b *ExecBackend) nestedSeedGone(ctx context.Context, rt registry.Route) error {
	if b.Control == nil {
		return fmt.Errorf("%s: instance control channel not configured", rt.Host)
	}
	sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
	if err := b.Control.Stop(sctx, rt.Host, instancectl.RoleSeed); err != nil {
		log.Printf("activator: %s: stop seed unit: %v", rt.Host, err)
	}
	cancel()
	wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
	defer wcancel()
	return b.waitSeedCgroupEmpty(wctx, rt)
}

// seedCreatedTheDB decides, from the stat taken just BEFORE a seed ran,
// whether the database at that path can only be the seed's own work — the
// single condition under which a failed seed's cleanup may delete it.
//
// Fail-safe by construction: ONLY a definite "not there" says yes. A stat that
// merely failed — a permission error, an I/O error, a path that stopped being
// traversable — means "there might be data here", and a guard on deleting live
// data must never arm itself because we couldn't look.
func seedCreatedTheDB(statErr error) bool { return os.IsNotExist(statErr) }

// seedTimeout bounds a first-boot seed run — a wedged seed must fail the wake
// (which releases the lease) rather than sit on it. seedDrainGrace is how long
// Wait then waits for the killed seed's output pipes to close before giving up
// on them (cmd.WaitDelay). Both vars so tests can shrink them.
var (
	seedTimeout    = 2 * time.Minute
	seedDrainGrace = 10 * time.Second
)

// capBuffer keeps at most capBufferMax bytes of what's written to it and
// discards the rest, so quoting a child's stderr in an error can never turn a
// chatty seed into an unbounded string. Locked because it is read after
// the seed's Wait returns while os/exec's own copy goroutine may still be
// writing: on a WaitDelay expiry (the wedged-seed path) Wait returns without
// waiting for that copy to finish.
type capBuffer struct {
	mu sync.Mutex
	b  []byte
}

const capBufferMax = 4 << 10

func (c *capBuffer) Write(p []byte) (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if room := capBufferMax - len(c.b); room > 0 {
		if len(p) < room {
			room = len(p)
		}
		c.b = append(c.b, p[:room]...)
	}
	return len(p), nil
}

func (c *capBuffer) String() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return string(c.b)
}

// seedUnit is Seed's nested body: the same contract as seedChild, namely
// the merged env, the CARLOS_ADMIN_TOKEN mint, and the cleanup on a failure
// this call itself caused, run through the root instance-control channel
// instead of a direct exec (design §6.2). Argv and env come from the same
// seedArgv and withoutVars calls seedChild uses, so the two cannot drift.
//
// A nil exe seeds on the live link (Seed). A staged exe is the build the
// wake recorded (SeedExe): root validates it and binds the seed unit's app
// mount to it, so the live link need not exist and a repoint of it
// changes nothing. A control channel that cannot name it refuses the seed.
// With LiveExecGate set a nil exe pins the live link (declinedLivePin) and
// seeds that file the same way. tier is that build's own (ruling R-TIER);
// zero is the live tier.
func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe, tier cgroup.Limit) error {
	if b.Cfg.HostWrites.Enabled {
		return ErrClaimIsolationUnavailable
	}
	host := rt.Host
	if err := b.nestedGate(rt); err != nil {
		return err
	}
	if exe == nil && b.LiveExecGate != nil {
		// A declined box names the build its gate passed to root, at that
		// build's tier (declinedLivePin), as SeedExe does. Before anything
		// is prepared and before the database stat below, so a refusal asks
		// root for nothing and removes nothing.
		id, err := b.declinedLivePin(ctx, rt)
		if err != nil {
			return err
		}
		exe, tier = stagedExeOf(id), b.GenTier(rt, id.Version)
	}
	if tier == (cgroup.Limit{}) {
		tier = b.tier(rt)
	}
	var xc StagedExeControl
	if exe == nil {
		if err := b.ensureLiveLink(ctx, host); err != nil {
			return err
		}
	} else {
		var err error
		if xc, err = b.stagedExeControl(); err != nil {
			return fmt.Errorf("%s: not seeded with %s: %w", host, exe.Version, err)
		}
	}
	p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSeed)
	if err != nil {
		return err
	}
	defer logs.Close()
	dbPath := p.dbPath
	seedArgs, err := seedArgv(rt, dbPath)
	if err != nil {
		return err
	}
	if seedArgs == nil {
		return nil // defensive: doWake only calls us for a non-empty seed
	}
	argv := append([]string{b.liveLink(host)}, seedArgs...)
	env := childEnv(seedVars(fileEnv, host))
	if err := b.writeLaunch(p, instancectl.RoleSeed, host, argv, env); err != nil {
		return err
	}
	// Whether the database existed BEFORE the seed ran decides whether a
	// failed seed may clean up after itself below, seedChild's rule,
	// unchanged for the unit path.
	_, statErr := os.Stat(dbPath)
	seedOwnsDB := seedCreatedTheDB(statErr)
	// The obligation is written ahead, as seedChild's is, before root is
	// asked to run anything; it names the seed unit as the proof to ask.
	if err := recordSeedObligation(host, dbPath, seedObligation{ownsDB: seedOwnsDB, layout: LayoutNested}); err != nil {
		return err
	}
	// Then the wake's pending record (amadan #272), the last step before
	// root is asked, as the flat seed's comes at its exec after its
	// obligation (Astra review of zdd-owed-items, P2): an obligation that
	// cannot be written records nothing. A record that fails, or a recheck
	// that refuses after it, runs no seed, so the obligation just written
	// is discharged and the database, which no seed touched, is kept.
	if _, err := runPendingRecord(ctx); err != nil {
		cause := fmt.Errorf("seed for %s: %w", host, err)
		if derr := dischargeSeedObligation(dbPath, false); derr != nil {
			return fmt.Errorf("%w; and %w", cause, derr)
		}
		return cause
	}

	log.Printf("activator: seeding %s: %s", host, strings.Join(argv, " "))
	// The client already bounds a seed call at instancectl.DeadlineSeed.
	// Wrapping ctx any tighter here (seedTimeout, at 2m, is shorter) would
	// guarantee the failure path on every seed that legitimately runs
	// longer than that. Wrapping at the SAME deadline is redundant with the
	// client's own timeout but harmless, and keeps this call visibly
	// bounded without second-guessing how long a seed may take.
	sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineSeed)
	var res instancectl.SeedResponse
	if exe != nil {
		res, err = xc.SeedExe(sctx, host, tier.MemoryMax, *exe)
	} else {
		res, err = b.Control.Seed(sctx, host, tier.MemoryMax)
	}
	cancel()
	if errors.Is(err, instancectl.ErrStagedExe) {
		// Root refused the recorded build before it laid out or started
		// anything (403), so no seed ran: nothing to stop, and the
		// database is not the seed's work. Its obligation is discharged.
		cause := fmt.Errorf("seed for %s: %w", host, err)
		if derr := dischargeSeedObligation(dbPath, false); derr != nil {
			return fmt.Errorf("%w; and %w", cause, derr)
		}
		return cause
	}
	if err != nil {
		return b.seedUnitFailed(ctx, rt, dbPath, seedOwnsDB, fmt.Errorf("seed for %s: %w", host, err))
	}
	if res.Result != "success" {
		return b.seedUnitFailed(ctx, rt, dbPath, seedOwnsDB, fmt.Errorf("seed for %s failed: exit %s (%s)", host, res.Status, res.Result))
	}
	// A successful seed is not done until its cgroup is empty (design
	// §6.2): the control service already waited before answering, but the
	// activator's own proof is what lets the wake go on to Start with no
	// seed process still holding the database. Unproven, the obligation
	// written ahead stays.
	wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout)
	emptyErr := b.waitSeedCgroupEmpty(wctx, rt)
	wcancel()
	if emptyErr != nil {
		return fmt.Errorf("seed for %s: %w: processes remain after a successful seed: %v", host, ErrSeedUnproven, emptyErr)
	}
	if derr := dischargeSeedObligation(dbPath, false); derr != nil {
		return fmt.Errorf("seed for %s: %w", host, derr)
	}
	log.Printf("activator: seeded %s", host)
	return nil
}

// seedUnitFailed is seedUnit's shared failure tail: on EITHER failure
// shape, whether Control.Seed itself errored or it answered with a
// non-success result, the seed unit is stopped and its cgroup is proven
// empty (nestedSeedGone) BEFORE any database cleanup runs, never after.
// Running the cleanup on the bare result instead could race a seed process
// SIGKILL hadn't actually reaped yet, exactly the class of bug Stop's own
// ErrStillRunning discipline exists to rule out elsewhere.
// The database goes only when seedOwnsDB AND the proof held, and then the
// obligation (dischargeSeedObligation); when the proof failed, nothing is
// removed, the obligation stays and the returned error says so, because a
// database is not safe to delete out from under a process that might still
// be writing it.
func (b *ExecBackend) seedUnitFailed(ctx context.Context, rt registry.Route, dbPath string, seedOwnsDB bool, cause error) error {
	if emptyErr := b.nestedSeedGone(ctx, rt); emptyErr != nil {
		return fmt.Errorf("%w: %w (the seed's processes remain after stopping it; its database was left in place: %v)", cause, ErrSeedUnproven, emptyErr)
	}
	if derr := dischargeSeedObligation(dbPath, seedOwnsDB); derr != nil {
		return fmt.Errorf("%w; and %w", cause, derr)
	}
	return cause
}

// waitSeedCgroupEmpty polls the seed unit's cgroup.events until it reports
// empty or roleEmptyBound elapses, honouring ctx and b.ObservePoll like the
// exit observer (unitproc.go). A timeout or a read error is returned to the
// caller as a log line, not a failure: it is the deciding result's
// confidence that is at stake here, not the wake itself.
func (b *ExecBackend) waitSeedCgroupEmpty(ctx context.Context, rt registry.Route) error {
	return b.waitRoleCgroupEmpty(ctx, rt, instancectl.RoleSeed)
}

// roleEmptyBound is how long waitRoleCgroupEmpty waits for a role's cgroup
// to empty: actStopTimeout. A var so tests can shrink it.
var roleEmptyBound = actStopTimeout

// waitRoleCgroupEmpty polls role's unit's cgroup.events until it reports
// empty or roleEmptyBound elapses, honouring ctx and b.ObservePoll.
func (b *ExecBackend) waitRoleCgroupEmpty(ctx context.Context, rt registry.Route, role instancectl.Role) error {
	events := b.CgroupEvents
	if events == nil {
		events = cgroupEventsUnder(b.cgroupRoot())
	}
	poll := b.ObservePoll
	if poll <= 0 {
		poll = time.Second
	}
	slice := instancectl.SliceName(rt.Host)
	unit := instancectl.UnitName(role, rt.Host)
	bound := roleEmptyBound
	deadline := time.Now().Add(bound)
	for {
		exists, populated, err := events(slice, unit)
		if err != nil {
			return err
		}
		if !exists || !populated {
			return nil
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("processes remained after %s", bound)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(poll):
		}
	}
}

// NextDue asks the (still-running) instance for its earliest pending scheduled
// delivery over its socket. The endpoint is admin-gated like every instance
// data endpoint, so present the instance's own admin token — read from the
// same env file the spawn path hands the child (the activator holds it either
// way).
func (b *ExecBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) {
	return b.NextDueVia(ctx, rt, rt.Socket)
}

// NextDueVia is NextDue dialling socket instead of rt.Socket: the activator
// passes the instance's current generation socket (AppSocket, zero-downtime
// spec §7.1), because with generations on nothing listens on the canonical
// path. With them off the activator passes rt.Socket, which is NextDue.
func (b *ExecBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) {
	paths := b.pathsFor(rt)
	// A nested host's env file lives on tenant-writable ground (design
	// §3.2), the same as instanceEnv's read of it: go through instancefs
	// rather than a plain read, so a name planted there cannot be anything
	// but a regular file, on pain of an error instead of a hang.
	read := readEnvFile
	if paths.layout == LayoutNested {
		read = func(path string) (map[string]string, error) {
			return readEnvFileBeneath(filepath.Dir(path), filepath.Base(path))
		}
	}
	fileEnv, err := read(paths.envFile)
	if err != nil {
		return 0, fmt.Errorf("read env file: %w", err)
	}
	token := fileEnv["CARLOS_ADMIN_TOKEN"]
	if token == "" {
		return 0, errors.New("no CARLOS_ADMIN_TOKEN in env file")
	}
	req, err := http.NewRequestWithContext(ctx, "GET", "http://carlos/api/next-due", nil)
	if err != nil {
		return 0, err
	}
	req.Header.Set("Authorization", "Bearer "+token)
	resp, err := socketClient(socket).Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return 0, fmt.Errorf("GET /api/next-due: HTTP %d", resp.StatusCode)
	}
	var body struct {
		Due int64 `json:"due"`
	}
	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<12)).Decode(&body); err != nil {
		return 0, fmt.Errorf("decode /api/next-due: %w", err)
	}
	return body.Due, nil
}

// ensureLiveLink verifies host's live symlink (<OptDir>/live/<host> → a staged
// version) exists before the exec, routing a missing one through the
// Materialize hook when configured. A materialize failure (e.g. the staging
// mechanism unreachable) must propagate — the exact live symptom found
// 2026-07-13 in Eleven (a 503 "instance failed to wake"), never silently
// swallowed into "looks fine, wakes anyway."
func (b *ExecBackend) ensureLiveLink(ctx context.Context, host string) error {
	bin := b.liveLink(host)
	if _, err := os.Stat(bin); err == nil {
		return nil
	}
	if b.Materialize != nil {
		if err := b.Materialize(ctx, host); err != nil {
			return fmt.Errorf("materialize live symlink for %s: %w", host, err)
		}
		return nil
	}
	return fmt.Errorf("no live binary for %s (%s missing and no Materialize hook)", host, bin)
}

// LiveIdentity is the one file a cold wake will run (settings spec §4.2):
// the live link, materialized first if it is missing (ensureLiveLink, so
// the Materialize hook and its root-stage relay run exactly as a wake's
// spawn would run them), then resolved ONCE into the file at the end of
// its chain and that file's digest, as StartWorkerStaged does. Version is
// "" for a link outside the version store; the caller decides whether that
// may run (Service.recordBuild refuses it while a build state is kept).
func (b *ExecBackend) LiveIdentity(ctx context.Context, rt registry.Route) (release.StagedIdentity, error) {
	if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
		return release.StagedIdentity{}, err
	}
	id, err := release.ResolveLive(b.OptDir, rt.Host)
	if err != nil {
		return release.StagedIdentity{}, fmt.Errorf("%s: %w", rt.Host, err)
	}
	return id, nil
}

// declinedLivePin is a live-link start's identity on a box with
// LiveExecGate set (a declined box that honours failure records or hold-ran
// markers): live/<host> materialized and resolved ONCE, hashed
// (LiveIdentity), and the gate asked about that version. The caller runs
// exactly that file: a flat start execs it by path, and a nested one names
// it to root (StartExe, StartGenExe, StartSidecarExe, SeedExe), which
// validates it and binds it in place of the link. So the build the gate
// passed is the build that runs, and root repointing the link after this
// read changes nothing.
//
// A link outside the version store names no build the gate could judge or
// root could validate (instancectl.validateStagedExe), and is refused in
// recordBuild's words, as the activator already refuses every exec row's
// unversioned capture while a build state is kept (ruling
// R-EXEC-NOVERSION). Nothing has been prepared, launched or asked of root
// when it returns an error.
func (b *ExecBackend) declinedLivePin(ctx context.Context, rt registry.Route) (release.StagedIdentity, error) {
	id, err := b.LiveIdentity(ctx, rt)
	if err != nil {
		return release.StagedIdentity{}, err
	}
	if id.Version == "" {
		log.Printf("activator: %s not started: the executable its live link names (%s) is outside the version store, so this box cannot check it against its holds; move the instance onto a promoted build", rt.Host, id.StagedPath)
		return release.StagedIdentity{}, fmt.Errorf("%s: %w: the executable is outside the version store", rt.Host, errBuildStateUnwritten)
	}
	if err := b.LiveExecGate(rt, id.Version); err != nil {
		return release.StagedIdentity{}, err
	}
	return id, nil
}

// BuildIdentity is buildIdentifier's answer: build's staged executable of
// rt's app on this box, at the path root's validation expects
// (release.VersionPath under the row's account; instancectl's
// validateStagedExe), hashed now. It reads no link. An invalid version, an
// app name that is not one path segment, or a copy that is not a regular
// file it can read names nothing. Whether build may run is the caller's
// question (the live exec gate), not this one's.
func (b *ExecBackend) BuildIdentity(rt registry.Route, build string) (release.StagedIdentity, error) {
	if !release.ValidVersion(build) {
		return release.StagedIdentity{}, fmt.Errorf("%s: %q is not a build version", rt.Host, build)
	}
	if rt.App == "" || rt.App == "." || rt.App == ".." || filepath.Base(rt.App) != rt.App {
		return release.StagedIdentity{}, fmt.Errorf("%s: app %q names no staged copy", rt.Host, rt.App)
	}
	path := release.VersionPath(b.OptDir, release.RouteAccount(rt), build, rt.App)
	fi, err := os.Lstat(path)
	if err != nil {
		return release.StagedIdentity{}, fmt.Errorf("%s: build %s: %w", rt.Host, build, err)
	}
	if !fi.Mode().IsRegular() {
		return release.StagedIdentity{}, fmt.Errorf("%s: build %s: %s is not a regular file", rt.Host, build, path)
	}
	sum, err := release.FileSHA256(path)
	if err != nil {
		return release.StagedIdentity{}, fmt.Errorf("%s: build %s: %w", rt.Host, build, err)
	}
	return release.StagedIdentity{Version: build, StagedPath: path, Digest: sum}, nil
}

// Start dispatches a wake by layout (design §8.1): a nested row starts as a
// unit through the root instance-control channel, a fenced or conflicting
// row refuses outright, and a flat or foreign row runs today's exec child.
func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
	p := b.pathsFor(rt)
	switch p.layout {
	case LayoutNested:
		return b.startUnitAt(ctx, rt, rt.Socket, 0, nil, cgroup.Limit{})
	case LayoutFenced, LayoutConflict:
		return nil, "", fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
	}
	return b.startChildAt(ctx, rt, rt.Socket, 0, "", cgroup.Limit{})
}

// StartGen is Start for generation n (zero-downtime spec §7.1). perGen false
// is Start exactly: the canonical socket, the canonical control socket, the
// same argv. perGen true binds the generation's own socket: GenSocketPath
// for a flat row, g<n>/app.sock in the instance run dir for a nested one
// (rev 19, rule 1). Only the directory differs from the flag-off launch:
// the file name, the rest of the argv and the agent-owned environment are
// the same (rule 2). A foreign row keeps its canonical socket either way:
// it is outside the two layouts the spec gives generation paths to, and
// keeping it is the conservative reading (plan 1, "Spec questions"). So
// does a flat row whose generation path would not fit a unix socket path
// (genLayout, rule 4).
//
// The stale-file rule (spec §7.1): before the app binds, the generation's
// directory is made and the file at THIS generation's own path is removed
// (left there by an earlier agent process, whose counter started at 1 like
// this one's). A sibling generation's path is never touched.
func (b *ExecBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
	return b.startGenFrom(ctx, rt, n, perGen, nil, cgroup.Limit{})
}

// StartGenExe is StartGen on the identity a wake recorded (settings spec
// §4.2): the same generation, socket and tier, but the process runs
// id.StagedPath, checked against id.Digest first. A copy that changed is a
// failed start, and nothing runs (never the live link instead). A flat or
// foreign row execs the file by path. A nested row asks root to bind it in
// place of the live link: the original run unit while GenContainment is
// off (generation 0), generation n's unit while it is on. The handle names
// id's build, path, digest, demand and tier, so a sidecar can bind the same
// file.
//
// The tier is id's own (GenTier, as StartStaged takes it; ruling R-TIER),
// never b.tier(rt): that reads whatever the live link names when it is
// asked, and root may have repointed the link since the record, which
// would run A's bytes under B's limits.
func (b *ExecBackend) StartGenExe(ctx context.Context, rt registry.Route, n uint64, perGen bool, id release.StagedIdentity) (Proc, Gen, error) {
	if err := release.VerifyStaged(id); err != nil {
		return nil, Gen{}, fmt.Errorf("%s: not started on %s: %w", rt.Host, id.Version, err)
	}
	tier := b.GenTier(rt, id.Version)
	proc, g, err := b.startGenFrom(ctx, rt, n, perGen, &id, tier)
	if err != nil {
		return proc, g, err
	}
	g.Build, g.StagedPath, g.Digest, g.Demand, g.Tier = id.Version, id.StagedPath, id.Digest, id.Demand, tier
	return proc, g, nil
}

// startGenFrom is StartGen's body. A nil id is the live link (StartGen); a
// non-nil one is the verified file StartGenExe names. tier is that file's
// own; zero is the live tier (b.tier).
func (b *ExecBackend) startGenFrom(ctx context.Context, rt registry.Route, n uint64, perGen bool, id *release.StagedIdentity, tier cgroup.Limit) (Proc, Gen, error) {
	var (
		flatExe string
		unitExe *instancectl.StagedExe
	)
	if id != nil {
		flatExe, unitExe = id.StagedPath, stagedExeOf(*id)
	}
	p := b.pathsFor(rt)
	sock := rt.Socket
	if perGen {
		switch genLayout(p.layout, rt.Host) {
		case LayoutFlat:
			sock = b.flatGenSocket(rt.Host, n)
		case LayoutNested:
			sock = nestedGenSocketIn(p.runDir, n)
		case LayoutForeign:
			if p.layout == LayoutFlat {
				log.Printf("activator: %s: a generation socket path would exceed %d bytes; generation %d keeps the canonical socket", rt.Host, maxUnixPath, n)
			}
		}
	}
	var (
		proc Proc
		ver  string
		err  error
	)
	switch p.layout {
	case LayoutNested:
		// prepareAt makes sock's generation directory and removes the
		// stale file at sock for the nested layout.
		proc, ver, err = b.startUnitAt(ctx, rt, sock, n, unitExe, tier)
	case LayoutFenced, LayoutConflict:
		return nil, Gen{}, fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
	default:
		if sock == rt.Socket {
			// A link left at the canonical path by an agent that ran with
			// generations on would make the app's bind fail. That holds for
			// every spawn on the canonical socket, not only a flag-off one:
			// a foreign row, or a flat row too long for a generation path
			// (rev 19 rule 4) whose box ran the old naming, binds it with
			// perGen true too (Astra, review of 19c07b2b), as StartStaged's
			// foreign arm already handles.
			if err := clearLeftoverCanonicalLink(rt.Socket); err != nil {
				return nil, Gen{}, fmt.Errorf("%s: leftover canonical link: %w", rt.Host, err)
			}
		}
		if sock != rt.Socket {
			if err := clearOwnGenSocket(sock); err != nil {
				return nil, Gen{}, fmt.Errorf("%s: generation socket: %w", rt.Host, err)
			}
		}
		proc, ver, err = b.startChildAt(ctx, rt, sock, n, flatExe, tier)
	}
	g := Gen{N: n, Socket: sock, Control: sock + ".control"}
	if err != nil {
		// The handle goes back with the error (P2-13): a spawn that failed
		// may have left generation n running, and its number is what the
		// caller proves it stopped by (KillGen). Refusals above spawned
		// nothing and answer no handle.
		return proc, g, err
	}
	g.Build = ver
	return proc, g, nil
}

// flatGenSocket is GenSocketPath, under RunDirOverride/g in a test.
func (b *ExecBackend) flatGenSocket(host string, n uint64) string {
	if b.RunDirOverride != "" {
		return genSocketPathIn(filepath.Join(b.RunDirOverride, "g"), host, n)
	}
	return GenSocketPath(host, n)
}

// clearOwnGenSocket makes the flat generation's own directory,
// /run/carlos/g/<n>, and removes a stale file at sock, and only at sock (rev
// 19, rule 3). The directory is 0750 and inherits /run/carlos's setgid
// carlos-sockets group, as the nested run dirs do.
func clearOwnGenSocket(sock string) error {
	if err := os.MkdirAll(filepath.Dir(sock), 0o750|os.ModeSetgid); err != nil {
		return err
	}
	if err := os.Remove(sock); err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}
	return nil
}

// startChildAt is Start's exec-child body (flat and foreign layouts):
// exec of the versioned binary directly, exactly as before nested layouts
// existed. sock is the app socket this spawn binds: rt.Socket, or a
// generation socket (StartGen). n is the generation this spawn is (StartGen's
// n; 0 from Start). It only places the spawn's cgroup, and only under
// GenContainment (placeFor).
//
// exe is "" for the live link, or the staged file a wake recorded
// (StartGenExe). spawnChildFrom leaves a staged spawn's cache purge to the
// swap's switch, but a wake's staged spawn is the host's only process, so
// its purge happens here, at the same moment, before the health wait.
// tier is exe's own; zero is the live tier (b.tier).
func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock string, n uint64, exe string, tier cgroup.Limit) (Proc, string, error) {
	p, spawnVer, err := b.spawnChildFrom(ctx, rt, sock, n, exe, tier)
	if err != nil {
		return p, "", err
	}
	if exe != "" && b.Cache != nil {
		if purged := b.Cache.PurgeHost(rt.Host); purged > 0 {
			log.Printf("activator: purged %d cached response(s) for %s on spawn", purged, rt.Host)
		}
	}
	if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil {
		stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		defer cancel()
		_ = p.Stop(stopCtx)
		if b.GenContainment && n != 0 {
			// The spawn was placed in g<n>, and Stop signals only the
			// leader: a process it forked can live on there (spec §7.3).
			// The generation's own kill ends it and proves the child
			// empty. Unproven, the process goes back with an error that
			// says so, and the caller keeps the lease until the sweep
			// proves it (P2-13, as startUnitAt does).
			if kerr := b.killGenAfterFailedStart(ctx, rt, n); kerr != nil {
				log.Printf("activator: %s: kill generation %d after a failed health wait: %v", rt.Host, n, kerr)
				return p, "", fmt.Errorf("%w; %w: kill generation %d after a failed health wait: %v", err, errGenUnproven, n, kerr)
			}
		}
		// The handle goes back with the health error, as startUnitAt's does
		// (P2-13). The leader's exit is not the end of what it started: with
		// no g<n>, a process it forked lives on in carlos-exec@<host>, and
		// the caller's failed-wake cleanup proves that cgroup empty before
		// the lease goes (Task 15 fix round 1, ruling a; Astra IMPL-3 F6).
		// The leader is already stopped, so the cleanup's Stop is a no-op.
		return p, "", err
	}
	b.normalizeSocketMode(rt, sock, p)
	return p, spawnVer, nil
}

// killGenAfterFailedStart is KillGen for a spawn path's own cleanup, on a
// budget the caller's context cannot cancel: a wake cancelled by a fence or
// Shutdown still proves what it started. nil is that proof.
func (b *ExecBackend) killGenAfterFailedStart(ctx context.Context, rt registry.Route, n uint64) error {
	kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
	defer cancel()
	if b.killGen != nil {
		return b.killGen(kctx, rt, n)
	}
	return b.KillGen(kctx, rt, n)
}

// spawnChildFrom is the flat spawn with the executable and the tier named,
// and nothing after the exec: it returns the moment the process runs. Empty
// exe is the live link, which is every wake (startChildAt adds today's
// health wait). A staged path is a swap's successor or a pinned wake
// (zero-downtime spec §7.5 step 3): the bytes the attempt chose, whatever
// root does to the live link meanwhile, with the caller owning the health
// deadline and the termination. tier is that artifact's own tier; zero is
// the live tier (b.tier). n is the generation number plan 2's containment
// places the child by.
func (b *ExecBackend) spawnChildFrom(ctx context.Context, rt registry.Route, sock string, n uint64, exe string, tier cgroup.Limit) (Proc, string, error) {
	host := rt.Host
	paths := b.pathsFor(rt)
	dbPath := paths.dbPath
	child, ownTier, err := b.placeFor(spawnInstance, n)
	if err != nil {
		return nil, "", err
	}

	// Exec exactly what a systemd unit would: the per-instance live symlink
	// (<OptDir>/live/<host> → a staged version), same --socket/--db flags,
	// same EnvironmentFile.
	bin := exe
	if bin == "" {
		bin = b.liveLink(host)
		if err := b.ensureLiveLink(ctx, host); err != nil {
			return nil, "", err
		}
	}

	// Mint the instance's admin token BEFORE the env is read for the spawn, so
	// the child is handed the same secret the agent will present back to it
	// (NextDue, the scheduled-work tick) and the same one its control socket
	// checks bearers against. A failure here fails the wake on purpose: an
	// instance with no token can neither answer /api/next-due nor accept a
	// tick, and a wake that quietly skipped it would report a healthy instance
	// that silently never runs its scheduled work — the #110 kind of lie.
	if err := EnsureAdminToken(paths.envFile); err != nil {
		return nil, "", fmt.Errorf("admin token for %s: %w", host, err)
	}
	// The instance key is minted here for the same reason and fails the
	// wake the same way: an app that takes its key from the platform
	// refuses to start without one (L21). It returns only once the key is
	// durable, so no child starts with a key a power cut could take back.
	if _, err := b.ensureInstanceKeyFor(ctx, rt, paths); err != nil {
		return nil, "", fmt.Errorf("instance key for %s: %w", host, err)
	}

	// One read gives the env and the layer its CARLOS_INSTANCE_KEY came
	// from, for the replica gate below (ruling 29).
	fileEnv, keyLayer, err := b.instanceEnvLayer(rt)
	if err != nil {
		return nil, "", err
	}

	// The control socket reaches the child through its ENVIRONMENT, never as a
	// CLI flag: every app binary built before this feature existed parses its
	// flags with flag.ExitOnError, which exits 2 on an unrecognized flag — so
	// passing --control-socket would refuse to start every app that hasn't
	// also shipped support for it, a fleet-wide outage on agent upgrade. An
	// environment variable a binary doesn't know about is simply ignored.
	controlSocket := sock + ".control"
	// Bind the control channel BEFORE the spawn, so CARLOS_CONTROL_SOCKET can
	// follow the OUTCOME rather than the intention. It used to be set from
	// "b.Cache != nil" and the listener started afterwards; now that the
	// channel exists for every exec child, the only thing that can make the
	// variable a lie is a bind that failed (windows has no secureUnixListen at
	// all, and a stale path can refuse), and binding first is what lets the
	// else-branch below strip the name in exactly that case.
	//
	// A bind failure is still NOT fatal to the wake — same reasoning as before,
	// the instance serves requests either way; it just goes without its control
	// verbs until the next restart, a stuck-stale cache self-heals via the
	// safety-net TTL, and a missed tick is retried by the runner.
	envPath := paths.envFile
	// Read on EVERY call, never captured: the env file can be rewritten under a
	// running instance (`carlos env`, a re-mint on a later spawn), and a token
	// cached at bind time would keep authorising the old secret.
	tokenReader := func() (string, error) {
		env, err := readEnvFile(envPath)
		if err != nil {
			return "", err
		}
		return env["CARLOS_ADMIN_TOKEN"], nil
	}
	var mount func(*http.ServeMux)
	if b.Timers != nil || b.UpdatesReader != nil {
		mount = func(mux *http.ServeMux) {
			if b.Timers != nil {
				schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.writeFence(rt.Host, n))
			}
			if b.UpdatesReader != nil {
				instanceupdates.Mount(mux, rt.Host, tokenReader, b.UpdatesReader)
			}
		}
	}
	ctl, err := edge.ListenControl(controlSocket, rt.Host, b.Cache, mount)
	if err != nil {
		log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err)
		ctl = nil
	}
	// A replica is started only with its home's key, verified, or with
	// none and a control socket its helper can refuse on (ruling 26).
	if err := b.replicaKeyGate(rt, fileEnv, keyLayer, ctl != nil); err != nil {
		if ctl != nil {
			ctl.Close()
		}
		return nil, "", err
	}
	// Both control-channel vars are AGENT-OWNED in BOTH directions: set when the
	// channel exists, and REMOVED when it does not. The removal half is not
	// decoration. fileEnv is tenant-supplied — <host>.config.env comes from the
	// bucket via `carlos env`, <host>.env is the instance's own file — so
	// without the else branches a tenant could put CARLOS_PURGE=header (or worse,
	// CARLOS_CONTROL_SOCKET=/some/path) in its own config and have it survive
	// into the child on a box with NO cache. The app would then be told a purge
	// channel exists when none does: an app that trusts the var skips its own
	// invalidation and serves stale content forever, and a socket path a tenant
	// chose is a name the agent never vouched for.
	//
	// "A variable naming a channel must not be set where that channel does
	// nothing" still holds — the two names just answer to different channels
	// now. CARLOS_CONTROL_SOCKET names the SOCKET, which exists for every exec
	// child once the bind above succeeded (the purge verb is only one of the
	// things served on it; the scheduled-work verbs are served there on a box
	// with no cache at all). CARLOS_PURGE names the PURGE channel, which still
	// exists only where there is a cache to purge. Both are set from what was
	// actually created, and stripped otherwise; neither is set from a tenant's
	// say-so.
	//
	// CARLOS_PURGE is ADVISORY: it tells the app that this box's edge reads the
	// Carlos-Purge response header (internal/edge/purge_header.go), so the app
	// can log or branch on it — but the channel works whether or not the app
	// ever reads the variable, because a header the edge understands is inert
	// when the app never emits it and honoured when it does. That is what makes
	// the rollout ordering harmless. It does not replace the socket channel;
	// both are live for exec-backed instances, and converging them is deferred
	// (spec §10.3).
	childVars := b.agentOwnedVars(rt, controlSocket, fileEnv, ctl != nil)
	// Record the build we're about to exec — read the symlink immediately before
	// the start so it names the version the kernel will resolve. Reading it here,
	// rather than back in wake() after the health wait, closes a real window: a
	// concurrent ship's repoint during the up-to-30s wait would otherwise make
	// us record the NEW target while this child runs the OLD binary, and the
	// sweep would then think it's already converged and never cycle it. The
	// residual sub-exec race (a repoint between this read and the start) is in
	// the safe direction — we'd record the older version, so the sweep cycles
	// once too often rather than never. Empty on a read error: the sweep won't
	// cycle.
	//
	// A staged spawn names its build from the attempt; the caller fills
	// Gen.Build. Reading the live link here would name whatever root
	// installed, not what this process runs.
	//
	// With LiveExecGate set (a declined box that honours failure records or
	// hold-ran markers, Task 9), the live link is resolved ONCE here, path
	// only, with no hash (release.ResolveLivePath, fix round 1 ruling 4b),
	// the gate is asked about that build at the
	// launch, and the staged file the link named is what is exec'd, not the
	// link. So the build the gate passed is the build that runs: a repoint
	// between the gate and the exec changes nothing, where exec'ing the link
	// would run whatever it named by then (plan review #6). A link that
	// cannot be resolved fails the wake, as a missing one does. With the gate
	// nil this is today's read and today's exec of the link.
	spawnVer := ""
	var liveGate func() error
	if exe == "" && b.LiveExecGate != nil {
		id, err := release.ResolveLivePath(b.OptDir, host)
		if err != nil {
			if ctl != nil {
				ctl.Close()
			}
			return nil, "", fmt.Errorf("%s: %w", host, err)
		}
		bin, spawnVer = id.StagedPath, id.Version
		liveGate = func() error { return b.LiveExecGate(rt, id.Version) }
	} else if exe == "" {
		spawnVer, _ = b.LiveVersion(host)
	}
	// Spawned through the cgroup manager (nil-tolerant: no delegation means
	// a plain start, exactly as before) so the child lands in
	// carlos-exec@<host> and the sampler can account it. build may run
	// twice — a flat start's PID-move fallback after a failed placement —
	// so everything about the Cmd is configured inside it.
	//
	// The caller's exec gate (final review F1) is the manager's before-start
	// hook (launchGate): asked after every step above (the token, the config
	// read, the control socket's bind) AND the cgroup preparation (the
	// parent, the child, its limits, and the shim's placement), immediately
	// before each exec, the flat PID-move fallback's included. The lease, a
	// fence or the attempt can change during any of it. A refusal runs
	// nothing of the command; the manager drops
	// the child cgroup and the error path below unbinds the listener, as for
	// a failed spawn. With no gate on ctx (every live-link spawn) there is no
	// hook and the spawn is exactly as it was, unless LiveExecGate is set:
	// then its check of the resolved build (liveGate) is the hook, asked at
	// the same point.
	cmd, settle, err := b.spawnInCgroupAtTier(rt, child, ownTier, tier, func() *exec.Cmd {
		cmd := exec.Command(bin, "--socket", sock, "--db", dbPath)
		cmd.Env = childEnv(childVars)
		// The child logs through the agent's own stderr/stdout (journald
		// under systemd); there's no per-instance unit to own a log stream
		// anymore. With Logs set the same bytes are ALSO teed line-wise
		// into the log sink (`carlos logs`); teeStd owns the WaitDelay
		// that piping entails.
		b.teeStd(cmd, rt)
		return cmd
	}, b.launchGateWith(ctx, host, liveGate))
	if err != nil {
		// The listener was bound before the spawn; nothing will ever answer on
		// it now, so unbind it rather than leaving a live socket for a process
		// that does not exist.
		if ctl != nil {
			ctl.Close()
		}
		return nil, "", err
	}

	p := &childProc{cmd: cmd, done: make(chan struct{}), ctl: ctl}
	// A staged successor spawns while the incumbent still serves, and its
	// purge happens at the switch (GenSwitched), not here.
	if b.Cache != nil && exe == "" {
		// A fresh process for this host is exactly the moment its cached
		// content becomes suspect: a deploy/redeploy means new code and
		// potentially new pages, and nothing else would invalidate them until
		// their TTL ran out. Drop the whole host's entries before the new
		// process serves its first request.
		if n := b.Cache.PurgeHost(rt.Host); n > 0 {
			log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
		}
	}
	go func() {
		p.err = cmd.Wait()
		// The member-facing record of the exit: a crash shows an "exited"
		// line with no "stopping" line before it — that pairing, not the
		// exit code alone, is what tells a member crash from lifecycle.
		if p.err != nil {
			emitEvent(b.Logs, rt, "warn", "instance exited: "+RedactPaths(p.err.Error()))
		} else {
			emitEvent(b.Logs, rt, "info", "instance exited (exit 0)")
		}
		// A strict child's group is drained and closed before anything
		// reads this exit as a stop (p.done closes below).
		settle()
		b.Cgroups.RemoveChild(host, child) // no-op for "" (flag off)
		// Last exiter wins: rmdir refuses while the sidecar (same host
		// cgroup) is still alive, and that failure is tolerated.
		b.cleanupHost(host)
		close(p.done)
	}()
	return p, spawnVer, nil
}

// agentOwnedNames are the variables agentOwnedVars decides: it sets or
// removes each of them on every launch, whatever the instance's files say.
// moves.AgentOwnedVars is the same set (TestAgentOwnedVarsAreMovesBoxOwned
// pins both against agentOwnedVars itself).
var agentOwnedNames = []string{"CARLOS_CONTROL_SOCKET", "CARLOS_PURGE", "CARLOS_REGION", "CARLOS_HOST"}

// agentOwnedVars composes the child's agent-owned environment variables,
// namely the control socket, the purge channel, and the region, on top of
// fileEnv, exactly as startChild did inline before the unit spawn path
// needed the same composition. Extracted so the two paths cannot drift
// (design §7.2): haveControl is whether this call's control listener
// actually bound. childVars must follow the OUTCOME of the bind, not the
// intention, per the long comment this replaced, still true here.
func (b *ExecBackend) agentOwnedVars(rt registry.Route, control string, fileEnv map[string]string, haveControl bool) map[string]string {
	childVars := fileEnv
	if haveControl {
		childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", control)
	} else {
		childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET")
	}
	if b.Cache != nil {
		childVars = withVar(childVars, "CARLOS_PURGE", "header")
	} else {
		childVars = withoutVars(childVars, "CARLOS_PURGE")
	}
	if b.Region != "" {
		childVars = withVar(childVars, "CARLOS_REGION", b.Region)
	} else {
		childVars = withoutVars(childVars, "CARLOS_REGION")
	}
	// The hostname, always, from the row (discussion #256 item 2). An app
	// used to learn it from its socket's file name, /run/carlos/<host>.sock,
	// and the nested layout's app.sock names nothing; three instances
	// crashed at start over that on the first roll. Agent-owned like the
	// three above: a tenant's env file does not get to tell an app it is
	// some other host.
	return withVar(childVars, "CARLOS_HOST", rt.Host)
}

// seedVars is the seed's environment over fileEnv: the control-channel
// variables stripped (a seed has no listener for either; see seedChild's
// comment), and CARLOS_HOST set, since a seed initialising a host's
// database is as entitled to know which host as the instance is. One
// composer for both seed paths, as agentOwnedVars is for both run paths.
func seedVars(fileEnv map[string]string, host string) map[string]string {
	return withVar(withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE"), "CARLOS_HOST", host)
}

// normalizeSocketMode is the PLATFORM's 0660 on a spawned instance's socket
// (issue #103), not the app's. Extracted from startChild so the unit path
// (whose own bind is the trampoline's, not this process's exec) applies the
// identical rule rather than a second copy that can drift. So a deployment's
// enrolled cross-user consumers (the carlos-sockets group, group-inherited
// from the setgid /run/carlos, see host.tmpfilesConf) can dial, no matter
// what the child's umask or its own post-bind chmod produced. titogo
// tightened its socket to 0600 and stranded the sibling router that proxies
// to it. Called after the health wait, so the app's own bind/chmod has
// already happened and this wins.
//
// The two guards make "only a socket THIS backend's child bound" true
// rather than merely intended. waitHealthy tolerates a squatter (its own
// contract: it only dials), so a child that lost the bind race with a
// stray supervisor (EADDRINUSE, exits) can still see the health wait
// succeed off the SQUATTER's answers; a dead child means the socket is
// not ours to widen (that route is the #17 external-socket advisory's
// territory). The Lstat keeps the chmod from ever following a
// non-socket: a symlink planted at the route's path must not become a
// mode change somewhere else. On a box with no consumers group the
// socket's group is just `carlos`, so the wider mode grants nobody
// anything. All three branches log and never fail the wake: a box with
// no consumers must not lose a healthy wake to a normalization it
// doesn't need.
func (b *ExecBackend) normalizeSocketMode(rt registry.Route, sock string, proc Proc) {
	host := rt.Host
	if proc.Exited() {
		log.Printf("activator: %s: child exited during the health wait (external process serving its socket?) — not normalizing the socket mode", host)
	} else if fi, statErr := os.Lstat(sock); statErr != nil || fi.Mode()&os.ModeSocket == 0 {
		log.Printf("activator: %s: %s is not a socket we can normalize (stat: %v)", host, sock, statErr)
	} else if err := os.Chmod(sock, 0o660); err != nil {
		log.Printf("activator: normalize socket mode for %s: %v (enrolled consumers would get permission denied)", host, err)
	}
}

// spawnUnitFrom is the nested spawn with the executable and the tier named,
// and nothing after root has started the unit: it returns the moment the
// unit runs, as spawnChildFrom does for a flat host. A nil exe is the live
// link, which is every wake (startUnitAt adds today's health wait). A staged
// exe is a swap's successor or a pinned wake (zero-downtime spec §7.5 step
// 3): root validates the path and mounts it in place of the live link, so
// whatever happens to the live link meanwhile changes nothing, and the
// caller owns the health deadline and the termination. tier is that
// artifact's own tier; zero is the live tier (b.tier). With LiveExecGate
// set a nil exe pins the live link (declinedLivePin) and starts that file as
// a staged one, at its own tier.
//
// n is the generation this spawn is (StartGen's n; 0 from Start). Under
// GenContainment it selects the carlos-instance-gen@<id>:<n> unit and its
// generation-keyed files (unitGen); otherwise it is ignored. A failure
// whose kill could not prove the unit empty wraps errGenUnproven.
func (b *ExecBackend) spawnUnitFrom(ctx context.Context, rt registry.Route, sock string, n uint64, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, string, error) {
	// The launch's own split, for the journal (wake latency, pass 2).
	var steps wakeSteps
	steps.begin(time.Now())
	if err := b.nestedGate(rt); err != nil {
		return nil, "", err
	}
	gen, err := b.unitGen(n)
	if err != nil {
		return nil, "", err
	}
	pinned := ""
	if exe == nil && b.LiveExecGate != nil {
		// A declined box that honours failure records or hold-ran markers
		// (Task 9) pins the live link once and names that file to root at
		// its own tier (declinedLivePin), so root binds the build the gate
		// passed and a repoint after this read changes nothing. Before
		// anything is prepared: a refusal binds no listener and writes no
		// launch file.
		id, err := b.declinedLivePin(ctx, rt)
		if err != nil {
			return nil, "", err
		}
		exe, tier, pinned = stagedExeOf(id), b.GenTier(rt, id.Version), id.Version
	} else if exe == nil {
		// A staged exe is mounted in place of the live link, which then
		// need not exist (as spawnChildFrom does for a flat host).
		if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
			return nil, "", err
		}
	}
	steps.mark(time.Now(), "build")
	var prep wakeSteps
	p, ctl, logs, fileEnv, keyLayer, err := b.prepareKeyedAt(rt, instancectl.RoleRun, sock, gen, &prep)
	if err != nil {
		return nil, "", err
	}
	fail := func(err error) (Proc, string, error) {
		if ctl != nil {
			ctl.Close()
		}
		logs.Close()
		return nil, "", err
	}
	childVars := b.agentOwnedVars(rt, sock+".control", fileEnv, ctl != nil)
	argv := []string{b.liveLink(rt.Host), "--socket", sock, "--db", p.dbPath}
	steps.mark(time.Now(), "prepare ("+prep.String()+")")
	if err := b.writeLaunchFor(p, instancectl.RoleRun, gen, rt.Host, argv, childEnv(childVars)); err != nil {
		return fail(err)
	}
	steps.mark(time.Now(), "launch file")
	// Read the symlink immediately before the start, same fail-safe
	// direction as startChild: a concurrent repoint at worst records the
	// older version (one extra convergence cycle), never the newer. A
	// pinned start names the build it pinned, which is what root binds.
	spawnVer := pinned
	if spawnVer == "" {
		spawnVer, _ = b.LiveVersion(rt.Host)
	}
	// The caller's exec gate (final review F1), right before root is asked
	// to start the unit: the config read, the listeners and the launch file
	// above all take time. A refusal starts nothing; the generation's own
	// launch file (it carries the instance's environment) goes, and the
	// listeners close, as after root's own refusal (ErrStagedExe below).
	// With no gate on ctx (every live-link spawn) this does nothing.
	if b.launchHook != nil {
		b.launchHook(rt.Host)
	}
	err = runExecGate(ctx)
	if err != nil {
		if gen != 0 {
			_ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen)))
		}
		return fail(err)
	}
	steps.mark(time.Now(), "gate")
	inv, err := b.startRole(ctx, rt, instancectl.RoleRun, gen, exe, tier)
	steps.mark(time.Now(), "root start")
	// Past this call root may have started the unit: every failure below
	// but ErrStagedExe (root's refusal before it laid out or started
	// anything) and errNoStagedExeControl (root was never asked: the
	// channel cannot name the file) is marked startUncertain, since the
	// build may have run before it failed or exited (final review wave 2,
	// N4).
	if errors.Is(err, errNoStagedExeControl) {
		// Nothing was sent to root. Proven not run, so never a failure
		// record. A generation's own launch file goes as after a refusal.
		if gen != 0 {
			_ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen)))
		}
		return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
	}
	if errors.Is(err, instancectl.ErrStagedExe) {
		// Root refused the staged build before it laid out or started
		// anything, so there is nothing to kill and nothing unproven. The
		// generation's own launch file, which this start wrote and which
		// carries the instance's environment, goes too (final review T9).
		// The legacy launch.run.json (gen 0) is shared with the original
		// template and is left as it is.
		if gen != 0 {
			_ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen)))
		}
		return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
	}
	if gen != 0 && (err != nil || inv == "") {
		if err == nil {
			// The unit ran and exited before root observed it: the build
			// crashed at its start (errStartedThenExited, ruling 206).
			err = exitedUnobserved{unit: unitName(instancectl.RoleRun, rt.Host, gen)}
		}
		if kerr := b.killFailedGen(ctx, rt, gen); kerr != nil {
			// Nothing is released: the generation may still be running.
			return nil, "", markStartUncertain(fmt.Errorf("%s: start unit: %w; %w: %v", rt.Host, err, errGenUnproven, kerr))
		}
		_ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen)))
		return fail(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err)))
	}
	if err != nil {
		// The original run unit (generation 0). Root was asked to start
		// it, and a start job systemd queued (behind an edge restart, say)
		// outlives this call's cancellation, so the wake's cleanup stops
		// the unit and proves it gone before the lease goes (#292).
		return fail(markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err))))
	}
	if inv == "" {
		// The server read the invocation after the start and found none:
		// the unit already exited (plan 2 final review). Nothing is
		// running, so there is nothing to stop; the wake failed.
		return fail(markStartUncertain(fmt.Errorf("%s: %w", rt.Host, exitedUnobserved{unit: unitName(instancectl.RoleRun, rt.Host, gen)})))
	}
	started := tier
	if started == (cgroup.Limit{}) {
		started = b.tier(rt)
	}
	log.Printf("activator: %s: started %s memory.max=%d (%s)", rt.Host, unitName(instancectl.RoleRun, rt.Host, gen), started.MemoryMax, steps.String())
	up := b.newUnitProcFor(rt, instancectl.RoleRun, gen, p, inv, ctl, logs)
	// The file environment this generation launched with, kept on its
	// handle: a run against it composes its environment from this, never
	// from the files config sync has delivered since (merge ruling M3).
	// It carries the generation's CARLOS_INSTANCE_KEY, minted or
	// configured before this launch, and the layer that key came from is
	// kept with it, decided by the same read (instance-key ruling 29), so
	// a run's command passes the replica gate on exactly what it is given.
	up.launchEnv, up.launchKeyLayer = maps.Clone(fileEnv), keyLayer
	if b.Cache != nil {
		// A fresh process for this host is exactly the moment its cached
		// content becomes suspect, same rule as startChild.
		if n := b.Cache.PurgeHost(rt.Host); n > 0 {
			log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
		}
	}
	return up, spawnVer, nil
}

// startUnitAt is Start's nested body (design §6): the same admin token, env
// composition and health gate as startChild, but the process itself is
// started by the root instance-control channel from a launch file rather
// than exec'd directly. The agent has no privilege to enter the tenant's
// namespace, only the control service does. It is spawnUnitFrom, then the
// health wait. exe is nil for the live link, or the staged file a wake
// recorded (StartGenExe), which root binds in place of the link. tier is
// exe's own; zero is the live tier (b.tier).
func (b *ExecBackend) startUnitAt(ctx context.Context, rt registry.Route, sock string, n uint64, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, string, error) {
	began := time.Now()
	p, spawnVer, err := b.spawnUnitFrom(ctx, rt, sock, n, exe, tier)
	if err != nil {
		return p, "", err
	}
	up := p.(*unitProc)
	launched := time.Now()
	if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil {
		// Set before the Stop call below, closing the same race Stop's own
		// internal set leaves open by a few instructions: observe runs
		// concurrently from the moment newUnitProc started it, and setting
		// this here rather than waiting for it to happen inside up.Stop
		// means the observer can never see a foreign invocation appear and
		// fire its own redundant stop for a unit we already know we are
		// about to stop.
		up.stopRequested.Store(true)
		stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		defer cancel()
		if serr := up.Stop(stopCtx); serr != nil {
			// The health error, not this one, is the wake's failure. The
			// observer stands down on its own once the stop actually
			// lands (stopRequested above), so an ErrStillRunning here
			// just means it had not landed yet within actStopTimeout, not
			// that the wake should be retried differently.
			log.Printf("activator: %s: stop after failed health wait: %v", rt.Host, serr)
			// The unit may still run, so its process goes back with the
			// health error and the caller's failed-wake cleanup proves it
			// stopped before the lease goes (P2-13). Under GenContainment
			// the error also says so (errGenUnproven), so that cleanup
			// asks for the proof with generations off as well.
			if up.gen != 0 {
				return up, "", fmt.Errorf("%w; %w: stop after failed health wait: %v", err, errGenUnproven, serr)
			}
			return up, "", err
		}
		return nil, "", err
	}
	// The spawn's own split, for the journal: the unit's launch (instance
	// control's half included) and then the app's boot to its socket.
	log.Printf("activator: %s: %s launched in %s, healthy %s later", rt.Host, unitName(instancectl.RoleRun, rt.Host, n),
		launched.Sub(began).Round(time.Millisecond), time.Since(launched).Round(time.Millisecond))
	b.normalizeSocketMode(rt, sock, up)
	return up, spawnVer, nil
}

// LegacyRunGone stops rt's original run unit (carlos-instance@, generation
// 0) and proves it gone (#292, Astra P1): the failed-wake cleanup's proof
// before an aborted start's lease goes back. The stop goes through
// instance-control's /stop, which runs `systemctl stop`. systemd keeps one
// job per unit and a stop job replaces a queued start job, so a successful
// stop also means that no start job is still queued, even while the edge
// restart that queued it is in progress. Then the unit must read inactive
// (or failed) with its cgroup empty or gone, the facts the exit observer
// trusts (observe, design §6.5). Bounded by ctx, which the caller gives a
// budget of its own; anything short of the proof is an error, and the
// caller keeps the lease.
func (b *ExecBackend) LegacyRunGone(ctx context.Context, rt registry.Route) error {
	if b.Control == nil {
		return fmt.Errorf("%s: instance control channel not configured", rt.Host)
	}
	unit := unitName(instancectl.RoleRun, rt.Host, 0)
	if err := b.Control.Stop(ctx, rt.Host, instancectl.RoleRun); err != nil {
		return fmt.Errorf("stop %s: %w", unit, err)
	}
	state := b.UnitState
	if state == nil {
		state = systemctlUnitState
	}
	events := b.CgroupEvents
	if events == nil {
		events = cgroupEventsUnder(b.cgroupRoot())
	}
	poll := b.ObservePoll
	if poll <= 0 {
		poll = time.Second
	}
	slice := instancectl.SliceName(rt.Host)
	var last error
	for {
		active, _, err := state(ctx, unit)
		switch {
		case err != nil:
			last = fmt.Errorf("unit state: %w", err)
		case active != "inactive" && active != "failed":
			last = fmt.Errorf("unit is %s", active)
		default:
			exists, populated, eerr := events(slice, unit)
			switch {
			case eerr != nil:
				last = fmt.Errorf("cgroup.events: %w", eerr)
			case exists && populated:
				last = errors.New("processes remain in its cgroup")
			default:
				return nil
			}
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("%s not proven stopped: %v", unit, last)
		case <-time.After(poll):
		}
	}
}

// killFailedGen ends a generation whose start failed or answered no
// invocation. The start may have succeeded with its answer lost, so the
// generation is killed and proven empty (StopGen with kill) on a budget the
// caller's context cannot cancel. nil is that proof. With no generation
// channel StartGen was never sent, so nothing was started.
func (b *ExecBackend) killFailedGen(ctx context.Context, rt registry.Route, gen uint64) error {
	gc, err := b.genControl()
	if err != nil {
		return nil
	}
	kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
	defer cancel()
	if err := gc.StopGen(kctx, rt.Host, gen, true); err != nil {
		log.Printf("activator: %s: kill %s after a failed start: %v", rt.Host, unitName(instancectl.RoleRun, rt.Host, gen), err)
		return err
	}
	return nil
}

// nestedGate is the pre-flight check every nested spawn path runs first: no
// root instance-control channel configured, or the box itself is not fit
// to run an isolated instance (NestedOK's own mounts and /ready check),
// refuses before anything else runs. Shared so seedUnit and
// startSidecarUnit gate on NestedOK exactly like startUnit does.
func (b *ExecBackend) nestedGate(rt registry.Route) error {
	if b.Control == nil {
		return fmt.Errorf("%s: instance control channel not configured", rt.Host)
	}
	if b.NestedOK != nil {
		if err := b.NestedOK(); err != nil {
			return fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err)
		}
	}
	return nil
}

// instanceEnv composes the host's two EnvironmentFile layers, both optional
// like a unit's "-" prefix: the sync-owned bucket config (<host>.config.env)
// underneath, the instance's own file (<host>.env) on top — it wins conflicts,
// see mergeEnv. Shared by Start and Seed precisely so a seed sees the SAME
// environment the instance will: a seed that read different credentials, or a
// different database URL, than the process it initializes would be a trap.
// (Start layers its agent-owned control-channel vars on top of this; those
// belong to a serving process, not to a one-shot init. Both callers mint the
// instance's CARLOS_ADMIN_TOKEN into <host>.env first, so "the same
// environment" includes the same secret.)
func (b *ExecBackend) instanceEnv(rt registry.Route) (map[string]string, error) {
	cfgEnv, ownEnv, err := b.instanceEnvParts(rt)
	if err != nil {
		return nil, err
	}
	return mergeEnv(cfgEnv, ownEnv), nil
}

// instanceEnvParts is instanceEnv's two reads, before they are merged:
// the sync-owned config layer and the instance's own file. Extracted so
// LaunchEnvironment can see which names the own file shadows; instanceEnv
// is exactly mergeEnv over these two, as it was.
func (b *ExecBackend) instanceEnvParts(rt registry.Route) (cfgEnv, ownEnv map[string]string, err error) {
	paths := b.pathsFor(rt)
	// A nested host's files live under a tenant-writable directory (design
	// §3.2): read them without following a symlink. A flat host's files sit
	// outside any tenant's reach, so the plain read stays exactly as it was.
	read := readEnvFile
	if paths.layout == LayoutNested {
		read = func(path string) (map[string]string, error) {
			return readEnvFileBeneath(filepath.Dir(path), filepath.Base(path))
		}
	}
	cfgEnv, err = read(paths.configEnv)
	if err != nil && !os.IsNotExist(err) {
		return nil, nil, fmt.Errorf("read config env file: %w", err)
	}
	ownEnv, err = read(paths.envFile)
	if err != nil && !os.IsNotExist(err) {
		return nil, nil, fmt.Errorf("read env file: %w", err)
	}
	return cfgEnv, ownEnv, nil
}

// LaunchEnvironment is the composed environment the next Start of rt
// would spawn with: childEnv over agentOwnedVars over instanceEnv, the
// exact composition Start uses (agentOwnedVars(rt, fileEnv, haveControl)),
// decoded back to a map. agentOwnedVars REMOVES tenant-provided
// CARLOS_CONTROL_SOCKET, CARLOS_PURGE and CARLOS_REGION when the agent
// does not provide them, and always sets CARLOS_HOST, so a report over
// mergeEnv alone would misstate the environment. haveControl is true: startChild calls
// edge.ListenControl for EVERY exec child, and b.Control is the nested
// root-control client, which says nothing about the flat channel, so
// before the spawn the honest composition is "the agent will bind it"; a
// bind that fails at spawn is logged there and the probe's spawn-time
// check compares the actual outcome. CARLOS_PURGE follows b.Cache != nil
// exactly as agentOwnedVars does.
//
// shadowed is the sorted names present in both the rendered configuration
// file and the instance's own <host>.env: the override mergeEnv applies
// (moves spec §7.3 T3 reports it), collected here because this is the only
// layer that sees both files. The one exception is mergeEnv's: an empty
// CARLOS_INSTANCE_KEY in the own file overrides nothing, so it is not
// listed. The move's launch report (§7.3, codex R2-8)
// is computed over exactly this env, so a target compares what would run,
// not a reconstruction of it. The values are raw: the caller HMACs them
// and must never log or return them.
func (b *ExecBackend) LaunchEnvironment(rt registry.Route) (map[string]string, []string, error) {
	env, _, shadowed, err := b.LaunchLayers(rt)
	return env, shadowed, err
}

// LaunchLayers is LaunchEnvironment with the names whose running value the
// instance's own layers supplied: every name of the rendered configuration
// file and the instance's own <host>.env, as mergeEnv merged them, except
// the names agentOwnedVars decides (agentOwnedNames), whose running value
// is the agent's whatever the files say. A name in the composed env that is
// not in instance came from the agent: its own variables or its process
// environment (childEnv). The move's env pin needs the difference
// (moves.ReportLaunch): a box variable such as HOSTNAME is the box's when
// inherited and the instance's when its files set it.
func (b *ExecBackend) LaunchLayers(rt registry.Route) (env map[string]string, instance map[string]bool, shadowed []string, err error) {
	cfgEnv, ownEnv, err := b.instanceEnvParts(rt)
	if err != nil {
		return nil, nil, nil, err
	}
	for k := range ownEnv {
		if _, rendered := cfgEnv[k]; rendered {
			if k == instancekey.Var && ownEnv[k] == "" {
				continue // an empty own key shadows nothing (mergeEnv)
			}
			shadowed = append(shadowed, k)
		}
	}
	sort.Strings(shadowed)
	fileEnv := mergeEnv(cfgEnv, ownEnv)
	instance = make(map[string]bool, len(fileEnv))
	for k := range fileEnv {
		instance[k] = true
	}
	for _, k := range agentOwnedNames {
		delete(instance, k)
	}
	out := map[string]string{}
	for _, kv := range childEnv(b.agentOwnedVars(rt, rt.Socket+".control", fileEnv, true)) {
		if i := strings.IndexByte(kv, '='); i > 0 {
			out[kv[:i]] = kv[i+1:]
		}
	}
	return out, instance, shadowed, nil
}

// childEnv is the child's complete environment: the agent's own, minus
// every CARLOS_* variable, with the instance's file values layered on top.
// One entry per key, last value wins: os/exec deduplicates cmd.Env that
// way, but on the unit path the trampoline hands this list to execve as
// is, duplicates and all, and a Go child then keeps the FIRST occurrence
// (design §6.2, codex R1-16). Composing it deduplicated here keeps the
// two spawn paths identical.
func childEnv(fileEnv map[string]string) []string {
	order := make([]string, 0, 32)
	values := make(map[string]string, 32)
	set := func(k, v string) {
		if _, seen := values[k]; !seen {
			order = append(order, k)
		}
		values[k] = v
	}
	for _, kv := range os.Environ() {
		if strings.HasPrefix(kv, "CARLOS_") {
			continue
		}
		k, v, ok := strings.Cut(kv, "=")
		if !ok {
			continue
		}
		set(k, v)
	}
	keys := make([]string, 0, len(fileEnv))
	for k := range fileEnv {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		set(k, fileEnv[k])
	}
	env := make([]string, 0, len(order))
	for _, k := range order {
		env = append(env, k+"="+values[k])
	}
	return env
}

// withVar returns a copy of env with key=value added — a copy, so the
// caller's map (an EnvironmentFile's parsed contents) is never mutated.
func withVar(env map[string]string, key, value string) map[string]string {
	out := make(map[string]string, len(env)+1)
	for k, v := range env {
		out[k] = v
	}
	out[key] = value
	return out
}

// withoutVars returns a copy of env with keys removed — the mirror of withVar,
// and the half that makes an agent-owned variable actually agent-owned. Absent
// it, "the agent decides whether this var is set" is only true when the agent
// sets it; a tenant supplying the same name in its own env file wins by
// default. Same copy discipline as withVar: the caller's map is never mutated.
//
// This package keeps its own agent-owned list inline (the two names above,
// at Start's control-channel branches) rather than sharing one with
// internal/appconfig's agentOwned (agentvars.go), which the unit-tenant sync
// path uses for the same purpose — the two lists must not drift apart.
func withoutVars(env map[string]string, keys ...string) map[string]string {
	out := make(map[string]string, len(env))
	for k, v := range env {
		out[k] = v
	}
	for _, k := range keys {
		delete(out, k)
	}
	return out
}

// mergeEnv composes the child's env-file layers: base (the sync-owned
// <host>.config.env, bucket-derived) under overlay (the instance's own
// <host>.env). The overlay wins every conflict but one — CARLOS_ADMIN_TOKEN
// lives only in the instance file and must never be clobberable from
// the bucket, and a box-local override deliberately beats central
// config. The exception: an empty CARLOS_INSTANCE_KEY in the overlay
// supplies no key, so a non-empty one in base stands (instancekey.Effective).
// Returns a fresh map; inputs are never mutated.
func mergeEnv(base, overlay map[string]string) map[string]string {
	out := make(map[string]string, len(base)+len(overlay))
	for k, v := range base {
		out[k] = v
	}
	for k, v := range overlay {
		out[k] = v
	}
	// An empty CARLOS_INSTANCE_KEY in the instance's own file supplies no
	// key, so it never shadows a configured one: the child is never handed
	// an empty key while a non-empty one exists (instance-key plan ruling
	// 13). instancekey.Effective is the rule the key's mint decides by.
	if key, _ := instancekey.Effective(base, overlay); key != "" {
		out[instancekey.Var] = key
	}
	return out
}

// HasSidecar: a host has a sidecar iff its sidecar EnvironmentFile exists.
// Derived from the filesystem on purpose, the way systemd units would do it:
// no registry column to drift from what's actually staged.
//
// This is the LEGACY file-staged co-spawn — Eleven runs on it today, so it
// is not going anywhere yet. The NATIVE path is a route with
// registry.Route.IsSidecar true: its own registry row, spawned by
// StartWorker instead of StartSidecar, with no parent instance and no file
// on disk. A host may carry one mechanism or the other but never both — the
// provisioning reconciler's mint-time check (sidecar_reconcile.go) refuses
// to mint a native row for a host this method already reports true for.
// Retiring this file path is a follow-up gated on Eleven's own migration to
// native sidecar records.
func (b *ExecBackend) HasSidecar(rt registry.Route) bool {
	_, err := os.Stat(b.pathsFor(rt).sidecarEnv)
	return err == nil
}

// StartSidecar dispatches by layout, the same way Start does.
func (b *ExecBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) {
	p := b.pathsFor(rt)
	switch p.layout {
	case LayoutNested:
		return b.startSidecarUnit(ctx, rt)
	case LayoutFenced, LayoutConflict:
		return nil, fmt.Errorf("%s: refusing to start sidecar: %s", rt.Host, p.reason)
	}
	return b.startSidecarChild(ctx, rt)
}

// startSidecarChild execs the host's live symlink with `sidecar run` under
// the sidecar's EnvironmentFile, CARLOS_*-clean like the instance spawn. No
// health wait — the sidecar is a pure outbound client with no socket to
// probe; its own connect/backoff loop absorbs a slow instance. The live
// symlink exists by now: the instance spawn checked it. The ctx is unused (an
// exec doesn't block); it's here for the backend seam's symmetry.
//
// This arm serves the LEGACY co-spawn only (see HasSidecar) — StartWorker is
// the native spawn path for a standalone registry.Route.IsSidecar row.
//
// With LiveExecGate set (a declined box that honours failure records or
// hold-ran markers; Task 9 fix round 1, ruling 1), the link is resolved once,
// without hashing (release.ResolveLivePath), the gate is asked about that
// build, and the file it resolved is exec'd, not the link. The legs leave a
// held instance running on its old build while root repoints the link, and
// every respawn of its sidecar (the wake tail, late attachment, the sweep's
// supervision) comes through here: exec'ing the link would run the held
// build beside it. A refusal starts nothing, and the supervisor's backoff
// keeps the sidecar down. With the gate nil it is today's exec of the link.
func (b *ExecBackend) startSidecarChild(ctx context.Context, rt registry.Route) (Proc, error) {
	if b.LiveExecGate == nil {
		return b.startSidecarChildFrom(ctx, rt, b.liveLink(rt.Host))
	}
	id, err := release.ResolveLivePath(b.OptDir, rt.Host)
	if err != nil {
		return nil, fmt.Errorf("%s: sidecar: %w", rt.Host, err)
	}
	if err := b.LiveExecGate(rt, id.Version); err != nil {
		return nil, err
	}
	return b.startSidecarChildFrom(ctx, rt, id.StagedPath)
}

// startSidecarChildFrom is startSidecarChild with the binary named. That is
// the live link for today's StartSidecar, and a generation's captured
// staged path for StartSidecarGen (zero-downtime spec §7.5). Placement is
// plan 2's: under GenContainment the sidecar lands in role-sidecar, with no
// limit of its own, under the generation parent's backstop (§7.3, plan 2
// D4). The parent holds no process and keeps memory.oom.group=0.
func (b *ExecBackend) startSidecarChildFrom(_ context.Context, rt registry.Route, bin string) (Proc, error) {
	fileEnv, err := readEnvFile(b.pathsFor(rt).sidecarEnv)
	if err != nil {
		return nil, fmt.Errorf("read sidecar env: %w", err)
	}
	// The sidecar runs the app binary with no control socket, so on a
	// replica host it is gated as the worker is (ruling 26): its home's key,
	// verified, or nothing runs. A key in its own file is checked too.
	if err := b.replicaKeyGate(rt, fileEnv, sidecarKeyLayer(fileEnv), false); err != nil {
		return nil, err
	}
	child, _, err := b.placeFor(spawnSidecar, 0)
	if err != nil {
		return nil, err
	}
	// Same host cgroup as the instance: a sidecar's load is that instance's
	// work, and attributing it elsewhere would be a lie in the other
	// direction.
	cmd, settle, err := b.spawnInCgroupAt(rt, child, false, func() *exec.Cmd {
		cmd := exec.Command(bin, "sidecar", "run")
		cmd.Env = childEnv(fileEnv)
		// Like the instance child: logs through the agent's own streams
		// (journald under systemd), teed into the log sink when armed — a
		// sidecar's output is the app's output.
		b.teeStd(cmd, rt)
		return cmd
	})
	if err != nil {
		return nil, err
	}
	p := &childProc{cmd: cmd, done: make(chan struct{})}
	go func() {
		p.err = cmd.Wait()
		settle()
		b.Cgroups.RemoveChild(rt.Host, child) // no-op for "" (flag off)
		b.cleanupHost(rt.Host)
		close(p.done)
	}()
	return p, nil
}

// StartWorker is StartWorkerStaged onto whatever live/<host> names now. Its
// string result is that build's version, as it always was.
func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra map[string]string) (Proc, string, error) {
	p, id, err := b.StartWorkerStaged(ctx, rt, extra, nil)
	return p, id.Version, err
}

// StartWorkerStaged execs a worker with the route's declared argv (rt.Args, a
// JSON array: a sidecar's `["sidecar","run"]` today) under the instance's own
// env layers, CARLOS_*-clean like every other child this backend spawns, and
// with NO health wait: a worker binds nothing, so there is nothing
// waitHealthyCtx could dial. extra (the per-wake sidecar token) is applied
// last and wins over the tenant's own <host>.env.
//
// What it execs is a staged build BY PATH, never live/<host> (zero-downtime
// deploys spec §7.9):
//   - onto == nil (a wake, or a converge onto a new build): the live link is
//     ensured, read ONCE, and the file it names is hashed and exec'd, once
//     LiveExecGate (when set) has passed its build;
//   - onto != nil (a crash or max_run recycle): onto.StagedPath is checked
//     against onto.Digest (release.VerifyStaged) and exec'd. A copy that has
//     changed is refused with release.ErrStagedCopyChanged and nothing runs.
//
// The identity returned is the build exec'd, the agent's serving identity for
// this child. What the child later reports over the agent socket is display
// metadata only.
//
// Between resolving that identity and the exec it runs ctx's before-exec hook
// (runBeforeExec), through which the activator publishes the build first; a
// hook error means nothing runs (ruling R19 item 3). ctx's exec gate
// (withExecGate), the caller's ownership check, is the cgroup manager's
// before-start hook (launchGate), as a flat instance spawn's is: asked after
// the cgroup's preparation and before each start attempt, and its refusal
// starts nothing and comes back as it is (Astra R3 §5 item 4).
//
// A worker on a replica row passes replicaKeyGate as a seed does, with no
// control socket, before that hook: its home's key, verified, or nothing
// runs (instance-key plan ruling 26). A worker never mints a key.
func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, extra map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
	var argv []string
	if err := json.Unmarshal([]byte(rt.Args), &argv); err != nil || len(argv) == 0 {
		return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: no runnable argv in args %q", rt.Host, rt.Args)
	}
	var id release.StagedIdentity
	if onto == nil {
		if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
			return nil, release.StagedIdentity{}, err
		}
		got, err := release.ResolveLive(b.OptDir, rt.Host)
		if err != nil {
			return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err)
		}
		id = got
		// A declined box that honours failure records or hold-ran markers
		// (Task 9) asks about the build just resolved, which is the build
		// exec'd (by its staged path), before it is published or anything is
		// prepared. A refusal runs nothing and comes back as it is. A recycle
		// onto the captured build (onto set) is no move onto the live link,
		// and is not asked.
		if b.LiveExecGate != nil {
			if err := b.LiveExecGate(rt, id.Version); err != nil {
				return nil, release.StagedIdentity{}, err
			}
		}
	} else {
		if err := release.VerifyStaged(*onto); err != nil {
			return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: not recycled onto %s: %w", rt.Host, onto.Version, err)
		}
		id = *onto
	}

	// One read gives the env and the layer its CARLOS_INSTANCE_KEY came
	// from, for the replica gate below (ruling 29). A worker never mints
	// (ruling 12): the key is whatever the env files already deliver.
	fileEnv, keyLayer, err := b.instanceEnvLayer(rt)
	if err != nil {
		return nil, release.StagedIdentity{}, err
	}
	for k, v := range extra {
		fileEnv = withVar(fileEnv, k, v)
	}
	// A worker has no control socket, so a replica row's worker is gated as
	// its seed is (ruling 26): its home's key, verified, or refused. extra is
	// agent-issued and has no business carrying a key; one that did would
	// not be the key the layer describes, so it is refused as of unknown
	// origin.
	if _, ok := extra[instancekey.Var]; ok {
		keyLayer = instancekey.LayerNone
	}
	if err := b.replicaKeyGate(rt, fileEnv, keyLayer, false); err != nil {
		return nil, release.StagedIdentity{}, err
	}

	// The identity is resolved and nothing has run: the activator publishes
	// the build now, before the exec, and a failure means no child (ruling
	// R19 item 3).
	if err := runBeforeExec(ctx, id); err != nil {
		return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err)
	}

	// Through spawnInCgroup, not Cgroups.Start directly: that seam is what
	// applies the route's resource tier, and a worker is a long-lived child
	// of a tenant's app like any other. Bypassing it compiles and silently
	// ships an uncapped process — the exact failure spawnInCgroup's own
	// comment names.
	//
	// Gated (spawnInCgroupGated): ownership can be lost while the cgroup is
	// prepared and the process placed, or between a failed placement and
	// its PID-move fallback, so the caller's gate is asked at each exec, not
	// once before all of it.
	// With no gate on ctx (an unplanned wake's spawn passes none) launchGate
	// is nil and this is spawnInCgroup exactly. The zero tier is
	// spawnInCgroup's too: the route's live tier (b.tier), as a worker has
	// always had.
	cmd, settle, err := b.spawnInCgroupGated(rt, cgroup.Limit{}, func() *exec.Cmd {
		cmd := exec.Command(id.StagedPath, argv...)
		cmd.Env = childEnv(fileEnv)
		b.teeStd(cmd, rt)
		return cmd
	}, b.launchGate(ctx, rt.Host))
	if err != nil {
		return nil, release.StagedIdentity{}, err
	}
	p := &childProc{cmd: cmd, done: make(chan struct{})}
	go func() {
		p.err = cmd.Wait()
		// Mirrors Start's exit record (#5 review, closed here): a worker's
		// crash is a sidecar's health word ("Crashed") the moment it lands,
		// so the member needs the SAME explanation an instance crash gets,
		// not silence. This fires once per actual exit, and the caller's
		// own doubling-backoff restart (2s up to a minute, activator.go)
		// already bounds how often an exit can happen at all — a
		// crash-looping worker cannot make this any noisier than the
		// restart log line it already earns internally.
		if p.err != nil {
			emitEvent(b.Logs, rt, "warn", "worker exited: "+RedactPaths(p.err.Error()))
		} else {
			emitEvent(b.Logs, rt, "info", "worker exited (exit 0)")
		}
		settle()
		b.cleanupHost(rt.Host)
		close(p.done)
	}()
	return p, id, nil
}

// startSidecarUnit is StartSidecar's nested body: a sidecar has no health
// wait either way, and no control listener of its own. Sidecar env is a
// single layer, read straight from its own file, never merged with the
// instance's two-layer config.
func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error) {
	return b.startSidecarUnitFrom(ctx, rt, nil, cgroup.Limit{})
}

// startSidecarUnitFrom is startSidecarUnit with the executable and the tier
// named. A nil exe is the live link (today's start), pinned instead on a box
// with LiveExecGate set (declinedLivePin). A staged exe is the serving
// generation's captured build (zero-downtime spec §7.5), which root
// validates before it lays out the mount, and tier that generation's own.
func (b *ExecBackend) startSidecarUnitFrom(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, error) {
	if b.Cfg.HostWrites.Enabled {
		return nil, ErrClaimIsolationUnavailable
	}
	if err := b.nestedGate(rt); err != nil {
		return nil, err
	}
	if exe == nil && b.LiveExecGate != nil {
		// A declined box that honours failure records or hold-ran markers
		// (Task 9 fix round 1, ruling 1) pins the live link once and names
		// that file to root at its own tier (declinedLivePin), as the nested
		// instance start does (spawnUnitFrom): root binds the build the gate
		// passed, never the link. Before the launch file is written, so a
		// refusal leaves nothing behind.
		id, err := b.declinedLivePin(ctx, rt)
		if err != nil {
			return nil, err
		}
		exe, tier = stagedExeOf(id), b.GenTier(rt, id.Version)
	}
	p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSidecar)
	if err != nil {
		return nil, err
	}
	argv := []string{b.liveLink(rt.Host), "sidecar", "run"}
	if err := b.writeLaunch(p, instancectl.RoleSidecar, rt.Host, argv, childEnv(fileEnv)); err != nil {
		logs.Close()
		return nil, err
	}
	var inv string
	if exe != nil {
		// The serving generation's captured staged build, which root
		// validates before it lays out the mount, at that generation's own
		// tier, never the live link's.
		gc, gerr := b.genControl()
		if gerr != nil {
			logs.Close()
			return nil, gerr
		}
		if tier == (cgroup.Limit{}) {
			tier = b.tier(rt)
		}
		inv, err = gc.StartSidecarExe(ctx, rt.Host, tier.MemoryMax, *exe)
	} else {
		inv, err = b.Control.Start(ctx, rt.Host, instancectl.RoleSidecar, b.tier(rt).MemoryMax)
	}
	if err != nil {
		logs.Close()
		return nil, fmt.Errorf("%s: start sidecar unit: %w", rt.Host, err)
	}
	if inv == "" {
		logs.Close()
		return nil, fmt.Errorf("%s: %s exited before it was observed", rt.Host, instancectl.UnitName(instancectl.RoleSidecar, rt.Host))
	}
	return b.newUnitProc(rt, instancectl.RoleSidecar, p, inv, nil, logs), nil
}

// socketClient returns an HTTP client that dials an instance's unix socket
// (request URLs use the placeholder host http://carlos).
func socketClient(socket string) *http.Client {
	return &http.Client{
		Timeout: 3 * time.Second,
		Transport: &http.Transport{
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				return unixdial.Checked(ctx, &net.Dialer{}, socket)
			},
		},
	}
}

// waitHealthy blocks until the instance answers an HTTP request on its socket,
// or the timeout elapses. It retries, so it's robust to the restart window
// (socket briefly absent, or a stale socket left by an abruptly-killed old
// process) — and a broken binary that never becomes healthy fails the wake.
func waitHealthy(socket string, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	client := socketClient(socket)
	var last error
	var poll time.Duration
	for time.Now().Before(deadline) {
		resp, err := client.Get("http://carlos/")
		if err == nil {
			resp.Body.Close()
			if resp.StatusCode < 500 {
				return nil
			}
			last = fmt.Errorf("unhealthy: HTTP %d", resp.StatusCode)
		} else {
			last = err
		}
		poll = nextHealthPoll(poll)
		time.Sleep(poll)
	}
	return fmt.Errorf("instance not healthy within %s: %v", timeout, last)
}

// The wait between health probes. A request is held open for the whole of
// a wake, and a fixed 150 ms gap cost it 75 ms on average after the app was
// already listening. A refused dial costs microseconds, so the wait starts
// short and doubles to a ceiling that keeps a slow boot from being probed
// in a busy loop.
const (
	healthPollMin = 5 * time.Millisecond
	healthPollMax = 50 * time.Millisecond
)

// nextHealthPoll is the wait after one of d; zero gives the first.
func nextHealthPoll(d time.Duration) time.Duration {
	return min(max(2*d, healthPollMin), healthPollMax)
}

// waitHealthyCtx is waitHealthy bounded by ctx as well as its own timeout, so
// an aborted wake doesn't keep probing a socket nobody wants anymore.
func waitHealthyCtx(ctx context.Context, socket string, timeout time.Duration) error {
	done := make(chan error, 1)
	go func() { done <- waitHealthy(socket, timeout) }()
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// readEnvFile parses a KEY=value EnvironmentFile (blank lines and #comments
// ignored).
func readEnvFile(path string) (map[string]string, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	return parseEnvBytes(raw), nil
}

// childProc is a spawned child: SIGTERM starts the instance's own graceful
// drain; SIGKILL only if it overstays.
type childProc struct {
	cmd  *exec.Cmd
	done chan struct{}
	err  error
	ctl  *edge.ControlListener // nil only if the pre-spawn bind failed
}

func (p *childProc) Exited() bool {
	select {
	case <-p.done:
		return true
	default:
		return false
	}
}

// Crashed answers the activator's procCrasher seam: a DEFINITELY non-zero
// exit status. Read from ProcessState rather than from p.err because a clean
// exit whose pipes outlive it comes back as exec.ErrWaitDelay (cmd.WaitDelay
// is set on every child here) — an error that says nothing about the exit
// status, and reading it as a crash would restart a worker that finished.
// Anything this cannot establish reads as NOT a crash, which is procCrashed's
// own default and for its reasons.
func (p *childProc) Crashed() bool {
	if !p.Exited() {
		return false
	}
	st := p.cmd.ProcessState // safe: written before done closed, read after
	return st != nil && !st.Success()
}

func (p *childProc) Stop(ctx context.Context) error {
	// The control listener OUTLIVES the drain, and the ordering is the whole
	// point. Closing it first — which is what this did — unbound and unlinked
	// the socket before the child had even seen SIGTERM, so an app that
	// registers or re-asserts its once-timers on the way down (spec §7.2's
	// lifetime decision — the socket brackets the PROCESS, bound before the
	// spawn and closed after the exit — serving §7.1's re-assert story) got
	// connection-refused on the channel the platform had just told it about,
	// during precisely the window it was told to use. Signal first, close
	// after the process is gone.
	//
	// Deferred, so it is unconditional on every path out of here — the
	// already-exited return, the lost-race return, the kill-timeout return,
	// and even a Signal error we hand back to the caller. A listener left
	// bound to a socket no process is behind is the stale-socket wake trap
	// (#129) waiting to happen, and this is the only close a childProc ever
	// gets — Start's own is on the path where the spawn failed and no
	// childProc was ever built.
	defer func() {
		if p.ctl != nil {
			p.ctl.Close()
		}
	}()
	if p.Exited() {
		return nil
	}
	if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
		if errors.Is(err, os.ErrProcessDone) { // lost the race with its own exit
			<-p.done
			return nil
		}
		return err
	}
	select {
	case <-p.done:
		return nil
	case <-ctx.Done():
		_ = p.cmd.Process.Kill()
		<-p.done
		return fmt.Errorf("instance did not exit on SIGTERM; killed")
	}
}

// movesPrefix is where a residence move stages a replica in the TARGET's
// data bucket (instance-move spec): moves/<host>/ beside the ordinary
// instances/ prefix. A free function so the host's move worker and this
// package spell it identically.
func MovesPrefix(host string) string { return "moves/" + host + "/" }

// MoveReceiptsKey is where a residence move's mirror keeps its copy
// receipts (#175, lease.Receipts) in the SOURCE box's own data bucket. The
// source cannot read the target's moves/ prefix (its grant there is
// push-only), so the receipts cannot live beside the staged copy. This key
// is under neither MovesPrefix nor the hibernate prefix, so no restore,
// promotion or staging cleanup ever sees it.
func MoveReceiptsKey(host string) string { return "move-receipts/" + host + ".json" }

// PurgeLocalDBArtifacts removes the local litestream footprint next to a
// DB — WAL, SHM and both meta dirs — and, when full is true, the DB file
// itself. Restore's pre-restore cleanup uses the partial form (the DB's
// absence is what got us here); a residence move's cut uses the full form:
// the activator's warm-cache trust is exactly right for crashes and exactly
// wrong after a cut, where a stale local file would resurrect a moved-away
// instance's history. The full form removes the database's <db>.seed
// marker too (final review carry item): a marker that outlived its
// database would name a seed for whatever database is put there later.
func PurgeLocalDBArtifacts(dbPath string, full bool) {
	for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
		_ = os.Remove(p)
	}
	_ = os.RemoveAll(dbPath + "-litestream")
	_ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
	if full {
		_ = os.Remove(dbPath)
		_ = os.Remove(dbPath + SeedMarkerSuffix)
	}
}

// promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this
// box's OWN bucket whenever anything is staged: staging wins when present,
// whatever the ordinary prefix holds (an exact mirror, so own keys the
// staging lacks are deleted). A no-op (nil error) when there is nothing
// staged.
func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]lease.MirrorObject, error) {
	// STAGING WINS when present (review round 1). A staged replica exists
	// only while an inbound move is landing or has just landed: it was
	// mirrored from the then-live data at cut, so it is by construction
	// newer than any lineage this box kept from a PREVIOUS residence — the
	// boomerang case (ie→uk→ie) that "own lineage never loses" restored
	// months-stale data from. An abandoned move's staging can't linger to
	// poison this: abort is two-phase, and the source deletes staging
	// before the move clears (MoveAborting).
	//
	// The trailing slash on the guard listing matters: hibernatePrefix has
	// none (path.Join), and a slashless List would let app.example.com's
	// keys answer for app.example.co.
	staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host))
	if err != nil {
		return nil, err
	}
	if len(staged) == 0 {
		return nil, nil
	}
	log.Printf("activator: %s: promoting %d staged object(s) from %s (residence move landing)", host, len(staged), MovesPrefix(host))
	// No receipts: a staged object is the source box's single-part copy,
	// and an in-bucket copy of it keeps its ETag, so plain ETag equality
	// is exact here. A deferred object (staging overwritten mid-promote,
	// which nothing should do once the move has cut) fails the wake
	// rather than restoring from a promotion that is not byte-final.
	round, err := lease.MirrorPrefix(ctx, b.Mirror, b.Cfg.Bucket, MovesPrefix(host), b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), lease.Receipts{})
	if err != nil {
		return nil, err
	}
	if round.Deferred > 0 {
		return nil, fmt.Errorf("promote staged replica for %s: %d staged object(s) changed during the copy; retrying on the next wake", host, round.Deferred)
	}
	return staged, nil
}

// dropStagedReplica deletes the staged keys promote just returned — no
// third listing of a prefix this restore has already walked twice.
// Best-effort: a leftover staging copy is storage, not a correctness
// problem.
func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]lease.MirrorObject) {
	for k := range staged {
		_ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k)
	}
}