package instancectl

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/netip"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/cliflag"
	"github.com/carlosframework/platform/internal/controlsock"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/version"
)

// Server is the root instance-control service (design §5). It does four
// things and nothing else: starts, stops and sweeps the four templates'
// instances, sets a host slice's memory tier before a start, runs the
// readiness probe, and answers /ready. Every systemctl argument it passes
// is built from a validated host by UnitName/SliceName; nothing from a
// request reaches systemctl as a string.
type Server struct {
	ClaimStop http.Handler
	// ClaimHeldRelease is the moves held-release route's handler (moves
	// Task 7b), registered only when set: the host sets it with both
	// CARLOS_HOST_CLAIM and CARLOS_MOVES_PHASE1 on, so with moves off the
	// route does not exist and main's 404 stands (whole-branch fixes,
	// item 6).
	ClaimHeldRelease http.Handler
	// Systemctl runs one systemctl invocation and returns its combined
	// output; production is exec.Command, tests record.
	Systemctl func(ctx context.Context, args ...string) (string, error)
	// PeerOK is asked once per accepted connection, before a byte is read.
	// Production requires SO_PEERCRED uid carlos and pid == MainPID of
	// carlos-edge.service (design §5.4). A refused connection is closed
	// with no response.
	PeerOK func(conn net.Conn) error
	// CgroupRoot is /sys/fs/cgroup; tests point it at a fixture tree.
	CgroupRoot string
	// ProbeOut is where the probe unit's StandardOutput=truncate: lands.
	ProbeOut string
	// FixtureOK verifies /etc/carlos/probe-secret is root:carlos 0640
	// and /dev/shm/carlos-probe is a root-owned file before a probe
	// verdict is accepted (design §8.5, codex R11-1, #126 R15).
	FixtureOK func() error
	// ProbeEvery is the re-run cadence (five minutes in production).
	ProbeEvery time.Duration
	Logf       func(string, ...any)

	// ExeRoot is where ensureExe lays out each instance's view of its own
	// build (#214 item 2); empty skips it, for tests of other verbs.
	ExeRoot string
	// OptDir is the version store's root (/opt/carlos), and HostApp reads a
	// host's account and app from the registry. Together they are what a
	// nested generation's staged executable is validated against
	// (validateStagedExe). Either one unset refuses every staged start.
	OptDir  string
	HostApp func(host string) (acct, app string, err error)
	// DropInDir is where generation units' runtime drop-ins go
	// (/run/systemd/system). Empty refuses every generation start: the
	// template alone runs nothing (zero-downtime plan 2 D1).
	DropInDir string
	// ResolvConf is the box's resolv.conf and ResolvedConf resolved's
	// upstream list; whichever an instance sees (resolverFile) has its
	// nameservers re-opened through the egress fence before every probe
	// (#214 item 3).
	ResolvConf   string
	ResolvedConf string
	// LocalAddrs lists what routes to this box itself; a nameserver
	// inside it is never opened (#126). Production reads procfs
	// (localPrefixes); nil closes every resolver.
	LocalAddrs func() ([]netip.Prefix, error)
	// DataRoot is the instance data tree, and Statfs reports its free and
	// total bytes; a nil Statfs skips the disk floors (#214 item 1).
	DataRoot  string
	Statfs    func(path string) (avail, total uint64, err error)
	DiskEvery time.Duration

	// cmdPollEvery, cmdWaitBound and cmdStartBound pace the cmd routes:
	// how often GET /cmd/{host} re-reads the unit (250 ms), how long it
	// holds a request (30 s, run spec §5.4), and how long POST /cmd waits
	// for the queued start to show a new invocation (10 s). Zero is the
	// default; tests shorten them.
	cmdPollEvery, cmdWaitBound, cmdStartBound time.Duration
	// cmdLocks holds one *sync.Mutex per host, taken by handleCmd from its
	// ActiveState read to its answer, so two starts for one host cannot both
	// pass the check.
	cmdLocks sync.Map

	// Run executes one command as root and returns its combined output;
	// production is exec.CommandContext, tests record. Only
	// inspectMountpoints uses it, and every argument it passes is a
	// constant: nothing from a request reaches it.
	Run func(ctx context.Context, args ...string) (string, error)

	// Build is the build this process runs (version.Build), answered on
	// every /ready as the BuildHeader so the agent can see it talks to
	// its own release (#256). Empty sends no header, which a client of
	// this release reads as an older service.
	Build string
	// ExeCurrent reports whether this process still runs the installed
	// platform binary; nil skips the check. Production compares
	// /proc/self/exe with /opt/carlos/carlos by identity
	// (installedBinaryCurrent). Asked on every /ready, not once at probe
	// time, because the binary can change at any moment after the probe
	// ran: that is exactly the stale-service case of #256.
	ExeCurrent func() error

	// cgroupWaitBudget bounds a wait for an empty cgroup; tests shorten it.
	cgroupWaitBudget time.Duration

	// genMu guards genStarting, the generation units a start is preparing
	// now. The sweep never removes the files of one of them.
	genMu       sync.Mutex
	genStarting map[string]bool

	// Legacy run start attempts per host (legacystart.go, #292): the
	// starts in flight, the stops in progress, and each host's barrier,
	// the accept order of the latest stop's request. connSeq numbers
	// connections in accept order.
	legacyMu       sync.Mutex
	legacyStarts   map[string]map[*legacyAttempt]struct{}
	legacyStopping map[string]int
	legacyBarrier  map[string]uint64
	connSeq        atomic.Uint64
	// legacyJoin overrides legacyJoinBudget; tests shorten it.
	legacyJoin time.Duration

	mu     sync.Mutex
	report ProbeReport
	probed bool
	// startLines is inspectMountpoints' verdict, taken once at start and
	// folded into every report RunProbe stores. A failing line keeps the
	// box not ready whatever the probe says, until an operator clears the
	// files and restarts the service.
	startLines []VerdictLine
}

