package host

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"slices"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/cliflag"
	"github.com/carlosframework/platform/internal/hostfloor"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rootparticipation"
	"github.com/carlosframework/platform/internal/version"
)

// The root half of the platform self-update (spec §6). Runs as root from
// carlos-system-update.service (or --recover from carlos-system-recover).
// Every box-changing step is journaled first; every run begins with
// recovery; root reads ONE file under /data and treats it as a hint.
//
// The layout guard (layoutReversalReasons) also touches /data: it opens
// the registry, lists /data/layout, and Lstats the journals there. That
// grants no new privilege. Root's adopt, stage and restart passes already
// open the same registry, the listing and the Lstats follow no symlink
// and read no journal's contents, and all the guard can do with what it
// finds is refuse the install.

const (
	systemEdgeUnit  = "carlos-edge"
	systemRelayUnit = "carlos-relay"
	// systemControlUnit is the root instance-control service (isolated-exec
	// design §5). It execs /opt/carlos/carlos and stays up for weeks once
	// the first wake activated it, so every install that changes the
	// binary restarts it (restartInstanceControl; discussion #256).
	systemControlUnit = "carlos-instance-control.service"
	systemLockFile    = "lock"
	systemMaxRequest  = 4 << 10
	systemGateWait    = 20 * time.Second
	// systemClaimsGateWait is the gate's window on a box running with
	// CARLOS_HOST_CLAIM on. There the agent finishes its whole claim
	// startup (assembly, wiring, the inline activator and duties) before
	// the edge's listener binds, which takes about 22 s on a small
	// deployment and grows with the deployment's app count. A fixed 20 s
	// rolled every update back, and its rollback too. The longer window
	// comes with fail-fast (gateFailFast), so a build that crashes still
	// rolls back in seconds. Claims-off boxes keep systemGateWait exactly.
	systemClaimsGateWait = 120 * time.Second
	systemGateSoak       = 15 * time.Second
	systemVersionWait    = 30 * time.Second
	systemBootstrapWait  = 120 * time.Second

	// How long --recover waits for the apply lock, and how often it asks.
	// See waitForLock.
	systemRecoverLockWait = 30 * time.Second
	systemRecoverLockPoll = 500 * time.Millisecond
)

// systemQuiesceUnits are the root units that exec /opt/carlos/carlos
// against the running edge. Stopped (services too — stopping a timer does
// not stop the pass it already started) before the rename, re-armed after
// the gate.
var systemQuiesceUnits = []string{
	"carlos-adopt.timer", "carlos-adopt.path", "carlos-stage.path", "carlos-restart.path",
	"carlos-adopt.service", "carlos-config-sync.service", "carlos-stage.service", "carlos-restart.service",
}

var (
	errSystemRefused  = errors.New("refused")
	errSystemDeferred = errors.New("deferred")

	// errSystemUnsupported is what the non-linux build's primitives return.
	// It is declared here so Run can tell it from a lock another apply holds
	// without knowing which build it is in.
	errSystemUnsupported = errors.New("carlos system apply: unsupported on this platform")
)

// sysOps is the seam between the apply and the box: systemctl, exec with a
// deadline, the clock, sleep, the edge probe. Tests record and answer.
type sysOps struct {
	systemctl           func(args ...string) (string, error)
	run                 func(ctx context.Context, timeout time.Duration, name string, args ...string) (string, error)
	now                 func() time.Time
	sleep               func(time.Duration)
	probeEdge           func() error
	edgeStartedThisBoot func() bool
	logf                func(format string, args ...any)
}

// RunSystemApply is `carlos system apply [-recover]`: the root oneshot
// behind carlos-system-update.service, and with -recover the boot and
// OnFailure= unit that puts an interrupted apply back together.
func RunSystemApply(args []string) {
	fs := flag.NewFlagSet("system apply", flag.ExitOnError)
	recover := fs.Bool("recover", false, "recover an interrupted apply (files only) and exit: the boot and OnFailure unit")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)
	if os.Geteuid() != 0 {
		log.Fatal("carlos system apply is the root oneshot behind carlos-system-update.service; it is not for operators")
	}
	if err := registerProductionLaunchVerifier(); err != nil {
		log.Fatal(err)
	}
	ctx := context.Background()
	channel, invalidChannel := systemChannel(os.Getenv)
	if invalidChannel != "" {
		log.Printf("system apply: %s=%q is not a channel; this box follows %s", systemChannelEnv, invalidChannel, channel)
	}
	claims, gateWait := systemApplyGate(os.Getenv)
	a := &systemApplier{
		rootMoves: rootMovesFloorIdentity, ops: productionSysOps(), openStore: openSystemStore, channel: channel,
		optDir: release.OptDir(), dataDir: dataDir(), etcDir: "/etc", build: version.Build,
		platform: runtime.GOOS + "-" + runtime.GOARCH, gateWait: gateWait, soak: systemGateSoak,
		regionHA: regionHAEnabled(os.Getenv), claims: claims,
	}
	a.floorHook = installFloorHook // decided under the apply lock, in Run
	runErr := a.runWithRelease(ctx, *recover)
	if runErr != nil {
		log.Fatalf("%v", runErr)
	}
}

func (a *systemApplier) runWithRelease(ctx context.Context, recover bool) error {
	if _, _, err := a.rootMoves.read(); err != nil {
		return err
	}
	return a.participationDisk().WithParticipation(ctx, func(ctx context.Context, _ rootparticipation.Ledger) error {
		return a.Run(ctx, recover)
	}, func() error { return a.Run(ctx, recover) })
}

// openSystemStore is the production openStore: the deploy store and the
// release public key, opened at the moment the pointer is about to be read
// and not one step sooner.
//
// The lateness is the point. `carlos system apply --recover` runs at boot,
// before the network is up (its unit deliberately carries no
// After=network-online.target, because the whole job is to put the box's
// FILES back so the edge can start at all), and reads nothing from any
// store. Opening a bucket and parsing CARLOS_RELEASE_PUBKEY up front made
// recovery — the one path that must work on a broken box — depend on two
// things a broken box is most likely to be missing.
func openSystemStore(ctx context.Context) (release.Store, ed25519.PublicKey, error) {
	store, err := openDeployStore(ctx)
	if err != nil {
		return nil, nil, err
	}
	pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		return nil, nil, err
	}
	return store, pub, nil
}

// openSystemRegistry keeps root's SQLite access in the service-user child.
func openSystemRegistry(dataDir string) (*registry.Registry, error) {
	path := filepath.Join(dataDir, "registry.db")
	reg, err := openServiceRegistry(path)
	if err != nil {
		return nil, fmt.Errorf("%s: %w", path, err)
	}
	return reg, nil
}

