package host

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"runtime"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/metrics"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// expectedUnits is the contract for what bootstrap installs — a new unit file
// must be added here (and a removed one taken out) or TestUnitsEmbedComplete
// fails, so the embed directive and the fleet's unit set can't drift apart.
var expectedUnits = []string{
	"carlos-adopt.path",
	"carlos-adopt.service",
	"carlos-adopt.timer",
	"carlos-app@.service",
	"carlos-app@.socket",
	"carlos-config-sync.service",
	"carlos-dbread.socket",
	"carlos-dbread@.service",
	"carlos-edge.service",
	"carlos-hibernate-verify.service",
	"carlos-hibernate-verify.timer",
	"carlos-instance-cmd@.service",
	"carlos-instance-control.service",
	"carlos-instance-control.socket",
	"carlos-instance-gen@.service",
	"carlos-instance-probe.service",
	"carlos-instance-seed@.service",
	"carlos-instance-sidecar@.service",
	"carlos-instance@.service",
	"carlos-instances.slice",
	"carlos-litestream.service",
	"carlos-relay.service",
	"carlos-restart.path",
	"carlos-restart.service",
	"carlos-restore-verify.service",
	"carlos-restore-verify.timer",
	"carlos-stage.path",
	"carlos-stage.service",
	"carlos-system-join.path",
	"carlos-system-join.service",
	"carlos-system-recover.service",
	"carlos-system-update.path",
	"carlos-system-update.service",
	"carlos.slice",
	"data-instances.mount",
	"run-carlos-instances.mount",
	"run-carlos-logs.mount",
}

// requireUnixSockets skips a test when the sandbox forbids unix-socket
// creation (probe: net.Listen("unix") in a temp dir). None of the current
// tests need one; the helper is here for the host tests that will (agent
// serving loops), matching the pattern used across the repo.
func requireUnixSockets(t *testing.T) {
	t.Helper()
	dir, err := os.MkdirTemp("", "unix-probe-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	l, err := net.Listen("unix", filepath.Join(dir, "probe.sock"))
	if err != nil {
		if strings.Contains(err.Error(), "operation not permitted") {
			t.Skipf("sandbox forbids unix sockets: %v", err)
		}
		t.Fatalf("unix socket probe: %v", err)
	}
	l.Close()
}

func TestUnitsEmbedComplete(t *testing.T) {
	names, err := unitNames()
	if err != nil {
		t.Fatal(err)
	}
	got := map[string]bool{}
	for _, n := range names {
		got[n] = true
	}
	for _, want := range expectedUnits {
		if !got[want] {
			t.Errorf("unit %s missing from the embed FS", want)
		}
	}
	if len(names) != len(expectedUnits) {
		t.Errorf("embed FS has %d units %v, expected list has %d — keep them in step", len(names), names, len(expectedUnits))
	}
}

// TestBootstrapStartsRelayUnit pins the one service bootstrap starts and the
// one it must not. The relay is safe to start fleet-wide because its
// ConditionPathExists gates it on a masters file; the edge is not, and a
// change that quietly promoted every service into the start line would arm
// half-staged boxes.
func TestBootstrapStartsRelayUnit(t *testing.T) {
	plan := systemctlPlan(enabledUnits)
	var started []string
	for _, step := range plan {
		if step[0] == "start" {
			started = step[1:]
		}
	}
	if !slices.Contains(started, "carlos-relay.service") {
		t.Fatalf("relay not started by plan: %v", plan)
	}
	if slices.Contains(started, "carlos-edge.service") {
		t.Fatalf("edge must stay start-by-hand: %v", plan)
	}
}

// TestRelayUnitIsGatedAndLosesFirst pins the three lines that make it safe
// to enable and start carlos-relay.service on every box in the fleet. The
// condition is what keeps an unarmed box from ever listening; the OOM
// adjustment and the absence of the bind capability are what keep the relay
// from costing the edge anything when they share a box (relay spec §3).
func TestRelayUnitIsGatedAndLosesFirst(t *testing.T) {
	b, err := unitsFS.ReadFile("units/" + relayUnitName)
	if err != nil {
		t.Fatal(err)
	}
	unit := string(b)
	for _, want := range []string{
		"ConditionPathExists=/etc/carlos/relay-masters",
		"OOMScoreAdjust=1000",
		"MemoryMax=256M",
		"ExecStart=/opt/carlos/carlos relay serve",
		"ReadWritePaths=/data/relay /run/carlos",
	} {
		if !strings.Contains(unit, want) {
			t.Errorf("%s: missing %q", relayUnitName, want)
		}
	}
	// 3478 is unprivileged; only the edge may bind below 1024.
	if strings.Contains(unit, "CAP_NET_BIND_SERVICE") {
		t.Errorf("%s: must not carry CAP_NET_BIND_SERVICE — 3478 needs no capability", relayUnitName)
	}
}

func TestBootstrapWritesUnitsToStagingDir(t *testing.T) {
	root := t.TempDir()
	if err := bootstrap(root, nil); err != nil {
		t.Fatalf("bootstrap: %v", err)
	}

	unitDir := filepath.Join(root, "etc", "systemd", "system")
	for _, name := range expectedUnits {
		b, err := os.ReadFile(filepath.Join(unitDir, name))
		if err != nil {
			t.Errorf("unit not installed: %v", err)
			continue
		}
		content := string(b)
		// Only the edge binds privileged ports; every other unit must not
		// carry the capability.
		if name == "carlos-edge.service" {
			if !strings.Contains(content, "AmbientCapabilities=CAP_NET_BIND_SERVICE") {
				t.Errorf("%s: missing AmbientCapabilities=CAP_NET_BIND_SERVICE", name)
			}
		} else if strings.Contains(content, "CAP_NET_BIND_SERVICE") {
			t.Errorf("%s: must not carry CAP_NET_BIND_SERVICE", name)
		}
	}

	// The edge unit is the agent role.
	edge, err := os.ReadFile(filepath.Join(unitDir, "carlos-edge.service"))
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"ExecStart=/opt/carlos/carlos agent",
		"EnvironmentFile=-/etc/carlos/host.env",
		"ProtectSystem=strict",
		"ReadWritePaths=/data /run/carlos",
		"NoNewPrivileges=yes",
		"MemoryDenyWriteExecute=yes",
		"SystemCallFilter=@system-service",
		"User=carlos",
		"Restart=always",
		// Delegation is the SOLE mechanism behind exec-instance readings
		// fleet-wide (exec-child accounting spec): without it the agent's
		// cgroup manager degrades to nil and every exec-backed tenant goes
		// blind on the Instances page.
		"Delegate=cpu memory io pids",
		// The outer wall (memory-pressure spec §2.2) is now on carlos.slice
		// (isolated-exec design §4.3). The edge is in the slice, so it still
		// participates in the ceiling. OOMPolicy=continue is the line that
		// keeps one tenant's OOM kill from becoming a whole-edge restart
		// (systemd's default is stop).
		"Slice=carlos.slice",
		"OOMPolicy=continue",
	} {
		if !strings.Contains(string(edge), want) {
			t.Errorf("carlos-edge.service: missing %q", want)
		}
	}

	// The outer wall moved from the edge to carlos.slice (isolated-exec
	// design §4.3): the slice bounds the edge and every awake instance
	// as before, but instances are now sibling units instead of children.
	slice, err := os.ReadFile(filepath.Join(unitDir, "carlos.slice"))
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"MemoryHigh=70%",
		"MemoryMax=80%",
	} {
		if !strings.Contains(string(slice), want) {
			t.Errorf("carlos.slice: missing %q", want)
		}
	}
	// The counterpart of Delegate=: ProtectControlGroups would mount the
	// cgroup tree read-only and silently un-arm delegation. Its hardening is
	// retained by ownership (DAC) instead — see the unit file's comment. A
	// future "hardening cleanup" that re-adds it must fail here, not on the
	// fleet.
	if strings.Contains(string(edge), "ProtectControlGroups=") {
		t.Error("carlos-edge.service: the ProtectControlGroups= directive is deliberately absent (Delegate= needs the subtree writable); do not re-add it")
	}

	// The tenant socket unit: systemd owns the listener as carlos so the
	// edge can connect and tenants can't reach each other (the whole point).
	sockUnit, err := os.ReadFile(filepath.Join(unitDir, "carlos-app@.socket"))
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"ListenStream=/run/carlos/apps/%i.sock",
		"SocketUser=carlos",
		"SocketMode=0600",
		"WantedBy=sockets.target",
	} {
		if !strings.Contains(string(sockUnit), want) {
			t.Errorf("carlos-app@.socket: missing %q", want)
		}
	}

	// The config fast-convergence relay (issue #44). The invariant worth a
	// test is WHICH unit the path starts: carlos-adopt.service also runs
	// release.Pass and the canary-TTL nag, so aiming a per-config-write
	// trigger at it would let one app's `carlos env set` cut over another
	// app's pending release seconds later and nag on every write instead of
	// hourly. The relay must reach the config half alone.
	pathUnit, err := os.ReadFile(filepath.Join(unitDir, "carlos-adopt.path"))
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"PathChanged=/data/config-sync-requested",
		"Unit=carlos-config-sync.service",
		"WantedBy=multi-user.target",
	} {
		if !strings.Contains(string(pathUnit), want) {
			t.Errorf("carlos-adopt.path: missing %q", want)
		}
	}
	if strings.Contains(string(pathUnit), "Unit=carlos-adopt.service") {
		t.Error("carlos-adopt.path must not start carlos-adopt.service — that unit adopts releases and nags on expired canaries")
	}

	// The release fast-convergence relay, and the same invariant: which unit
	// the path starts. carlos-adopt.service prints the canary-TTL nag, an
	// HOURLY visibility feature; this trigger fires on agent request, so
	// aiming it there would repeat the same expired-canary lines every time a
	// promotion could not be staged.
	stagePath, err := os.ReadFile(filepath.Join(unitDir, "carlos-stage.path"))
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"PathChanged=/data/stage-requested",
		"Unit=carlos-stage.service",
		"WantedBy=multi-user.target",
	} {
		if !strings.Contains(string(stagePath), want) {
			t.Errorf("carlos-stage.path: missing %q", want)
		}
	}
	if strings.Contains(string(stagePath), "Unit=carlos-adopt.service") {
		t.Error("carlos-stage.path must not start carlos-adopt.service — that unit nags on expired canaries, and this trigger is per-request")
	}

	// The staging pass must run -stage-only, and must NOT be startable by
	// anything but its path unit.
	stageSvc, err := os.ReadFile(filepath.Join(unitDir, "carlos-stage.service"))
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(string(stageSvc), "carlos adopt --stage-only") {
		t.Error("carlos-stage.service must run `carlos adopt --stage-only` — the full pass would nag per request")
	}
	// A SECTION header, not any mention: the unit's own comment explains why
	// it has no [Install], and a substring match would fail on the
	// explanation. Same shape as carlos-config-sync.service, which documents
	// the same choice the same way.
	for _, line := range strings.Split(string(stageSvc), "\n") {
		if strings.TrimSpace(line) == "[Install]" {
			t.Error("carlos-stage.service must have no [Install] section: only carlos-stage.path should ever start it")
		}
	}

	syncUnit, err := os.ReadFile(filepath.Join(unitDir, "carlos-config-sync.service"))
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{
		"ExecStart=/opt/carlos/carlos adopt --config-only",
		"Type=oneshot",
		"EnvironmentFile=-/etc/carlos/host.env",
		// A path-triggered unit must never latch into start-limit-hit after a
		// burst of config writes and then ignore every later trigger.
		"StartLimitIntervalSec=0",
	} {
		if !strings.Contains(string(syncUnit), want) {
			t.Errorf("carlos-config-sync.service: missing %q", want)
		}
	}
	// Started only by the path unit: an [Install] section would invite an
	// operator to enable a oneshot that has nothing to do at boot. Matched
	// as a whole line, since the unit's own comment names the section.
	for _, line := range strings.Split(string(syncUnit), "\n") {
		if strings.TrimSpace(line) == "[Install]" {
			t.Error("carlos-config-sync.service must have no [Install] section — carlos-adopt.path is its only starter")
		}
	}

	// The standard dirs.
	for _, d := range []string{
		filepath.Join(root, "data"),
		filepath.Join(root, "opt", "carlos", "versions"),
		filepath.Join(root, "opt", "carlos", "live"),
		filepath.Join(root, "etc", "carlos"),
		filepath.Join(root, "run", "carlos"),
		filepath.Join(root, "run", "carlos", "apps"),
	} {
		fi, err := os.Stat(d)
		if err != nil {
			t.Errorf("dir not created: %v", err)
			continue
		}
		if !fi.IsDir() {
			t.Errorf("%s: not a directory", d)
		}
	}

	// The tmpfiles.d entry that keeps /run/carlos across reboots.
	tf, err := os.ReadFile(filepath.Join(root, "etc", "tmpfiles.d", "carlos.conf"))
	if err != nil {
		t.Fatalf("tmpfiles conf: %v", err)
	}
	if !strings.Contains(string(tf), "/run/carlos") {
		t.Errorf("tmpfiles conf %q does not mention /run/carlos", tf)
	}
	if !strings.Contains(string(tf), "/run/carlos/apps") {
		t.Errorf("tmpfiles conf %q does not keep the tenant socket dir", tf)
	}
}

