// Package host is the box side of the platform: `carlos bootstrap` provisions
// a fresh machine (units, dirs, service user), `carlos agent` runs the
// single-box role (edge + activator + release adoption in one process), and
// `carlos adopt` is one immediate adoption pass for the timer unit.
//
// Bootstrap ported from Eleven's bootstrap.go (elevenmessenger/messenger:
// bootstrap.go), radically slimmed: carlos has one long-running unit (the
// edge/agent) plus timers — no sibling-service zoo, no collectors, no SSH
// hardening verbs (those stay app-level concerns).
package host

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

	"bytes"
	"embed"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"os/user"
	"path/filepath"
	"strconv"
	"strings"
)

// The systemd units ship inside the binary (source of truth:
// internal/host/units), so `carlos bootstrap` can provision a fresh box with
// no separate files to copy. Ported from Eleven's deploy/systemd (renamed,
// same hardening) and titogo's deploy/systemd (litestream + restore-verify).
//
//go:embed units/*.service units/*.timer units/*.socket units/*.path units/*.slice units/*.mount
var unitsFS embed.FS

// serviceUser is the unprivileged account the edge unit runs as; bootstrap
// creates it and hands it the data + runtime dirs. The edge keeps
// CAP_NET_BIND_SERVICE to still bind :80/:443.
const serviceUser = "carlos"

// socketsGroup is the blessed cross-user consumers group for exec-backed
// instance sockets (issue #103): a host app's own process (titogo's router
// was the finding) enrolled here may dial every instance socket on the box.
// A DEDICATED group, never serviceUser's own: group `carlos` is also the
// group on /data (0750), so enrolling a consumer there would grant
// group-read of every instance database and env file — every
// CARLOS_ADMIN_TOKEN on the box — when the grant's whole meaning is "may
// dial sockets".
const socketsGroup = "carlos-sockets"

// tmpfilesConf keeps /run/carlos across reboots (/run is tmpfs). Owned by the
// service user so the non-root units can bind their sockets there; group is
// the consumers group with SETGID so every socket an exec child binds inside
// inherits it — that inheritance is what lets the activator's post-health
// 0660 normalization (backend_exec.go) grant exactly the enrolled consumers
// and nobody else, with no app aware the group exists. The apps subdir holds
// the carlos-app@ tenant sockets: root systemd binds them (SocketUser=carlos,
// 0600), tenants never need path access (the fd is passed in) — it stays
// carlos:carlos, so consumer enrollment deliberately does not reach unit
// tenants. other=0 on both dirs keeps every unenrolled uid out.
const tmpfilesConf = "d /data/.parked 2770 " + serviceUser + " " + serviceUser + " -\n" +
	"d /run/carlos 2750 " + serviceUser + " " + socketsGroup + " -\n" +
	"d /run/carlos/apps 0750 " + serviceUser + " " + serviceUser + " -\n" +
	"d /run/carlos/hostguard 2770 " + serviceUser + " " + serviceUser + " -\n" +
	"d /data/hostguard 2770 " + serviceUser + " " + serviceUser + " -\n" +
	// The isolated-exec trees (design §3.1, §3.2). Each protected view is
	// a nosymfollow bind of the dotted raw directory beside it (the mount
	// units); tmpfiles creates both every boot because /run is tmpfs and
	// because the mounts are ordered after it. The socket trees keep the
	// consumers group and setgid, exactly like /run/carlos above, so an
	// enrolled router can still dial a migrated instance (issue #103).
	"d /data/.instances 0700 " + serviceUser + " " + serviceUser + " -\n" +
	// /data/instances is a nosymfollow bind mount OF /data/.instances, so
	// once the mount is active the two paths name ONE inode: a mode that
	// differed between the two lines would flip back and forth on every
	// tmpfiles run, since each write lands on the same inode the other
	// line just set. Both are 0700 so re-runs converge instead of racing.
	"d /data/instances 0700 " + serviceUser + " " + serviceUser + " -\n" +
	"d /data/layout 0700 " + serviceUser + " " + serviceUser + " -\n" +
	// The relay's one writable state dir (carlos-relay.service's
	// ReadWritePaths). It is in bootstrap's dirs list too, but that list
	// only ever runs against a FRESH box or a staging root — a self-update
	// apply creates no directories, it re-runs tmpfiles. Without this line
	// a box bootstrapped before the relay existed gets the unit, the
	// binary and the master file and still fails to start with "Failed to
	// set up mount namespacing: /data/relay: No such file or directory"
	// (discussion #118, hit on the Ireland box 2026-09-07).
	"d /data/relay 0750 " + serviceUser + " " + serviceUser + " -\n" +
	"d /run/carlos/.instances 2750 " + serviceUser + " " + socketsGroup + " -\n" +
	"d /run/carlos/instances 2750 " + serviceUser + " " + socketsGroup + " -\n" +
	// The readiness probe's own instance directory (design §8.5): a name
	// no host can have, created every boot so the probe unit's bind
	// sources exist before it starts.
	"d /data/.instances/_probe 0700 " + serviceUser + " " + serviceUser + " -\n" +
	"d /run/carlos/.instances/_probe 2750 " + serviceUser + " " + socketsGroup + " -\n" +
	"d /run/carlos/.logs 0700 " + serviceUser + " " + serviceUser + " -\n" +
	"d /run/carlos/logs 0700 " + serviceUser + " " + serviceUser + " -\n" +
	// The root instance-control channel's directory (design §5.2): root's,
	// so no carlos process can create, rename or replace anything in it;
	// group carlos with search so the agent can connect.
	"d /run/carlos-control 0750 root " + serviceUser + " -\n" +
	// The readiness probe's fixture (design §8.5, codex R11-1): a file
	// root:carlos 0640 that an isolated instance must NOT be able to read.
	// Written here, not by the control service, whose ProtectSystem=strict
	// makes /etc read-only to it.
	"f /etc/carlos/probe-secret 0640 root " + serviceUser + " - probe\n" +
	// And one in the host's /dev/shm, which an isolated instance must NOT
	// see: its own /dev/shm is a private tmpfs (#126). Without a file
	// here, an empty listing inside the probe would prove nothing.
	"f /dev/shm/carlos-probe 0644 root root - probe\n"