const (
	probeUnit    = "carlos-instance-probe.service"
	edgeUnit     = "carlos-edge.service"
	parentSlices = "carlos.slice/carlos-instances.slice"
)

// Run is `carlos instance-control`.
func Run(args []string) { runWithClaim(args, nil, nil, nil) }

func RunWithRegistry(args []string, hostApp func(string) (string, string, error)) {
	runWithClaim(args, nil, nil, hostApp)
}

func runWithClaim(args []string, claim, heldRelease http.Handler, hostApp func(string) (string, string, error)) {
	fs := flag.NewFlagSet("instance-control", flag.ExitOnError)
	fs.Usage = cliflag.Usage(fs, "usage: carlos instance-control (socket-activated by carlos-instance-control.socket)")
	_ = fs.Parse(args)
	if os.Geteuid() != 0 {
		log.Fatal("instance-control: this is the root service behind carlos-instance-control.socket; it is not for operators")
	}
	s := &Server{
		ClaimStop:        claim,
		ClaimHeldRelease: heldRelease,
		Systemctl: func(ctx context.Context, args ...string) (string, error) {
			out, err := exec.CommandContext(ctx, "systemctl", args...).CombinedOutput()
			return string(out), err
		},
		Run: func(ctx context.Context, args ...string) (string, error) {
			out, err := exec.CommandContext(ctx, args[0], args[1:]...).CombinedOutput()
			return string(out), err
		},
		PeerOK:       productionPeerOK(),
		CgroupRoot:   "/sys/fs/cgroup",
		ProbeOut:     filepath.Join(controlsock.Dir, "probe.out"),
		FixtureOK:    probeFixturesOK,
		ProbeEvery:   5 * time.Minute,
		Logf:         log.Printf,
		ExeRoot:      filepath.Join(controlsock.Dir, "exe"),
		OptDir:       "/opt/carlos",
		HostApp:      hostApp,
		DropInDir:    "/run/systemd/system",
		ResolvConf:   "/etc/resolv.conf",
		ResolvedConf: ResolvedConf,
		LocalAddrs:   func() ([]netip.Prefix, error) { return localPrefixes(fibTriePath, ifInet6Path) },
		DataRoot:     "/data/instances",
		Statfs:       statfsBytes,
		DiskEvery:    30 * time.Second,
		Build:        version.Build,
		ExeCurrent:   installedBinaryCurrent,
	}
	ln, activated, err := activatedListener()
	if err != nil {
		log.Fatalf("instance-control: %v", err)
	}
	if !activated {
		// Not socket-activated (a hand run): bind the path ourselves with
		// the same ownership the socket unit gives it.
		ln, err = bindControlSocket(controlsock.Path)
		if err != nil {
			log.Fatalf("instance-control: %v", err)
		}
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	s.InspectMountpoints(ctx)
	go s.probeLoop(ctx)
	go s.diskLoop(ctx)
	if err := s.Serve(ctx, ln); err != nil {
		log.Fatalf("instance-control: %v", err)
	}
}

// Serve accepts connections, asks PeerOK about each, and hands the accepted
// ones to the HTTP server one at a time through a listener that only ever
// yields vetted connections.
func (s *Server) Serve(ctx context.Context, ln net.Listener) error {
	vetted := &vettedListener{Listener: ln, ok: s.peerOK, logf: s.logf}
	srv := &http.Server{Handler: s.Handler(), ReadHeaderTimeout: 5 * time.Second, ConnContext: s.connContext}
	// One request per connection (#292): a connection's accept order is the
	// request's (connContext), which the legacy start fence relies on, so a
	// client may not reuse one. The agent's client never does.
	srv.SetKeepAlivesEnabled(false)
	go func() {
		<-ctx.Done()
		srv.Close()
	}()
	err := srv.Serve(vetted)
	if errors.Is(err, http.ErrServerClosed) || ctx.Err() != nil {
		return nil
	}
	return err
}

// vettedListener runs the peer check inside Accept so a refused peer is
// closed before net/http ever reads from it.
type vettedListener struct {
	net.Listener
	ok   func(net.Conn) error
	logf func(string, ...any)

	// lastLog and suppressed rate-limit the refusal log to at most one
	// line per second: a refusing peer that reconnects in a tight loop
	// (the common shape while carlos-edge.service has no MainPID yet)
	// would otherwise flood the log at the same rate. Accept is only ever
	// called from the one goroutine net/http.Server.Serve runs its accept
	// loop on, so these need no lock of their own.
	lastLog    time.Time
	suppressed int
}

func (v *vettedListener) Accept() (net.Conn, error) {
	for {
		c, err := v.Listener.Accept()
		if err != nil {
			return nil, err
		}
		if err := v.ok(c); err != nil {
			v.logRefusal(err)
			c.Close()
			continue
		}
		return c, nil
	}
}

func (v *vettedListener) logRefusal(err error) {
	now := time.Now()
	if !v.lastLog.IsZero() && now.Sub(v.lastLog) < time.Second {
		v.suppressed++
		return
	}
	if v.suppressed > 0 {
		v.logf("instance-control: refused connection: %v (%d more refusals suppressed)", err, v.suppressed)
	} else {
		v.logf("instance-control: refused connection: %v", err)
	}
	v.lastLog = now
	v.suppressed = 0
}

func (s *Server) logf(format string, args ...any) {
	if s.Logf != nil {
		s.Logf(format, args...)
	}
}

// peerOK calls through s.PeerOK per connection (never captured into a
// closure at Serve start) so a caller that sets PeerOK after Serve has
// begun — a test, or a production reconfiguration — takes effect on the
// very next Accept instead of being silently ignored (codex review,
// fix round 2 item 1).
func (s *Server) peerOK(c net.Conn) error { return s.PeerOK(c) }

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	if s.ClaimStop != nil {
		mux.Handle("POST /claim/stop", s.ClaimStop)
		mux.Handle("POST /claim/retire", s.ClaimStop)
		mux.Handle("POST /claim/handover", s.ClaimStop)
		mux.Handle("POST /claim/recovery-rebind", s.ClaimStop)
		mux.Handle("POST /claim/journal", s.ClaimStop)
		mux.Handle("POST /claim/execution", s.ClaimStop)
		mux.Handle("POST /claim/writers-stopped", s.ClaimStop)
		mux.Handle("POST /claim/launch", s.ClaimStop)
	}
	if s.ClaimHeldRelease != nil {
		// moves Task 7b: the held release under the claim, only with both
		// flags on (item 6); off, the route is absent as on main.
		mux.Handle("POST /claim/held-release", s.ClaimHeldRelease)
	}
	mux.HandleFunc("POST /start", s.claimAwareLegacy(s.handleStart))
	mux.HandleFunc("POST /seed", s.claimAwareLegacy(s.handleSeed))
	mux.HandleFunc("POST /stop", s.claimAwareLegacy(s.handleStop))
	mux.HandleFunc("POST /reset", s.claimAwareLegacy(s.handleReset))
	mux.HandleFunc("POST /sweep", s.claimAwareLegacy(s.handleSweep))
	mux.HandleFunc("POST /parent", s.claimAwareLegacy(s.handleParent))
	mux.HandleFunc("POST /cmd", s.claimAwareLegacy(s.handleCmd))
	mux.HandleFunc("GET /cmd/{host}", s.claimAwareLegacy(s.handleCmdStatus))
	mux.HandleFunc("GET /ready", s.handleReady)
	return mux
}