// TestResolveStatusBucketDefaultsToDeploymentBucket pins the I2 fix: the
// status loop and the deployment (release-adoption) loop must agree on which
// bucket to serve status out of by default, since the edge/console read
// status/ from the deployment bucket — an explicit CARLOS_STATUS_BUCKET
// still wins when set, and a CARLOS_DEPLOYMENT_DIR-only deployment (no S3
// bucket to fall back to) gets no default at all.
func TestResolveStatusBucketDefaultsToDeploymentBucket(t *testing.T) {
	env := func(m map[string]string) func(string) string {
		return func(k string) string { return m[k] }
	}

	cases := []struct {
		name string
		env  map[string]string
		want string
	}{
		{"explicit status bucket wins", map[string]string{
			"CARLOS_STATUS_BUCKET": "status-bucket", "CARLOS_DEPLOYMENT_BUCKET": "deploy-bucket"}, "status-bucket"},
		{"falls back to deployment bucket when status unset", map[string]string{
			"CARLOS_DEPLOYMENT_BUCKET": "deploy-bucket"}, "deploy-bucket"},
		{"neither set is still neither set", map[string]string{}, ""},
		{"deployment dir gives no fallback (S3-only)", map[string]string{
			"CARLOS_DEPLOYMENT_DIR": "/data/deploy"}, ""},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			if got := resolveStatusBucket(env(c.env)); got != c.want {
				t.Errorf("resolveStatusBucket(%v) = %q, want %q", c.env, got, c.want)
			}
		})
	}
}

// ---- Task 8: statusTick — the loop's pure part -----------------------------

// TestStatusTickWritesBoxAndPublishesOnlyWhenAsked is the plan's Step-1 test:
// statusTick against a DirStore + fixture registry must always land the
// box's own snapshot under status/boxes/, but only land the merged
// status/fleet.json (+ the embedded page assets PublishFleet ships beside
// it) when publish=true — publish=false is the "another box holds the
// lease" tick, which must still self-report without touching the fleet-wide
// objects.
func TestStatusTickWritesBoxAndPublishesOnlyWhenAsked(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "static"}); err != nil {
		t.Fatal(err)
	}

	store := release.NewDirStore(t.TempDir())
	probes := status.Probes{MachineID: "test-machine"}
	ctx := context.Background()

	if err := statusTick(ctx, store, reg, probes, false, ""); err != nil {
		t.Fatalf("statusTick (publish=false): %v", err)
	}
	boxes, err := store.List(ctx, "status/boxes/")
	if err != nil {
		t.Fatalf("List status/boxes/: %v", err)
	}
	if len(boxes) != 1 {
		t.Fatalf("status/boxes/ = %v, want exactly one snapshot", boxes)
	}
	if _, _, err := store.Get(ctx, "status/fleet.json"); err == nil {
		t.Error("publish=false must not write status/fleet.json")
	}

	if err := statusTick(ctx, store, reg, probes, true, ""); err != nil {
		t.Fatalf("statusTick (publish=true): %v", err)
	}
	for _, key := range []string{"status/fleet.json", "status/index.html", "status/status.css"} {
		if _, _, err := store.Get(ctx, key); err != nil {
			t.Errorf("publish=true must write %s: %v", key, err)
		}
	}
	// The box snapshot count must stay at one: the second tick republishes
	// the SAME box's key (WriteBox is last-writer-wins on the label), not a
	// second file.
	boxes, err = store.List(ctx, "status/boxes/")
	if err != nil {
		t.Fatalf("List status/boxes/ after publish tick: %v", err)
	}
	if len(boxes) != 1 {
		t.Errorf("status/boxes/ = %v after second tick, want still exactly one snapshot", boxes)
	}
}

// ---- Task 5 (customer-fleets plan): publisher guard + fleet-prefixed store -

// TestShouldSkipPublisherElection pins the CARLOS_FLEET contract: a fleet
// box (CARLOS_FLEET set to anything non-empty) must skip the publisher
// election, and a platform box (unset, the every-deployment-today case)
// must not. statusLoop checks this BEFORE calling lease.Acquire — see the
// call site — so this also documents that the decision needs no lease, no
// bucket, no network at all: it's a pure function of one env var.
func TestShouldSkipPublisherElection(t *testing.T) {
	env := func(m map[string]string) func(string) string {
		return func(k string) string { return m[k] }
	}
	cases := []struct {
		name string
		env  map[string]string
		want bool
	}{
		{"fleet box skips the election", map[string]string{"CARLOS_FLEET": "pi"}, true},
		{"platform box still elects", map[string]string{}, false},
		{"empty CARLOS_FLEET is not set", map[string]string{"CARLOS_FLEET": ""}, false},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			if got := shouldSkipPublisherElection(env(c.env)); got != c.want {
				t.Errorf("shouldSkipPublisherElection(%v) = %v, want %v", c.env, got, c.want)
			}
		})
	}
}

// TestFleetBoxHeartbeatLandsUnderFleetPrefixNeverPublishes is the end-to-end
// pin for Task 5's whole contract, composing release.NewPrefixStore (whose
// own contract is pinned in isolation by internal/release's
// TestNewPrefixStoreRebasesKeys) with statusTick: a fleet box's tick (store
// wrapped in CARLOS_STATUS_PREFIX="fleets/pi/", publish forced false the way
// shouldSkipPublisherElection drives statusLoop) writes ITS OWN Box
// snapshot at "fleets/pi/status/boxes/<label>.json" in the underlying
// bucket — never at the shared "status/boxes/<label>.json" a platform box
// uses, and never status/fleet.json either.
//
// Critical 1b (Opus review, fix round, controller ruling): <label> here MUST
// be the fleet box's operator-chosen fleet_boxes.label — resolveFleetIdentity's
// result, passed as statusTick's labelOverride — NEVER
// status.Label(probes.MachineID)'s hashed pseudonym. The public status
// page's anonymization rationale for that hash doesn't apply to a path
// (fleets/<fleet>/status/boxes/…) no public page ever reads; using the
// hashed value here would leave the heartbeat filename unrelated to the
// name anything else (the edge's tunnel hub, a fleet binding) calls this
// box by.
func TestFleetBoxHeartbeatLandsUnderFleetPrefixNeverPublishes(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "static"}); err != nil {
		t.Fatal(err)
	}

	under := release.NewDirStore(t.TempDir())
	fleetStore := release.NewPrefixStore(under, "fleets/pi/")
	ctx := context.Background()
	// MachineID deliberately differs from the fleet-box label passed below —
	// proving the filename tracks the identity override, not the probe.
	probes := status.Probes{MachineID: "some-hashed-machine-id"}

	// publish=false is what shouldSkipPublisherElection forces statusLoop
	// into for a fleet box — lease.Acquire is never even called.
	const fleetBoxLabel = "amber-gliding"
	if err := statusTick(ctx, fleetStore, reg, probes, false, fleetBoxLabel); err != nil {
		t.Fatalf("statusTick: %v", err)
	}

	// The filename is pinned to the CHANNEL label, not a hash of MachineID.
	if _, _, err := under.Get(ctx, "fleets/pi/status/boxes/"+fleetBoxLabel+".json"); err != nil {
		t.Fatalf("Get fleets/pi/status/boxes/%s.json: %v", fleetBoxLabel, err)
	}

	boxes, err := under.List(ctx, "fleets/pi/status/boxes/")
	if err != nil {
		t.Fatal(err)
	}
	if len(boxes) != 1 {
		t.Fatalf("fleets/pi/status/boxes/ = %v, want exactly one snapshot", boxes)
	}

	// The shared platform namespace must stay untouched: no box snapshot,
	// and definitely no fleet.json (which would only ever appear had
	// publish been true).
	sharedBoxes, err := under.List(ctx, "status/boxes/")
	if err != nil {
		t.Fatal(err)
	}
	if len(sharedBoxes) != 0 {
		t.Errorf("status/boxes/ = %v, want none — a fleet box must never write the shared namespace", sharedBoxes)
	}
	if _, _, err := under.Get(ctx, "status/fleet.json"); err == nil {
		t.Error("a fleet box must never publish status/fleet.json")
	}
}

// ---- Task 6 (multi-upstream plan): per-upstream probe state in the snapshot -