// productionSysOps is the real box behind the seam.
func productionSysOps() sysOps {
	return sysOps{
		systemctl: func(args ...string) (string, error) {
			out, err := exec.Command("systemctl", args...).CombinedOutput()
			if err != nil {
				return string(out), fmt.Errorf("systemctl %s: %w: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
			}
			return string(out), nil
		},
		run: func(ctx context.Context, timeout time.Duration, name string, args ...string) (string, error) {
			ctx, cancel := context.WithTimeout(ctx, timeout)
			defer cancel()
			out, err := exec.CommandContext(ctx, name, args...).Output()
			if err != nil {
				// Output() keeps stderr out of the returned bytes so the
				// caller can parse them; a command that failed is the one
				// case where what it said matters more than what it printed.
				var ee *exec.ExitError
				if errors.As(err, &ee) && len(ee.Stderr) > 0 {
					return string(out), fmt.Errorf("%s %s: %w: %s", name, strings.Join(args, " "), err, strings.TrimSpace(string(ee.Stderr)))
				}
				return string(out), fmt.Errorf("%s %s: %w", name, strings.Join(args, " "), err)
			}
			return string(out), nil
		},
		now:   time.Now,
		sleep: time.Sleep,
		probeEdge: func() error {
			// The edge's :80 listener answers on every box — ACME HTTP-01
			// and the redirect to https — so any HTTP answer on loopback
			// proves the process serves. A redirect is an answer, and a box
			// with no route yet must still pass (spec §6 step 10).
			c := &http.Client{Timeout: 3 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
			resp, err := c.Get("http://127.0.0.1/")
			if err != nil {
				return err
			}
			resp.Body.Close()
			return nil
		},
		edgeStartedThisBoot: func() bool {
			out, err := exec.Command("systemctl", "show", "-p", "ActiveEnterTimestampMonotonic", "--value", systemEdgeUnit).Output()
			return err == nil && strings.TrimSpace(string(out)) != "" && strings.TrimSpace(string(out)) != "0"
		},
		logf: log.Printf,
	}
}

type systemApplier struct {
	floorAlerted  bool
	rootMoves     rootFloorIdentitySource
	participation *rootparticipation.Ledger
	floor         *hostfloor.Disk
	// floorHook, when set, decides legacy or floor-guarded state once Run
	// holds the apply lock, so a concurrent run's marker write is never read.
	floorHook func(*systemApplier) error
	guardOpen func(string) (*os.File, error)
	ops       sysOps
	// openStore is called once, by verifyPointer, and only on a run that
	// has a request to answer. --recover never calls it: see
	// openSystemStore for why that matters at boot.
	openStore func(ctx context.Context) (release.Store, ed25519.PublicKey, error)
	// openRegistry is called only when the candidate predates the instance
	// layout, to see whether anything on the box is still nested. Nil means
	// openSystemRegistry over dataDir.
	openRegistry   func() (*registry.Registry, error)
	channel        string
	optDir         string
	dataDir        string
	etcDir         string // "/etc" in production; a temp dir in tests
	build          string // version.Build of THIS process
	platform       string // "<goos>-<goarch>"; the caller sets it from runtime
	gateWait, soak time.Duration
	// regionHA is CARLOS_REGION_HA: root writes its update-begun marker and
	// refuses an update while this box takes part in a claim operation
	// (region HA spec 1 §5.10; systemparticipation.go).
	regionHA bool
	// claims is CARLOS_HOST_CLAIM, read with hostwrite.Enabled from the
	// unit's environment (carlos-system-update.service and
	// carlos-system-recover.service both load host.env). It changes only the
	// gate: the claims window and fail-fast (systemGateWaitFor,
	// gateFailFast). Off, the gate is exactly the 20 s one.
	claims bool
	// store is the deploy store verifyPointer opened, kept for armFloor.
	// Nil on --recover, which reads no store.
	store release.Store
	// beforeSwap, when set, runs inside activationGuard after the rule has
	// allowed and just before the step it guards: a test's schedule hook.
	// Nil in production.
	beforeSwap func()
}

func (a *systemApplier) rootDir() string         { return systemRootDir(a.optDir) }
func (a *systemApplier) stagingDir() string      { return filepath.Join(a.rootDir(), "staging") }
func (a *systemApplier) lockPath() string        { return filepath.Join(a.rootDir(), systemLockFile) }
func (a *systemApplier) file(name string) string { return filepath.Join(a.rootDir(), name) }

// Run is the whole oneshot. It returns an error only for a failure the
// caller should exit non-zero on (an apply that rolled back, a rollback
// that failed, a broken box); refusals and no-ops are recorded in
// result.json and return nil.
//
// A run whose RECOVERY rolled the box back exits 0, unlike a rollback
// inside install: the result and the heartbeat carry the truth, the
// journal is closed, and re-firing OnFailure= over a finished recovery
// would only start it again with nothing left to do.
func (a *systemApplier) Run(ctx context.Context, recover bool) (retErr error) {
	defer func() {
		if errors.Is(retErr, context.DeadlineExceeded) {
			if required, err := a.participationDisk().ParticipationRequired(context.WithoutCancel(ctx)); err == nil && required {
				retErr = errors.Join(errSystemDeferred, retErr)
			}
		}
	}()
	reapStagingWrites(a.optDir, a.dataDir, a.etcDir, a.ops.logf)
	if err := os.MkdirAll(a.rootDir(), 0o755); err != nil {
		return err
	}
	lock, err := os.OpenFile(a.lockPath(), os.O_CREATE|os.O_RDWR, 0o600)
	if err != nil {
		return err
	}
	defer lock.Close()
	if recover {
		got, err := a.waitForLock(lock)
		if err != nil {
			return err
		}
		if !got {
			a.ops.logf("system apply: an apply holds the lock; it recovers at its own start")
			return nil
		}
	} else if err := flockExclusiveNonblock(lock); err != nil {
		if errors.Is(err, errSystemUnsupported) {
			a.ops.logf("system apply: %v", err)
			return err
		}
		a.ops.logf("system apply: another apply is running; leaving it to finish")
		return nil
	}
	if a.floorHook != nil {
		if err := a.floorHook(a); err != nil {
			return err
		}
	}
	// Rejected floor recovery must retain staging and the journal. A
	// settlement deferred with no journal open (errSettlementDeferred) lets
	// --recover go on and exit 0; it is not tried again at the end.
	deferred := false
	if err := a.checkGuardRecovery(ctx); err != nil {
		if !a.deferSettlement(recover, err) {
			return err
		}
		deferred = true
	}
	defer func() {
		active, err := a.activeFloor()
		if err != nil {
			retErr = errors.Join(retErr, err)
			return
		}
		if !active {
			if !deferred {
				a.settleMarker(ctx, recover, &retErr)
			}
			_ = os.RemoveAll(a.stagingDir())
			return
		}
		if j, e := readSystemJournal(a.optDir); e == nil && j == nil && !deferred {
			if err := a.checkGuardRecovery(ctx); err != nil {
				if !a.deferSettlement(recover, err) {
					retErr = errors.Join(retErr, err)
				}
				return
			}
			a.settleMarker(ctx, recover, &retErr)
			_ = os.RemoveAll(a.stagingDir())
		}
	}()
	// Root passes an earlier run paused and could not start again come
	// first: they are the box's convergence, whatever this run is for.
	// After main's floor hook and guard recovery check (merge of claim
	// Plan 8): a release-floor recovery it rejects keeps the box exactly as
	// main leaves it, and the triggers wait for the run that gets past it.
	a.rearmPending()

	// A settlement the census refused after recovery durably removed the
	// journal is deferred like one with no journal open: --recover exits 0
	// and does not try it again at the end of the run.
	recovered, err := a.recoverIfNeeded(ctx)
	if err != nil {
		if !a.deferSettlement(recover, err) {
			return err
		}
		deferred = true
	}
	if recover {
		return nil
	}
	if recovered {
		// The recovery moved the box and wrote the result this run answers
		// with — an apply on top of that would overwrite that result and
		// restart the edge a second time. The agent re-touches the marker
		// if it still wants the update, and by then it can read the result.
		return nil
	}
	req, err := a.readRequest()
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		a.ops.logf("system apply: request: %v", err)
		return nil // nothing to record against: the request itself is unreadable
	}
	// A tuple this box has already rolled back is not tried again: the
	// build failed its gate here, and re-applying it on every marker touch
	// would restart the edge in a loop. Only those two outcomes stick —
	// refused, deferred and noted are all worth another go.
	if r, _ := readSystemResult(a.optDir); r.matches(a.channel, req.Version, req.PromotedAt) &&
		(r.Outcome == systemOutcomeRolledBack || r.Outcome == systemOutcomeRollbackFailed) {
		a.ops.logf("system apply: %s %s was already %s on this box; not applying it again", a.channel, req.Version, r.Outcome)
		return nil
	}
	p, err := a.verifyPointer(ctx)
	if err != nil {
		return a.record(req, err)
	}
	// The moves protocol floor (moves spec §11), written to disk from the
	// store before anything is staged, so every check below and every later
	// recovery enforces what is published.
	if err := a.armFloor(ctx); err != nil {
		return a.record(req, err)
	}
	if req.Version != p.Current.Version || !req.PromotedAt.Equal(p.PromotedAt) {
		return a.record(req, fmt.Errorf("%w: request names %s@%s, pointer is %s@%s", errSystemRefused,
			req.Version, req.PromotedAt.Format(time.RFC3339), p.Current.Version, p.PromotedAt.Format(time.RFC3339)))
	}
	seen, err := readSystemSeen(a.optDir)
	if err != nil {
		return a.record(req, fmt.Errorf("%w: seen.json unreadable: %v", errSystemRefused, err))
	}
	if wm, ok := seen[a.channel]; ok && p.PromotedAt.Before(wm) {
		return a.record(req, fmt.Errorf("%w: replay — pointer promoted %s is older than the %s watermark %s", errSystemRefused,
			p.PromotedAt.Format(time.RFC3339), a.channel, wm.Format(time.RFC3339)))
	}
	seen[a.channel] = p.PromotedAt
	if err := writeJSONDurable(a.file("seen.json"), seen, 0o644); err != nil {
		return err
	}
	if req.NoOp || p.Current.Version == a.onDiskBuild(ctx) {
		return a.writeResult(req, systemOutcomeNoted, "")
	}
	// The claim release floor's transaction (Plan 8) first, as on main;
	// then the moves activation check. Either one refusing refuses; a
	// moves refusal after beginGuard leaves no journal, so Run's deferred
	// checkGuardRecovery closes the guard transaction as it does for any
	// refusal before the journal.
	if err := a.beginGuard(ctx, p); err != nil {
		return err
	}
	// The install's activation check, before the candidate is copied in or
	// run at all: a build below the floor, or a downgrade while a row is
	// held, is refused with the box exactly as it was.
	if err := a.activationAllowed(ctx, p.Current.Version); err != nil {
		return a.record(req, fmt.Errorf("%w: %v", errSystemRefused, err))
	}
	staged, err := a.stageCandidate(ctx, p, req)
	if err != nil {
		return a.record(req, fmt.Errorf("%w: %v", errSystemRefused, err))
	}
	return a.install(ctx, req, p, staged)
}

// waitForLock is --recover's bounded wait for the apply lock: poll
// systemRecoverLockPoll apart for up to systemRecoverLockWait, then give up.
// It reports whether it got the lock; an unsupported platform is the one
// error it returns.
//
// The bound is the fix for a real deadlock. An update run holds this lock
// across `systemctl restart carlos-edge`, and carlos-system-recover.service
// is ordered Before=carlos-edge.service — so if a recover unit's start job
// is queued while an apply is mid-restart, systemd makes the edge's restart
// wait on the recovery, and a recovery that blocked on the lock would wait
// on the apply that is waiting on the restart. Both then sit there until
// their unit timeouts fire. There is nothing for a recovery to do in that
// window anyway: the apply owns the journal it would read, and finishes by
// recovering from it itself if it dies.
func (a *systemApplier) waitForLock(lock *os.File) (bool, error) {
	deadline := a.ops.now().Add(systemRecoverLockWait)
	for {
		err := flockExclusiveNonblock(lock)
		if err == nil {
			return true, nil
		}
		if errors.Is(err, errSystemUnsupported) {
			return false, err
		}
		if !a.ops.now().Before(deadline) {
			return false, nil
		}
		a.ops.sleep(systemRecoverLockPoll)
	}
}

// install is steps 6–12: decide .prev, stage both unit sets, then the
// box-changing dance — each step journaled before it happens, and any
// failure from the quiesce on going to the rollback of step 13.
func (a *systemApplier) install(ctx context.Context, req systemRequest, p *release.SystemPointer, staged string) error {
	target := p.Current.Version
	if err := a.guardImage(staged); err != nil {
		return err
	}
	prevOK, err := a.decidePrev(ctx)
	if err != nil {
		return a.record(req, fmt.Errorf("%w: %v", errSystemRefused, err))
	}
	if !prevOK {
		return a.record(req, fmt.Errorf("%w: no known-good carlos.prev to fall back to", errSystemRefused))
	}
	newDir, prevDir := filepath.Join(a.stagingDir(), "new"), filepath.Join(a.stagingDir(), "prev")
	newPlan, newUnits, err := a.stagePlan(ctx, staged, newDir)
	if err != nil {
		return a.record(req, fmt.Errorf("%w: candidate bootstrap: %v", errSystemRefused, err))
	}
	// A release before the instance layout ships no data-instances.mount,
	// so installing it disables the mount units and then serves every
	// nested row from an empty mountpoint directory (design §8.2, "That
	// order is enforced"). The candidate's own staged unit set is the
	// evidence of which kind of release it is. Refused here, before the
	// journal, so a refusal leaves the box exactly as it was.
	//
	// The unit set is not the whole story: v0.20.0 to v0.24.2 ship the
	// mounts and still cannot wake a nested row (their edge unit keeps
	// RestrictSUIDSGID=, so the agent's openat2 fails; release.LayoutFloor),
	// and `carlos system update --file` reaches this guard with no console
	// check ahead of it. So a candidate whose reported version parses and
	// is below the floor is refused on the same terms. A version that
	// cannot be parsed (a local build named by a sha) is judged by its
	// unit set alone, as before; docs/site/cli.md says so.
	if !slices.Contains(newUnits, layoutMountUnit) || release.BelowFloor(target, release.LayoutFloor) {
		if reasons := a.layoutReversalReasons(); len(reasons) > 0 {
			return a.record(req, fmt.Errorf("%w: "+systemApplyLayoutRefusal, errSystemRefused, target, strings.Join(reasons, ", ")))
		}
	}
	// Only the unit set is taken from the return: a rollback picks this
	// staging up as it stands, plan.json and all, rather than exec'ing the
	// previous binary again on a box that is already unhealthy.
	_, prevUnits, err := a.stagePlan(ctx, systemPrevPath(a.optDir), prevDir)
	if err != nil {
		return a.record(req, fmt.Errorf("%w: carlos.prev bootstrap: %v", errSystemRefused, err))
	}
	// Region HA (spec 1 §5.10): the marker, then the participation check,
	// before the first step that changes the box. Either failing defers the
	// update, which the agent retries; the box is left exactly as it was.
	if err := a.beginExecutableChange(ctx, target, req.PromotedAt); err != nil {
		if !errors.Is(err, errSystemDeferred) {
			err = fmt.Errorf("%w: update-begun marker: %v", errSystemDeferred, err)
		}
		return a.record(req, err)
	}
	// The release on disk, for the activation guard at the rename: read now,
	// because reading it runs the binary and the guard holds locks.
	running := a.onDiskBuild(ctx)
	// PromotedAt rides along for the whole transaction: it is half of the
	// tuple a recovery has to answer for.
	j := systemJournal{Phase: systemPhaseInstalling, Target: target, PromotedAt: req.PromotedAt, PrevBuild: a.prevBuild(ctx),
		UnitsAdded: unitsMinus(newUnits, prevUnits), UnitsRemoved: unitsMinus(prevUnits, newUnits), StartedAt: a.ops.now().UTC()}
	if err := a.journal(j); err != nil {
		return err
	}
	fail := func(step string, err error) error {
		return a.rollback(ctx, req, fmt.Sprintf("%s: %v", step, err), &j)
	}
	// Stop the root passes that exec the old binary against the old edge
	// before the rename; one that is already running is killed, and those
	// passes are idempotent. Which of their triggers were active is kept, so
	// a refused rename can put them back.
	prior := a.quiesce()
	// The activation rule again, at the step it guards (moves spec §11): the
	// check before staging can have gone stale, since the agent may have
	// raised the floor or held a row since. activationGuard holds the floor
	// and the holds still across the rename.
	renamed := false
	if err := a.activationGuard(ctx, target, running, func(ctx context.Context) error {
		if err := a.rootProtocolAllowsBinary(staged); err != nil {
			return err
		}
		renamed = true
		return os.Rename(staged, systemBinaryPath(a.optDir))
	}); err != nil {
		if !renamed {
			return a.abandonInstall(req, err, prior)
		}
		return fail("install", err)
	}
	if err := os.Chown(systemBinaryPath(a.optDir), 0, 0); err != nil && !errors.Is(err, os.ErrPermission) {
		a.ops.logf("system apply: chown root: %v", err)
	}
	j.Phase = systemPhaseInstalled
	if err := a.journal(j); err != nil {
		return fail("journal", err)
	}
	if err := a.installUnits(ctx, newDir, j.UnitsRemoved); err != nil {
		return fail("units", err)
	}
	j.Phase = systemPhaseUnits
	if err := a.journal(j); err != nil {
		return fail("journal", err)
	}
	// The control service before the edge; see restartInstanceControl for
	// why that order, and why a restart that may drop a wake is safe here.
	if err := a.restartInstanceControl(); err != nil {
		return fail("instance-control restart", err)
	}
	if err := a.restartEdge(); err != nil {
		return fail("restart", err)
	}
	a.restartRelay()
	j.Phase = systemPhaseRestarted
	if err := a.journal(j); err != nil {
		return fail("journal", err)
	}
	if err := a.gate(ctx, target); err != nil {
		return fail("gate", err)
	}
	if err := a.acceptRootFloorHealthy(ctx); err != nil {
		return fail("gate", err)
	}
	j.Phase = systemPhaseGated
	if err := a.journal(j); err != nil {
		return fail("journal", err)
	}
	// Outside activationGuard, deliberately: the floor protects activation,
	// and the guarded rename above was this install's activation. A floor
	// raised since applies to the next one. (Recovery's forward finish is
	// different: it has no rename of its own, so the finish is its
	// activation, and it is guarded.)
	return a.finish(ctx, req, target, newPlan)
}

// abandonInstall undoes an install that the activation guard refused at the
// rename, when nothing but the quiesce has happened: the binary, the units
// and carlos.prev are all as they were.
//
// The paused triggers are started again FIRST, with the journal still
// open, and the journal is removed only once they are (fix round 2). When
// starting them fails, what is owed goes to rearm-pending.json, durably,
// before the journal goes, and the run fails with a reason that says so;
// every later run starts with rearmPending, and this failure fires the
// recover unit at once. The journal itself is not what is kept: while it
// exists the system units pick carlos.prev for that recovery, and a
// carlos.prev older than this build reads an install journal as an
// interrupted install and restores itself over a binary that never changed.
// Only when even the retry state cannot be written does the journal stay.
func (a *systemApplier) abandonInstall(req systemRequest, refusal error, prior []string) error {
	// A guard that could not get the floor lock is deferred (retried at the
	// next run); every other stop is a refusal.
	kind := errSystemRefused
	var deferred activationDeferred
	if errors.As(refusal, &deferred) {
		kind = errSystemDeferred
	}
	rerr := a.rearm(prior)
	if rerr == nil {
		a.removeJournal()
		return a.record(req, fmt.Errorf("%w: %v", kind, refusal))
	}
	reason := fmt.Errorf("%w: %v; the root passes paused for the install could not be started again: %v. the next apply or recovery starts them",
		kind, refusal, rerr)
	if err := a.notePendingRearm(prior); err != nil {
		reason = fmt.Errorf("%w: %v; the root passes paused for the install could not be started again: %v, and the retry could not be recorded: %v. the journal stays for the next recovery",
			kind, refusal, rerr, err)
	} else {
		a.removeJournal()
	}
	_ = a.record(req, reason)
	return fmt.Errorf("system apply: %s: %v", req.Version, reason)
}

// rearmPendingFile records root-pass triggers a refused step paused and
// could not start again (abandonInstall), for the next run to start.
const rearmPendingFile = "rearm-pending.json"

type rearmPending struct {
	Units []string  `json:"units"`
	At    time.Time `json:"at"`
}

// notePendingRearm adds units to rearm-pending.json, durably.
func (a *systemApplier) notePendingRearm(units []string) error {
	var p rearmPending
	if err := readJSONFile(a.file(rearmPendingFile), &p); err != nil && !errors.Is(err, os.ErrNotExist) {
		p = rearmPending{}
	}
	for _, u := range units {
		if !slices.Contains(p.Units, u) {
			p.Units = append(p.Units, u)
		}
	}
	p.At = a.ops.now().UTC()
	return writeJSONDurable(a.file(rearmPendingFile), p, 0o600)
}

// rearmPending starts what rearm-pending.json records and removes it once
// they are started. A failure is logged and the file stays for the next run.
func (a *systemApplier) rearmPending() {
	var p rearmPending
	if err := readJSONFile(a.file(rearmPendingFile), &p); err != nil {
		if !errors.Is(err, os.ErrNotExist) {
			a.ops.logf("system apply: %s: %v", rearmPendingFile, err)
		}
		return
	}
	if err := a.rearm(p.Units); err != nil {
		a.ops.logf("system apply: the root passes paused by a refused install could not be started again: %v; the next run tries again", err)
		return
	}
	if err := removeDurable(a.file(rearmPendingFile)); err != nil {
		a.ops.logf("system apply: remove %s: %v", rearmPendingFile, err)
	}
}

// quiesce stops the root passes (systemQuiesceUnits) and returns the
// timers and paths among them that were active, for rearm to start again
// if the step they were paused for is refused. When their state cannot be
// read, all of them are returned: active is what an armed box has.
func (a *systemApplier) quiesce() []string {
	var triggers []string
	for _, u := range systemQuiesceUnits {
		if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") {
			triggers = append(triggers, u)
		}
	}
	prior := triggers
	// is-active exits non-zero when any unit is not active; the output still
	// has one state per unit, in order.
	out, _ := a.ops.systemctl(append([]string{"is-active"}, triggers...)...)
	if states := strings.Fields(out); len(states) == len(triggers) {
		prior = nil
		for i, u := range triggers {
			if states[i] == "active" || states[i] == "activating" || states[i] == "reloading" {
				prior = append(prior, u)
			}
		}
	}
	if _, err := a.ops.systemctl(append([]string{"stop"}, systemQuiesceUnits...)...); err != nil {
		a.ops.logf("system apply: quiesce: %v (continuing)", err)
	}
	return prior
}