// journaldConfName is the journald drop-in every box gets (discussion #111).
//
// The name matches the one placed by hand on the flagship during the
// 2026-09-06 relief, and must keep matching it: drop-ins are read in name
// order and the LAST assignment wins, so a differently-named file would sit
// beside the hand-placed one and whichever sorted later would decide the cap.
// Same name means this file supersedes it on every box that has it, and the
// hand-placed copy can be deleted or left alone with the same result.
const journaldConfName = "50-cap.conf"

// journaldConf caps the journal. Uncapped, journald's default is 10% of the
// filesystem — 2 GB of a 20 GB root, of which the flagship had spent 1.2 GB
// by the time it filled. A cap is not a fix for a chatty box, it is the
// difference between a chatty box losing old log lines and a chatty box
// losing the ability to accept a push.
//
// Both halves are set explicitly. SystemMaxUse governs /var/log/journal, the
// persistent journal, which is the one that filled; RuntimeMaxUse governs
// /run/log/journal, which is tmpfs — so its default 10% is 10% of RAM, and on
// a small box a journal storm there is memory pressure rather than disk. One
// drop-in that leaves the volatile half at the default would have to be
// reasoned about again the first time that matters.
const journaldConf = "# Written by carlos bootstrap (discussion #111). Do not edit by hand:\n" +
	"# the next bootstrap or self-update apply rewrites this file.\n" +
	"[Journal]\n" +
	"SystemMaxUse=256M\n" +
	"RuntimeMaxUse=64M\n"

// enabledUnits are what a live bootstrap turns on: the always-on front door,
// the hourly adoption timer, and the path units that let the sandboxed
// agent ask a root pass for the halves it cannot do itself: carlos-adopt.path
// (config fast convergence, issue #44), carlos-stage.path (release staging
// — the agent sees a promotion in seconds but cannot write /opt),
// carlos-restart.path (a member's `carlos restart`, issue #113 — the agent
// bounces its own exec children but cannot cycle a unit) and
// carlos-system-update.path (the platform self-update relay, spec §4 — the
// agent can fetch and verify a candidate but cannot write /opt or /etc).
// carlos-system-recover.service is enabled (not armed by a path — it also
// runs Before=carlos-edge.service on every boot, and on the update oneshot's
// own OnFailure=) so recovery from an interrupted self-update never depends
// on a marker file surviving a crash. The litestream + restore-verify units install
// everywhere but stay inert until an operator stages /data/litestream.env and
// generates /etc/litestream.yml (`carlos ops litestream`) — a box with no
// replication bucket is unaffected. carlos-relay.service (relay spec §3) is
// enabled AND started fleet-wide, unlike the edge: its
// ConditionPathExists=/etc/carlos/relay-masters makes the start a no-op on
// every box the operator has not armed with a region master.
//
// relayUnitName is spelled once because two places must agree on it: this
// list and the systemctl plan's start set.
const relayUnitName = "carlos-relay.service"