// TestStatusTickCarriesProbedUpstreams is Task 6's Step-1 test: the state the
// edge's probers hold for a multi-upstream route must reach the box's OWN
// snapshot object through the real tick — statusTick over a DirStore, read
// back as the bytes an operator (or the console) would parse — not a
// hand-built Box.
//
// It drives the same seam production does: setUpstreamSource is what RunAgent
// hands the live *edge.Edge's UpstreamStates method to from the edge-ready
// callback, and upstreamStates() is what statusLoop reads on every tick. The
// fixture stands in for the Edge here for the same reason every other probe in
// this file is injected — a real prober would need a live listener and a
// goroutine this package cannot stop.
func TestStatusTickCarriesProbedUpstreams(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "ha.example.com", Kind: "service", Status: "ready"}); err != nil {
		t.Fatal(err)
	}

	store := release.NewDirStore(t.TempDir())
	ctx := context.Background()
	since := time.Now().Add(-90 * time.Second).UTC().Truncate(time.Second)

	setUpstreamSource(func() map[string][]status.UpstreamState {
		return map[string][]status.UpstreamState{"ha.example.com": {
			{Addr: "10.0.0.11:8080", Up: true, Since: since},
			{Addr: "10.0.0.12:8080", Up: false, Since: since, Fails: 3, LastError: "dial: connection refused"},
		}}
	})
	t.Cleanup(func() { setUpstreamSource(nil) })

	// Exactly what statusLoop builds each tick.
	probes := status.Probes{MachineID: "test-machine", Upstreams: upstreamStates()}
	if err := statusTick(ctx, store, reg, probes, false, ""); err != nil {
		t.Fatalf("statusTick: %v", err)
	}

	key := "status/boxes/" + status.Label("test-machine") + ".json"
	raw, _, err := store.Get(ctx, key)
	if err != nil {
		t.Fatalf("Get %s: %v", key, err)
	}
	var box status.Box
	if err := json.Unmarshal(raw, &box); err != nil {
		t.Fatalf("unmarshal %s: %v", key, err)
	}
	if box.Metrics == nil {
		t.Fatalf("metrics block missing from %s: %s", key, raw)
	}
	got := box.Metrics.Upstreams["ha.example.com"]
	want := []status.UpstreamState{
		{Addr: "10.0.0.11:8080", Up: true, Since: since},
		{Addr: "10.0.0.12:8080", Up: false, Since: since, Fails: 3, LastError: "dial: connection refused"},
	}
	if !reflect.DeepEqual(got, want) {
		t.Errorf("upstreams for ha.example.com = %+v, want %+v: %s", got, want, raw)
	}

	// Staleness: the map is rebuilt from the prober's State() every tick, so a
	// box that stops probing a route (route deleted, addrs cleared) must leave
	// NO ghost entry behind in the next snapshot — and with nothing probed at
	// all the key must be absent, not an empty object.
	setUpstreamSource(func() map[string][]status.UpstreamState { return nil })
	probes.Upstreams = upstreamStates()
	if err := statusTick(ctx, store, reg, probes, false, ""); err != nil {
		t.Fatalf("statusTick (no probers): %v", err)
	}
	raw, _, err = store.Get(ctx, key)
	if err != nil {
		t.Fatalf("Get %s after second tick: %v", key, err)
	}
	if strings.Contains(string(raw), "ha.example.com") || strings.Contains(string(raw), "upstreams") {
		t.Errorf("second snapshot still carries a retired route's probe state: %s", raw)
	}
}

// TestUpstreamStatesUnsetSourceIsNil pins the honest degrade: `carlos edge`
// and every test that never calls setUpstreamSource must read nil, not panic
// — statusLoop calls upstreamStates() on every tick regardless of whether the
// edge-ready callback has fired yet (it runs its first tick a minute in, but
// the ordering is not something this loop gets to assume).
func TestUpstreamStatesUnsetSourceIsNil(t *testing.T) {
	setUpstreamSource(nil)
	if got := upstreamStates(); got != nil {
		t.Errorf("upstreamStates() with no source = %+v, want nil", got)
	}
}

// TestWrapListeningProbeNullsUnitBackedSockets is fix round 1's regression
// test: listeningProbe's raw /proc/net/unix read is confidently WRONG for
// systemd-socket-activated (registry.BackingUnit) routes, since the
// carlos-app@.socket unit holds the kernel-level listener open at all times
// regardless of whether the paired service is actually awake — so a raw
// probe would report "awake" forever for a sleeping unit-backed tenant. This
// exercises the pure seam (unitBackedSockets + wrapListeningProbe) statusLoop
// wires together, with a stand-in raw probe standing in for
// /proc/net/unix's "yes, bound" answer on both sockets — exactly what a real
// kernel-table read would say, since the unit's socket is always bound.
func TestWrapListeningProbeNullsUnitBackedSockets(t *testing.T) {
	routes := []registry.Route{
		{Host: "exec-app", Kind: "instance", Hibernate: true, Socket: "/run/carlos/exec-app.sock"},
		{Host: "unit-app", Kind: "instance", Hibernate: true, Socket: "/run/carlos/apps/unit-app.sock", Backing: registry.BackingUnit},
	}

	// rawAlwaysListening stands in for listeningProbe: both sockets read as
	// bound/listening, which is the honest kernel-table answer for BOTH
	// routes here (the unit-backed one's systemd .socket unit never closes
	// its listener) — the wrapper's job is to know that answer can't be
	// trusted for the unit-backed one.
	rawAlwaysListening := func(string) *bool { v := true; return &v }
	wrapped := wrapListeningProbe(unitBackedSockets(routes), rawAlwaysListening)

	if got := wrapped(routes[0].Socket); got == nil || !*got {
		t.Errorf("exec-backed socket must pass through to the raw probe unchanged, got %v", got)
	}
	if got := wrapped(routes[1].Socket); got != nil {
		t.Errorf("unit-backed socket must report nil (unknown), got %v — its .socket unit's always-open listener can't distinguish awake from asleep", got)
	}

	// One level up, the observable effect on a published snapshot: CollectBox
	// nulls the WHOLE box's Awake/Asleep pair the instant any hibernating
	// route's Listening call returns nil (snapshot.go's partial-knowledge
	// rule — "some known, some not" publishes as null, never a partial
	// count). So even though the exec-backed route's own state IS known
	// here, the box-wide totals still come back nil because the unit-backed
	// route's is not. That's the correct, if blunt, behavior this fix
	// produces: a box carrying even one systemd-socket-activated hibernating
	// tenant can't yet publish a trustworthy awake/asleep split at all.
	b := status.CollectBox(routes, status.Probes{MachineID: "m", Listening: wrapped}, time.Now())
	if b.Routes.Awake != nil || b.Routes.Asleep != nil {
		t.Errorf("box awake=%v asleep=%v, want both nil: one unit-backed hibernating route must null the whole box's counts, not just its own", b.Routes.Awake, b.Routes.Asleep)
	}
}

// ---- domains: the reconcile loop's pure config builder ----------------------

// TestDomainsTickConfig pins the env contract: PLATFORM_APPS_DOMAIN unset
// means a zero Config (Reconcile is inert on it — the feature does not
// exist); set means every seam is populated, with PLATFORM_EDGE_IPS parsed
// comma-separated and the box's public IP as the fallback.
//
// That parsed list is a RETURN VALUE, not a Config field, and this test
// asserts it there. Config.EdgeIPs is now a func answering the deployment's
// LIVE edge set — every box's published edges file folded against its status
// snapshot — which needs a deployment store, and this builder must stay pure
// and getenv-injectable. domainsLoop unions the two once its store is open,
// so what is pinned here is exactly the ENV half of that union.
//
// It follows that this builder never populates EdgeIPs at all, and that
// Resolve no longer waits on this box knowing an edge address: a box with no
// PLATFORM_EDGE_IPS whose fold supplies edges must still stamp. The
// stand-down when nothing is knowable moved to domains.Reconcile's section 4,
// where the set is actually known — see TestReconcileStandsDownWithNoEdgeIPs.
func TestDomainsTickConfig(t *testing.T) {
	env := func(m map[string]string) func(string) string {
		return func(k string) string { return m[k] }
	}

	if got, envIPs, err := domainsTickConfig(env(map[string]string{}), "9.9.9.9"); !reflect.DeepEqual(got, domains.Config{}) || envIPs != nil || err != nil {
		t.Errorf("unset PLATFORM_APPS_DOMAIN: config = %+v, envIPs = %v, err = %v, want zero, nil, nil", got, envIPs, err)
	}

	got, envIPs, err := domainsTickConfig(env(map[string]string{
		"PLATFORM_APPS_DOMAIN": "oncarlos.test",
		"PLATFORM_EDGE_IPS":    "203.0.113.7, 198.51.100.9",
		"CARLOS_DATA_DIR":      "/custom",
	}), "9.9.9.9")
	if err != nil {
		t.Errorf("configured deployment with no release key: err = %v, want nil", err)
	}
	if got.AppsDomain != "oncarlos.test" {
		t.Errorf("AppsDomain = %q", got.AppsDomain)
	}
	if !reflect.DeepEqual(envIPs, []string{"203.0.113.7", "198.51.100.9"}) {
		t.Errorf("envIPs = %v, want the parsed PLATFORM_EDGE_IPS (public IP must not override an explicit list)", envIPs)
	}
	if got.EdgeIPs != nil {
		t.Error("EdgeIPs must be left nil here: only domainsLoop, which has the store, can answer the live set")
	}
	if got.AcmeDir != filepath.Join("/custom", "acme") {
		t.Errorf("AcmeDir = %q, want the edge's cache dir under CARLOS_DATA_DIR", got.AcmeDir)
	}
	if got.Resolve == nil || got.Now == nil {
		t.Error("Resolve/Now seams not populated on a configured deployment")
	}

	// No explicit edge IPs: the box's own public IP (IMDS) is the default.
	got, envIPs, _ = domainsTickConfig(env(map[string]string{"PLATFORM_APPS_DOMAIN": "oncarlos.test"}), "9.9.9.9")
	if !reflect.DeepEqual(envIPs, []string{"9.9.9.9"}) {
		t.Errorf("envIPs = %v, want the public-IP fallback", envIPs)
	}
	if got.AcmeDir != filepath.Join("/data", "acme") {
		t.Errorf("AcmeDir = %q, want the /data default", got.AcmeDir)
	}

	// No edge IP knowable from the ENVIRONMENT at all. The env half is empty
	// — but Resolve is still armed, because the fold may well know edges this
	// box's variables never named. Minting runs regardless.
	got, envIPs, _ = domainsTickConfig(env(map[string]string{"PLATFORM_APPS_DOMAIN": "oncarlos.test"}), "")
	if envIPs != nil {
		t.Errorf("envIPs = %v, want nothing: neither PLATFORM_EDGE_IPS nor IMDS answered", envIPs)
	}
	if got.Resolve == nil {
		t.Error("Resolve must stay armed with no env edge IPs — the fold, not this box's variables, decides whether a pass can stamp")
	}
	if got.AppsDomain != "oncarlos.test" {
		t.Error("minting must stay configured even without edge IPs")
	}
}