// request is the validated form of a verb's query: nothing else reaches a
// unit name.
type request struct {
	host string
	role Role
	tier cgroup.Limit
	gen  uint64 // a generation of the run role (zero-downtime spec §7.6); 0 is none
	kill bool
	exe  StagedExe // a staged build to mount (zero-downtime plan 3); Path "" is none
}

// verbShape is what a verb's query may carry.
type verbShape struct {
	roles []Role
	tier  bool
	gen   bool
	kill  bool
	exe   bool // a start, seed or cmd verb: the staged build its unit binds
	// seed is the /seed verb, which names no role: its unit is the seed
	// unit, which may bind a staged build (settings spec §4.2).
	seed bool
	// cmd is /cmd, whose staged build is the serving generation's, as the
	// legacy sidecar's is (run spec §5.4 under zero-downtime spec §7.5).
	cmd bool
}

func parseRequest(r *http.Request, v verbShape) (request, error) {
	q := r.URL.Query()
	host := q.Get("host")
	if !provision.ValidHostShape(host) {
		return request{}, fmt.Errorf("host %q is not a host", host)
	}
	req := request{host: host}
	if len(v.roles) > 0 {
		role, ok := ParseRole(q.Get("role"))
		if !ok {
			return request{}, fmt.Errorf("role %q is not run, seed, sidecar or cmd", q.Get("role"))
		}
		allowed := false
		for _, r := range v.roles {
			allowed = allowed || r == role
		}
		if !allowed {
			return request{}, fmt.Errorf("role %s is not valid for this verb", role)
		}
		req.role = role
	}
	if v.tier {
		n, err := strconv.ParseInt(q.Get("memory_max"), 10, 64)
		if err != nil {
			return request{}, fmt.Errorf("memory_max %q is not a number", q.Get("memory_max"))
		}
		switch n {
		case cgroup.TierDefault.MemoryMax, cgroup.TierChildren.MemoryMax:
			req.tier = cgroup.Limit{MemoryMax: n}
		default:
			return request{}, fmt.Errorf("memory_max %d is not a platform tier", n)
		}
	}
	if q.Has("gen") {
		raw := q.Get("gen")
		if !v.gen {
			return request{}, errors.New("this verb takes no generation")
		}
		n, err := strconv.ParseUint(raw, 10, 64)
		if err != nil || n == 0 || strconv.FormatUint(n, 10) != raw {
			return request{}, fmt.Errorf("gen %q is not a generation", raw)
		}
		if req.role != RoleRun {
			return request{}, fmt.Errorf("role %s has no generations", req.role)
		}
		req.gen = n
	}
	if q.Has("kill") {
		if !v.kill || req.gen == 0 || q.Get("kill") != "1" {
			return request{}, errors.New("kill=1 belongs to a generation stop only")
		}
		req.kill = true
	}
	if q.Has("exe") || q.Has("version") || q.Has("digest") {
		if !v.exe {
			return request{}, errors.New("this verb takes no staged executable")
		}
		if x := q.Get("exe"); x != "" {
			// A generation start; the legacy sidecar (zero-downtime spec
			// §7.5: it runs its instance's serving artifact); a run's
			// command, which runs that same artifact; or the original
			// template's run unit and the seed unit, which run the build a
			// cold wake recorded (settings spec §4.2). Every one is
			// validated before anything is laid out (refuseStagedExe).
			if req.gen == 0 && !v.seed && !v.cmd && req.role != RoleRun && req.role != RoleSidecar {
				return request{}, errors.New("exe is only for a run, seed or sidecar unit, a run's command, or a generation start")
			}
			req.exe = StagedExe{Path: x, Version: q.Get("version"), Digest: q.Get("digest")}
		} else {
			return request{}, errors.New("version and digest need an exe")
		}
	}
	return req, nil
}