var enabledUnits = []string{"carlos-edge.service", relayUnitName, "carlos-adopt.timer", "carlos-adopt.path", "carlos-stage.path", "carlos-restart.path",
	"carlos-system-update.path", "carlos-system-recover.service",
	// Region HA joining (spec 2 §5 step 5.3): root's half, on the agent's
	// request. Idle on every box that is not joining.
	"carlos-system-join.path",
	// The isolated-exec mounts (design §3.2): the edge Requires= them.
	"data-instances.mount", "run-carlos-instances.mount", "run-carlos-logs.mount",
	// The root instance-control channel (design §5.2); socket-activated, so
	// only the socket is enabled and started.
	"carlos-instance-control.socket"}

// inertUnits are installed everywhere and deliberately never armed by
// bootstrap or by a self-update apply: restore-verify until an operator
// stages litestream, hibernate-verify by long-standing choice.
var inertUnits = []string{"carlos-restore-verify.timer", "carlos-hibernate-verify.timer"}

// bootstrapPlan is what `bootstrap -root <dir> -plan-json` prints: the
// arming list of THIS binary, so a self-update apply arms what the new
// binary's bootstrap would arm and never a list an older binary remembers.
//
// Armed is the other half of that promise, and it was missing until
// discussion #118. Enabled alone says which units to `enable`; WHICH of them
// also get a `start` was decided by the applier's own systemctlPlan — code
// from the OLD binary, which cannot know about a unit the new build just
// introduced. That is how the relay landed on the Ireland box enabled but
// never started on 2026-09-07: the applying build predated
// carlos-relay.service, so its systemctlPlan had no reason to start it.
// Carrying the list the TARGET build would start closes that gap for every
// future unit, not just the relay.
//
// Additive on purpose: a plan.json written by a build older than this field
// (a rollback stages carlos.prev's plan) carries no "armed" key, and arm()
// then decides locally exactly as it always did. The key is NOT omitempty:
// nil ("this build is too old to say") and empty ("start nothing") are
// different answers, and omitempty would collapse them into the same absent
// key, making "start nothing" impossible for a build to express.
type bootstrapPlan struct {
	Enabled []string `json:"enabled"`
	Inert   []string `json:"inert"`
	Armed   []string `json:"armed"`
}

func currentBootstrapPlan() bootstrapPlan {
	return bootstrapPlan{
		Enabled: append([]string(nil), enabledUnits...),
		Inert:   append([]string(nil), inertUnits...),
		Armed:   armedUnits(enabledUnits),
	}
}

// armedUnits is the "and start these too" rule in one place: the unit types
// whose enable-without-start is a silent no-op until the next boot, plus
// carlos-relay.service, whose ConditionPathExists makes starting it safe on a
// box with no region master. See systemctlPlan for the long version.
func armedUnits(units []string) []string {
	var armed []string
	for _, u := range units {
		if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") ||
			strings.HasSuffix(u, ".socket") || strings.HasSuffix(u, ".mount") || u == relayUnitName {
			armed = append(armed, u)
		}
	}
	return armed
}

// systemctlPlan is the exact systemctl sequence a live bootstrap runs — and
// the same sequence a staging run (-root <dir>) PRINTS instead of running, so
// the two can never drift into describing different behaviour.
//
// Timers and path units get an explicit `start` after the `enable`, and that
// line is the point of this function: `systemctl enable` writes a .wants
// symlink and stops there, which arms the unit at the NEXT boot and not a
// moment sooner. carlos-adopt.timer sat enabled-but-inactive on the flagship
// for weeks for exactly this reason — symlink present, nothing rebooted, no
// release ever adopted itself, every deploy silently needing a hand-run
// adoption pass. A path unit fails the same way and just as quietly: an
// unstarted carlos-adopt.path watches nothing, so the agent's config-sync
// marker lands in a file no one is reading. A re-run is harmless: starting an
// already-running timer or path unit is a no-op, which keeps bootstrap
// idempotent.
//
// Services are deliberately NOT started, with one exception. carlos-edge
// needs its binary at /opt/carlos/carlos and /etc/carlos/host.env staged
// first — neither of which bootstrap can do for the operator, which is why
// it ends by telling them to. Starting timers is safe in a way starting
// services is not: a timer that fires before the binary lands just fails one
// adoption pass and tries again on the next tick.
//
// carlos-relay.service is the exception (relay spec §3). It carries
// ConditionPathExists=/etc/carlos/relay-masters, so on a box the operator
// has not armed with a region master the start is a no-op systemd records as
// a failed condition, not a failure — which is what makes starting it safe
// fleet-wide. On a box that IS armed, starting it here is what saves a
// reboot.
//
// Socket and mount units are armed the same way (isolated-exec design
// §8.6, codex R1-14): an enabled-but-unstarted socket is a control channel
// nobody is listening on, and an enabled-but-unmounted bind is an instance
// tree with no protection, until the next reboot. Starting either is
// idempotent.
func systemctlPlan(units []string) [][]string {
	return systemctlPlanArming(units, armedUnits(units))
}