// TestDomainsTickConfigReleasePubKey pins the auto-purge's arming condition.
// The reconcile pass destroys objects only when it can VERIFY an app's
// production flag, so the key is what switches purging on — and a missing or
// unparseable one leaves it off rather than failing the loop, because hiding
// a trashed app (reversible) must keep working on a box that may not destroy
// anything (irreversible).
func TestDomainsTickConfigReleasePubKey(t *testing.T) {
	env := func(m map[string]string) func(string) string {
		return func(k string) string { return m[k] }
	}
	_, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	got, _, err := domainsTickConfig(env(map[string]string{
		"PLATFORM_APPS_DOMAIN":  "oncarlos.test",
		"CARLOS_RELEASE_PUBKEY": pubB64,
	}), "9.9.9.9")
	if err != nil {
		t.Fatalf("a good key returned err = %v", err)
	}
	want, err := release.ParsePublicKey(pubB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	if !got.ReleasePub.Equal(want) {
		t.Errorf("ReleasePub = %x, want %x", got.ReleasePub, want)
	}

	// UNSET and UNPARSEABLE both leave purging off, and the caller must be
	// able to tell them apart — release.ParsePublicKey says the same sentence
	// for both, so the distinction lives here: unset is (nil, nil), a typo is
	// (nil, error). domainsLoop logs the error verbatim, which is the one line
	// an operator reads when purging silently is not happening.
	for _, key := range []string{"", "   "} {
		got, _, err = domainsTickConfig(env(map[string]string{
			"PLATFORM_APPS_DOMAIN":  "oncarlos.test",
			"CARLOS_RELEASE_PUBKEY": key,
		}), "9.9.9.9")
		if got.ReleasePub != nil || err != nil {
			t.Errorf("CARLOS_RELEASE_PUBKEY=%q: ReleasePub = %x, err = %v — unset must be silent, not an error", key, got.ReleasePub, err)
		}
	}
	for _, key := range []string{"not-base64!!", base64.StdEncoding.EncodeToString([]byte("short"))} {
		got, _, err = domainsTickConfig(env(map[string]string{
			"PLATFORM_APPS_DOMAIN":  "oncarlos.test",
			"CARLOS_RELEASE_PUBKEY": key,
		}), "9.9.9.9")
		if got.ReleasePub != nil {
			t.Errorf("CARLOS_RELEASE_PUBKEY=%q armed purging with %x", key, got.ReleasePub)
		}
		if err == nil {
			t.Errorf("CARLOS_RELEASE_PUBKEY=%q returned no error — a typo would be logged as \"not set\"", key)
		} else if !strings.Contains(err.Error(), "CARLOS_RELEASE_PUBKEY") {
			t.Errorf("CARLOS_RELEASE_PUBKEY=%q: err = %v, want it to name the variable", key, err)
		}
		// The rest of the pass is unaffected: hiding trashed apps is
		// reversible and must not stop because a key is mistyped.
		if got.AppsDomain != "oncarlos.test" || got.Resolve == nil {
			t.Errorf("CARLOS_RELEASE_PUBKEY=%q disabled the whole pass: %+v", key, got)
		}
	}
}

// ---- instance provisioning: the reconcile loop's pure config builder ------

// TestProvisionTickConfig pins the CARLOS_REGION/CARLOS_POOLS env contract.
// Unlike domainsTickConfig, there is no inert zero-Config case: an unset
// region/pools is a legal, still-armed state — see provision.Reconcile's
// doc comment (a box with no region still serves ""-scoped records) — so
// DataDir always defaults to /data regardless of whether region/pools are
// set.
func TestProvisionTickConfig(t *testing.T) {
	env := func(m map[string]string) func(string) string {
		return func(k string) string { return m[k] }
	}

	got := provisionTickConfig(env(map[string]string{}))
	if got.Region != "" {
		t.Errorf("unset CARLOS_REGION: Region = %q, want empty", got.Region)
	}
	if got.Pools != nil {
		t.Errorf("unset CARLOS_POOLS: Pools = %v, want nil", got.Pools)
	}
	if got.DataDir != "/data" {
		t.Errorf("DataDir = %q, want /data default", got.DataDir)
	}
	if got.Now == nil {
		t.Error("Now seam not populated")
	}

	got = provisionTickConfig(env(map[string]string{
		"CARLOS_REGION": "ie",
		"CARLOS_POOLS":  "tito, shared",
	}))
	if got.Region != "ie" {
		t.Errorf("Region = %q, want ie", got.Region)
	}
	if !reflect.DeepEqual(got.Pools, []string{"tito", "shared"}) {
		t.Errorf("Pools = %v, want [tito shared] (whitespace trimmed)", got.Pools)
	}

	// Whitespace-only and empty entries between commas are dropped, not
	// preserved as "".
	got = provisionTickConfig(env(map[string]string{
		"CARLOS_POOLS": " , tito ,, shared ,",
	}))
	if !reflect.DeepEqual(got.Pools, []string{"tito", "shared"}) {
		t.Errorf("Pools = %v, want empties dropped and whitespace trimmed", got.Pools)
	}

	got = provisionTickConfig(env(map[string]string{"CARLOS_DATA_DIR": "/custom"}))
	if got.DataDir != "/custom" {
		t.Errorf("DataDir = %q, want /custom", got.DataDir)
	}

	// Region set with no pools: Pools stays nil (the shared "" pool only —
	// NOT "every pool").
	got = provisionTickConfig(env(map[string]string{"CARLOS_REGION": "ie"}))
	if got.Region != "ie" || got.Pools != nil {
		t.Errorf("Region = %q Pools = %v, want ie / nil", got.Region, got.Pools)
	}

	// The release pubkey (#340): present and parseable arms the production
	// guard on trash-driven route retirement.
	if got.ReleasePub != nil {
		t.Error("ReleasePub set with CARLOS_RELEASE_PUBKEY unset")
	}
	_, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	got = provisionTickConfig(env(map[string]string{"CARLOS_RELEASE_PUBKEY": pubB64}))
	if got.ReleasePub == nil {
		t.Error("ReleasePub nil with a valid CARLOS_RELEASE_PUBKEY")
	}

	// A key SET but unparseable must not be mistaken for a key that parsed.
	// It lands on nil — the same unguarded end state domains.Config accepts
	// for the same decision — and says so in the log rather than silently.
	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	got = provisionTickConfig(env(map[string]string{"CARLOS_RELEASE_PUBKEY": "not-a-key"}))
	if got.ReleasePub != nil {
		t.Error("ReleasePub set from an unparseable CARLOS_RELEASE_PUBKEY")
	}
	if !strings.Contains(logged.String(), "CARLOS_RELEASE_PUBKEY") {
		t.Errorf("an unparseable key was swallowed; log was:\n%s", logged.String())
	}
}

// TestProvisionTickConfigReadsInbound pins CARLOS_INBOUND's contract: an
// unset var is the SAME apps-domain-style inert gate CARLOS_REGION/
// CARLOS_POOLS already use — a legal "no constraint" state, never a
// "feature not configured" one — so it must read back as "" (today's
// behaviour, byte for byte), and only an explicit "none" changes Config.
func TestProvisionTickConfigReadsInbound(t *testing.T) {
	env := func(m map[string]string) func(string) string {
		return func(k string) string { return m[k] }
	}

	got := provisionTickConfig(env(map[string]string{
		"CARLOS_REGION": "ie", "CARLOS_POOLS": "eleven", "CARLOS_INBOUND": "none",
	}))
	if got.Inbound != "none" {
		t.Errorf("Inbound = %q, want none", got.Inbound)
	}

	got = provisionTickConfig(env(map[string]string{}))
	if got.Inbound != "" {
		t.Fatal("an unset CARLOS_INBOUND must be an inert default — today's behaviour exactly")
	}

	// Review fix (Task 10 finding): provisionTickConfig must fold case and
	// whitespace, and must read an unrecognised value as restrictive rather
	// than letting it silently fall through as "any". These pin the actual
	// production reading path (getenv -> provision.NormalizeInbound), not
	// just the pure function in the provision package.
	got = provisionTickConfig(env(map[string]string{"CARLOS_INBOUND": "  NONE  "}))
	if got.Inbound != "none" {
		t.Errorf("Inbound = %q, want none (case/whitespace folded)", got.Inbound)
	}
	got = provisionTickConfig(env(map[string]string{"CARLOS_INBOUND": "off"}))
	if got.Inbound != "none" {
		t.Errorf("Inbound = %q, want none (unrecognised value reads restrictive)", got.Inbound)
	}
}

func TestBootstrapIdempotent(t *testing.T) {
	root := t.TempDir()
	if err := bootstrap(root, nil); err != nil {
		t.Fatalf("first bootstrap: %v", err)
	}
	if err := bootstrap(root, nil); err != nil {
		t.Fatalf("second bootstrap: %v", err)
	}
	// Still all present after the re-run.
	for _, name := range expectedUnits {
		if _, err := os.Stat(filepath.Join(root, "etc", "systemd", "system", name)); err != nil {
			t.Errorf("after re-run: %v", err)
		}
	}
}

func TestIsCanaryRoute(t *testing.T) {
	cases := []struct {
		channel string
		want    bool
	}{
		{"canary/x", true},
		{"canary/pr-142", true},
		{"stable", false},
		{"beta", false},
		{"canary", false}, // no slash — same convention as bake.go's DefaultHolds wildcard
		{"", false},
	}
	for _, c := range cases {
		rt := registry.Route{Channel: c.channel}
		if got := isCanaryRoute(rt); got != c.want {
			t.Errorf("isCanaryRoute(Channel=%q) = %v, want %v", c.channel, got, c.want)
		}
	}
}

// ---- Pipelines Task 5: the install path arms timers -------------------------

// TestSystemctlPlanArmsTimers is the regression test for the flagship's
// silent-adoption bug: carlos-adopt.timer was enabled — the .wants symlink was
// right there — but never active, because `systemctl enable` only writes the
// symlink and the timer does not arm until something starts it or the box
// reboots. Nothing had rebooted, so no release ever adopted itself and every
// deploy quietly needed a hand-run `systemctl start carlos-adopt.service`.
// Every timer bootstrap enables must therefore also be started, and no
// SERVICE may be (carlos-edge needs its binary and host.env staged first —
// bootstrap's own closing line tells the operator to start it). The one
// exemption is carlos-relay.service, whose ConditionPathExists makes the
// start a no-op on any box the operator has not armed with a region master
// (relay spec §3); TestBootstrapStartsRelayUnit pins that it IS started.
//
// PATH units are the same trap with the same fix: an enabled-but-unstarted
// carlos-adopt.path watches nothing until the next reboot, so the agent's
// config-sync marker would be written to a file no one reads and unit-backed
// apps would silently keep only the hourly floor.
func TestSystemctlPlanArmsTimers(t *testing.T) {
	plan := systemctlPlan(enabledUnits)
	if len(plan) == 0 || plan[0][0] != "daemon-reload" {
		t.Fatalf("plan must reload units first, got %v", plan)
	}

	started := map[string]bool{}
	enabled := map[string]bool{}
	for _, args := range plan {
		for _, unit := range args[1:] {
			switch args[0] {
			case "start":
				started[unit] = true
			case "enable":
				enabled[unit] = true
			}
		}
	}
	for _, u := range enabledUnits {
		if !enabled[u] {
			t.Errorf("%s: enabled by the plan? no — bootstrap must still enable everything it did before", u)
		}
		if (strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path")) && !started[u] {
			t.Errorf("%s: enabled but never started — that is exactly the bug (enable only symlinks; the unit arms at the next boot)", u)
		}
		if strings.HasSuffix(u, ".service") && u != relayUnitName && started[u] {
			t.Errorf("%s: bootstrap must not start a service — the binary and /etc/carlos/host.env may not be staged yet", u)
		}
	}
}

// TestBootstrapStagingRunsNoSystemctl guards the other half of the same
// change: `carlos bootstrap -root <dir>` is how the unit-refresh path stages
// units into a temp root on a LIVE box (see pinfra's
// update-platform-binary.sh, which stages into /tmp/carlos-unitstage). Arming
// timers from a staging run would touch the live system from a command whose
// entire contract is "write files, change nothing" — so a staging run must
// print what a live one would do and execute none of it. The fake systemctl on
// PATH records any invocation; the file must never appear.
func TestBootstrapStagingRunsNoSystemctl(t *testing.T) {
	bin := t.TempDir()
	log := filepath.Join(t.TempDir(), "systemctl-calls")
	script := "#!/bin/sh\necho \"$@\" >> " + log + "\n"
	if err := os.WriteFile(filepath.Join(bin, "systemctl"), []byte(script), 0o755); err != nil {
		t.Fatal(err)
	}
	t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))

	if err := bootstrap(t.TempDir(), nil); err != nil {
		t.Fatalf("bootstrap: %v", err)
	}
	if b, err := os.ReadFile(log); err == nil {
		t.Errorf("staging bootstrap ran systemctl: %s", b)
	}
}