// rearm starts units again (nothing to do for none).
func (a *systemApplier) rearm(units []string) error {
	if len(units) == 0 {
		return nil
	}
	_, err := a.ops.systemctl(append([]string{"start"}, units...)...)
	return err
}

// finish is steps 11–12, and what recovery runs forward from phase gated.
func (a *systemApplier) finish(ctx context.Context, req systemRequest, target string, plan bootstrapPlan) error {
	reason := ""
	if err := a.arm(plan); err != nil {
		// The gate passed, so this run is an apply and rolling back now
		// would undo a build that proved itself. But `enable` is one
		// batched call: one bad unit leaves the box on the new binary with
		// nothing armed, and a fleet-visible reason is the only way anyone
		// finds out.
		a.ops.logf("system apply: arm: %v", err)
		reason = "armed with errors: " + err.Error()
	}
	if err := writeJSONDurable(a.file("applied.json"), systemApplied{Version: target, Channel: a.channel, At: a.ops.now().UTC()}, 0o644); err != nil {
		return err
	}
	if err := a.writeResult(req, systemOutcomeApplied, reason); err != nil {
		return err
	}
	return a.closeGuardJournal(ctx)
}

// removeJournal closes the transaction, durably: an unsynced removal can
// come back after a power loss, and a journal that outlives the run it
// belongs to is one the next run recovers from.
func (a *systemApplier) removeJournal() {
	os.Remove(a.file("journal.json"))
	if err := syncDir(a.rootDir()); err != nil {
		a.ops.logf("system apply: fsync %s: %v", a.rootDir(), err)
	}
}

// decidePrev refreshes carlos.prev from the on-disk binary only when that
// binary is known good — applied.json names it, or nothing has ever been
// applied and there is no .prev (a hand-rolled box whose running binary
// passed the script's own post-condition). Otherwise it leaves .prev alone
// and reports whether there is one to fall back to at all.
//
// The backup is deliberately NOT journaled, though every other box-changing
// step is. copyFileDurable is whole-or-nothing — it writes carlos.prev.tmp
// and renames — so a kill here leaves either the old .prev or the new one,
// never a torn file, and a stray .prev.tmp is truncated by the next write.
// A journal would make it WORSE: a phase written before the copy is a phase
// the exec shim in carlos-system-recover.service reads at the next boot,
// and it would then run a .prev that is one release older than the box, for
// a whole apply, on the strength of a backup that had nothing to recover.
// (recoverIfNeeded still knows systemPhaseBackingUp: journals written by
// the version that did write one outlive the upgrade to this one.)
func (a *systemApplier) decidePrev(ctx context.Context) (bool, error) {
	prev := systemPrevPath(a.optDir)
	if err := a.guardImage(systemBinaryPath(a.optDir)); err != nil {
		return false, err
	}
	_, prevErr := os.Stat(prev)
	applied, err := readSystemApplied(a.optDir)
	if err != nil {
		return false, err
	}
	disk := a.onDiskBuild(ctx)
	known := (applied != nil && applied.Version == disk && disk != "") || (applied == nil && prevErr != nil)
	if !known {
		if prevErr == nil {
			if err := a.guardImage(prev); err != nil {
				return false, err
			}
		}
		return prevErr == nil, a.notePrev(ctx, false, "")
	}
	if err := a.notePrev(ctx, true, disk); err != nil {
		return false, err
	}
	return true, nil
}