// setTier writes the whole tier onto the host slice in one call: the
// memory caps, and the CPU and IO shares that follow from the same tier
// (#214 item 1). Seed, instance and sidecar share all of it.
func (s *Server) setTier(ctx context.Context, host string, tier cgroup.Limit) error {
	return sliceTier(ctx, s.Systemctl, host, tier)
}

// sliceTier is setTier's one systemctl call, shared with the claim launch.
func sliceTier(ctx context.Context, systemctl func(context.Context, ...string) (string, error), host string, tier cgroup.Limit) error {
	max := strconv.FormatInt(tier.MemoryMax, 10)
	high := strconv.FormatInt(cgroup.HighOf(tier.MemoryMax), 10)
	sh := cgroup.SharesOf(tier)
	_, err := systemctl(ctx, "set-property", "--runtime", SliceName(host),
		"MemoryMax="+max, "MemoryHigh="+high, "MemorySwapMax="+max,
		"CPUWeight="+strconv.Itoa(sh.CPUWeight), "CPUQuota="+strconv.Itoa(sh.CPUQuotaPercent)+"%",
		"IOWeight="+strconv.Itoa(sh.IOWeight))
	return err
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}

func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
	req, err := parseRequest(r, verbShape{roles: []Role{RoleRun, RoleSidecar}, tier: true, gen: true, exe: true})
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if req.gen != 0 {
		s.startGen(w, r, req)
		return
	}
	ctx := r.Context()
	if req.role == RoleRun {
		// An attempt a /stop of this host can cancel and join, and that a
		// stop which has already run refuses (legacystart.go, #292).
		attempt, finish, err := s.beginLegacyStart(ctx, req.host, requestSeq(ctx))
		if err != nil {
			http.Error(w, "start: "+err.Error(), http.StatusConflict)
			return
		}
		defer finish()
		ctx = attempt
	}
	unit := UnitName(req.role, req.host)
	// A sidecar naming its generation's staged build, or a run unit naming
	// the build its cold wake recorded, is validated before anything is
	// laid out, under the generation rule (validateStagedExe).
	if s.refuseStagedExe(w, req, unit) {
		return
	}
	if req.role == RoleRun {
		if err := s.clearLegacyClaimLaunch(ctx, req.host); err != nil {
			http.Error(w, "start: "+err.Error(), http.StatusBadGateway)
			return
		}
	}
	if s.refuseStart(w, req.host, unit, func() error { return s.ensureHostExeAt(req.host, req.exe) }) {
		return
	}
	// A sidecar start names its instance's staged build with generations
	// on or off (settings ruling R-EXEC-SIDECAR). With them on, the host
	// slice is the generation parent (spec §7.3): a swap may have raised it
	// to two tiers, and lowering it here would put both generations under
	// one tier's limit. Its restore is the swap owner's, so a populated
	// slice is left alone, exactly as startGen does; a cold slice still gets
	// the tier. With them off, a populated slice was tiered by the
	// instance's own run start of that same recorded build, so leaving it
	// alone keeps the right tier too. A start without exe (a live-link
	// sidecar) sets the tier as before. The exemption is the sidecar's
	// alone: a run unit naming its recorded build starts a cold instance,
	// and sets its tier as a run start always has.
	if req.role == RoleSidecar && req.exe.Path != "" && s.slicePopulated(req.host) {
		s.logf("instance-control: %s: slice populated, leaving its limit to the generation owner", req.host)
	} else if err := s.setTier(ctx, req.host, req.tier); err != nil {
		http.Error(w, "set-property: "+err.Error(), http.StatusBadGateway)
		return
	}
	// Never submitted once cancelled (#292): a /stop of this host cancels
	// the attempt before it joins it, and the client may have gone.
	if err := ctx.Err(); err != nil {
		http.Error(w, "start: cancelled before it was submitted: "+err.Error(), http.StatusConflict)
		return
	}
	if out, err := s.Systemctl(ctx, "start", unit); err != nil {
		http.Error(w, "start: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
		return
	}
	inv, err := s.Systemctl(ctx, "show", "-p", "InvocationID", "--value", unit)
	if err != nil {
		http.Error(w, "show: "+err.Error(), http.StatusBadGateway)
		return
	}
	s.logf("instance-control: %s: started %s memory.max=%d", req.host, unit, req.tier.MemoryMax)
	writeJSON(w, http.StatusOK, StartResponse{InvocationID: strings.TrimSpace(inv)})
}