// TestBootstrapTmpfilesBlessesConsumersGroup pins the instance-socket
// consumers contract (issue #103): /run/carlos carries the dedicated
// carlos-sockets group with the setgid bit — so every socket an exec child
// binds inside inherits the consumers group — while /run/carlos/apps stays
// carlos:carlos (unit tenants get their fd passed; enrollment deliberately
// does not reach them). The group is dedicated, never `carlos`: membership
// in `carlos` also grants group-read of /data, i.e. every instance's DB and
// CARLOS_ADMIN_TOKEN.
func TestBootstrapTmpfilesBlessesConsumersGroup(t *testing.T) {
	root := t.TempDir()
	if err := bootstrap(root, nil); err != nil {
		t.Fatalf("bootstrap: %v", err)
	}
	tf, err := os.ReadFile(filepath.Join(root, "etc", "tmpfiles.d", "carlos.conf"))
	if err != nil {
		t.Fatalf("tmpfiles conf: %v", err)
	}
	if want := "d /run/carlos 2750 carlos carlos-sockets -"; !strings.Contains(string(tf), want) {
		t.Errorf("tmpfiles conf %q missing %q", tf, want)
	}
	if want := "d /run/carlos/apps 0750 carlos carlos -"; !strings.Contains(string(tf), want) {
		t.Errorf("tmpfiles conf %q missing %q — the tenant socket dir must stay out of the consumers group", tf, want)
	}

	// Bootstrap's own dir handling must agree with the tmpfiles line: setgid
	// on /run/carlos (mode is what makes group inheritance real between
	// boots), and NOT on apps/.
	fi, err := os.Stat(filepath.Join(root, "run", "carlos"))
	if err != nil {
		t.Fatal(err)
	}
	if fi.Mode()&os.ModeSetgid == 0 {
		t.Errorf("/run/carlos mode %v: missing setgid — bound sockets would not inherit the consumers group", fi.Mode())
	}
	if fi.Mode().Perm() != 0o750 {
		t.Errorf("/run/carlos perm %o, want 750", fi.Mode().Perm())
	}
	afi, err := os.Stat(filepath.Join(root, "run", "carlos", "apps"))
	if err != nil {
		t.Fatal(err)
	}
	if afi.Mode()&os.ModeSetgid != 0 {
		t.Errorf("/run/carlos/apps mode %v: setgid must not leak onto the tenant socket dir", afi.Mode())
	}

	// The sysusers.d snippet is what makes the tmpfiles line safe at boot on
	// a box whose conf arrived via a staged-tree refresh (never a live
	// groupadd): systemd-sysusers runs Before=systemd-tmpfiles-setup, and a
	// tmpfiles line whose group doesn't resolve is DROPPED — /run/carlos then
	// isn't created and carlos-edge (ReadWritePaths, no "-") can't start.
	su, err := os.ReadFile(filepath.Join(root, "etc", "sysusers.d", "carlos.conf"))
	if err != nil {
		t.Fatalf("sysusers conf: %v", err)
	}
	if want := "g carlos-sockets -"; !strings.Contains(string(su), want) {
		t.Errorf("sysusers conf %q missing %q", su, want)
	}
}

// TestBootstrapPrivilegedPlanEnrollsConsumers pins the printed-equals-run
// contract for the group half, the same discipline systemctlPlan established:
// a staging run prints exactly the commands a live run executes, so the two
// can never drift. -socket-consumer users each get a usermod enrollment into
// the dedicated group; no consumers means no usermod lines but the group and
// dir convergence always run.
func TestBootstrapPrivilegedPlanEnrollsConsumers(t *testing.T) {
	got := privilegedPlan([]string{"titogo", "titogo-relay"})
	// The /run/carlos chown is recursive AND targets the consumers group: a
	// re-run is the documented enrollment path, and a recursive chown back
	// to carlos:carlos would reset every live instance socket's group,
	// stranding already-enrolled consumers until each instance rebinds.
	// apps/ is put back to carlos:carlos right after (unit tenants stay out
	// of consumer reach).
	want := [][]string{
		{"useradd", "--system", "--user-group", "--no-create-home", "--shell", "/usr/sbin/nologin", "carlos"},
		{"groupadd", "--system", "carlos-sockets"},
		{"chown", "-R", "carlos:carlos", "/data"},
		{"chown", "-R", "carlos:carlos-sockets", "/run/carlos"},
		{"chown", "-R", "carlos:carlos", "/run/carlos/apps"},
		{"chmod", "2750", "/run/carlos"},
		{"usermod", "-aG", "carlos-sockets", "titogo"},
		{"usermod", "-aG", "carlos-sockets", "titogo-relay"},
	}
	if !reflect.DeepEqual(got, want) {
		t.Errorf("privilegedPlan = %v, want %v", got, want)
	}
	// No consumers: same plan minus the usermod tail.
	if got := privilegedPlan(nil); !reflect.DeepEqual(got, want[:6]) {
		t.Errorf("privilegedPlan(nil) = %v, want %v", got, want[:6])
	}
}

// TestBootstrapOffcloudRequiresCredentialsFile pins the off-cloud precondition
// (issue #53 point 4): a box with no EC2 instance profile must have its scoped
// credentials staged — and staged at 0600 — before bootstrap proceeds. This is
// the staging-root path (no root privileges available in tests), so the
// root:service ownership repair never runs; that only happens on a live root.
func TestBootstrapOffcloudRequiresCredentialsFile(t *testing.T) {
	root := t.TempDir()
	if err := bootstrapOffcloud(root, nil); err == nil {
		t.Fatal("want error when <root>/etc/carlos/credentials is absent")
	}
	credDir := filepath.Join(root, "etc", "carlos")
	os.MkdirAll(credDir, 0o755)
	credFile := filepath.Join(credDir, "credentials")
	os.WriteFile(credFile, []byte("[default]\naws_access_key_id=X\naws_secret_access_key=Y\n"), 0o644)
	if err := bootstrapOffcloud(root, nil); err == nil {
		t.Fatal("want error on 0644 credentials (must be 0600)")
	}
	os.Chmod(credFile, 0o600)
	if err := bootstrapOffcloud(root, nil); err != nil {
		t.Fatalf("offcloud bootstrap with staged 0600 credentials: %v", err)
	}
	// The normal bootstrap artifacts must exist too — offcloud is a superset.
	if _, err := os.Stat(filepath.Join(root, "etc", "systemd", "system", "carlos-edge.service")); err != nil {
		t.Fatalf("units not installed: %v", err)
	}
}

// TestAdoptTimerStateReading pins the probe's honesty rules at the seam that
// needs no systemd to test: systemd's own word for the timer's state maps to
// running/not-running, and anything this probe cannot interpret — no systemd
// at all, a systemctl that failed to run, a word from a future version —
// reports nil (unknown), never a reassuring true.
func TestAdoptTimerStateReading(t *testing.T) {
	cases := []struct {
		out  string
		want *bool
	}{
		{"active\n", boolPtr(true)},
		{"inactive\n", boolPtr(false)},
		{"failed\n", boolPtr(false)},
		{"deactivating\n", boolPtr(false)},
		{"unknown\n", boolPtr(false)}, // systemd answered: it has no such unit loaded
		{"", nil},                     // systemctl never ran (not a systemd box)
		{"something-new\n", nil},      // a word this probe does not understand
	}
	for _, c := range cases {
		got := parseIsActive(c.out)
		switch {
		case c.want == nil && got != nil:
			t.Errorf("parseIsActive(%q) = %v, want nil", c.out, *got)
		case c.want != nil && got == nil:
			t.Errorf("parseIsActive(%q) = nil, want %v", c.out, *c.want)
		case c.want != nil && *got != *c.want:
			t.Errorf("parseIsActive(%q) = %v, want %v", c.out, *got, *c.want)
		}
	}
}

func boolPtr(v bool) *bool { return &v }

func TestParseMeminfoBytes(t *testing.T) {
	total, avail := parseMeminfoBytes(strings.NewReader(
		"MemTotal:       3961472 kB\nMemFree:  100 kB\nMemAvailable:   1980736 kB\n"))
	if total == nil || *total != 3961472*1024 || avail == nil || *avail != 1980736*1024 {
		t.Fatalf("got %v %v", total, avail)
	}
	if tt, aa := parseMeminfoBytes(strings.NewReader("garbage")); tt != nil || aa != nil {
		t.Fatal("unparseable meminfo must yield nil, not zero")
	}
}

func TestParseLoad1(t *testing.T) {
	if l := parseLoad1("0.42 0.36 0.30 1/234 5678\n"); l == nil || *l != 0.42 {
		t.Fatalf("got %v", l)
	}
	if l := parseLoad1(""); l != nil {
		t.Fatal("empty loadavg must yield nil")
	}
}