// notePrev is decidePrev's half of the moves floor (moves spec §11), under
// the lock the adopt pass's judge takes too. With replace, carlos.prev is
// refreshed from the on-disk binary (version is what that binary reported):
// the sidecar and the verdict of the carlos.prev being replaced are removed
// first, so a crash mid-copy leaves a carlos.prev that nothing vouches for
// (refused while a floor is armed), never one vouched for by its
// predecessor's files. Then the sidecar is written and judgePrev runs, all
// before the journal.
//
// The copy's own failure is decidePrev's error as it always was. The floor
// steps are best-effort while no floor is armed (logged, never fatal: an
// unarmed deployment must update exactly as main does), and hard once one
// is (floorStep).
func (a *systemApplier) notePrev(ctx context.Context, replace bool, version string) error {
	var copyErr error
	ran := false
	steps := func() error {
		ran = true
		if replace {
			if err := a.floorStep(errors.Join(removeDurable(prevAllowedPath(a.optDir)), removeDurable(seedsPrevAllowedPath(a.optDir)), removeDurable(prevVersionPath(a.optDir)))); err != nil {
				return err
			}
			if copyErr = copyFileDurable(systemBinaryPath(a.optDir), systemPrevPath(a.optDir), 0o755); copyErr != nil {
				return nil
			}
			if version != "" {
				if err := a.floorStep(writePrevVersion(a.optDir, version)); err != nil {
					return err
				}
			}
		}
		if id, required, err := a.rootMoves.read(); err != nil {
			return err
		} else if required {
			f, err := openRootMovesFloors(a.optDir, id, false)
			if err != nil {
				return err
			}
			defer f.close()
			j := rootFloorJudge{floors: f, current: systemBinaryPath(a.optDir), prev: systemPrevPath(a.optDir), idle: rootFloorDowngradeIdle, binary: func(path string) (rootFloorBinary, error) { return readRootFloorBinary(a.optDir, path) }}
			return errors.Join(j.judge(ctx), judgeRootSeedsPrev(ctx, a.optDir, id))
		}
		return a.floorStep(errors.Join(judgePrevWith(a.optDir, a.dataDir, a.freezeHolds, func() string { return a.prevBuild(ctx) }), judgeSeedsPrev(a.optDir, a.dataDir)))
	}
	err := withPrevLock(a.optDir, steps)
	if !ran {
		// The lock itself failed. Armed, that is the apply's error; with no
		// floor the steps run anyway, because the backup is main's step.
		if err = a.floorStep(err); err != nil {
			return err
		}
		err = steps()
	}
	if copyErr != nil {
		return copyErr
	}
	return err
}

// floorArmed reports whether a moves protocol floor is armed on this box:
// a local floor file that reads, or one that is there and cannot be read
// (armed, value unknown). An armed seeds floor (seedsfloor.go) counts too,
// since the floor's steps write and judge both.
func (a *systemApplier) floorArmed() bool {
	f, err := readEffectiveMovesFloor(a.rootMoves, a.optDir, a.dataDir)
	sf, se := readEffectiveSeedsFloor(a.rootMoves, a.optDir, a.dataDir)
	return err != nil || f != nil || se != nil || sf != nil
}

// floorStep is the dormancy rule for the floor's own steps (moves Task 13,
// ruling F15): while no floor is armed a failure is a log line and the
// apply goes on as main's would; once one is, it is the apply's error.
func (a *systemApplier) floorStep(err error) error {
	if err == nil {
		return nil
	}
	if a.floorArmed() {
		return err
	}
	a.ops.logf("system apply: %v (no moves protocol floor is armed; continuing)", err)
	return nil
}

// armFloor writes the published floor to this box's local floor files
// (syncLocalFloor), first thing once the store is open. What is published
// and could not be written down defers the apply: the box would otherwise
// enforce less than the floor. Anything else (the store not answering, no
// object, an unusable one) keeps what is on disk and is one log line, so a
// deployment that never published a floor updates exactly as main does.
// The seeds floor is armed first, by the same rule (armSeedsFloor).
func (a *systemApplier) armFloor(ctx context.Context) error {
	if id, required, err := a.rootMoves.read(); err != nil {
		return err
	} else if required {
		if err := refreshRootMovesFloor(ctx, a.store, a.optDir, a.dataDir, id); err != nil {
			return err
		}
		return a.armSeedsFloor(ctx)
	}

	if err := a.armSeedsFloor(ctx); err != nil {
		return err
	}
	f, err := syncLocalFloor(ctx, a.store, a.dataDir)
	if err == nil {
		return nil
	}
	var fw *floorWriteError
	if errors.As(err, &fw) {
		return fmt.Errorf("%w: %s", errSystemDeferred, fmt.Sprintf("moves protocol floor: could not write it on this box: %v", fw.cause))
	}
	onDisk := "none"
	if f != nil {
		onDisk = f.MinRelease
	}
	a.ops.logf("system apply: moves protocol floor: %v (the floor on this box stays %s)", err, onDisk)
	return nil
}

// activationAllowed is the applier's own rule for every path that puts an
// executable in place (moves spec §11): the install (for the candidate),
// and restoreFiles (for carlos.prev, by its sidecar), which both rollback
// and recovery go through. Two halves, both only once a floor is armed:
//
//   - the floor: version below the armed min_release is refused. The floor
//     is read from the local file alone, never the store: boot recovery
//     runs before the network.
//   - the hold: while any registry row is held by an operation, a version
//     below the release running now is refused (§11: a participant in a
//     held operation does not downgrade).
//
// With no floor armed it reads neither the registry nor any version and
// allows, with one log line: main's behaviour.
//
// The seeds floor (seedsActivationRefusal) is checked first, on every one
// of these paths and the same way as the floor half; it has no hold half.
//
// This is the EARLY check, made before anything is staged so a refused
// build is never copied in or run. It can go stale by the time the build is
// put in place, so the step that puts it there runs under activationGuard,
// which applies the same rule again with its inputs held still.
//
// How the hold half relates to systemparticipation.go (region HA H14): that
// guard defers any executable change, up or down, while this box takes
// part in a claim operation (.hostexec), and only under CARLOS_REGION_HA,
// and only on the install path. This one refuses only a downgrade, while a
// moves hold_op row exists, once a floor is armed, and on every activating
// path, rollback and recovery included. They overlap on a downgrade
// install during a move that also runs a claim operation; there both run
// (this check before staging, H14's marker just before the journal) and
// either refusal stands. Neither reads or relaxes the other.
func (a *systemApplier) activationAllowed(ctx context.Context, version string) error {
	return a.activationRule(version, func() string { return a.onDiskBuild(ctx) })
}

// activationRefused is a refusal by the activation rule, told apart from a
// failure of the step it guards.
type activationRefused struct{ error }

func (e activationRefused) Unwrap() error { return e.error }

func refuse(format string, args ...any) error {
	return activationRefused{fmt.Errorf(format, args...)}
}

// activationDeferred is the guard unable to get floor.lock because another
// process holds it: a writer is active and may be about to publish a floor.
// The step is put off to the next run, never taken unlocked (fix round 3).
type activationDeferred struct{ error }

func (e activationDeferred) Unwrap() error { return e.error }

// activationGuard is the ONE guard for every step that makes an executable
// this box's (moves spec §11; fix round 3): the install's rename, the
// restore's rename, and each finish of a journal that has no rename of its
// own (recovery forward and backward, and rollback's). Nothing else renames
// an executable into place or finishes a journal; the scan in
// activationguard_test.go fails if something does.
//
// After release and participation exclusion, it takes floor.lock, creating .moves on a box with
// no floor yet, so no writer can arm or raise a floor until fn has run.
// Then, with a floor armed, it always freezes the held rows
// (registry.WithHoldsFrozen), whether or not target is the running release,
// and checks the floor and the hold rule inside, immediately before fn.
// With no floor armed, under the lock none can appear, the hold rule does
// not apply, and the registry is not consulted (index D13: an unarmed
// deployment behaves as main).
//
// Lock failures:
//   - contention (the wait ran out): activationDeferred. Never unlocked: a
//     writer holding the lock may not have published yet.
//   - the lock cannot exist (.moves or floor.lock cannot be created:
//     read-only, full or unwritable data dir) and no floor is armed: one
//     log line and fn runs as main's step would (F15). No writer can arm a
//     floor on such a box either.
//   - anything else, or a floor armed: a refusal.
//
// Run holds the apply flock. Census activation takes release exclusion,
// then participation, then the effective floor and registry write locks.
// Journal close reuses the held release and participation contexts.
// Nothing here may take the apply flock.
//
// running is the release on disk, read by the caller before any lock,
// since reading it runs a binary. A refusal or deferral comes back typed;
// fn's own error comes back as it is.
func (a *systemApplier) activationGuard(ctx context.Context, target, running string, fn func(context.Context) error) error {
	guard := func(ctx context.Context) error {
		if _, required, err := a.rootMoves.read(); err != nil {
			return err
		} else if required {
			bound, e := readEffectiveMovesFloor(a.rootMoves, a.optDir, a.dataDir)
			if e != nil {
				return e
			}
			if at, _ := moves.ReleaseAtOrAbove(target, running); bound != nil && !at {
				if err := rootFloorDowngradeIdle(ctx); err != nil {
					return err
				}
			}
		}
		step := func() error {
			if a.beforeSwap != nil {
				a.beforeSwap()
			}
			return fn(ctx)
		}
		locked := false
		err := withEffectiveMovesFloorLock(a.rootMoves, a.optDir, a.dataDir, true, func() error {
			locked = true
			if err := a.seedsActivationRefusal(target); err != nil {
				return err
			}
			floor, err := readEffectiveMovesFloor(a.rootMoves, a.optDir, a.dataDir)
			if err != nil {
				return refuse("moves protocol floor: the floor on this box cannot be read: %v", err)
			}
			if floor == nil {
				return step()
			}
			if err := floorRefusal(target, a.activationFloor(floor, false)); err != nil {
				return err
			}
			frozen := false
			err = a.freezeHolds(func(held []registry.Route) error {
				frozen = true
				if err := holdRefusal(target, running, held); err != nil {
					return err
				}
				return step()
			})
			if !frozen {
				return refuse("a participant in a held operation does not downgrade, and the registry cannot say whether this box is one: %v", err)
			}
			return err
		})
		if locked {
			return err
		}
		if errors.Is(err, errFloorLockBusy) {
			return activationDeferred{errors.New("moves protocol floor: another writer holds the floor lock on this box; this waits for the next run")}
		}
		if floorLockCannotExist(err) && !a.floorArmed() {
			a.ops.logf("system apply: moves protocol floor lock: %v (no floor is armed, and none can be on this box; continuing)", err)
			return step()
		}
		return refuse("moves protocol floor: the floor on this box cannot be read: %v", err)
	}
	return a.participationDisk().WithParticipation(ctx, func(ctx context.Context, l rootparticipation.Ledger) error { return l.WithTransaction(ctx, guard) }, func() error { return guard(ctx) })
}

// floorRefusal is the floor half of the rule.
func floorRefusal(version string, floor *moves.Floor) error {
	if floor == nil {
		return nil
	}
	if version == "" {
		return refuse("moves protocol floor: carlos.prev records no version, so it is below %s", floor.MinRelease)
	}
	if ok, _ := moves.ReleaseAtOrAbove(version, floor.MinRelease); !ok {
		return refuse("moves protocol floor: %s is below %s", version, floor.MinRelease)
	}
	return nil
}

// holdRefusal is the hold half: while any row is held, nothing below the
// running release. The same release, or a later one, is not a downgrade.
func holdRefusal(version, running string, held []registry.Route) error {
	if len(held) == 0 || version == running {
		return nil
	}
	if ok, _ := moves.ReleaseAtOrAbove(version, running); ok {
		return nil
	}
	if running == "" {
		return refuse("a participant in a held operation does not downgrade, and the running release is unknown")
	}
	return refuse("a participant in a held operation does not downgrade: %s is below the running %s", version, running)
}