// systemctlPlanArming is systemctlPlan with the start list supplied rather
// than derived — the shape a self-update apply needs, where the units come
// from the target build's plan.json and so does the decision about which of
// them to start (discussion #118).
func systemctlPlanArming(units, armed []string) [][]string {
	plan := [][]string{
		{"daemon-reload"},
		append([]string{"enable"}, units...),
	}
	if len(armed) > 0 {
		plan = append(plan, append([]string{"start"}, armed...))
	}
	return plan
}

// journaldReload is the one systemctl call that makes a freshly written
// journald drop-in take effect. Kept out of systemctlPlan on purpose: that
// plan is "enable these units and arm the triggers among them", and it is
// also what the self-update relay's arm step and ensureSystemUnits run over
// narrow unit lists — neither of which should be bouncing the box's logging
// daemon. Both places that install the drop-in call this explicitly instead.
var journaldReload = []string{"restart", "systemd-journald"}

// RunBootstrap is the `carlos bootstrap` verb. Idempotent — safe to re-run
// (which is also how a box enrolls a new socket consumer after the fact).
func RunBootstrap(args []string) {
	fs := flag.NewFlagSet("bootstrap", flag.ExitOnError)
	root := fs.String("root", "/", "install root; a staging dir writes files only (no user creation, no systemctl)")
	offcloud := fs.Bool("offcloud", false, "off-cloud box: require a staged 0600 credentials file instead of an EC2 instance profile")
	planJSON := fs.Bool("plan-json", false, "with -root: print the arming plan as JSON on stdout instead of the human summary")
	var consumers repeatableFlag
	fs.Var(&consumers, "socket-consumer", "host-app user to enroll in the "+socketsGroup+" group (repeatable); members may dial every instance socket on the box")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)
	if *planJSON && filepath.Clean(*root) == "/" {
		log.Fatal("bootstrap: -plan-json is for a staging root")
	}
	if *offcloud {
		if err := bootstrapOffcloud(*root, consumers); err != nil {
			log.Fatalf("bootstrap: %v", err)
		}
		return
	}
	if err := bootstrapWith(*root, consumers, *planJSON); err != nil {
		log.Fatalf("bootstrap: %v", err)
	}
}

// repeatableFlag collects every occurrence of a string flag.
type repeatableFlag []string

func (f *repeatableFlag) String() string { return strings.Join(*f, ",") }
func (f *repeatableFlag) Set(v string) error {
	*f = append(*f, v)
	return nil
}

// bootstrapOffcloud is bootstrap for a box with no EC2 instance profile
// (issue #53 point 4): identical structural work, plus the credential
// preconditions — a staged 0600 credentials file scoped (at IAM) to the
// deployment bucket + this host's own region bucket and nothing else, so
// wrong-region-fails-at-IAM holds off-cloud too. The flag is explicit; no
// IMDS sniffing (the agent's one IMDS probe is already gated for non-EC2).
//
// Credential shape and mode checks precede bootstrap mutations. A rejected
// credential precondition leaves the box untouched.
func bootstrapOffcloud(root string, consumers []string) error {
	cred := filepath.Join(root, "etc", "carlos", "credentials")
	fi, err := checkOffcloudCredentials(cred)
	if err != nil {
		return fmt.Errorf("offcloud: stage the scoped credentials file first: %s (0600, see infra/remote-host.md): %w", cred, err)
	}
	if fi.Mode().Perm() != 0o600 && fi.Mode().Perm() != 0o640 {
		return fmt.Errorf("offcloud: %s must be mode 0600 or 0640, is %o", cred, fi.Mode().Perm())
	}
	if err := bootstrap(root, consumers); err != nil {
		return err
	}
	fmt.Printf("offcloud bootstrap complete.\n  ensure /etc/carlos/host.env sets:\n    AWS_SHARED_CREDENTIALS_FILE=/etc/carlos/credentials\n  then stage /opt/carlos/carlos and start carlos-edge.service (see infra/remote-host.md)\n")
	return nil
}