// parseShow reads systemctl show's KEY=value lines.
func parseShow(out string) map[string]string {
	m := map[string]string{}
	sc := bufio.NewScanner(strings.NewReader(out))
	for sc.Scan() {
		if k, v, ok := strings.Cut(sc.Text(), "="); ok {
			m[k] = strings.TrimSpace(v)
		}
	}
	return m
}

// handleSeed runs the seed oneshot once and returns its retained result.
// The unit's InvocationID is read before and after the first `start`: if
// start left the SAME invocation in place, an earlier request's stop
// never ran (its client disconnected before this fix existed, or the
// agent restarted mid-seed), so this handler stops the stale run and
// starts again rather than reading and returning someone else's result.
//
// The request context ctx is used only up to and including that first
// `start`. From the moment it returns, the unit is this server's
// responsibility to clean up, whether or not the client that asked for
// it is still connected: `start` is the blocking call, and production's
// Systemctl is exec.CommandContext, which kills the systemctl process
// outright when its context is cancelled. A client that hung up while
// `start` was running would otherwise cancel every systemctl call after
// it too: the very next `show` would fail, the handler would 502, and
// the cleanup `stop` after it would never run (fix round 1, item 4). So
// every call from here to the end of the handler runs on ONE context
// (dctx) detached from the request instead. The order is show (the
// fields RemainAfterExit retains), then stop, then wait for the cgroup
// to empty: a populated cgroup after the wait is still reported as a 502
// naming the unit.
//
// A seed may name its executable (settings spec §4.2): the staged build
// the wake recorded. Root validates it as it does a generation's (403 on
// a refusal, before anything is laid out or started) and binds the host's
// app mount to it. Without one the mount is the live link, as before.
func (s *Server) handleSeed(w http.ResponseWriter, r *http.Request) {
	req, err := parseRequest(r, verbShape{tier: true, exe: true, seed: true})
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	ctx := r.Context()
	unit := UnitName(RoleSeed, req.host)
	if s.refuseStagedExe(w, req, unit) {
		return
	}
	if s.refuseStart(w, req.host, unit, func() error { return s.ensureHostExeAt(req.host, req.exe) }) {
		return
	}
	if err := s.setTier(ctx, req.host, req.tier); err != nil {
		http.Error(w, "set-property: "+err.Error(), http.StatusBadGateway)
		return
	}

	// A oneshot's start job completes when ExecStart exits; RemainAfterExit
	// keeps the unit loaded with its result until the stop below.
	invBefore := s.showValue(ctx, "InvocationID", unit)
	startOut, startErr := s.Systemctl(ctx, "start", unit)

	// See the handler doc comment: everything past this point belongs to
	// the server, not the client, so it runs on a context of the server's
	// own rather than one the client's disconnect can cancel.
	dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
	defer cancel()

	invAfter := s.showValue(dctx, "InvocationID", unit)
	if invBefore != "" && invBefore == invAfter {
		s.stopUnit(dctx, unit)
		startOut, startErr = s.Systemctl(dctx, "start", unit)
		if retry := s.showValue(dctx, "InvocationID", unit); retry == "" || retry == invAfter {
			// Still this server's unit to clean up even though the request
			// is about to fail: leaving it running would strand the NEXT
			// /seed on this same stale-invocation guard forever.
			s.stopUnit(dctx, unit)
			http.Error(w, "seed: "+unit+" would not leave its stale invocation "+invAfter, http.StatusBadGateway)
			return
		}
	}

	out, err := s.Systemctl(dctx, "show", "-p", "InvocationID,Result,ExecMainStatus,ExecMainCode", unit)
	if err != nil {
		http.Error(w, "show: "+err.Error(), http.StatusBadGateway)
		return
	}
	m := parseShow(out)
	s.stopUnit(dctx, unit)
	wctx, waitCancel := context.WithTimeout(dctx, s.waitBudget())
	defer waitCancel()
	if err := WaitCgroupEmpty(wctx, s.CgroupRoot, SliceName(req.host), unit, 50*time.Millisecond); err != nil {
		http.Error(w, "seed: "+err.Error(), http.StatusBadGateway)
		return
	}
	if startErr != nil && m["Result"] == "" {
		http.Error(w, "start: "+strings.TrimSpace(startOut+" "+startErr.Error()), http.StatusBadGateway)
		return
	}
	writeJSON(w, http.StatusOK, SeedResponse{InvocationID: m["InvocationID"], Result: m["Result"], Status: m["ExecMainStatus"], Code: m["ExecMainCode"]})
}

// showValue reads one systemd property's --value for unit, returning "" on
// any error. It is a probe, not a hard failure path: the caller's own
// later show call is what surfaces a real systemctl error to the client.
func (s *Server) showValue(ctx context.Context, prop, unit string) string {
	out, err := s.Systemctl(ctx, "show", "-p", prop, "--value", unit)
	if err != nil {
		return ""
	}
	return strings.TrimSpace(out)
}