// activationRule is the early, unlocked check (activationAllowed, and
// restoreFiles before its quiesce): the same two halves, read once, so a
// refused build is never staged and a refused restore never quiesces. It
// guards nothing by itself; the step it precedes runs under
// activationGuard.
func (a *systemApplier) activationRule(version string, running func() string) error {
	if err := a.seedsActivationRefusal(version); err != nil {
		return err
	}
	floor, err := readEffectiveMovesFloor(a.rootMoves, a.optDir, a.dataDir)
	if err != nil {
		return refuse("moves protocol floor: the floor on this box cannot be read: %v", err)
	}
	if floor == nil {
		a.ops.logf("system apply: no moves protocol floor is armed on this box; %q is not checked against one", version)
		return nil
	}
	if err := floorRefusal(version, a.activationFloor(floor, false)); err != nil {
		return err
	}
	run := running()
	if version == run {
		return nil
	}
	if ok, _ := moves.ReleaseAtOrAbove(version, run); ok {
		return nil
	}
	var held []registry.Route
	called := false
	err = a.freezeHolds(func(h []registry.Route) error { called = true; held = h; return nil })
	if !called {
		return refuse("a participant in a held operation does not downgrade, and the registry cannot say whether this box is one: %v", err)
	}
	return holdRefusal(version, run, held)
}

// floorAllowsForward is the floor half alone, for recovery's forward path:
// the gated build is already the one on disk, so only the floor can refuse
// it. Unarmed: allowed.
func (a *systemApplier) floorAllowsForward(version string) error {
	if err := a.seedsActivationRefusal(version); err != nil {
		return err
	}
	floor, err := readEffectiveMovesFloor(a.rootMoves, a.optDir, a.dataDir)
	if err != nil {
		return fmt.Errorf("moves protocol floor: the floor on this box cannot be read: %v", err)
	}
	floor = a.activationFloor(floor, false)
	if floor == nil {
		return nil
	}
	if ok, _ := moves.ReleaseAtOrAbove(version, floor.MinRelease); !ok {
		return fmt.Errorf("moves protocol floor: %s is below %s", version, floor.MinRelease)
	}
	return nil
}

// freezeHolds opens the registry (openRegistry, as the layout guard does)
// and runs fn with the held rows frozen (registry.WithHoldsFrozen).
func (a *systemApplier) freezeHolds(fn func(held []registry.Route) error) error {
	open := a.openRegistry
	if open == nil {
		open = func() (*registry.Registry, error) { return openSystemRegistry(a.dataDir) }
	}
	reg, err := open()
	if err != nil {
		return err
	}
	defer reg.Close()
	return reg.WithHoldsFrozen(fn)
}

func (a *systemApplier) prevBuild(ctx context.Context) string {
	out, err := a.guardRun(ctx, systemVersionWait, systemPrevPath(a.optDir), "version")
	if err != nil {
		return ""
	}
	return carlosVersionOf(out)
}

// stagePlan runs `<binary> bootstrap -root <dir> -plan-json` and returns
// that binary's arming plan and the unit files it staged. The plan is left
// in <dir>/plan.json next to them, so a later step can pick up the whole
// staging (stagedPlan) without exec'ing the binary a second time.
func (a *systemApplier) stagePlan(ctx context.Context, binary, dir string) (bootstrapPlan, []string, error) {
	os.RemoveAll(dir)
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return bootstrapPlan{}, nil, err
	}
	out, err := a.guardRun(ctx, systemBootstrapWait, binary, "bootstrap", "-root", dir, "-plan-json")
	if err != nil {
		return bootstrapPlan{}, nil, err
	}
	var plan bootstrapPlan
	if err := json.Unmarshal([]byte(out), &plan); err != nil {
		return bootstrapPlan{}, nil, fmt.Errorf("plan: %w", err)
	}
	entries, err := os.ReadDir(filepath.Join(dir, "etc", "systemd", "system"))
	if err != nil {
		return bootstrapPlan{}, nil, err
	}
	var units []string
	for _, e := range entries {
		if !e.IsDir() {
			units = append(units, e.Name())
		}
	}
	if err := writeJSONDurable(filepath.Join(dir, "plan.json"), plan, 0o600); err != nil {
		return bootstrapPlan{}, nil, err
	}
	return plan, units, nil
}

// layoutMountUnit is the unit whose presence in a candidate's staged set
// marks a release that understands the nested layout.
const layoutMountUnit = "data-instances.mount"

// systemApplyLayoutRefusal takes the release and the rows in its way.
const systemApplyLayoutRefusal = "%s predates the instance layout, and this box still has: %s. revert the layout and wait for every row to report flat before installing it"

// layoutReversalReasons opens the registry and lists what stands in the
// way of a release before the layout. A registry that will not open cannot
// vouch that nothing is nested, so its error is itself a reason.
func (a *systemApplier) layoutReversalReasons() []string {
	if _, required, err := a.rootMoves.read(); err != nil {
		return []string{err.Error()}
	} else if required {
		if err := rootFloorDowngradeIdle(context.Background()); err != nil {
			return []string{err.Error()}
		}
		snapshot, err := productionGenerationInventory(context.Background())
		if err != nil {
			return []string{err.Error()}
		}
		for _, view := range snapshot.Installations {
			if !view.Installation.Empty {
				return []string{errRootFloorHistory.Error()}
			}
		}
	}

	open := a.openRegistry
	if open == nil {
		open = func() (*registry.Registry, error) { return openSystemRegistry(a.dataDir) }
	}
	reg, err := open()
	if err != nil {
		return []string{err.Error()}
	}
	defer reg.Close()
	return layoutReversalIncomplete(reg, a.dataDir)
}

// layoutReversalIncomplete names, once each and sorted, every host a
// release before the layout would mishandle: a row whose paths are nested
// by shape (provision.NestedPaths, whatever its flags, so a row that left
// management is still seen), a row the layout pass's own classification
// calls fenced or in conflict, and a host with a migration journal whether
// or not its row still exists. A registry that cannot be listed, or a
// journal directory that cannot be read, is named as such rather than
// taken for an empty box.
func layoutReversalIncomplete(reg *registry.Registry, dataDir string) []string {
	named := map[string]bool{}
	var extra []string
	routes, err := reg.List()
	if err != nil {
		extra = append(extra, err.Error())
	}
	pass := layout.Pass{Registry: reg, DataDir: dataDir}
	for _, rt := range routes {
		if provision.NestedPaths(dataDir, rt) {
			named[rt.Host] = true
			continue
		}
		if l, _ := pass.Classify(rt, routes); l == activator.LayoutFenced || l == activator.LayoutConflict {
			named[rt.Host] = true
		}
	}
	entries, err := os.ReadDir(filepath.Join(dataDir, "layout"))
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		extra = append(extra, err.Error())
	}
	for _, e := range entries {
		if host, ok := strings.CutSuffix(e.Name(), ".json"); ok && !strings.HasPrefix(host, ".") && layout.JournalPresent(dataDir, host) {
			named[host] = true
		}
	}
	reasons := make([]string, 0, len(named)+len(extra))
	for h := range named {
		reasons = append(reasons, h)
	}
	slices.Sort(reasons)
	return append(reasons, extra...)
}

func (a *systemApplier) journal(j systemJournal) error {
	return writeJSONDurable(a.file("journal.json"), j, 0o600)
}

// installUnits removes what the target build dropped, copies its units in,
// then sysusers BEFORE tmpfiles — tmpfiles silently drops a line whose
// group does not resolve, and carlos-edge never starts again after the
// next boot — then daemon-reload. Never the staging root's data/: it holds
// a 0-byte registry.db.
func (a *systemApplier) installUnits(ctx context.Context, stagedRoot string, remove []string) error {
	claimSettings, settingsErr := readClaimUnitSettings(filepath.Join(a.etcDir, "carlos", "host.env"))
	if settingsErr != nil {
		a.ops.logf("system apply: instance-control unit settings: %v", settingsErr)
	}
	unitDir := filepath.Join(a.etcDir, "systemd", "system")
	if err := os.MkdirAll(unitDir, 0o755); err != nil {
		return err
	}
	for _, u := range remove {
		if _, err := a.ops.systemctl("disable", "--now", u); err != nil {
			a.ops.logf("system apply: disable %s: %v", u, err)
		}
		os.Remove(filepath.Join(unitDir, u))
	}
	beforeControl, err := os.ReadFile(filepath.Join(unitDir, claimControlUnit))
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}
	// A host.env this build cannot parse says nothing about the claim
	// (6b-N2). A control unit already rendered for an enabled box stays as
	// it is, unrestarted, as adopt leaves it; a flag-off one is rendered as
	// main renders it.
	keepControl := settingsErr != nil && bytes.Contains(beforeControl, []byte(claimControlEnvironment))
	entries, err := os.ReadDir(filepath.Join(stagedRoot, "etc", "systemd", "system"))
	if err != nil {
		return err
	}
	for _, e := range entries {
		if e.IsDir() {
			continue
		}
		if e.Name() == claimControlUnit {
			if keepControl {
				a.ops.logf("system apply: %s kept as installed: host.env cannot be parsed", claimControlUnit)
				continue
			}
			b, err := os.ReadFile(filepath.Join(stagedRoot, "etc", "systemd", "system", e.Name()))
			if err != nil {
				return err
			}
			if err := writeFileDurable(filepath.Join(unitDir, e.Name()), claimUnitContent(e.Name(), b, claimSettings), 0644); err != nil {
				return err
			}
			continue
		}
		if err := copyFileDurable(filepath.Join(stagedRoot, "etc", "systemd", "system", e.Name()), filepath.Join(unitDir, e.Name()), 0o644); err != nil {
			return err
		}
	}
	sys := filepath.Join(stagedRoot, "etc", "sysusers.d", "carlos.conf")
	if _, err := os.Stat(sys); err == nil {
		if err := os.MkdirAll(filepath.Join(a.etcDir, "sysusers.d"), 0o755); err != nil {
			return err
		}
		dst := filepath.Join(a.etcDir, "sysusers.d", "carlos.conf")
		if err := copyFileDurable(sys, dst, 0o644); err != nil {
			return err
		}
		if _, err := a.ops.run(ctx, systemVersionWait, "systemd-sysusers", dst); err != nil {
			return fmt.Errorf("systemd-sysusers: %w", err)
		}
	}
	// The journal cap (discussion #111). This is how a box that was
	// bootstrapped before the cap existed actually gets it: the drop-in is a
	// named file in the staged etc/ tree, exactly like sysusers and tmpfiles
	// below, and copying it here means the fleet converges on the cap through
	// the self-update relay rather than one box at a time by hand.
	//
	// Stat-guarded like sysusers: a ROLLBACK stages carlos.prev, and a
	// previous build whose bootstrap wrote no drop-in must leave whatever is
	// on the box alone rather than fail the restore.
	//
	// journald is restarted, not just reloaded, because that is the only way
	// it re-reads its configuration; failure is logged and the apply
	// continues, since an uncapped journal must not roll back a good build.
	jrnl := filepath.Join(stagedRoot, "etc", "systemd", "journald.conf.d", journaldConfName)
	if _, err := os.Stat(jrnl); err == nil {
		if err := os.MkdirAll(filepath.Join(a.etcDir, "systemd", "journald.conf.d"), 0o755); err != nil {
			return err
		}
		if err := copyFileDurable(jrnl, filepath.Join(a.etcDir, "systemd", "journald.conf.d", journaldConfName), 0o644); err != nil {
			return err
		}
		if _, err := a.ops.systemctl(journaldReload...); err != nil {
			a.ops.logf("system apply: restart systemd-journald: %v (continuing; the journal cap takes effect at the next restart)", err)
		}
	}
	tmp := filepath.Join(stagedRoot, "etc", "tmpfiles.d", "carlos.conf")
	if err := os.MkdirAll(filepath.Join(a.etcDir, "tmpfiles.d"), 0o755); err != nil {
		return err
	}
	dst := filepath.Join(a.etcDir, "tmpfiles.d", "carlos.conf")
	if err := copyFileDurable(tmp, dst, 0o644); err != nil {
		return err
	}
	if _, err := a.ops.run(ctx, systemVersionWait, "systemd-tmpfiles", "--create", dst); err != nil {
		a.ops.logf("system apply: systemd-tmpfiles: %v (continuing, as the script does)", err)
	}
	if _, err = a.ops.systemctl("daemon-reload"); err != nil {
		return err
	}
	afterControl, err := os.ReadFile(filepath.Join(unitDir, claimControlUnit))
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}
	return restartClaimControl(filepath.Join(unitDir, claimControlUnit), beforeControl, afterControl, a.ops.systemctl)
}