// bootstrap provisions (or re-converges) a box under root: units into
// <root>/etc/systemd/system, the standard dirs, the tmpfiles.d entry, and —
// only when root is "/" — the service user, ownership, and the systemctl
// sequence (daemon-reload, enable, and arming the timers; see systemctlPlan).
// A non-"/" root is the staging/test path: files only, with the privileged
// steps printed instead of run.
func bootstrap(root string, consumers []string) error {
	return bootstrapWith(root, consumers, false)
}

// bootstrapWith is bootstrap's body, plus the -plan-json staging path: when
// planJSON is set (staging root only — RunBootstrap refuses it with -root /),
// every "wrote"/"created" line below is redirected to stderr so the JSON
// plan currentBootstrapPlan() prints is alone on stdout.
func bootstrapWith(root string, consumers []string, planJSON bool) error {
	live := filepath.Clean(root) == "/"
	hostEnv := filepath.Join(root, "etc", "carlos", "host.env")
	if live {
		hostEnv = claimBootstrapHostEnv
	}
	claimSettings, settingsErr := readClaimUnitSettings(hostEnv)
	if settingsErr != nil {
		fmt.Fprintf(os.Stderr, "bootstrap: instance-control unit settings: %v\n", settingsErr)
		claimSettings = claimUnitSettings{}
	}
	beforeControl, err := os.ReadFile(filepath.Join(root, "etc", "systemd", "system", 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))
	var afterControl []byte
	out := io.Writer(os.Stdout)
	if planJSON {
		out = os.Stderr
	}
	if live {
		// Validate consumer users BEFORE any mutation (the offcloud
		// precondition's discipline): a typo in -socket-consumer must leave
		// the box untouched, not abort bootstrap halfway with units written
		// but timers never armed.
		for _, u := range consumers {
			if _, err := user.Lookup(u); err != nil {
				return fmt.Errorf("--socket-consumer %s: no such user (create the host app's user first): %w", u, err)
			}
		}
	}
	unitDir := filepath.Join(root, "etc", "systemd", "system")
	dataDir := filepath.Join(root, "data")
	runDir := filepath.Join(root, "run", "carlos")

	type dir struct {
		path string
		mode os.FileMode
	}
	dirs := []dir{
		{dataDir, 0o750},
		{filepath.Join(dataDir, "acme"), 0o750},
		// The relay's one writable state dir (carlos-relay.service's
		// ReadWritePaths). Created on every box, armed or not, so the unit
		// never fails for want of a directory the moment a master lands.
		{filepath.Join(dataDir, "relay"), 0o750},
		// Setgid: sockets bound inside inherit the dir's group — the
		// consumers group, once a live run (or the tmpfiles line) sets it.
		// See socketsGroup/tmpfilesConf.
		{runDir, 0o750 | os.ModeSetgid},
		// The carlos-app@ tenant sockets (see tmpfilesConf for the perms
		// rationale).
		{filepath.Join(runDir, "apps"), 0o750},
		// The isolated-exec trees (design §3.1). The dotted names are the
		// raw directories the mount units bind onto the undotted views;
		// see tmpfilesConf for the modes and why both exist.
		{filepath.Join(dataDir, ".instances"), 0o700},
		// One inode once the bind is active (see tmpfilesConf): both this
		// entry and the tmpfiles line for /data/instances must agree, or
		// the mode flips on every tmpfiles run.
		{filepath.Join(dataDir, "instances"), 0o700},
		{filepath.Join(dataDir, "layout"), 0o700},
		{filepath.Join(runDir, ".instances"), 0o750 | os.ModeSetgid},
		{filepath.Join(runDir, "instances"), 0o750 | os.ModeSetgid},
		// The readiness probe's own instance directory (design §8.5): a
		// name no host can have, created here so a staging root has it
		// too (see tmpfilesConf for the modes and why both exist).
		{filepath.Join(dataDir, ".instances", "_probe"), 0o700},
		{filepath.Join(runDir, ".instances", "_probe"), 0o750 | os.ModeSetgid},
		{filepath.Join(runDir, ".logs"), 0o700},
		{filepath.Join(runDir, "logs"), 0o700},
		// Root-owned on a live box (privilegedPlan chowns /run/carlos
		// recursively to the service user; this one sits BESIDE it, not
		// under it, precisely so that chown never reaches it).
		{filepath.Join(root, "run", "carlos-control"), 0o750},
		// The version store + per-host live symlinks (release/versions.go).
		// World-readable/executable so sandboxed instance units can exec their
		// binary; only root (the adopt pass) writes here.
		{filepath.Join(root, "opt", "carlos", "versions"), 0o755},
		{filepath.Join(root, "opt", "carlos", "live"), 0o755},
		// Host-wide config (host.env, 0600 root-owned, staged by the operator).
		{filepath.Join(root, "etc", "carlos"), 0o755},
		// Per-app env files for carlos-app@ tenants (systemd reads them as
		// root before dropping to the unit's DynamicUser, so 0600 root works).
		{filepath.Join(root, "etc", "carlos", "apps"), 0o755},
		{unitDir, 0o755},
		// The journal cap's drop-in dir. systemd creates it on demand for a
		// hand-placed file, but nothing creates it on a fresh box, and the
		// write below would fail ENOENT if this line were missing.
		{filepath.Join(root, "etc", "systemd", "journald.conf.d"), 0o755},
		{filepath.Join(root, "etc", "tmpfiles.d"), 0o755},
		{filepath.Join(root, "etc", "sysusers.d"), 0o755},
	}
	for _, d := range dirs {
		if err := bootstrapMakeDir(d.path, d.mode); err != nil {
			return fmt.Errorf("mkdir/chmod %s: %v", d.path, err)
		}
	}

	// The registry file, created here and by nothing else on a fresh box.
	//
	// It has to exist BEFORE the units below are armed, because four of them
	// (carlos-adopt, -stage, -restart, -config-sync) run `carlos adopt` as
	// root and open this same file — and on a fresh box adopt's .path and
	// .timer can fire before the edge has ever started. Whoever creates the
	// file owns it, and an edge running as serviceUser cannot write a
	// root-owned registry: that is discussion #8, three region boxes left
	// unstartable on 2026-08-31. Creating it here puts it inside the
	// descriptor-based ownership walk that privilegedPlan runs, so it is
	// serviceUser's before anything can race for it.
	//
	// An empty file is a valid empty SQLite database — the first opener
	// (edge or adopt) runs the migrations into it. Deliberately NOT opened
	// as a registry here: on a re-run over a live box this must not touch
	// the schema of a database the running edge holds open, and "the file
	// exists, owned by the right user" is the whole job.
	//
	// One constraint this puts on the deploy side: a STAGING run takes this
	// branch too. `carlos bootstrap -root <dir>` is how a live box refreshes
	// its unit files (pinfra's update-platform-binary.sh stages into
	// /tmp/carlos-unitstage), so a staging tree now contains a 0-byte
	// data/registry.db. Harmless while that script copies only etc/ out of
	// the tree — and destructive the day anything rsyncs the whole tree over
	// /, because a 0-byte file would land on top of the live registry and
	// empty it. The script lives in the private platform-infrastructure repo;
	// whoever widens what it copies has to exclude data/.
	regPath := filepath.Join(dataDir, "registry.db")
	created, err := bootstrapCreateRegistry(regPath)
	if err != nil {
		return fmt.Errorf("create %s: %v", regPath, err)
	}
	if created {
		fmt.Fprintf(out, "  created %s\n", regPath)
	}

	names, err := unitNames()
	if err != nil {
		return err
	}
	for _, name := range names {
		content, err := unitsFS.ReadFile("units/" + name)
		if err != nil {
			return fmt.Errorf("read embedded unit %s: %v", name, err)
		}
		path := filepath.Join(unitDir, name)
		if name == claimControlUnit {
			if keepControl {
				afterControl = beforeControl
				fmt.Fprintf(out, "  kept %s (host.env cannot be parsed)\n", path)
				continue
			}
			afterControl = claimUnitContent(name, content, claimSettings)
		}
		if err := os.WriteFile(path, claimUnitContent(name, content, claimSettings), 0o644); err != nil {
			return fmt.Errorf("write %s: %v", path, err)
		}
		fmt.Fprintf(out, "  wrote %s\n", path)
	}

	// On a live root the privileged plan runs BEFORE the tmpfiles conf is
	// written: the conf's /run/carlos line names socketsGroup, and
	// systemd-tmpfiles DROPS a line whose group doesn't resolve — so a
	// bootstrap interrupted between "conf written" and "groupadd ran" would
	// leave a box that looks healthy until its next reboot, when /run/carlos
	// simply isn't created and carlos-edge (ReadWritePaths=/run/carlos, no
	// "-" prefix) fails its namespace setup forever. Group first, then the
	// conf that names it.
	if live {
		for _, args := range privilegedPlan(consumers) {
			if err := runPrivileged(args); err != nil {
				return err
			}
		}
	}

	// The sysusers.d snippet closes the same hole for confs that arrive
	// WITHOUT a live bootstrap run: a staging root is how a live box
	// refreshes its unit files (pinfra's update-platform-binary.sh), and a
	// refresh that installs the staged etc/ tree ships the tmpfiles line to
	// a box that never ran groupadd. systemd-sysusers.service is ordered
	// Before=systemd-tmpfiles-setup.service, so the group exists by the time
	// the tmpfiles line is parsed, however the conf got there.
	sysusersPath := filepath.Join(root, "etc", "sysusers.d", "carlos.conf")
	if err := os.WriteFile(sysusersPath, []byte("g "+socketsGroup+" -\n"), 0o644); err != nil {
		return fmt.Errorf("write %s: %v", sysusersPath, err)
	}
	fmt.Fprintf(out, "  wrote %s\n", sysusersPath)

	// The journal cap (discussion #111). Written on every root, live or
	// staging, because a staging root is how a live box refreshes its etc/
	// tree — both the self-update apply (systemapply.go's installUnits) and
	// pinfra's update-platform-binary.sh stage with `bootstrap -root <dir>`
	// and copy named files out of it. The apply path names this one; the
	// pinfra script copies only units, sysusers and tmpfiles today, so a box
	// updated by hand gets the cap on its next self-update or bootstrap
	// rather than on that script's next run.
	journaldPath := filepath.Join(root, "etc", "systemd", "journald.conf.d", journaldConfName)
	if err := os.WriteFile(journaldPath, []byte(journaldConf), 0o644); err != nil {
		return fmt.Errorf("write %s: %v", journaldPath, err)
	}
	fmt.Fprintf(out, "  wrote %s\n", journaldPath)

	tmpfilesPath := filepath.Join(root, "etc", "tmpfiles.d", "carlos.conf")
	if err := os.WriteFile(tmpfilesPath, []byte(tmpfilesConf), 0o644); err != nil {
		return fmt.Errorf("write %s: %v", tmpfilesPath, err)
	}
	fmt.Fprintf(out, "  wrote %s\n", tmpfilesPath)

	if !live {
		if planJSON {
			return json.NewEncoder(os.Stdout).Encode(currentBootstrapPlan())
		}
		fmt.Printf("bootstrap: staged under %s — on a real box (--root /) it would also:\n", root)
		// Printed, never run: a staging root is how a LIVE box refreshes its
		// unit files (pinfra's update-platform-binary.sh stages into
		// /tmp/carlos-unitstage), so arming anything from here would reach out
		// of the temp root and touch the running system from a command whose
		// entire contract is "write files, change nothing".
		for _, args := range privilegedPlan(consumers) {
			fmt.Printf("  %s\n", strings.Join(args, " "))
		}
		for _, args := range systemctlPlan(enabledUnits) {
			fmt.Printf("  systemctl %s\n", strings.Join(args, " "))
		}
		fmt.Printf("  systemctl %s\n", strings.Join(journaldReload, " "))
		return nil
	}

	// Repair before arming any root pass that may load these credentials.
	if err := repairOffcloudCredentials("/etc/carlos/credentials"); err != nil {
		return err
	}
	for _, args := range systemctlPlan(enabledUnits) {
		if err := systemctl(args...); err != nil {
			return err
		}
	}
	if err := restartClaimControl(filepath.Join(unitDir, claimControlUnit), beforeControl, afterControl, func(args ...string) (string, error) { return "", systemctl(args...) }); err != nil {
		return err
	}
	// A journald drop-in is inert until journald re-reads its configuration,
	// and journald re-reads it only on restart — so a bootstrap that writes
	// the cap and stops leaves a box whose journal is capped from its next
	// reboot, which on a long-lived box is never. Non-fatal: the cap is a
	// safety margin, and refusing to finish bootstrap over it would be a
	// worse outcome than an uncapped journal on one box.
	if err := systemctl(journaldReload...); err != nil {
		fmt.Printf("bootstrap: could not restart systemd-journald (%v) — the journal cap in %s takes effect at the next restart\n", err, journaldConfName)
	}
	fmt.Printf("bootstrap: units installed, %s enabled (timers and path units armed, not just symlinked)\n", strings.Join(enabledUnits, " + "))
	fmt.Println("  next: copy the carlos binary to /opt/carlos/carlos, stage /etc/carlos/host.env, then: systemctl start carlos-edge")
	return nil
}