// stopUnit issues `systemctl stop` for unit on ctx, logging rather than
// surfacing a failure to the caller: every call site in handleSeed is
// already past the point where the unit is this server's responsibility,
// not something a failed stop should turn into a different HTTP error for
// the client.
func (s *Server) stopUnit(ctx context.Context, unit string) {
	if _, err := s.Systemctl(ctx, "stop", unit); err != nil {
		s.logf("instance-control: stop %s: %v", unit, err)
	}
}

func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) {
	req, err := parseRequest(r, verbShape{roles: []Role{RoleRun, RoleSeed, RoleSidecar, RoleCmd}, gen: true, kill: true})
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if req.gen != 0 {
		s.stopGen(w, r, req)
		return
	}
	if req.role == RoleRun {
		// Every start of this host's run unit in flight is cancelled and
		// joined first, and earlier requests' starts are refused from here
		// on (legacystart.go, #292): once this answers, no start from an
		// earlier request can still be submitted, and the stop below
		// replaces any job a joined start had submitted.
		release, err := s.fenceLegacyStarts(r.Context(), req.host)
		defer release()
		if err != nil {
			// Stopped anyway, but not proven final: a start may follow.
			_, _ = s.Systemctl(r.Context(), "stop", UnitName(req.role, req.host))
			http.Error(w, "stop: "+err.Error(), http.StatusBadGateway)
			return
		}
	}
	if out, err := s.Systemctl(r.Context(), "stop", UnitName(req.role, req.host)); err != nil {
		http.Error(w, "stop: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
		return
	}
	if req.role == RoleRun {
		if err := s.clearLegacyClaimLaunch(r.Context(), req.host); err != nil {
			http.Error(w, "stop: "+err.Error(), http.StatusBadGateway)
			return
		}
	}
	if req.role == RoleCmd {
		s.cleanCmdExe(r.Context(), req.host)
	}
	w.WriteHeader(http.StatusNoContent)
}

func (s *Server) handleReset(w http.ResponseWriter, r *http.Request) {
	req, err := parseRequest(r, verbShape{roles: []Role{RoleRun, RoleSeed, RoleSidecar, RoleCmd}, gen: true})
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	unit := UnitName(req.role, req.host)
	if req.gen != 0 {
		unit = GenUnitName(req.host, req.gen)
	}
	if out, err := s.Systemctl(r.Context(), "reset-failed", unit); err != nil && !strings.Contains(out+err.Error(), "not loaded") {
		http.Error(w, "reset-failed: "+strings.TrimSpace(out+" "+err.Error()), http.StatusBadGateway)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// handleSweep stops every instance of the four templates that is not
// inactive or has a job, keyed on the names systemd reports, so a unit
// whose identifier is hashed and whose row is gone is still stopped
// (design §6.7, codex R7-1).
func (s *Server) handleSweep(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	targets := map[string]bool{}
	out, err := s.Systemctl(ctx, "list-units", "--all", "--plain", "--no-legend", "--no-pager",
		UnitPrefixRun+"*.service", UnitPrefixSeed+"*.service", UnitPrefixSidecar+"*.service", UnitPrefixCmd+"*.service", UnitPrefixGen+"*.service")
	if err != nil {
		http.Error(w, "list-units: "+err.Error(), http.StatusBadGateway)
		return
	}
	for _, line := range strings.Split(out, "\n") {
		f := strings.Fields(line)
		if len(f) < 3 {
			continue
		}
		if _, _, ok := ParseUnitName(f[0]); !ok {
			continue
		}
		if f[2] != "inactive" {
			targets[f[0]] = true
		}
	}
	jobs, err := s.Systemctl(ctx, "list-jobs", "--plain", "--no-legend", "--no-pager")
	if err != nil {
		http.Error(w, "list-jobs: "+err.Error(), http.StatusBadGateway)
		return
	}
	for _, line := range strings.Split(jobs, "\n") {
		f := strings.Fields(line)
		if len(f) < 2 {
			continue
		}
		if _, _, ok := ParseUnitName(f[1]); ok {
			targets[f[1]] = true
		}
	}
	// Every target gets stopped and waited on; a stuck unit is recorded and
	// the sweep continues to the rest rather than aborting on the first one
	// (design §6.7, codex review fix round 2 item 3): a sweep exists to
	// clear the box for the next thing, and giving up after unit one would
	// leave every unit after it running.
	var stopped []string
	failed := map[string]string{}
	for unit := range targets {
		wctx, cancel := context.WithTimeout(ctx, s.waitBudget())
		if _, err := s.Systemctl(ctx, "stop", unit); err != nil {
			failed[unit] = "stop: " + err.Error()
			cancel()
			continue
		}
		// A unit that is still activating has no cgroup yet; wait for
		// systemd to settle it to inactive or failed before trusting an
		// absent cgroup as "swept" (codex review fix round 2 item 4).
		if err := s.waitInactiveOrFailed(wctx, unit, 50*time.Millisecond); err != nil {
			failed[unit] = err.Error()
			cancel()
			continue
		}
		_, esc, _ := ParseUnitName(unit)
		err := WaitCgroupEmpty(wctx, s.CgroupRoot, "carlos-instances-"+esc+".slice", unit, 50*time.Millisecond)
		cancel()
		if err != nil {
			failed[unit] = err.Error()
			continue
		}
		stopped = append(stopped, unit)
		s.removeGenFiles(unit)
		if role, esc, _ := ParseUnitName(unit); role == RoleCmd {
			if iid, ok := Unescape(esc); ok {
				s.removeCmdExe(iid)
			}
		}
	}
	s.sweepGenLeftovers(ctx, failed)
	if len(failed) > 0 {
		writeJSON(w, http.StatusBadGateway, SweepResponse{Units: stopped, Failed: failed})
		return
	}
	writeJSON(w, http.StatusOK, SweepResponse{Units: stopped})
}

// waitInactiveOrFailed polls a unit's ActiveState until systemd reports it
// settled (inactive after a clean stop, failed after one that wasn't),
// bounded by ctx.
func (s *Server) waitInactiveOrFailed(ctx context.Context, unit string, every time.Duration) error {
	var last string
	for {
		if out, err := s.Systemctl(ctx, "show", "-p", "ActiveState", "--value", unit); err == nil {
			last = strings.TrimSpace(out)
			if last == "inactive" || last == "failed" {
				return nil
			}
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("%s: ActiveState still %q", unit, last)
		case <-time.After(every):
		}
	}
}

func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
	if s.Build != "" {
		w.Header().Set(BuildHeader, s.Build)
	}
	rep, ready := s.Ready()
	if ready {
		w.WriteHeader(http.StatusNoContent)
		return
	}
	writeJSON(w, http.StatusServiceUnavailable, rep)
}

// Ready returns the last probe verdict, plus the one check that cannot
// wait for the next probe: that this process still runs the installed
// binary (ExeCurrent). Before the first probe has ever run, there is no
// verdict to report, so a synthetic line names that reason explicitly
// rather than answering with an empty Lines (final review, plan 2, item
// 6): a caller reading /ready during startup should see why, not a report
// that looks like every check silently passed.
func (s *Server) Ready() (ProbeReport, bool) {
	s.mu.Lock()
	rep, probed := s.report, s.probed
	s.mu.Unlock()
	if !probed {
		rep = ProbeReport{Lines: []VerdictLine{{Check: "probe", Verdict: "fail", Detail: "not yet run"}}}
	}
	if s.ExeCurrent != nil {
		if err := s.ExeCurrent(); err != nil {
			// A copy: the stored report must not grow a line per call.
			rep = ProbeReport{Ready: false, Lines: append(append([]VerdictLine(nil), rep.Lines...),
				VerdictLine{Check: "binary", Verdict: "fail", Detail: err.Error()})}
		}
	}
	return rep, rep.Ready
}

// installedBinaryCurrent is ExeCurrent in production: /proc/self/exe
// resolves to the file this process was started from, even after that
// file was renamed over (the kernel reports it as deleted), and
// os.SameFile compares device and inode, so a service started from the
// previous release's binary answers a mismatch the moment carlos system
// apply renames the new one into place, and keeps answering it until it
// is restarted (#256).
func installedBinaryCurrent() error {
	running, err := os.Stat("/proc/self/exe")
	if err != nil {
		return fmt.Errorf("stat /proc/self/exe: %v", err)
	}
	installed, err := os.Stat(SystemBinary)
	if err != nil {
		return fmt.Errorf("stat %s: %v", SystemBinary, err)
	}
	if !os.SameFile(running, installed) {
		return fmt.Errorf("instance control is running a binary that is no longer %s; restart carlos-instance-control.service", SystemBinary)
	}
	return nil
}

func (s *Server) waitBudget() time.Duration {
	if s.cgroupWaitBudget > 0 {
		return s.cgroupWaitBudget
	}
	return 30 * time.Second
}

// WaitCgroupEmpty waits until the unit's cgroup under the host slice is
// absent or reports "populated 0" in cgroup.events, polling every `every`,
// bounded by ctx and the caller's budget (design §6.5, codex R2-5).
func WaitCgroupEmpty(ctx context.Context, root, slice, unit string, every time.Duration) error {
	events := filepath.Join(root, parentSlices, slice, unit, "cgroup.events")
	for {
		b, err := os.ReadFile(events)
		switch {
		case errors.Is(err, os.ErrNotExist):
			return nil
		case err != nil:
			return fmt.Errorf("%s: %w", unit, err)
		case strings.Contains(string(b), "populated 0"):
			return nil
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("processes remain in %s", unit)
		case <-time.After(every):
		}
	}
}

// probeLoop runs the probe at start and every ProbeEvery.
func (s *Server) probeLoop(ctx context.Context) {
	s.RunProbe(ctx)
	t := time.NewTicker(s.ProbeEvery)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			s.RunProbe(ctx)
		}
	}
}