// restartEdge resets the failure counter FIRST — a unit that hit its start
// limit refuses restart, and so would the rollback — then restarts.
func (a *systemApplier) restartEdge() error {
	a.ops.systemctl("reset-failed", systemEdgeUnit)
	_, err := a.ops.systemctl("restart", systemEdgeUnit)
	return err
}

// restartInstanceControl restarts carlos-instance-control.service, the
// root service that starts every nested instance (design §5). It is
// socket-activated and then stays up: on the flagship it had run the
// 2026-09-16 binary for three weeks when v0.25.1 was installed around it
// (discussion #256). A control service of the previous release serves the
// new release's unit templates, which bind sources only the new code
// writes (/run/carlos-control/exe/<iid>/), so every nested start failed
// at namespace setup and the probe unit, run by the same stale service,
// failed the same way: the box reported not ready with the right binary
// on disk and nothing to say why.
//
// Order: after the new units are installed and daemon-reload has run, and
// BEFORE the edge restarts. The new service then runs its start-up probe
// against the old edge, which is up, so the verdict it records is real
// (a probe with no edge cgroup fails on cgroup-ro); and by the time the
// new agent starts, the service it calls is already the new build, so
// there is no window in which a new agent asks an old service to start
// units it cannot. A wake in flight at this moment is lost, and that is
// safe: the edge restart one step later stops every instance anyway
// (BindsTo=carlos-edge.service), so the wake fails either way; the
// systemctl start the dropped request was running is a job systemd
// finishes on its own; the agent reads the dropped connection as a failed
// wake and sends the stop, which the socket unit, up throughout, queues
// for the restarted service. Flat hosts never pass through this service.
//
// A build that ships no control service (a downgrade below v0.20, whose
// unit installUnits has just disabled and removed) has nothing to
// restart; that is the one case this returns nil without calling
// systemctl. reset-failed first, as restartEdge does, so a unit that hit
// its start limit can be restarted at all.
func (a *systemApplier) restartInstanceControl() error {
	if _, err := os.Stat(filepath.Join(a.etcDir, "systemd", "system", systemControlUnit)); err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil
		}
		return err
	}
	a.ops.systemctl("reset-failed", systemControlUnit)
	_, err := a.ops.systemctl("restart", systemControlUnit)
	return err
}

// restartRelay mirrors restartEdge for the relay (relay spec §3), with one
// difference that decides its shape: the relay is optional. A box with no
// /etc/carlos/relay-masters must not be handed a relay it never asked for.
//
// What keeps that promise is the unit's own ConditionPathExists, not the
// verb: on an unarmed box `restart` starts nothing, because the condition
// fails and systemd records the unit as skipped — which is exactly what
// bootstrap already relies on when it starts the unit outright. So the verb
// here is `restart`, the same one restartEdge uses, and `try-restart` is
// wrong for a reason that only shows up on a bad day: reset-failed turns a
// crash-looped relay from failed into inactive, and try-restart does nothing
// at all to an inactive unit. A relay that died at boot would then stay down
// until the NEXT platform update happened to run arm(), with the box's
// heartbeat the only sign.
//
// The error is still logged and swallowed. A relay that will not start is not
// a failure of the update — the edge, the binary and the gate are all fine —
// and `carlos relay status` is where a dead region is meant to be read.
func (a *systemApplier) restartRelay() {
	a.ops.systemctl("reset-failed", systemRelayUnit)
	if _, err := a.ops.systemctl("restart", systemRelayUnit); err != nil {
		a.ops.logf("system apply: relay restart: %v", err)
	}
}

// gate: within gateWait the edge is active, answers on loopback, and the
// installed binary reports the target; then a soak across which NRestarts
// stays 0 — step 9's reset-failed zeroed it, so anything above 0 is a
// crash the new build caused — and ActiveEnterTimestampMonotonic does not
// move. A build that starts, answers, and dies a second later is caught
// here, and so is one that crashed once and then settled.
func (a *systemApplier) gate(ctx context.Context, target string) error {
	wait, soak := a.gateWait, a.soak
	if wait <= 0 {
		wait = systemGateWaitFor(a.claims)
	}
	if soak <= 0 {
		soak = systemGateSoak
	}
	deadline := a.ops.now().Add(wait)
	var lastErr error
	for {
		var state string
		state, lastErr = a.gateCheck(ctx, target)
		if lastErr == nil {
			break
		}
		if a.claims {
			if err := a.gateFailFast(state, lastErr); err != nil {
				return err
			}
		}
		if !a.ops.now().Before(deadline) {
			return lastErr
		}
		a.ops.sleep(time.Second)
	}
	before, err := a.edgeRunState()
	if err != nil {
		return err
	}
	if err := before.steady(); err != nil {
		return err
	}
	a.ops.sleep(soak)
	after, err := a.edgeRunState()
	if err != nil {
		return err
	}
	if err := after.steady(); err != nil {
		return err
	}
	if before != after {
		return fmt.Errorf("edge restarted during the soak (%s → %s)", before, after)
	}
	return a.gateOnce(ctx, target)
}

// systemApplyGate is how RunSystemApply reads claims mode for the gate: the
// unit's CARLOS_HOST_CLAIM through hostwrite.Enabled, the test the agent
// and the claim wiring make, and the window that follows from it.
func systemApplyGate(getenv func(string) string) (bool, time.Duration) {
	claims := hostwrite.Enabled(getenv)
	return claims, systemGateWaitFor(claims)
}

// systemGateWaitFor is the gate's window: systemClaimsGateWait with
// CARLOS_HOST_CLAIM on, systemGateWait without. The install's gate and the
// rollback's gate both use it, so a claims-on box can roll back to a
// previous build that also needs more than 20 s to listen.
func systemGateWaitFor(claims bool) time.Duration {
	if claims {
		return systemClaimsGateWait
	}
	return systemGateWait
}

// gateFailFast is what makes the claims window "wait while the same process
// is still starting" rather than "wait 120 s for anything". After a failed
// check it returns an error, ending the gate before its deadline, when the
// edge unit reports a failed or inactive state (from the is-active the
// check just made) or NRestarts above 0 (from the same systemctl show the
// soak reads). A build that crashes rolls back in seconds; one that is
// still starting, active with no restart, gets the whole window. A show
// that cannot be read says nothing either way, so the wait goes on to the
// deadline. Claims-off boxes never call it.
func (a *systemApplier) gateFailFast(state string, cause error) error {
	switch state {
	case "failed", "inactive":
		return fmt.Errorf("%w (failing fast: %s is %s)", cause, systemEdgeUnit, state)
	}
	rs, err := a.edgeRunState()
	if err != nil {
		return nil
	}
	if err := rs.steady(); err != nil {
		return fmt.Errorf("%w (failing fast: %v)", cause, err)
	}
	return nil
}

// edgeRunState is the edge's NRestarts and ActiveEnterTimestampMonotonic,
// in that order — the order `show -p` is asked for them.
type edgeRunState struct{ restarts, activeEnter string }

func (s edgeRunState) String() string { return s.restarts + "/" + s.activeEnter }

// steady is the half of the soak that a comparison cannot make: a restart
// that happened before the first sample is still a restart.
func (s edgeRunState) steady() error {
	if s.restarts != "0" {
		return fmt.Errorf("edge has crash-restarted since the install (NRestarts=%s)", s.restarts)
	}
	return nil
}

func (a *systemApplier) edgeRunState() (edgeRunState, error) {
	out, err := a.ops.systemctl("show", "-p", "NRestarts", "-p", "ActiveEnterTimestampMonotonic", "--value", systemEdgeUnit)
	if err != nil {
		return edgeRunState{}, err
	}
	restarts, rest, _ := strings.Cut(strings.TrimSpace(out), "\n")
	return edgeRunState{restarts: strings.TrimSpace(restarts), activeEnter: strings.TrimSpace(rest)}, nil
}

func (a *systemApplier) gateOnce(ctx context.Context, target string) error {
	_, err := a.gateCheck(ctx, target)
	return err
}

// gateCheck is gateOnce that also returns what is-active said about the
// edge, trimmed, for gateFailFast. The checks, their order and their
// errors are gateOnce's.
func (a *systemApplier) gateCheck(ctx context.Context, target string) (string, error) {
	out, err := a.ops.systemctl("is-active", systemEdgeUnit)
	state := strings.TrimSpace(out)
	if err != nil || state != "active" {
		if err != nil {
			return state, fmt.Errorf("%s is %s: %w", systemEdgeUnit, state, err)
		}
		return state, fmt.Errorf("%s is %s", systemEdgeUnit, state)
	}
	if err := a.ops.probeEdge(); err != nil {
		return state, fmt.Errorf("edge probe: %w", err)
	}
	if got := a.onDiskBuild(ctx); got != target {
		return state, fmt.Errorf("installed binary reports %q, not %s", got, target)
	}
	return state, nil
}

// arm enables every unit in the plan and starts the ones the plan says to
// start: the unit types whose enable-without-start is a silent no-op until
// the next boot, and of the services only carlos-relay, whose
// ConditionPathExists makes the start a no-op on a box with no region master
// (relay spec §3).
//
// It never starts the edge, which is running this very code — that is why
// apply can call it from inside itself. Before #118 that was structural: the
// applier derived the start list itself and could not name a service other
// than the relay. Now the list comes from the target build, so the guarantee
// is enforced here instead, explicitly.
func (a *systemApplier) arm(plan bootstrapPlan) error {
	// plan.Armed is the TARGET build's answer to "and which of these get a
	// start" (discussion #118). Preferring it is what lets a build introduce
	// a unit that must be started at apply time rather than at the next
	// boot; falling back to the local rule keeps a plan.json from an older
	// build — which is what a rollback stages — working unchanged.
	armed := plan.Armed
	if armed == nil {
		armed = armedUnits(plan.Enabled)
	}
	// Two guards on a list that now comes from another build. The edge is
	// never started here, whatever the plan says. And a unit that is not in
	// Enabled is dropped rather than passed to systemctl: `start` is ONE
	// batched call, so a single name this box has no unit file for would
	// fail the whole call and leave nothing armed at all.
	enabled := make(map[string]bool, len(plan.Enabled))
	for _, u := range plan.Enabled {
		enabled[u] = true
	}
	kept := armed[:0:0]
	for _, u := range armed {
		switch {
		case u == systemEdgeUnit || u == systemEdgeUnit+".service":
			a.ops.logf("system apply: arm: refusing to start %s, which is running this code", u)
		case !enabled[u]:
			a.ops.logf("system apply: arm: %s is armed by the plan but not enabled by it; skipping", u)
		default:
			kept = append(kept, u)
		}
	}
	for _, args := range systemctlPlanArming(plan.Enabled, kept) {
		if _, err := a.ops.systemctl(args...); err != nil {
			return fmt.Errorf("systemctl %s: %w", strings.Join(args, " "), err)
		}
	}
	return nil
}