// TestProbedUnitsAreTheOnesBootstrapArms pins the pairing that makes both
// systemd probes meaningful: each one must name a unit bootstrap actually
// installs AND arms. A probe pointed at a unit no box has answers "unknown"
// (systemd's own word for "no such unit"), which this package maps to a
// confident false — so a typo here would report every box in the fleet as
// broken, and a renamed unit would report a live box as dead.
func TestProbedUnitsAreTheOnesBootstrapArms(t *testing.T) {
	for _, unit := range []string{adoptTimerUnit, configPathUnit, systemUpdatePathUnit} {
		if !slices.Contains(enabledUnits, unit) {
			t.Errorf("%s: probed but not in enabledUnits — bootstrap never arms it", unit)
		}
		if !slices.Contains(expectedUnits, unit) {
			t.Errorf("%s: probed but not an installed unit", unit)
		}
	}
}

func TestBootstrapPlanJSONNamesWhatALiveRunArms(t *testing.T) {
	p := currentBootstrapPlan()
	want := map[string]bool{"carlos-edge.service": true, "carlos-adopt.timer": true, "carlos-adopt.path": true,
		"carlos-stage.path": true, "carlos-restart.path": true, "carlos-system-update.path": true, "carlos-system-recover.service": true,
		"data-instances.mount": true, "run-carlos-instances.mount": true, "run-carlos-logs.mount": true}
	got := map[string]bool{}
	for _, u := range p.Enabled {
		got[u] = true
	}
	for u := range want {
		if !got[u] {
			t.Errorf("plan lacks %s", u)
		}
	}
	if got["carlos-system-update.service"] {
		t.Error("the path-triggered oneshot must not be enabled: it is started by its .path")
	}
	inert := map[string]bool{}
	for _, u := range p.Inert {
		inert[u] = true
	}
	if !inert["carlos-restore-verify.timer"] || !inert["carlos-hibernate-verify.timer"] {
		t.Errorf("inert = %v", p.Inert)
	}
	// And systemctlPlan starts no service but the relay, whatever is
	// enabled — the relay is safe because its ConditionPathExists gates it
	// on a masters file the operator stages (relay spec §3).
	for _, args := range systemctlPlan(p.Enabled) {
		if args[0] == "start" {
			for _, u := range args[1:] {
				if strings.HasSuffix(u, ".service") && u != relayUnitName {
					t.Errorf("plan starts a service: %v", args)
				}
			}
		}
	}
}

func TestSystemUnitsAreShapedLikeTheStageRelay(t *testing.T) {
	read := func(name string) string {
		b, err := unitsFS.ReadFile("units/" + name)
		if err != nil {
			t.Fatal(err)
		}
		return string(b)
	}
	p := read("carlos-system-update.path")
	for _, want := range []string{"PathChanged=/data/system-update-requested", "Unit=carlos-system-update.service", "WantedBy=multi-user.target"} {
		if !strings.Contains(p, want) {
			t.Errorf("path unit lacks %q", want)
		}
	}
	u := read("carlos-system-update.service")
	for _, want := range []string{"Type=oneshot", "StartLimitIntervalSec=0", "TimeoutStartSec=15min", "OnFailure=carlos-system-recover.service",
		"EnvironmentFile=-/etc/carlos/host.env", "system apply", "carlos.prev", "MemoryMax=256M",
		"floor.min-release", "carlos.prev.allowed", floorGateCall, "seeds-floor.min-release", "carlos.prev.seeds-allowed", seedsFloorGateCall} {
		if !strings.Contains(u, want) {
			t.Errorf("update unit lacks %q", want)
		}
	}
	if strings.Contains(u, "cmp ") {
		t.Errorf("update unit still compares the floor file with cmp, which follows a symlink and blocks on a FIFO")
	}
	for _, never := range []string{"ProtectSystem", "User="} {
		if strings.Contains(u, never) {
			t.Errorf("update unit must run as root with /opt writable; found %q", never)
		}
	}
	r := read("carlos-system-recover.service")
	for _, want := range []string{"Type=oneshot", "Before=carlos-edge.service", "WantedBy=multi-user.target", "TimeoutStartSec=10min", "system apply --recover", "carlos.prev",
		"floor.min-release", "carlos.prev.allowed", floorGateCall, "seeds-floor.min-release", "carlos.prev.seeds-allowed", seedsFloorGateCall} {
		if !strings.Contains(r, want) {
			t.Errorf("recover unit lacks %q", want)
		}
	}
	if strings.Contains(r, "cmp ") {
		t.Errorf("recover unit still compares the floor file with cmp, which follows a symlink and blocks on a FIFO")
	}
}

// TestStatusPublishNeverReadsMetricsPrefixes is the privacy tripwire the
// monitoring spec commits to: operator-grade needles live under metrics/ and
// alerts/, and nothing the public status publisher writes may ever be
// derived from them. Sentinel values planted in metrics objects must not
// appear anywhere under status/ after a full publish tick.
func TestStatusPublishNeverReadsMetricsPrefixes(t *testing.T) {
	ctx := context.Background()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	store := release.NewDirStore(t.TempDir())
	sentinel := 0.987654321
	chunk := metrics.RawChunk{V: 1, Box: "box-a", Hour: "2026-08-02-10", Samples: []metrics.Sample{
		{At: time.Now(), Host: metrics.HostSample{CPUFrac: &sentinel}},
	}}
	if err := metrics.WriteRaw(ctx, store, chunk); err != nil {
		t.Fatal(err)
	}

	// Plant sentinel in alerts/boxes/ as well
	alertState := metrics.AlertState{
		V:         1,
		Box:       "box-a",
		UpdatedAt: time.Now(),
		Rules: map[string]metrics.RuleState{
			"disk": {Firing: true, Value: sentinel},
		},
	}
	alertBody, err := json.Marshal(alertState)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.Put(ctx, metrics.AlertKey("box-a"), alertBody, ""); err != nil {
		t.Fatal(err)
	}

	if err := statusTick(ctx, store, reg, status.Probes{MachineID: "m"}, true, ""); err != nil {
		t.Fatal(err)
	}
	keys, err := store.List(ctx, "status/")
	if err != nil {
		t.Fatal(err)
	}
	// Without this the loop below has nothing to inspect and the test passes
	// no matter what statusTick publishes — a tripwire across a doorway
	// nobody walks through.
	if len(keys) == 0 {
		t.Fatal("no status/ keys after a publish tick — the tripwire is not armed")
	}
	for _, k := range keys {
		body, _, err := store.Get(ctx, k)
		if err != nil {
			t.Fatal(err)
		}
		if strings.Contains(string(body), "987654321") {
			t.Errorf("%s contains the metrics sentinel — the status publisher read a metrics object", k)
		}
	}
}

// ---- Pipelines: canary TTL, the status loop's expired-canary count --------

// wireWrapper mirrors release's unexported signedWrapper wire shape
// (payload/sig) — duplicated here rather than imported (it isn't exported,
// deliberately: forging a pointer's PromotedAt is not something production
// code anywhere should be able to do) so this test can write a validly
// SIGNED but arbitrarily-dated channel pointer straight into a DirStore,
// the only way to get an actually-expired pointer under test since every
// public promotion path (release.Promote) stamps the real clock.
type wireWrapper struct {
	Payload string `json:"payload"`
	Sig     string `json:"sig"`
}

// writeBackdatedPointer signs and writes a channel pointer with an
// arbitrary PromotedAt directly to key release.ChannelKey(acct, app,
// channel), bypassing Promote's clock stamp.
func writeBackdatedPointer(t *testing.T, s release.Store, priv ed25519.PrivateKey, acct, app, channel, version string, promotedAt time.Time) {
	t.Helper()
	p := struct {
		Schema     int       `json:"schema"`
		App        string    `json:"app"`
		Channel    string    `json:"channel"`
		Version    string    `json:"version"`
		PromotedAt time.Time `json:"promoted_at"`
	}{Schema: 1, App: app, Channel: channel, Version: version, PromotedAt: promotedAt}
	payload, err := json.Marshal(p)
	if err != nil {
		t.Fatalf("marshal backdated pointer: %v", err)
	}
	b64 := base64.StdEncoding.EncodeToString(payload)
	sig := ed25519.Sign(priv, []byte(b64))
	wrapper, err := json.Marshal(wireWrapper{Payload: b64, Sig: base64.StdEncoding.EncodeToString(sig)})
	if err != nil {
		t.Fatalf("marshal backdated wrapper: %v", err)
	}
	if _, err := s.Put(context.Background(), release.ChannelKey(acct, app, channel), wrapper, ""); err != nil {
		t.Fatalf("write backdated pointer: %v", err)
	}
}

func testCanaryKeys(t *testing.T) (ed25519.PrivateKey, ed25519.PublicKey) {
	t.Helper()
	seedB64, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatalf("GenerateKey: %v", err)
	}
	priv, err := release.ParseSigningKey(seedB64, "test")
	if err != nil {
		t.Fatalf("ParseSigningKey: %v", err)
	}
	pub, err := release.ParsePublicKey(pubB64, "test")
	if err != nil {
		t.Fatalf("ParsePublicKey: %v", err)
	}
	return priv, pub
}

// No canary routes at all needs no store to answer "none" — a confident,
// known 0, not the unknown/nil path a missing store would otherwise imply.
func TestCountExpiredCanariesNoCanaryRoutesIsZero(t *testing.T) {
	routes := []registry.Route{
		{Host: "a.example.com", Kind: "instance", App: "myapp", Channel: "stable"},
		{Host: "b.example.com", Kind: "static"},
	}
	got := countExpiredCanaries(context.Background(), routes, nil, nil, time.Now())
	if got == nil || *got != 0 {
		t.Errorf("count = %v, want 0 (no canary routes, no store needed)", got)
	}
}

// A canary route exists but there is no deployment store to resolve its
// pointer against — the "routes managed by hand" box shape adoptLoop's own
// doc comment names. That must publish unknown, never a guessed 0.
func TestCountExpiredCanariesNilStoreIsUnknown(t *testing.T) {
	routes := []registry.Route{{Host: "rc.example.com", Kind: "instance", App: "myapp", Channel: "canary/x"}}
	if got := countExpiredCanaries(context.Background(), routes, nil, nil, time.Now()); got != nil {
		t.Errorf("count = %v, want nil (no store to resolve the pointer against)", *got)
	}
}

// A canary route whose channel was never actually promoted (no pointer at
// that key) is the same "can't answer" case as no store at all — release.
// Channel returns os.ErrNotExist, and that must null the count too, not
// silently skip the route and undercount.
func TestCountExpiredCanariesUnresolvablePointerIsUnknown(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, pub := testCanaryKeys(t)
	routes := []registry.Route{{Host: "rc.example.com", Kind: "instance", App: "myapp", Channel: "canary/x", Account: "bab"}}
	if got := countExpiredCanaries(context.Background(), routes, s, pub, time.Now()); got != nil {
		t.Errorf("count = %v, want nil (channel was never promoted)", *got)
	}
}