// RunProbe starts the probe unit (a oneshot: start blocks until it exits),
// reads the report it left, and stores the verdict. A fixture that is not
// root:carlos 0640, a probe that could not be started, or a report that
// could not be read all leave the box not ready, with the reason on a
// verdict line; readiness is never the default.
func (s *Server) RunProbe(ctx context.Context) {
	set := func(r ProbeReport) {
		s.mu.Lock()
		s.report, s.probed = r, true
		s.mu.Unlock()
	}
	if err := s.FixtureOK(); err != nil {
		set(ProbeReport{Lines: []VerdictLine{{Check: "fixture", Verdict: "fail", Detail: err.Error()}}})
		return
	}
	// The probe binds the same /opt view an instance does, so its sources
	// must exist (#214 item 2), and the resolver allow it checks must be
	// in place (#214 item 3).
	if s.ExeRoot != "" {
		if err := ensureExe(s.ExeRoot, ProbeDirName, SystemBinary, ProbeDirName); err != nil {
			set(ProbeReport{Lines: []VerdictLine{{Check: "probe-exe", Verdict: "fail", Detail: err.Error()}}})
			return
		}
	}
	var resolverLines []VerdictLine
	if s.ResolvConf != "" {
		refused, err := s.allowResolvers(ctx)
		if err != nil {
			set(ProbeReport{Lines: []VerdictLine{{Check: "resolver-allow", Verdict: "fail", Detail: err.Error()}}})
			return
		}
		resolverLines = append(resolverLines, resolverLocalLine(refused))
	}
	if out, err := s.Systemctl(ctx, "start", probeUnit); err != nil {
		set(ProbeReport{Lines: []VerdictLine{{Check: "probe-unit", Verdict: "fail", Detail: strings.TrimSpace(out + " " + err.Error())}}})
		return
	}
	b, err := os.ReadFile(s.ProbeOut)
	if err != nil {
		set(ProbeReport{Lines: []VerdictLine{{Check: "probe-output", Verdict: "fail", Detail: err.Error()}}})
		return
	}
	var r ProbeReport
	if err := json.Unmarshal(b, &r); err != nil {
		set(ProbeReport{Lines: []VerdictLine{{Check: "probe-output", Verdict: "fail", Detail: err.Error()}}})
		return
	}
	for _, l := range resolverLines {
		r.Lines = append(r.Lines, l)
		if l.Verdict == "fail" {
			r.Ready = false
		}
	}
	for _, l := range r.Lines {
		if l.Verdict != "ok" {
			s.logf("instance-control: probe: %s %s %s", l.Check, l.Verdict, l.Detail)
		}
	}
	set(s.withStartLines(r))
}