// rollback is step 13: back to carlos.prev and its units, restart, gate,
// arm the previous plan. applied.json is untouched — the .prev build is
// what passed a gate, and it is what applied.json already names.
//
// A rollback that cannot finish leaves the journal open: the box is then
// between two builds — the previous binary under the failed build's units,
// say — and only another recovery puts it right. The one exception is a
// complete restore whose GATE failed: the files are all the previous
// build's, restoring them again would change nothing, and the box needs a
// human rather than a loop.
func (a *systemApplier) rollback(ctx context.Context, req systemRequest, reason string, j *systemJournal) error {
	a.ops.logf("system apply: %s %s: rolling back: %s", a.channel, j.Target, reason)
	failed := func(what string, err error) error {
		_ = a.writeResult(req, systemOutcomeRollbackFailed, reason+"; "+what+": "+err.Error())
		return fmt.Errorf("system apply: rollback of %s failed: %s: %v", j.Target, what, err)
	}
	plan, err := a.restoreFiles(ctx, j)
	if err != nil {
		return failed("restore", err) // journal kept: the restore is half done
	}
	// carlos.prev, now on disk: what this rollback's journal finish makes
	// current, checked again under activationGuard at that finish.
	restored, err := a.restoreVersion()
	if err != nil {
		return failed("identity", err)
	}
	// The previous binary is on disk, so the control service goes back
	// too, before the edge, as on the way forward. Logged, not fatal: a
	// rollback's job is a good edge, and a control service that would
	// not restart is caught by the agent's readiness check (it refuses
	// nested wakes until the service runs the installed binary).
	if err := a.restartInstanceControl(); err != nil {
		a.ops.logf("system apply: rollback: instance-control restart: %v (nested wakes refuse until it is restarted)", err)
	}
	if err := a.restartEdge(); err != nil {
		return failed("restart", err) // journal kept: the edge is still to come back
	}
	a.restartRelay()
	if err := a.gate(ctx, j.PrevBuild); err != nil {
		err := failed("previous build failed its gate too", err)
		// The journal finish, under activationGuard (moves floor and hold
		// rule first), closed through closeGuardRollback (the claim release
		// floor's completion of a rolled-back transaction) inside it.
		var closeErr error
		if gerr := a.activationGuard(ctx, restored, restored, func(ctx context.Context) error {
			closeErr = a.closeGuardRollback(ctx)
			return nil
		}); gerr != nil {
			return fmt.Errorf("%v; the journal stays: %v", err, gerr)
		}
		if closeErr != nil {
			return errors.Join(err, closeErr)
		}
		return err
	}
	if err := a.arm(plan); err != nil {
		a.ops.logf("system apply: arm previous plan: %v", err)
	}
	// closeGuardRollback's own error is main's: returned as it is, not as a
	// rollback failure (the journal is already gone by then, unless the
	// armed floor refused to record the rollback, which keeps it).
	var closeErr error
	if err := a.activationGuard(ctx, restored, restored, func(ctx context.Context) error {
		if err := a.writeResult(req, systemOutcomeRolledBack, reason); err != nil {
			return err
		}
		closeErr = a.closeGuardRollback(ctx)
		return nil
	}); err != nil {
		return failed("finish", err) // journal kept
	}
	if closeErr != nil {
		return closeErr
	}
	return fmt.Errorf("system apply: %s rolled back: %s", j.Target, reason)
}

// restoreFiles is the files-only half shared by rollback and recovery:
// carlos.prev over carlos (keeping .prev), units_added disabled and
// removed, the previous unit set copied back (bringing units_removed back),
// sysusers, tmpfiles, daemon-reload, and `enable` of the previous plan so
// it comes up at the next boot. No restart, no start: what to do about the
// running edge is the caller's, and at boot the answer is nothing.
func (a *systemApplier) restoreFiles(ctx context.Context, j *systemJournal) (bootstrapPlan, error) {
	prev := systemPrevPath(a.optDir)
	if err := a.guardImage(prev); err != nil {
		return bootstrapPlan{}, err
	}
	if _, err := os.Stat(prev); err != nil {
		return bootstrapPlan{}, fmt.Errorf("no carlos.prev: %w", err)
	}
	// The moves floor and the hold rule (moves spec §11), before anything
	// on the box changes, for the version carlos.prev's sidecar records (no
	// sidecar is below any armed floor). A refusal keeps the journal: the
	// box stays on the build it is on, and the caller reports it.
	prevVersion, err := a.restoreVersion()
	if err != nil {
		return bootstrapPlan{}, err
	}
	running := a.onDiskBuild(ctx)
	if _, required, err := a.rootMoves.read(); err != nil {
		return bootstrapPlan{}, err
	} else if required {
		if current, err := readRootFloorBinary(a.optDir, systemBinaryPath(a.optDir)); err == nil {
			running = current.Version
		}
	}
	if err := a.activationRule(prevVersion, func() string { return running }); err != nil {
		return bootstrapPlan{}, err
	}
	// The same quiesce the install did: a root pass that is mid-flight
	// against the failed build must not run on across the swap back.
	prior := a.quiesce()
	// carlos.prev is copied beside the binary first, outside any lock, and
	// renamed over it under the activation guard, so the rule's inputs are
	// held still for a rename rather than a whole binary copy.
	restore := systemBinaryPath(a.optDir) + ".restore"
	if err := copyFileDurable(prev, restore, 0o755); err != nil {
		return bootstrapPlan{}, err
	}
	renamed := false
	if err := a.activationGuard(ctx, prevVersion, running, func(ctx context.Context) error {
		if err := a.rootProtocolAllowsBinary(restore); err != nil {
			return err
		}
		renamed = true
		if err := os.Rename(restore, systemBinaryPath(a.optDir)); err != nil {
			return err
		}
		return syncDir(a.optDir)
	}); err != nil {
		os.Remove(restore)
		// Stopped before the rename (refused or deferred: the floor or a hold
		// changed during the copy, or a writer holds the floor lock): nothing
		// was restored, so the paused triggers go back to how they were. If
		// they cannot be started, they go to rearm-pending.json, captured as
		// they were before the quiesce, and every run starts them first,
		// before any early refusal. The caller keeps the journal.
		if !renamed {
			if rerr := a.rearm(prior); rerr != nil {
				if nerr := a.notePendingRearm(prior); nerr != nil {
					err = fmt.Errorf("%w; the root passes paused for the restore could not be started again: %v, and the retry could not be recorded: %v", err, rerr, nerr)
				} else {
					err = fmt.Errorf("%w; the root passes paused for the restore could not be started again: %v", err, rerr)
				}
			}
		}
		return bootstrapPlan{}, err
	}
	// The unit set to put back is whatever the binary being restored ships,
	// so it comes from carlos.prev and never from the journal. The install
	// already staged it, and that staging is preferred: re-running a
	// bootstrap on a box that is mid-rollback is one more thing that can
	// fail, and it would wipe the good units before it tried.
	prevDir := filepath.Join(a.stagingDir(), "prev")
	plan, err := stagedPlan(prevDir)
	if err != nil {
		if plan, _, err = a.stagePlan(ctx, prev, prevDir); err != nil {
			return bootstrapPlan{}, fmt.Errorf("carlos.prev bootstrap: %w", err)
		}
	}
	if err := a.installUnits(ctx, prevDir, j.UnitsAdded); err != nil {
		return bootstrapPlan{}, err
	}
	if len(plan.Enabled) > 0 {
		if _, err := a.ops.systemctl(append([]string{"enable"}, plan.Enabled...)...); err != nil {
			return bootstrapPlan{}, err
		}
	}
	return plan, nil
}

// stagedPlan reads back what a stagePlan into dir left there: the plan the
// binary printed and the units it wrote. An empty unit directory or a
// missing plan.json means the staging is not there to be reused.
func stagedPlan(dir string) (bootstrapPlan, error) {
	entries, err := os.ReadDir(filepath.Join(dir, "etc", "systemd", "system"))
	if err != nil {
		return bootstrapPlan{}, err
	}
	if len(entries) == 0 {
		return bootstrapPlan{}, fmt.Errorf("no units staged in %s", dir)
	}
	var plan bootstrapPlan
	if err := readJSONFile(filepath.Join(dir, "plan.json"), &plan); err != nil {
		return bootstrapPlan{}, err
	}
	return plan, nil
}

// unitsMinus is have − drop, order preserved.
func unitsMinus(have, drop []string) []string {
	skip := map[string]bool{}
	for _, d := range drop {
		skip[d] = true
	}
	var out []string
	for _, h := range have {
		if !skip[h] {
			out = append(out, h)
		}
	}
	return out
}

// copyFileDurable is writeFileDurable for a file too large to hold in
// memory — the binary itself — and lands the same way.
func copyFileDurable(src, dst string, mode os.FileMode) error {
	in, err := os.Open(src)
	if err != nil {
		return err
	}
	defer in.Close()
	return writeDurable(dst, mode, func(w io.Writer) error {
		_, err := io.Copy(w, in)
		return err
	})
}