// unitNames lists every embedded unit, sorted (embed.FS ReadDir order).
func unitNames() ([]string, error) {
	entries, err := unitsFS.ReadDir("units")
	if err != nil {
		return nil, fmt.Errorf("read embedded units: %v", err)
	}
	names := make([]string, 0, len(entries))
	for _, e := range entries {
		names = append(names, e.Name())
	}
	return names, nil
}

// privilegedPlan is the exact non-systemctl privileged sequence a live
// bootstrap dispatches — and the same sequence a staging run PRINTS instead of
// running (systemctlPlan's discipline, extended to the user/group half when
// issue #103 added the consumers group): the two can never drift into
// describing different behaviour.
//
// The sequence: service user, consumers group, ownership. Every chown
// converges TOWARD the target state, never through an intermediate that
// undoes a previous run's work: /run/carlos is recursively
// serviceUser:socketsGroup. Unsafe entries, including live sockets, are
// logged and skipped; their creating units establish socket ownership.
// The apps/ directory (unit-tenant sockets, no path consumers) is then
// put back to serviceUser:serviceUser. The chmod re-asserts the setgid bit
// the tmpfiles line promises; consumers enroll last, one usermod each
// (`-aG` is append — an existing member is a no-op, so re-runs converge).
// It deliberately does NOT touch group `carlos` memberships: an interim
// grant made there (the tito workaround this replaces) is the operator's to
// revoke, per the deployment notes in the #103 spec.
func privilegedPlan(consumers []string) [][]string {
	plan := [][]string{
		{"useradd", "--system", "--user-group", "--no-create-home", "--shell", "/usr/sbin/nologin", serviceUser},
		{"groupadd", "--system", socketsGroup},
		{"chown", "-R", serviceUser + ":" + serviceUser, "/data"},
		{"chown", "-R", serviceUser + ":" + socketsGroup, "/run/carlos"},
		{"chown", "-R", serviceUser + ":" + serviceUser, "/run/carlos/apps"},
		{"chmod", "2750", "/run/carlos"},
	}
	for _, u := range consumers {
		plan = append(plan, []string{"usermod", "-aG", socketsGroup, u})
	}
	return plan
}