// A fresh canary (real Promote, real clock) counts as 0 — not expired, and
// known.
func TestCountExpiredCanariesFreshCanaryIsZero(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	ctx := context.Background()
	priv, pub := testCanaryKeys(t)
	shipVersion(t, s, "bab", "myapp", "v1")
	if err := release.Promote(ctx, s, priv, "bab", "myapp", "canary/x", "v1"); err != nil {
		t.Fatalf("Promote: %v", err)
	}
	routes := []registry.Route{{Host: "rc.example.com", Kind: "instance", App: "myapp", Channel: "canary/x", Account: "bab"}}
	got := countExpiredCanaries(ctx, routes, s, pub, time.Now())
	if got == nil || *got != 0 {
		t.Errorf("count = %v, want 0 (freshly promoted)", got)
	}
}

// The core case: a backdated (expired) canary pointer counts as 1; a second,
// unexpired canary route on a different channel does not add to it.
func TestCountExpiredCanariesCountsExpiredOnly(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	ctx := context.Background()
	priv, pub := testCanaryKeys(t)
	writeBackdatedPointer(t, s, priv, "bab", "myapp", "canary/old", "v1",
		time.Now().Add(-(release.CanaryTTL + 24*time.Hour)))
	shipVersion(t, s, "bab", "myapp", "v2")
	if err := release.Promote(ctx, s, priv, "bab", "myapp", "canary/fresh", "v2"); err != nil {
		t.Fatalf("Promote: %v", err)
	}
	routes := []registry.Route{
		{Host: "old.example.com", Kind: "instance", App: "myapp", Channel: "canary/old", Account: "bab"},
		{Host: "fresh.example.com", Kind: "instance", App: "myapp", Channel: "canary/fresh", Account: "bab"},
	}
	got := countExpiredCanaries(ctx, routes, s, pub, time.Now())
	if got == nil || *got != 1 {
		t.Errorf("count = %v, want 1 (only the old channel's route is expired)", got)
	}
}

// A platform-minted alias is a second NAME for a route already counted, not
// a second canary route — same convention CollectBox uses to avoid
// double-counting aliases (host_test.go's own
// TestCollectBoxSkipsPlatformMintedRoutes in internal/status covers that
// half; this pins the same rule for the expired-canary count).
func TestCountExpiredCanariesSkipsPlatformAliases(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	priv, pub := testCanaryKeys(t)
	writeBackdatedPointer(t, s, priv, "bab", "myapp", "canary/old", "v1",
		time.Now().Add(-(release.CanaryTTL + 24*time.Hour)))
	routes := []registry.Route{
		{Host: "alias.example.com", Kind: "instance", App: "myapp", Channel: "canary/old", Account: "bab", Platform: true},
	}
	got := countExpiredCanaries(context.Background(), routes, s, pub, time.Now())
	if got == nil || *got != 0 {
		t.Errorf("count = %v, want 0 — a platform alias must not be counted", got)
	}
}

// getCountingStore wraps a real release.Store and counts Get calls — used to
// prove countExpiredCanaries' per-call cache actually dedupes store reads,
// not just channel-pointer resolutions in memory.
type getCountingStore struct {
	release.Store
	gets int
}

func (s *getCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.gets++
	return s.Store.Get(ctx, key)
}

// TestCountExpiredCanariesReadsEachChannelOnce pins the caching half of the
// 2026-08-03 review: statusLoop calls countExpiredCanaries once per
// 5-minute tick, and this function must resolve a given canary channel's
// pointer AT MOST ONCE per call however many of this box's routes follow
// it — N routes sharing one channel costs one store.Get, not N.
func TestCountExpiredCanariesReadsEachChannelOnce(t *testing.T) {
	real := release.NewDirStore(t.TempDir())
	priv, pub := testCanaryKeys(t)
	writeBackdatedPointer(t, real, priv, "bab", "myapp", "canary/old", "v1",
		time.Now().Add(-(release.CanaryTTL + 24*time.Hour)))
	counting := &getCountingStore{Store: real}

	routes := []registry.Route{
		{Host: "a.example.com", Kind: "instance", App: "myapp", Channel: "canary/old", Account: "bab"},
		{Host: "b.example.com", Kind: "instance", App: "myapp", Channel: "canary/old", Account: "bab"},
		{Host: "c.example.com", Kind: "instance", App: "myapp", Channel: "canary/old", Account: "bab"},
	}
	got := countExpiredCanaries(context.Background(), routes, counting, pub, time.Now())
	if got == nil || *got != 3 {
		t.Fatalf("count = %v, want 3 (all three routes share the one expired channel)", got)
	}
	if counting.gets != 1 {
		t.Errorf("store.Get called %d times for 3 routes on ONE channel, want exactly 1 — the per-tick cache must dedupe by (account, app, channel)", counting.gets)
	}
}

// shipVersion ships a minimal one-artifact release so Promote has something
// real to point at — release.Promote refuses to point a channel at a
// version nothing shipped.
func shipVersion(t *testing.T, s release.Store, acct, app, version string) {
	t.Helper()
	m := release.Manifest{
		Schema: 1, App: app, Version: version, Kind: "binary",
		Artifacts: map[string]release.Artifact{"linux-amd64": {File: app}},
	}
	if err := release.Ship(context.Background(), s, acct, m, map[string][]byte{"linux-amd64": []byte("bin")}); err != nil {
		t.Fatalf("Ship %s/%s: %v", app, version, err)
	}
}

// Both restore drills must write their restored copies to DISK, never to the
// PrivateTmp tmpfs. On tmpfs the restored database is charged to the unit's
// own cgroup as memory, so the drill's ceiling has to exceed the fleet's
// largest tenant — and the drill starts failing on exactly the databases it
// exists to vouch for. Measured on the flagship 2026-08-15: a 412M tenant
// produced a 485.6M peak against MemoryMax=512M.
//
// This is a one-line property that would regress silently (the drill still
// passes, right up until a tenant grows), so it is pinned here.
// The tenant template must deny IMDS at the unit level (issue #351): the
// nftables skuid gate is operator territory and a box can simply not have
// one (tito didn't), at which point any app process could read the box
// instance-role credentials off 169.254.169.254. Apps get per-app scoped
// IAM keys via env; the one legitimate IMDS consumer under this template
// is a host_role tenant, whose drop-in re-opens it with IPAddressAllow
// (allow beats deny in systemd's BPF firewall) — see caps.RenderService.
// One-line property, silent regression, so it is pinned here.
func TestAppTemplateDeniesIMDS(t *testing.T) {
	body, err := unitsFS.ReadFile("units/carlos-app@.service")
	if err != nil {
		t.Fatalf("carlos-app@.service: %v", err)
	}
	if !strings.Contains(string(body), "IPAddressDeny=169.254.169.254/32 fd00:ec2::254/128") {
		t.Errorf("carlos-app@.service does not deny IMDS — app code on an ungated box can take the instance role's credentials (issue #351)")
	}
}

func TestVerifyDrillsRestoreToDiskNotTmpfs(t *testing.T) {
	for _, unit := range []string{"carlos-restore-verify.service", "carlos-hibernate-verify.service"} {
		body, err := unitsFS.ReadFile("units/" + unit)
		if err != nil {
			t.Fatalf("%s: %v", unit, err)
		}
		s := string(body)
		if !strings.Contains(s, "CacheDirectory=carlos-verify") {
			t.Errorf("%s has no CacheDirectory — nothing creates the on-disk scratch dir", unit)
		}
		if !strings.Contains(s, "Environment=TMPDIR=%C/carlos-verify") {
			t.Errorf("%s does not point TMPDIR at the on-disk dir; restores land on the tmpfs and count as memory", unit)
		}
		if !strings.Contains(s, "MemoryMax=") {
			t.Errorf("%s has no MemoryMax — a runaway drill should hit its own ceiling, not the box's", unit)
		}
	}
}

// The journal cap ships with every bootstrap (discussion #111): uncapped,
// journald's default is 10% of the filesystem, and 1.2 GB of the flagship's
// 20 GB root was journal by the time it filled.
func TestBootstrapWritesJournaldCap(t *testing.T) {
	root := t.TempDir()
	if err := bootstrap(root, nil); err != nil {
		t.Fatalf("bootstrap: %v", err)
	}
	path := filepath.Join(root, "etc", "systemd", "journald.conf.d", journaldConfName)
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("journald drop-in not written: %v", err)
	}
	got := string(b)
	// The section header is not optional: journald ignores a bare assignment.
	for _, want := range []string{"[Journal]", "SystemMaxUse=256M", "RuntimeMaxUse="} {
		if !strings.Contains(got, want) {
			t.Errorf("drop-in missing %q:\n%s", want, got)
		}
	}
	// The name has to match the one placed by hand on the flagship during the
	// 2026-09-06 relief, or the two files sit side by side and name order
	// decides which cap wins.
	if journaldConfName != "50-cap.conf" {
		t.Errorf("journaldConfName = %q; it must supersede the hand-placed 50-cap.conf", journaldConfName)
	}
	// A staged root is the delivery path to boxes that predate the cap: the
	// self-update apply copies this file out of one by name (installUnits).
	if err := bootstrap(root, nil); err != nil {
		t.Fatalf("re-run: %v", err)
	}
	if _, err := os.Stat(path); err != nil {
		t.Errorf("drop-in gone after a re-run: %v", err)
	}
}

// TestTmpfilesCarriesRelayStateDir is discussion #118 as a test. The relay's
// unit declares ReadWritePaths=/data/relay, and systemd refuses to start a
// unit whose ReadWritePaths names a directory that is not there — "Failed to
// set up mount namespacing: /data/relay: No such file or directory", which is
// what the Ireland box answered on 2026-09-07.
//
// bootstrap's own dirs list creates it, and that is not enough: the dirs list
// runs on a FRESH box or against a staging root, never against the live root
// of a self-update apply. What an apply re-runs is systemd-tmpfiles. So the
// directory has to be in tmpfilesConf, and this test is what keeps a future
// unit's writable state dir from being added to one list and not the other.
func TestTmpfilesCarriesRelayStateDir(t *testing.T) {
	b, err := unitsFS.ReadFile("units/" + relayUnitName)
	if err != nil {
		t.Fatal(err)
	}
	var rw string
	for _, line := range strings.Split(string(b), "\n") {
		if v, ok := strings.CutPrefix(line, "ReadWritePaths="); ok {
			rw = v
			break
		}
	}
	if rw == "" {
		t.Fatal("carlos-relay.service declares no ReadWritePaths")
	}
	for _, p := range strings.Fields(rw) {
		if !strings.HasPrefix(p, "/data/") {
			continue // /run/carlos is tmpfs and already has its own line
		}
		want := "d " + p + " "
		if !strings.Contains(tmpfilesConf, want) {
			t.Errorf("tmpfilesConf has no %q line, so an apply onto a box that predates the unit leaves it unable to start:\n%s", want, tmpfilesConf)
		}
	}
	// The mode must match bootstrap's dirs list, or a re-run of one flips
	// what the other just set (the /data/instances lesson, one line up).
	// Read the mode bootstrap actually uses rather than restating it: a
	// hardcoded "0750" here passes happily while the two lists disagree.
	root := t.TempDir()
	if err := bootstrapWith(root, nil, false); err != nil {
		t.Fatal(err)
	}
	fi, err := os.Stat(filepath.Join(root, "data", "relay"))
	if err != nil {
		t.Fatalf("bootstrap did not create /data/relay: %v", err)
	}
	want := fmt.Sprintf("d /data/relay %04o ", fi.Mode().Perm())
	if !strings.Contains(tmpfilesConf, want) {
		t.Errorf("tmpfilesConf must carry %q to match the mode bootstrap's dirs list creates:\n%s", want, tmpfilesConf)
	}
}