// recoverIfNeeded is step 0: an open journal means a previous run was cut
// off, and nothing else happens until the box is whole again. Before the
// gate → back, files only, restarting the edge only when the edge has
// already started this boot (at boot, systemd is about to start it and a
// restart from an ordering-before unit would deadlock). At gated → forward:
// that build proved itself, and undoing it would throw away a good apply.
// The journal is removed LAST, after the result is written and the restart
// issued, so a recovery that dies partway is recovered again.
//
// It reports whether it moved the box: a run that recovered has written the
// result it answers with, and does not then go on to apply a request.
func (a *systemApplier) recoverIfNeeded(ctx context.Context) (bool, error) {
	j, err := readSystemJournal(a.optDir)
	if err != nil || j == nil {
		return false, err
	}
	// The result belongs to the promote being recovered, which the journal
	// names in full. The request is used only when it is that same promote,
	// version and promoted_at both: a re-promote of the same version is a
	// new tuple, and recording this rollback against it would refuse the
	// operator's retry before it started.
	req, _ := a.readRequest()
	if req.Version != j.Target || !req.PromotedAt.Equal(j.promoted()) {
		req = systemRequest{Version: j.Target, PromotedAt: j.promoted()}
	}
	a.ops.logf("system apply: recovering an apply of %s interrupted at %s", j.Target, j.Phase)
	if j.Phase == systemPhaseBackingUp {
		// A legacy journal: this version no longer writes one for the
		// backup (see decidePrev), but a box killed mid-apply by the
		// version that did comes up running this one. Nothing on the box
		// had changed yet — only a half-written .prev copy can be left,
		// and it is not the file anything reads. The run carries on to the
		// request it was started for.
		os.Remove(systemPrevPath(a.optDir) + ".tmp")
		return false, a.closeGuardJournal(ctx)
	}
	// Forward from gated — but only while the build that passed the gate is
	// still the one on disk. A binary that has moved since means this is a
	// pre-gate state whatever the journal says.
	if j.Phase == systemPhaseGated {
		if got := a.onDiskBuild(ctx); got != j.Target {
			a.ops.logf("system apply: recovery: %s passed its gate but the binary on disk reports %q; rolling back instead", j.Target, got)
		} else if err := a.floorAllowsForward(j.Target); err != nil {
			// A floor raised since the gate (moves spec §11; fix round 1): the
			// build is not finished forward. Back to carlos.prev, which
			// restoreFiles judges in turn; if that is refused too, the
			// journal stays and the box stays as it is.
			a.ops.logf("system apply: recovery: %s passed its gate but %v; rolling back instead", j.Target, err)
		} else {
			plan, _, err := a.stagePlan(ctx, systemBinaryPath(a.optDir), filepath.Join(a.stagingDir(), "new"))
			if err != nil {
				return false, fmt.Errorf("system apply: recovery: %w", err)
			}
			// Finishing the journal is the forward path's activation: the
			// rule again, under the guard, at the step itself, since the
			// floor can rise while the plan is staged (fix round 2).
			finished := false
			err = a.activationGuard(ctx, j.Target, j.Target, func(ctx context.Context) error {
				if err := a.rootProtocolAllowsBinary(systemBinaryPath(a.optDir)); err != nil {
					return err
				}
				finished = true
				return a.finish(ctx, req, j.Target, plan)
			})
			if finished {
				return true, err
			}
			var deferred activationDeferred
			if errors.As(err, &deferred) {
				return false, fmt.Errorf("system apply: recovery: %w", err) // journal kept
			}
			a.ops.logf("system apply: recovery: %s passed its gate but %v; rolling back instead", j.Target, err)
		}
	}
	plan, err := a.restoreFiles(ctx, j)
	if err != nil {
		return false, fmt.Errorf("system apply: recovery: %w", err) // journal kept: the next run recovers again
	}
	// carlos.prev, now on disk: what the finish below makes current.
	restored, err := a.restoreVersion()
	if err != nil {
		return false, err
	}
	if a.ops.edgeStartedThisBoot() {
		// The control service first, as in install and rollback. Not at
		// boot: it is socket-activated and not yet running, and started
		// before the edge it would run its probe against an edge whose
		// cgroup does not exist yet and record a failing verdict for the
		// next five minutes. Blocking is safe: nothing orders this unit
		// against carlos-system-recover.service.
		if err := a.restartInstanceControl(); err != nil {
			a.ops.logf("system apply: recovery: instance-control restart: %v (nested wakes refuse until it is restarted)", err)
		}
		a.ops.systemctl("reset-failed", systemEdgeUnit)
		// --no-block: this can be running as the edge's own OnFailure=
		// handler, and a blocking restart of the unit that triggered it
		// would wait on the job that is waiting on this one.
		if _, err := a.ops.systemctl("restart", "--no-block", systemEdgeUnit); err != nil {
			return false, fmt.Errorf("system apply: recovery: restart: %w", err)
		}
		var armed []string
		for _, u := range plan.Enabled {
			if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") {
				armed = append(armed, u)
			}
		}
		if len(armed) > 0 {
			a.ops.systemctl(append([]string{"start"}, armed...)...)
		}
	}
	// The backward finish is guarded like every other: a floor raised since
	// the restore's rename keeps the journal, and the next run judges again.
	// closeGuardRollback's own error is main's, returned as main returns it.
	var closeErr error
	if err := a.activationGuard(ctx, restored, restored, func(ctx context.Context) error {
		if err := a.writeResult(req, systemOutcomeRolledBack, "interrupted at "+j.Phase); err != nil {
			return err
		}
		closeErr = a.closeGuardRollback(ctx)
		return nil
	}); err != nil {
		return false, fmt.Errorf("system apply: recovery: %w", err) // journal kept
	}
	return true, closeErr
}

// record turns a refusal or deferral into result.json and a nil return.
func (a *systemApplier) record(req systemRequest, err error) error {
	outcome := systemOutcomeRefused
	if errors.Is(err, errSystemDeferred) {
		outcome = systemOutcomeDeferred
	}
	a.ops.logf("system apply: %s %s: %s: %v", a.channel, req.Version, outcome, err)
	return a.writeResult(req, outcome, err.Error())
}

func (a *systemApplier) writeResult(req systemRequest, outcome, reason string) error {
	return writeJSONDurable(a.file("result.json"), systemResult{
		Channel: a.channel, Version: req.Version, PromotedAt: req.PromotedAt,
		Outcome: outcome, Reason: reason, At: a.ops.now().UTC(),
	}, 0o644)
}

// readRequest is the one read under /data: a fixed path reached without
// following a single symlink, a regular file under 4 KiB, a version that
// passes validSystemVersion. Anything else is an error, and root acts on
// nothing.
func (a *systemApplier) readRequest() (systemRequest, error) {
	f, err := openBeneath(a.dataDir, "system", systemComponent, "requested.json")
	if err != nil {
		return systemRequest{}, err
	}
	defer f.Close()
	st, err := f.Stat()
	if err != nil {
		return systemRequest{}, err
	}
	// openBeneath already checked the inode; bound the request size too.
	if !st.Mode().IsRegular() || st.Size() > systemMaxRequest {
		return systemRequest{}, fmt.Errorf("request: not a regular file under %d bytes", systemMaxRequest)
	}
	var req systemRequest
	if err := json.NewDecoder(io.LimitReader(f, systemMaxRequest)).Decode(&req); err != nil {
		return systemRequest{}, fmt.Errorf("request: %w", err)
	}
	if !validSystemVersion(req.Version) {
		return systemRequest{}, fmt.Errorf("request: version %q is not a version", req.Version)
	}
	if req.PromotedAt.IsZero() {
		return systemRequest{}, errors.New("request: promoted_at unset")
	}
	return req, nil
}

// verifyPointer fetches THIS box's channel pointer and verifies it with
// the release public key. The channel is root's own (host.env), never the
// request's. A store error is deferred, a bad pointer refused.
//
// This is where the store is opened and the pubkey parsed — a store the box
// cannot reach, or an unset CARLOS_RELEASE_PUBKEY, is deferred like any
// other "could not read the pointer", recorded in result.json where an
// operator sees it, rather than exiting before the run began.
func (a *systemApplier) verifyPointer(ctx context.Context) (*release.SystemPointer, error) {
	store, pub, err := a.openStore(ctx)
	if err != nil {
		return nil, fmt.Errorf("%w: %v", errSystemDeferred, err)
	}
	a.store = store
	p, _, err := release.GetSystemPointer(ctx, store, pub, release.SystemComponentRouter, a.channel)
	switch {
	case err == nil:
		if _, required, e := a.rootMoves.read(); e != nil {
			return nil, e
		} else if required {
			if e := rememberRootFloorSystemIdentity(ctx, store, pub, a.optDir, p); e != nil {
				return nil, e
			}
		}
		return p, nil
	case errors.Is(err, os.ErrNotExist):
		return nil, fmt.Errorf("%w: no %s pointer for this box's channel", errSystemRefused, a.channel)
	case errors.Is(err, release.ErrBadSystemPointer):
		return nil, fmt.Errorf("%w: %v", errSystemRefused, err)
	default:
		return nil, fmt.Errorf("%w: %v", errSystemDeferred, err)
	}
}

// onDiskBuild asks the installed binary what it is, with a deadline. Not
// version.Build: after an interrupted apply the two can differ.
func (a *systemApplier) onDiskBuild(ctx context.Context) string {
	out, err := a.guardRun(ctx, systemVersionWait, systemBinaryPath(a.optDir), "version")
	if err != nil {
		return ""
	}
	return carlosVersionOf(out)
}

// carlosVersionOf reads `carlos version` output — exactly "carlos <build>",
// nothing else. A build that merely starts with the expected version, or
// output with anything else on the line, yields "" and matches nothing.
func carlosVersionOf(out string) string {
	v, ok := strings.CutPrefix(strings.TrimSpace(out), "carlos ")
	if !ok {
		return ""
	}
	return v
}

// stageCandidate copies the agent's cache into staging/ (0700) as 0600,
// verifies the copy's sha against the pointer, makes it executable, and
// runs its `version`. Nothing under /data is read after the copy.
func (a *systemApplier) stageCandidate(ctx context.Context, p *release.SystemPointer, req systemRequest) (string, error) {
	art, ok := p.Current.Artifacts[a.platform]
	if !ok {
		return "", fmt.Errorf("pointer names no artifact for %s", a.platform)
	}
	os.RemoveAll(a.stagingDir())
	if err := os.MkdirAll(a.stagingDir(), 0o700); err != nil {
		return "", err
	}
	if err := os.Chmod(a.stagingDir(), 0o700); err != nil {
		return "", err
	}
	src, err := openBeneath(a.dataDir, systemCacheParts(req.Version)...)
	if err != nil {
		return "", fmt.Errorf("candidate: %w", err)
	}
	defer src.Close()
	// openBeneath checked the inode before opening it for I/O.
	if st, err := src.Stat(); err != nil || !st.Mode().IsRegular() {
		return "", errors.New("candidate: not a regular file")
	}
	staged := filepath.Join(a.stagingDir(), "carlos")
	dst, err := os.OpenFile(staged, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return "", err
	}
	h := sha256.New()
	n, err := io.Copy(io.MultiWriter(dst, h), io.LimitReader(src, systemArtifactLimit+1))
	if cerr := dst.Close(); err == nil {
		err = cerr // a close that fails is a copy that did not land
	}
	if err != nil {
		return "", fmt.Errorf("candidate: copying into staging: %w", err)
	}
	if n > systemArtifactLimit {
		return "", fmt.Errorf("candidate: larger than %d bytes", systemArtifactLimit)
	}
	if got := hex.EncodeToString(h.Sum(nil)); got != art.SHA256 {
		return "", fmt.Errorf("candidate: sha256 %s does not match the signed %s", got, art.SHA256)
	}
	if err := os.Chmod(staged, 0o755); err != nil {
		return "", err
	}
	out, err := a.guardRun(ctx, systemVersionWait, staged, "version")
	if err != nil {
		return "", fmt.Errorf("candidate: running its version: %w", err)
	}
	if carlosVersionOf(out) != p.Current.Version {
		return "", fmt.Errorf("candidate reports %q, not %s", strings.TrimSpace(out), p.Current.Version)
	}
	return staged, nil
}

func (a *systemApplier) restoreVersion() (string, error) {
	if _, required, err := a.rootMoves.read(); err != nil {
		return "", err
	} else if required {
		identity, err := readRootFloorBinary(a.optDir, systemPrevPath(a.optDir))
		return identity.Version, err
	}
	version, _ := readLine(prevVersionPath(a.optDir))
	return version, nil
}