// mountpoints are the two bind mounts whose underlying directories
// inspectMountpoints looks under (design §8.2, "Recovery for a layout that
// was unmounted anyway").
var mountpoints = []struct{ name, where string }{
	{"data-instances", "/data/instances"},
	{"run-carlos-instances", "/run/carlos/instances"},
}

// InspectMountpoints runs inspectMountpoints once and keeps its lines for
// every later report. Run calls it before the probe loop starts.
func (s *Server) InspectMountpoints(ctx context.Context) {
	lines := s.inspectMountpoints(ctx)
	for _, l := range lines {
		if l.Verdict != "ok" {
			s.logf("instance-control: %s %s %s", l.Check, l.Verdict, l.Detail)
		}
	}
	s.mu.Lock()
	s.startLines = lines
	s.mu.Unlock()
}

// inspectMountpoints looks UNDER the two bind mounts: in a private mount
// namespace it unmounts each bind and lists the raw mountpoint directory.
// Anything there was written while the layout was unmounted (an operator
// removing units by hand, a boot without the mounts), and the layout
// pass's destination check would otherwise never see it: the mount hides
// it, and the old binary would restore a second database over it on the
// next unmount. The verdict is taken once at start, as root, because
// unmounting in a private namespace needs CAP_SYS_ADMIN; it clears only
// when an operator moves or removes the files and restarts the service.
// An inspection that cannot run (no unshare, no permission) is
// unverified, and does not block readiness on its own: the probe's own
// checks stand, and the unverified line is in every report for the
// operator to see.
func (s *Server) inspectMountpoints(ctx context.Context) []VerdictLine {
	if s.Run == nil {
		return nil
	}
	var lines []VerdictLine
	for _, m := range mountpoints {
		// The listing runs after the umount so it reads the underlying
		// directory; a umount that fails because nothing is mounted there
		// is fine (the listing is then of the directory itself, which is
		// what an unmounted layout looks like), so its error is dropped.
		out, err := s.Run(ctx, "unshare", "-m", "--propagation", "private", "sh", "-c",
			"umount "+m.where+" 2>/dev/null; ls -A "+m.where)
		if err != nil {
			lines = append(lines, VerdictLine{Check: "mountpoint-" + m.name, Verdict: "unverified", Detail: strings.TrimSpace(out + " " + err.Error())})
			continue
		}
		if n := len(strings.Fields(out)); n > 0 {
			lines = append(lines, VerdictLine{Check: "mountpoint-" + m.name, Verdict: "fail",
				Detail: fmt.Sprintf("mountpoint held files while unmounted: %d entries under %s; move or remove them, then restart carlos-instance-control", n, m.where)})
			continue
		}
		lines = append(lines, VerdictLine{Check: "mountpoint-" + m.name, Verdict: "ok"})
	}
	return lines
}

// withStartLines prepends the start inspection to a probe report and
// recomputes Ready: the probe's own verdict, and no start line failing.
// A copy, so the stored report never aliases the probe's slice.
func (s *Server) withStartLines(r ProbeReport) ProbeReport {
	s.mu.Lock()
	start := append([]VerdictLine(nil), s.startLines...)
	s.mu.Unlock()
	if len(start) == 0 {
		return r
	}
	out := ProbeReport{Ready: r.Ready, Lines: append(start, r.Lines...)}
	for _, l := range start {
		if l.Verdict == "fail" {
			out.Ready = false
		}
	}
	return out
}

func (s *Server) claimAwareLegacy(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if s.ClaimStop != nil {
			w.WriteHeader(http.StatusServiceUnavailable)
			return
		}
		next(w, r)
	}
}