// TestPlanCarriesArmedList pins the other half of discussion #118: the plan a
// build prints says which units to START, not just which to enable, so an
// applier running an OLDER binary arms what the new build wants rather than
// what its own systemctlPlan happens to know about.
func TestPlanCarriesArmedList(t *testing.T) {
	plan := currentBootstrapPlan()
	if len(plan.Armed) == 0 {
		t.Fatal("plan carries no armed list")
	}
	if !slices.Contains(plan.Armed, relayUnitName) {
		t.Errorf("armed list omits the relay: %v", plan.Armed)
	}
	if slices.Contains(plan.Armed, "carlos-edge.service") {
		t.Errorf("edge must stay start-by-hand: %v", plan.Armed)
	}
	for _, u := range plan.Armed {
		if !slices.Contains(plan.Enabled, u) {
			t.Errorf("armed unit %s is not in the enabled list", u)
		}
	}
	// The armed list and the local rule must agree for THIS build; they
	// diverge only across builds, which is the point of carrying it.
	if got, want := plan.Armed, armedUnits(enabledUnits); !slices.Equal(got, want) {
		t.Errorf("plan.Armed = %v, systemctlPlan would start %v", got, want)
	}
}

// floorGateCall is how both system units ask whether carlos.prev may run
// against an armed floor: carlos.prev's own floor gate (root's trusted
// path, never a PATH lookup), with both files named in the unit.
const floorGateCall = "/opt/carlos/carlos.prev system floor-allows-prev /data/.moves/floor.min-release /opt/carlos/carlos.prev.allowed"

// seedsFloorGateCall is the same gate asked about the seeds floor's two
// files (seeded-instances spec §3.3): the units' second clause.
const seedsFloorGateCall = "/opt/carlos/carlos.prev system floor-allows-prev /data/.moves/seeds-floor.min-release /opt/carlos/carlos.prev.seeds-allowed"

// floorGateHelperEnv makes this test binary act as the floor gate for the
// unit stubs in TestSystemUnitsSelectPrevByTheFloor.
const floorGateHelperEnv = "CARLOS_TEST_FLOOR_GATE_HELPER"

// TestFloorGateHelperProcess is not a test: run with floorGateHelperEnv
// set, it is `carlos system floor-allows-prev` with the test's own uid as
// the expected owner of both files (a test cannot make root's or the
// service account's files).
func TestFloorGateHelperProcess(t *testing.T) {
	if os.Getenv(floorGateHelperEnv) != "1" {
		return
	}
	args := os.Args
	for i, a := range args {
		if a == "--" {
			args = args[i+1:]
			break
		}
	}
	uid := []uint32{uint32(os.Getuid())}
	os.Exit(floorGateMain(args, uid, uid, os.Stderr))
}

// TestSystemUnitsSelectPrevByTheFloor runs each system unit's ExecStart
// shell (moves spec §11, round-6 finding 4) against a fake /opt/carlos and
// /data, with stub binaries that record which one ran. Without a floor file
// the selection is main's exactly; with one, carlos.prev runs only when its
// verdict names the floor on disk, byte for byte, as carlos.prev's floor
// gate reads them (the stubs hand the gate verb to the real Go code).
func TestSystemUnitsSelectPrevByTheFloor(t *testing.T) {
	t.Parallel()
	if runtime.GOOS != "linux" {
		t.Skip("the floor gate refuses off Linux")
	}
	self, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	for _, unit := range []string{"carlos-system-recover.service", "carlos-system-update.service"} {
		t.Run(unit, func(t *testing.T) {
			b, err := unitsFS.ReadFile("units/" + unit)
			if err != nil {
				t.Fatal(err)
			}
			var line string
			for _, l := range strings.Split(string(b), "\n") {
				if strings.HasPrefix(l, "ExecStart=") {
					line = l
				}
			}
			// systemd expands %-specifiers and $VARIABLES in ExecStart before
			// the shell ever sees the line (systemd.service(5)).
			if strings.ContainsAny(line, "%$") {
				t.Fatalf("ExecStart carries a %% or a $, which systemd expands: %s", line)
			}
			for _, path := range []string{"/data/" + moves.FloorMinReleaseFile, "/opt/carlos/" + moves.PrevAllowedFile} {
				if !strings.Contains(line, path) {
					t.Fatalf("ExecStart does not read %s: %s", path, line)
				}
			}
			body, ok := strings.CutPrefix(line, "ExecStart=/bin/sh -c '")
			body, ok2 := strings.CutSuffix(body, "'")
			if !ok || !ok2 || strings.Contains(body, "'") {
				t.Fatalf("ExecStart is not one quoted sh -c body: %s", line)
			}

			for _, tc := range []struct {
				name           string
				journal        bool
				floor, verdict string // "" means no file
				floorKind      string // "" a regular file; or symlink, dangling, directory, fifo
				noPrev         bool
				want           string
				gate           string // which binary ran the floor gate, "" for none
				prevNoVerb     bool   // carlos.prev predates the gate and rejects its verb
			}{
				{"no journal", false, "", "", "", false, "carlos", "", false},
				{"no floor", true, "", "", "", false, "carlos.prev", "", false},
				{"no floor, no prev", true, "", "", "", true, "carlos", "", false},
				{"floor and a verdict naming it", true, "v0.30.0\n", "v0.30.0\n", "", false, "carlos.prev", "carlos.prev", false},
				{"a verdict for an older floor", true, "v0.30.0\n", "v0.29.0\n", "", false, "carlos", "carlos.prev", false},
				{"floor and no verdict", true, "v0.30.0\n", "", "", false, "carlos", "carlos.prev", false},
				{"no journal, armed and allowed", false, "v0.30.0\n", "v0.30.0\n", "", false, "carlos", "", false},
				{"armed, no prev", true, "v0.30.0\n", "v0.30.0\n", "", true, "carlos", "", false},
				// The agent owns /data: anything but a regular file at the
				// floor path is refused before the gate runs (#267), and the
				// gate refuses one swapped in after that test (it never
				// follows a symlink or opens a FIFO for a blocking read).
				// [ -s ] is false for a dangling symlink and a FIFO, so
				// without the shell's test both would read as "no floor".
				{"symlink floor with matching verdict", true, "v0.30.0\n", "v0.30.0\n", "symlink", false, "carlos", "", false},
				{"dangling symlink floor", true, "", "", "dangling", false, "carlos", "", false},
				{"directory floor", true, "", "", "directory", false, "carlos", "", false},
				{"fifo floor", true, "", "v0.30.0\n", "fifo", false, "carlos", "", false},
				// A carlos.prev older than the gate answers its verb with
				// an unknown-verb exit: a mismatch, so carlos runs.
				{"a carlos.prev without the gate verb", true, "v0.30.0\n", "v0.30.0\n", "", false, "carlos", "carlos.prev", true},
			} {
				t.Run(tc.name, func(t *testing.T) {
					root := t.TempDir()
					opt, data := filepath.Join(root, "opt"), filepath.Join(root, "data")
					ran, gate := filepath.Join(root, "ran"), filepath.Join(root, "gate")
					os.MkdirAll(filepath.Join(opt, "system", "router"), 0o755)
					os.MkdirAll(filepath.Join(data, ".moves"), 0o755)
					// Each stub answers the gate verb by handing it to this
					// test binary (TestFloorGateHelperProcess), noting which
					// stub was asked; anything else records the stub as run.
					stub := func(name string) {
						answer := "  shift 2\n" +
							"  exec " + self + " -test.run='^TestFloorGateHelperProcess$' -- \"$@\"\n"
						if name == "carlos.prev" && tc.prevNoVerb {
							answer = "  echo 'system: unknown verb \"floor-allows-prev\"' >&2\n  exit 2\n"
						}
						sh := "#!/bin/sh\n" +
							"if [ \"$1\" = system ] && [ \"$2\" = floor-allows-prev ]; then\n" +
							"  echo " + name + " >> " + gate + "\n" +
							answer +
							"fi\n" +
							"echo " + name + " > " + ran + "\n"
						os.WriteFile(filepath.Join(opt, name), []byte(sh), 0o755)
					}
					stub("carlos")
					if !tc.noPrev {
						stub("carlos.prev")
					}
					if tc.journal {
						os.WriteFile(filepath.Join(opt, "system", "router", "journal.json"), []byte("{}"), 0o600)
					}
					floorPath := filepath.Join(data, moves.FloorMinReleaseFile)
					switch tc.floorKind {
					case "symlink", "dangling":
						target := filepath.Join(root, "floor-target")
						if tc.floorKind == "symlink" {
							if err := os.WriteFile(target, []byte(tc.floor), 0o644); err != nil {
								t.Fatal(err)
							}
						}
						if err := os.Symlink(target, floorPath); err != nil {
							t.Fatal(err)
						}
					case "directory":
						if err := os.Mkdir(floorPath, 0o755); err != nil {
							t.Fatal(err)
						}
					case "fifo":
						if out, err := exec.Command("mkfifo", floorPath).CombinedOutput(); err != nil {
							t.Fatalf("mkfifo: %v: %s", err, out)
						}
					default:
						if tc.floor != "" {
							if err := os.WriteFile(floorPath, []byte(tc.floor), 0o644); err != nil {
								t.Fatal(err)
							}
						}
					}
					if tc.verdict != "" {
						os.WriteFile(filepath.Join(opt, moves.PrevAllowedFile), []byte(tc.verdict), 0o644)
					}
					script := strings.ReplaceAll(body, "/data/", data+"/")
					script = strings.ReplaceAll(script, "/opt/carlos/", opt+"/")
					cmd := exec.Command("/bin/sh", "-c", script)
					cmd.Env = append(os.Environ(), floorGateHelperEnv+"=1")
					if out, err := cmd.CombinedOutput(); err != nil {
						t.Fatalf("%v: %s", err, out)
					}
					got, _ := os.ReadFile(ran)
					if strings.TrimSpace(string(got)) != tc.want {
						t.Fatalf("ran %q, want %s", strings.TrimSpace(string(got)), tc.want)
					}
					asked, _ := os.ReadFile(gate)
					if strings.TrimSpace(string(asked)) != tc.gate {
						t.Fatalf("the floor gate ran from %q, want %q", strings.TrimSpace(string(asked)), tc.gate)
					}
				})
			}
		})
	}
}