// runPrivileged executes one privilegedPlan step, adding the existence
// guards that keep the raw commands idempotent: useradd and groupadd exit
// non-zero when their subject already exists, and "already provisioned" is
// bootstrap's normal re-run state, not an error. Everything else in the plan
// is idempotent. Ownership and mode steps use descriptor helpers; the
// command text is retained for staging output and is never sent to chown.
func runPrivileged(args []string) error {
	if len(args) == 4 && args[0] == "chown" && args[1] == "-R" {
		owner := strings.SplitN(args[2], ":", 2)
		if len(owner) != 2 {
			return fmt.Errorf("invalid ownership %q", args[2])
		}
		uid, gid, err := bootstrapOwner(owner[0], owner[1])
		if err != nil {
			return err
		}
		return bootstrapChownTree(args[3], uid, gid)
	}
	if len(args) == 3 && args[0] == "chmod" && args[1] == "2750" {
		return bootstrapMakeDir(args[2], 0o750|os.ModeSetgid)
	}
	// The subject is derived from the command line itself (both commands
	// take it as their final argument), never from a package constant — a
	// guard keyed to a constant would silently skip any SECOND useradd or
	// groupadd a later plan revision adds.
	switch subject := args[len(args)-1]; args[0] {
	case "useradd":
		if _, err := user.Lookup(subject); err == nil {
			return nil
		}
	case "groupadd":
		if _, err := user.LookupGroup(subject); err == nil {
			return nil
		}
	}
	out, err := exec.Command(args[0], args[1:]...).CombinedOutput()
	if err != nil {
		return fmt.Errorf("%s: %v: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
	}
	return nil
}

func bootstrapOwner(owner, group string) (int, int, error) {
	u, err := user.Lookup(owner)
	if err != nil {
		return 0, 0, err
	}
	g, err := user.LookupGroup(group)
	if err != nil {
		return 0, 0, err
	}
	uid, err := strconv.Atoi(u.Uid)
	if err != nil {
		return 0, 0, err
	}
	gid, err := strconv.Atoi(g.Gid)
	return uid, gid, err
}

func systemctl(args ...string) error {
	out, err := exec.Command("systemctl", args...).CombinedOutput()
	if err != nil {
		return fmt.Errorf("systemctl %s: %v: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
	}
	return nil
}