//go:build linux

package host

// The large-box startup, end to end (amadan #281, ask 2). On 2026-10-13 a
// box with 142 instances failed the 20 s update gate twice: the updater
// restarts carlos-instance-control just before the new edge, and the
// agent's startup sweep waited on that cold control service before the
// edge listened. The hotfix (6dfbff1) moved the sweep, layout recovery,
// hold restoration, Start and the key pass off the listen path, behind a
// startup latch that instance admission waits on.
//
// TestStartupServesWhileSweepAndRecoveryWait proves the latch with the
// steps stubbed. This test runs them for real on a box larger than the
// one that failed:
//
//   - the agent is a child process (this test binary, re-executed), so
//     what is timed is a process start up to the update gate's own probe;
//   - instance control is the real instancectl.Server over a unix socket,
//     made cold by its systemctl seam: every call waits until the test
//     warms it, then each of the box's 300 leftover units takes a moment
//     to stop, as a restarted control service does;
//   - the sweep is the real client call, layout recovery and hold
//     restoration are the real pass over real journals, hold markers and
//     files in a temp data dir, Start is the real activator's, and the
//     edge is edge.RunWithActivatorReady in its production (TLS) mode;
//   - every kind of admission the latch gates arrives during startup: an
//     instance request, a sidecar-marked request, a worker wake (from
//     runAgent's boot pass and from a direct call), and a scheduled
//     delivery through the real schedule runner and deliverer;
//   - the activator's backend is a fake that serves an app on the row's
//     socket and starts workers that report ready. It and the key pass's
//     writer are the only fakes in the agent, and neither is a startup
//     step.
//
// The child's startup is runAgent's legacy-box sequence: the same
// activatorStart, with runAgent's Sweep and Recover closures. runAgent
// itself cannot run in a test (its control socket path is a constant,
// it builds AWS clients, and it log.Fatals), so the child assembles the
// same pieces in the same order, and TestRunAgentStartupWiring reads
// runAgent's source to fail if the two drift apart. Sharing one
// orchestration function between them is a follow-up (#281). Two box
// facts differ from production and are set on the pass: its socket
// directory is the fixture's, not /run/carlos, and it has no
// carlos-sockets group.
//
// The child reports what it does as timestamped lines in an events file;
// the test reads them to order "a request reached the activator", "the
// latch opened" and "the backend acquired a host" across the two
// processes.

import (
	"bufio"
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"go/ast"
	"go/parser"
	"go/token"
	"go/types"
	"io"
	"log"
	"math/big"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/instancefs"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/schedule"
	"github.com/carlosframework/platform/internal/version"
)

const (
	// largeBoxAgentEnv names the child's config file. Set, the test binary
	// is the agent and nothing else.
	largeBoxAgentEnv = "CARLOS_TEST_LARGEBOX_AGENT"

	largeBoxRoutes      = 300 // instance routes; the box that failed had 142
	largeBoxJournals    = 8   // hosts 0..7: a layout move was in flight at the restart
	largeBoxHolds       = 4   // hosts 8..11: held down after a refused move
	largeBoxWakeHost    = 12  // an ordinary sleeping instance, woken by a request
	largeBoxDueHost     = 13  // a sleeping instance with a scheduled delivery due
	largeBoxSidecarHost = 14  // a sleeping instance its sidecar calls
	largeBoxWorker      = "worker-000.example.com"
	largeBoxDuePath     = "/jobs/tick"

	// largeBoxListenBound is how soon the gate's probe must be answered
	// after the agent process starts. The gate allows 20 s; the edge's own
	// startup on this box takes a fraction of a second.
	largeBoxListenBound = 3 * time.Second
	// largeBoxCold is how long instance control stays cold in the scenario
	// where it warms: longer than the listen bound, so the gate is answered
	// while the sweep is provably still waiting.
	largeBoxCold = 4 * time.Second
	// largeBoxUnitStop is one leftover unit's stop once control is warm.
	largeBoxUnitStop = 5 * time.Millisecond
	// largeBoxLatchBound is the activator's wakeWaitBound, which the
	// startup latch shares: a wake held longer is answered 503.
	largeBoxLatchBound = 30 * time.Second
	// largeBoxJoinBound is runAgent's join on the startup work after the
	// edge returns, and largeBoxDrain the edge's drain of in-flight
	// requests before it returns.
	largeBoxJoinBound = 5 * time.Second
	largeBoxDrain     = 10 * time.Second
	// largeBoxBindAttempts is how often the agent is started on fresh ports
	// when one it was handed is taken before it binds (see startAgent).
	largeBoxBindAttempts = 3
)

func largeBoxHost(i int) string { return fmt.Sprintf("app-%03d.example.com", i) }

// largeBoxConfig is what the child agent is told.
type largeBoxConfig struct {
	DataDir, SocketDir, StoreDir, AcmeDir string
	Control, Events, HTTP, HTTPS          string
}

func TestStartupLargeBoxEndToEnd(t *testing.T) {
	if path := os.Getenv(largeBoxAgentEnv); path != "" {
		os.Exit(runLargeBoxAgent(path))
	}
	// The parent stays serial on purpose. awaitListening holds each child
	// agent to a 3 s startup budget, so running beside the package's
	// parallel tests would put their load into that measurement. The
	// scenarios still run beside each other.
	t.Run("cold instance control", func(t *testing.T) {
		t.Parallel()
		testLargeBoxColdControl(t)
	})
	t.Run("startup overruns the latch", func(t *testing.T) {
		t.Parallel()
		testLargeBoxOverrun(t)
	})
	t.Run("shutdown mid-startup", func(t *testing.T) {
		t.Parallel()
		testLargeBoxShutdownMidStartup(t)
	})
}

// testLargeBoxColdControl is the 2026-10-13 shape: control restarted just
// before the agent, and slow to answer. The edge must answer the gate's
// probe while the sweep waits, hold every admission at the latch until
// sweep, recovery and hold restoration are done, and then serve each host
// only where it is safe to.
func testLargeBoxColdControl(t *testing.T) {
	b := newLargeBox(t)
	a := b.startAgent(t)
	t.Logf("gate probe answered %s after the agent started, %d routes, instance control cold", a.listen, largeBoxRoutes)
	if _, ok := a.event("ready"); ok {
		t.Fatal("the startup latch opened while instance control was still cold")
	}

	wake, held, moved, due, sidecar := largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxJournals), largeBoxHost(0), largeBoxHost(largeBoxDueHost), largeBoxHost(largeBoxSidecarHost)
	flatMoved := filepath.Join(b.socketDir, moved+".sock")
	wakeAns, heldAns, movedAns := b.fire(wake, false), b.fire(held, false), b.fire(moved, false)
	sidecarAns := b.fire(sidecar, true)
	enterWake := a.waitEvent(t, 10*time.Second, "enter", wake)
	a.waitEvent(t, 10*time.Second, "enter", held)
	enterSidecar := a.waitEvent(t, 10*time.Second, "enter", sidecar)
	enterMoved := a.waitEvent(t, 10*time.Second, "enter", moved)
	if enterMoved.arg(1) != flatMoved {
		t.Fatalf("the edge routed %s to %q before recovery, want its pre-move socket %q", moved, enterMoved.arg(1), flatMoved)
	}
	// The agent's own admissions: a direct worker wake and the scheduled
	// tick, both started before the edge listened.
	workerAsked := a.waitEvent(t, 10*time.Second, "worker-direct-begin")
	ticked := a.waitEvent(t, 10*time.Second, "tick-done")
	// Every admission is now waiting, and startup is not done.
	time.Sleep(200 * time.Millisecond)
	for host, ch := range map[string]chan largeBoxAnswer{wake: wakeAns, held: heldAns, moved: movedAns, sidecar: sidecarAns} {
		select {
		case got := <-ch:
			t.Fatalf("%s answered before startup finished: %d %q %v", host, got.status, got.body, got.err)
		default:
		}
	}
	for _, kind := range []string{"acquire", "start", "worker-start", "served", "worker-direct", "worker-boot", "activator-start"} {
		if evs := a.events(kind); len(evs) != 0 {
			t.Fatalf("%s while instance control was cold: %v", kind, evs)
		}
	}

	// Instance control warms: the sweep stops the 300 leftovers, then
	// recovery, holds, Start, and the latch.
	if d := time.Until(a.began.Add(largeBoxCold)); d > 0 {
		time.Sleep(d)
	}
	close(b.control.warm)
	ready := a.waitEvent(t, 30*time.Second, "ready")
	t.Logf("startup latch opened %s after the agent started (sweep stopped %d units)", ready.at.Sub(a.began), b.control.stops.Load())
	if got := b.control.stops.Load(); got != largeBoxRoutes {
		t.Errorf("the sweep stopped %d units, want %d", got, largeBoxRoutes)
	}
	for what, at := range map[string]time.Time{"wake": enterWake.at, "sidecar request": enterSidecar.at, "worker wake": workerAsked.at, "scheduled tick": ticked.at} {
		if !at.Before(ready.at) {
			t.Fatalf("the %s did not arrive before the latch opened; the test proves nothing about its latch", what)
		}
	}

	// Start ran only once the held hosts were fenced again and every
	// journal was complete.
	st := a.waitEvent(t, time.Second, "activator-start")
	if st.arg(0) != "held="+strconv.Itoa(largeBoxHolds) || st.arg(1) != "fenced="+strconv.Itoa(largeBoxHolds) || st.arg(2) != "journals=0" || st.at.After(ready.at) {
		t.Fatalf("at Start: %v", st)
	}

	// The wake was held, then served.
	got := b.answer(t, wakeAns)
	if got.status != http.StatusOK || got.body != "host="+wake+" socket="+filepath.Join(b.socketDir, wake+".sock") {
		t.Fatalf("held wake after startup: %d %q %v", got.status, got.body, got.err)
	}
	a.assertAfter(t, ready, "acquire", wake)

	// The sidecar's call was held too, then answered as the activator
	// answers a sidecar whose instance sleeps: it does not wake it.
	got = b.answer(t, sidecarAns)
	if got.status != http.StatusServiceUnavailable || got.body != "instance is asleep; sidecar requests do not wake it\n" || got.at.Before(ready.at) {
		t.Fatalf("sidecar request held through startup: %d %q %v at %s, latch at %s", got.status, got.body, got.err, got.at, ready.at)
	}

	// The worker: both wakes waited, then coalesced on one child.
	a.assertAfter(t, ready, "worker-direct", largeBoxWorker, "true")
	a.assertAfter(t, ready, "worker-boot", largeBoxWorker, "true")
	a.assertAfter(t, ready, "worker-start", largeBoxWorker)
	if n := len(a.events("worker-start")); n != 1 {
		t.Errorf("worker started %d times, want once", n)
	}

	// The scheduled delivery waited, then woke its instance and ran.
	a.assertAfter(t, ready, "served", due, largeBoxDuePath)
	a.assertAfter(t, ready, "acquire", due)

	// The held host was not started: its restored hold answers maintenance.
	got = b.answer(t, heldAns)
	if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" || got.header.Get("Cache-Control") != "no-store" {
		t.Fatalf("held host after startup: %d %q %v %v", got.status, got.body, got.header, got.err)
	}

	// The moved host was not served at the socket recovery moved it from:
	// the request answers as a wake in progress, and its retry is routed
	// afresh to where recovery put it.
	got = b.answer(t, movedAns)
	if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" {
		t.Fatalf("host whose recovery moved it, held through startup: %d %q %v", got.status, got.body, got.err)
	}
	nested, err := layout.PlanFor(b.dataDir, b.socketDir, moved, layout.ModeNested)
	if err != nil {
		t.Fatal(err)
	}
	got = b.get(moved, false)
	if got.status != http.StatusOK || got.body != "host="+moved+" socket="+nested.To.Socket {
		t.Fatalf("retry of the moved host: %d %q %v", got.status, got.body, got.err)
	}
	got = b.get(held, false)
	if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" {
		t.Fatalf("retry of the held host: %d %q %v", got.status, got.body, got.err)
	}

	// The background key pass ran after Start, and its slot refused the
	// held hosts.
	a.waitEvent(t, 30*time.Second, "startup-done")
	keys := a.events("key")
	if len(keys) != largeBoxRoutes-largeBoxHolds {
		t.Errorf("key pass wrote %d rows, want %d", len(keys), largeBoxRoutes-largeBoxHolds)
	}
	for _, e := range keys {
		if e.at.Before(ready.at) {
			t.Errorf("key pass reached %s before Start", e.arg(0))
		}
	}

	b.assertNoWrongService(t, a)
	b.assertRecovered(t)

	a.signal(t, syscall.SIGTERM)
	a.waitExit(t, largeBoxJoinBound+3*time.Second)
	a.assertCleanShutdown(t)
}

// testLargeBoxOverrun: instance control never warms, so startup never
// finishes. A wake must be answered with the wake-timeout 503 at the
// latch's bound: not an error, not a hang, and not a wake.
func testLargeBoxOverrun(t *testing.T) {
	b := newLargeBox(t)
	a := b.startAgent(t)
	t.Logf("gate probe answered %s after the agent started, instance control never warms", a.listen)

	wake := largeBoxHost(largeBoxWakeHost)
	began := time.Now()
	got := b.get(wake, false)
	took := time.Since(began)
	t.Logf("wake held %s by a startup that does not finish", took)
	if got.err != nil {
		t.Fatalf("wake during an overrunning startup: %v", got.err)
	}
	if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" {
		t.Fatalf("wake during an overrunning startup: %d %q %v", got.status, got.body, got.header)
	}
	if took < largeBoxLatchBound-time.Second || took > largeBoxLatchBound+5*time.Second {
		t.Fatalf("wake answered after %s, want the latch bound %s", took, largeBoxLatchBound)
	}
	a.assertNeverStarted(t)
	b.assertNoWrongService(t, a)

	// Shutting down with the sweep still waiting: nothing in flight, so the
	// edge returns at once and runAgent's join bounds the rest.
	at := time.Now()
	a.signal(t, syscall.SIGTERM)
	a.waitExit(t, largeBoxJoinBound+3*time.Second)
	t.Logf("agent exited %s after SIGTERM with the sweep still waiting", time.Since(at))
	a.assertNeverStarted(t)
	a.assertCleanShutdown(t)
}

// testLargeBoxShutdownMidStartup: SIGTERM with startup still waiting on
// the sweep and a wake held at the latch. The held request is answered
// 503 (never proxied), the background work is joined within runAgent's
// bound, and nothing of startup outlives it.
func testLargeBoxShutdownMidStartup(t *testing.T) {
	b := newLargeBox(t)
	a := b.startAgent(t)

	wake := largeBoxHost(largeBoxWakeHost)
	ans := b.fire(wake, false)
	a.waitEvent(t, 10*time.Second, "enter", wake)

	at := time.Now()
	a.signal(t, syscall.SIGTERM)
	// The edge drains in-flight requests for up to largeBoxDrain before it
	// returns, and the held wake is in flight until the activator shuts
	// down after the join.
	a.waitExit(t, largeBoxDrain+largeBoxJoinBound+3*time.Second)
	t.Logf("agent exited %s after SIGTERM with a wake held at the latch", time.Since(at))
	got := b.answer(t, ans)
	if got.err != nil || got.status != http.StatusServiceUnavailable {
		t.Fatalf("wake held through shutdown: %d %q %v", got.status, got.body, got.err)
	}
	a.assertNeverStarted(t)
	b.assertNoWrongService(t, a)
	a.assertCleanShutdown(t)
}

// --- runAgent's wiring ------------------------------------------------------------

// TestRunAgentStartupWiring reads the startup's source and fails if it
// drifts from what runLargeBoxAgent rebuilds, or if anything on runAgent's
// path to the edge waits for startup to finish. Calls are matched on the
// callee's own name (a selector's Sel), and order means execution order on
// one goroutine: a step inside a go or defer statement, or inside a
// function literal that is not invoked on the spot, does not count as
// taken. Where the hotfix does mean a step to run in the background, the
// one goroutine it runs on is found structurally and its order checked
// inside it.
func TestRunAgentStartupWiring(t *testing.T) {
	t.Parallel()
	fset := token.NewFileSet()
	decls := map[string]*ast.FuncDecl{} // runAgent, startActivator, recoverLayoutAtStart, activatorStart.Run
	files, err := filepath.Glob("*.go")
	if err != nil {
		t.Fatal(err)
	}
	for _, name := range files {
		if strings.HasSuffix(name, "_test.go") {
			continue
		}
		f, err := parser.ParseFile(fset, name, nil, 0)
		if err != nil {
			t.Fatal(err)
		}
		for _, d := range f.Decls {
			fd, ok := d.(*ast.FuncDecl)
			if !ok {
				continue
			}
			key := fd.Name.Name
			if fd.Recv != nil && len(fd.Recv.List) == 1 {
				key = types.ExprString(fd.Recv.List[0].Type) + "." + key
			}
			switch key {
			case "runAgent", "startActivator", "recoverLayoutAtStart", "activatorStart.Run", "joinAgentShutdown":
				decls[key] = fd
			}
		}
	}
	for _, key := range []string{"runAgent", "startActivator", "recoverLayoutAtStart", "activatorStart.Run", "joinAgentShutdown"} {
		if decls[key] == nil {
			t.Fatalf("%s not found in package host", key)
		}
	}
	str := types.ExprString

	// runAgent: the latch is set, the start sequence is handed the startup
	// context, the edge runs, then cancel, join and Shutdown, all on
	// runAgent's own goroutine.
	runAgent := decls["runAgent"].Body
	sync := syncCalls(runAgent)
	var run, edgeCall *ast.CallExpr
	var lit *ast.CompositeLit
	for _, c := range sync {
		if sel, ok := c.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Run" {
			if cl, ok := sel.X.(*ast.CompositeLit); ok && str(cl.Type) == "activatorStart" {
				run, lit = c, cl
			}
		}
		if calleeName(c) == "RunWithActivatorReady" {
			edgeCall = c
		}
	}
	if run == nil || edgeCall == nil {
		t.Fatalf("runAgent: activatorStart{...}.Run called synchronously: %v; edge.RunWithActivatorReady called synchronously: %v", run != nil, edgeCall != nil)
	}
	if len(run.Args) != 1 || str(run.Args[0]) != "startupCtx" {
		t.Errorf("activatorStart.Run is given %v, want startupCtx", run.Args)
	}
	var latch token.Pos
	ast.Inspect(runAgent, func(n ast.Node) bool {
		if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 && selName(a.Lhs[0]) == "Startup" && str(a.Rhs[0]) == "started" {
			latch = a.Pos()
		}
		return true
	})
	if latch == token.NoPos || latch > run.Pos() {
		t.Error("runAgent: svc.Startup = started is not set before activatorStart.Run")
	}
	after := func(name string, arg string) token.Pos {
		for _, c := range sync {
			if c.Pos() > edgeCall.Pos() && calleeName(c) == name && (arg == "" || (len(c.Args) > 0 && str(c.Args[len(c.Args)-1]) == arg)) {
				return c.Pos()
			}
		}
		return token.NoPos
	}
	prev, prevName := edgeCall.Pos(), "edge.RunWithActivatorReady"
	for _, step := range []struct{ name, arg string }{{"joinAgentShutdown", "dutiesDone"}, {"Shutdown", ""}} {
		p := after(step.name, step.arg)
		if p == token.NoPos || p < prev {
			t.Errorf("runAgent: %s is not called synchronously after %s", step.name, prevName)
			continue
		}
		prev, prevName = p, step.name
	}

	// Follow the synchronous helper without weakening cancel/join ordering or
	// allowing the caller to pass different startup channels or cancellation.
	for _, call := range sync {
		if calleeName(call) != "joinAgentShutdown" {
			continue
		}
		want := []string{"seedCtx", "seedCancel", "seedServer", "cancelStartup", "startupDone", "dutiesDone"}
		if len(call.Args) != len(want) {
			t.Fatal("shutdown helper argument count", len(call.Args))
		}
		for i, arg := range call.Args {
			if str(arg) != want[i] {
				t.Errorf("shutdown argument %d: %s, want %s", i, str(arg), want[i])
			}
		}
	}
	helper := syncCalls(decls["joinAgentShutdown"].Body)
	previous := token.NoPos
	for _, step := range []struct{ name, arg string }{{"cancelStartup", ""}, {"joinAgentStartup", "startupDone"}, {"joinAgentStartup", "dutiesDone"}} {
		found := token.NoPos
		for _, call := range helper {
			if calleeName(call) == step.name && (step.arg == "" || len(call.Args) > 0 && str(call.Args[len(call.Args)-1]) == step.arg) {
				found = call.Pos()
				break
			}
		}
		if found == token.NoPos || found <= previous {
			t.Fatalf("shutdown helper lost synchronous order at %s(%s)", step.name, step.arg)
		}
		previous = found
	}

	// The activatorStart fields the child copies, and the steps inside
	// them. Run calls each field on its background goroutine, so inside a
	// field each step must be synchronous.
	fields := map[string]ast.Expr{}
	for _, el := range lit.Elts {
		if kv, ok := el.(*ast.KeyValueExpr); ok {
			fields[str(kv.Key)] = kv.Value
		}
	}
	for field, want := range map[string]string{
		"Start": "Start", "Publish": "Store", "Slot": "PrepareScheduledToken",
		"Routes": "List", "Route": "Get", "Claims": "Enabled",
	} {
		if got := selName(fields[field]); got != want {
			t.Errorf("activatorStart.%s is %s, want a %s method value", field, exprOrNil(fields[field]), want)
		}
	}
	funcField := func(name string) ast.Node {
		fl, _ := fields[name].(*ast.FuncLit)
		if fl == nil {
			t.Errorf("activatorStart.%s is not a func literal", name)
			return nil
		}
		return fl.Body
	}
	if body := funcField("Recover"); body != nil {
		requireSteps(t, "activatorStart.Recover", body, "newLayoutPass", "startupRecoveryIntent", "recoverLayoutAtStart")
	}
	if body := funcField("Sweep"); body != nil {
		requireSteps(t, "activatorStart.Sweep", body, "Sweep")
	}
	if body := funcField("Ready"); body != nil {
		closes := false
		for _, c := range syncCalls(body) {
			closes = closes || (calleeName(c) == "close" && len(c.Args) == 1 && str(c.Args[0]) == "started")
		}
		if !closes {
			t.Error("activatorStart.Ready does not close started synchronously")
		}
	}

	// activatorStart.Run: on a legacy box, one goroutine runs the start
	// sequence and then the key pass, in that order; start itself is
	// Start, then Ready.
	runBody := decls["activatorStart.Run"].Body
	var background []*ast.FuncLit
	ast.Inspect(runBody, func(n ast.Node) bool {
		if g, ok := n.(*ast.GoStmt); ok {
			if fl, ok := g.Call.Fun.(*ast.FuncLit); ok && hasCall(syncCalls(fl.Body), "startActivator") {
				background = append(background, fl)
			}
		}
		return true
	})
	if len(background) != 1 {
		t.Errorf("activatorStart.Run: %d goroutines run startActivator, want the one background start", len(background))
	} else {
		requireSteps(t, "activatorStart.Run's background goroutine", background[0].Body, "startActivator", "backfillInstanceKeys")
	}
	var startBody ast.Node
	ast.Inspect(runBody, func(n ast.Node) bool {
		if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 && str(a.Lhs[0]) == "start" {
			if fl, ok := a.Rhs[0].(*ast.FuncLit); ok {
				startBody = fl.Body
			}
		}
		return true
	})
	if startBody == nil {
		t.Error("activatorStart.Run: no start closure")
	} else {
		requireSteps(t, "activatorStart.Run's start", startBody, "Start", "Ready")
	}

	// startActivator: its sweep, recoverLayout and start parameters, called
	// in that order, synchronously (matched by position, not name).
	sa := decls["startActivator"]
	var params []string
	for _, f := range sa.Type.Params.List {
		for _, n := range f.Names {
			params = append(params, n.Name)
		}
	}
	if len(params) != 4 {
		t.Errorf("startActivator takes %v, want (ctx, sweep, recoverLayout, start)", params)
	} else {
		requireSteps(t, "startActivator", sa.Body, params[1], params[2], params[3])
	}

	// recoverLayoutAtStart: journal recovery, then hold restoration.
	requireSteps(t, "recoverLayoutAtStart", decls["recoverLayoutAtStart"].Body, "Recover", "RestoreHolds")

	// The worker boot pass waits on the latch or the startup context, then
	// lists and wakes.
	var bootPass *ast.FuncLit
	ast.Inspect(runAgent, func(n ast.Node) bool {
		if g, ok := n.(*ast.GoStmt); ok {
			if fl, ok := g.Call.Fun.(*ast.FuncLit); ok && hasCall(syncCalls(fl.Body), "runWorkerBootPass") {
				bootPass = fl
			}
		}
		return true
	})
	if bootPass == nil {
		t.Error("runAgent: no goroutine runs runWorkerBootPass")
	} else if !receivesFrom(bootPass.Body, "started") || !hasCall(syncCalls(bootPass.Body), "Done") {
		t.Error("runAgent: the worker boot pass does not wait on started and startupCtx.Done")
	}

	// Nothing before the edge waits for startup. On runAgent's own
	// goroutine (closures invoked on the spot included): no receive at all,
	// no join, and a closure that waits on startup is only called directly
	// on a claim box, where Run has already started the activator before
	// it returns.
	waiters := map[string]bool{}
	ast.Inspect(runAgent, func(n ast.Node) bool {
		if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 {
			if fl, ok := a.Rhs[0].(*ast.FuncLit); ok && (receivesFrom(fl.Body, "started") || receivesFrom(fl.Body, "startupDone") || hasCall(syncCalls(fl.Body), "joinAgentStartup")) {
				waiters[str(a.Lhs[0])] = true
			}
		}
		return true
	})
	var walk func(n ast.Node, claimBranch bool)
	walk = func(n ast.Node, claimBranch bool) {
		ast.Inspect(n, func(n ast.Node) bool {
			if n == nil || n.Pos() >= edgeCall.Pos() {
				return false
			}
			switch n := n.(type) {
			case *ast.FuncLit, *ast.GoStmt, *ast.DeferStmt:
				return false
			case *ast.IfStmt:
				if n.Init != nil {
					walk(n.Init, claimBranch)
				}
				walk(n.Cond, claimBranch)
				walk(n.Body, claimBranch || selName(n.Cond) == "Enabled")
				if n.Else != nil {
					walk(n.Else, claimBranch)
				}
				return false
			case *ast.UnaryExpr:
				if n.Op == token.ARROW {
					t.Errorf("runAgent waits on %s at %s, before the edge listens", str(n.X), fset.Position(n.Pos()))
				}
			case *ast.CallExpr:
				name := calleeName(n)
				if name == "joinAgentStartup" {
					t.Errorf("runAgent joins startup at %s, before the edge listens", fset.Position(n.Pos()))
				}
				if waiters[str(n.Fun)] && !claimBranch {
					t.Errorf("runAgent calls %s, which waits on startup, at %s, before the edge listens", str(n.Fun), fset.Position(n.Pos()))
				}
				if fl, ok := n.Fun.(*ast.FuncLit); ok { // invoked on the spot: its body runs here
					walk(fl.Body, claimBranch)
					for _, a := range n.Args {
						walk(a, claimBranch)
					}
					return false
				}
			}
			return true
		})
	}
	walk(runAgent, false)
}

// syncCalls lists, in source order, the calls body makes on its own
// goroutine: none inside a go or defer statement, and none inside a
// function literal unless the literal is invoked right there.
func syncCalls(body ast.Node) []*ast.CallExpr {
	var out []*ast.CallExpr
	ast.Inspect(body, func(n ast.Node) bool {
		switch n := n.(type) {
		case *ast.GoStmt, *ast.DeferStmt, *ast.FuncLit:
			return false
		case *ast.CallExpr:
			out = append(out, n)
			if fl, ok := n.Fun.(*ast.FuncLit); ok {
				for _, a := range n.Args {
					out = append(out, syncCalls(a)...)
				}
				out = append(out, syncCalls(fl.Body)...)
				return false
			}
		}
		return true
	})
	return out
}

// requireSteps fails unless body calls each step synchronously, in order,
// and calls none of them anywhere else (in a go or defer statement, or a
// literal that is not invoked on the spot).
func requireSteps(t *testing.T, where string, body ast.Node, steps ...string) {
	t.Helper()
	sync := syncCalls(body)
	inSync := map[*ast.CallExpr]bool{}
	for _, c := range sync {
		inSync[c] = true
	}
	ast.Inspect(body, func(n ast.Node) bool {
		if c, ok := n.(*ast.CallExpr); ok && !inSync[c] {
			for _, s := range steps {
				if calleeName(c) == s {
					t.Errorf("%s: %s is called off its goroutine (go, defer, or a literal not invoked there)", where, s)
				}
			}
		}
		return true
	})
	i := 0
	for _, c := range sync {
		if i < len(steps) && calleeName(c) == steps[i] {
			i++
		}
	}
	if i < len(steps) {
		t.Errorf("%s: want %v called in that order on its goroutine; %s is missing or out of order", where, steps, steps[i])
	}
}

// calleeName is the name a call is matched on: the function's own name,
// whatever the receiver or package it is reached through.
func calleeName(c *ast.CallExpr) string {
	switch f := c.Fun.(type) {
	case *ast.Ident:
		return f.Name
	case *ast.SelectorExpr:
		return f.Sel.Name
	}
	return ""
}

func selName(e ast.Expr) string {
	switch e := e.(type) {
	case *ast.SelectorExpr:
		return e.Sel.Name
	case *ast.Ident:
		return e.Name
	}
	return ""
}

func hasCall(calls []*ast.CallExpr, name string) bool {
	for _, c := range calls {
		if calleeName(c) == name {
			return true
		}
	}
	return false
}

// receivesFrom reports whether body receives from the channel named ch
// (in a select case or a bare receive).
func receivesFrom(body ast.Node, ch string) bool {
	found := false
	ast.Inspect(body, func(n ast.Node) bool {
		if u, ok := n.(*ast.UnaryExpr); ok && u.Op == token.ARROW && types.ExprString(u.X) == ch {
			found = true
		}
		return !found
	})
	return found
}

func exprOrNil(e ast.Expr) string {
	if e == nil {
		return "unset"
	}
	return types.ExprString(e)
}

// --- the box ------------------------------------------------------------------

type largeBox struct {
	dataDir, socketDir, storeDir, acmeDir string
	reg                                   *registry.Registry
	control                               *largeBoxControl
	httpAddr, httpsAddr                   string
	client                                *http.Client
}

// largeBoxSocketDir makes the box's run directory. It must be short: a
// nested instance's socket is <dir>/instances/<instance id>/app.sock, the
// activator binds a ".control" socket beside it, and the whole path must
// fit a unix socket address (provision.MaxUnixPath). The default temp dir
// is tried first, then a directory of the test's own directly under /tmp;
// when neither is short enough the test is skipped, not failed, since
// nothing about the code under test is wrong.
func largeBoxSocketDir(t *testing.T) string {
	t.Helper()
	fits := func(dir string) bool {
		return len(filepath.Join(dir, "instances", provision.InstanceID(largeBoxHost(0)), "app.sock.control")) <= provision.MaxUnixPath
	}
	var tried []string
	for _, base := range []string{os.TempDir(), "/tmp"} {
		dir, err := os.MkdirTemp(base, "lb")
		if err != nil {
			tried = append(tried, base+": "+err.Error())
			continue
		}
		if fits(dir) {
			t.Cleanup(func() { os.RemoveAll(dir) })
			return dir
		}
		os.RemoveAll(dir)
		tried = append(tried, dir+": too long")
	}
	t.Skipf("no directory short enough for a nested instance's control socket (%d bytes at most): %s", provision.MaxUnixPath, strings.Join(tried, "; "))
	return ""
}

// newLargeBox builds the box the agent restarts on: largeBoxRoutes
// sleeping instances on flat paths, largeBoxJournals of them with a layout
// move cut short at three different points, largeBoxHolds held down after
// a refused move (the reason, a second name for the database, persists),
// a scheduled delivery due on one instance, an always-on worker the boot
// pass starts, a nested layout intent in the deployment store, a cached
// certificate for every host the test asks for, and a cold instance
// control.
func newLargeBox(t *testing.T) *largeBox {
	t.Helper()
	sd := largeBoxSocketDir(t)
	b := &largeBox{dataDir: t.TempDir(), storeDir: t.TempDir(), acmeDir: t.TempDir(), socketDir: sd}

	var err error
	b.reg, err = registry.Open(filepath.Join(b.dataDir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { b.reg.Close() })
	var hosts []string
	for i := 0; i < largeBoxRoutes; i++ {
		h := largeBoxHost(i)
		hosts = append(hosts, h)
		if err := b.reg.Upsert(registry.Route{
			Host: h, Kind: "instance", Account: "acct", App: "shop", Status: "ready", Hibernate: true,
			Socket: filepath.Join(sd, h+".sock"), DBPath: provision.DefaultDBPath(b.dataDir, h),
		}); err != nil {
			t.Fatal(err)
		}
	}
	if err := b.reg.Upsert(registry.Route{
		Host: largeBoxWorker, Kind: "sidecar", Account: "acct", App: "shop", Status: "ready", Args: `["serve"]`,
	}); err != nil {
		t.Fatal(err)
	}
	due := time.Now().Unix() - 5
	if err := b.reg.UpsertTimer(registry.Timer{
		Host: largeBoxHost(largeBoxDueHost), Name: "tick", Kind: schedule.KindSchedule, Spec: "every 6h0m0s",
		Path: largeBoxDuePath, NextAt: due, DueAt: due,
	}); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < largeBoxJournals; i++ {
		crash := "after-journal"
		switch {
		case i >= 6:
			crash = "before-remove"
		case i >= 4:
			crash = "mid-files"
		}
		b.crashedMove(t, largeBoxHost(i), crash)
	}
	for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ {
		b.heldMove(t, largeBoxHost(i), i)
	}
	appendIntent(t, release.NewDirStore(b.storeDir), release.LayoutModeNested, "")

	served := []string{largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxSidecarHost)}
	for i := 0; i < largeBoxJournals+largeBoxHolds; i++ {
		served = append(served, largeBoxHost(i))
	}
	pool := writeLargeBoxCerts(t, b.acmeDir, served)

	b.control = startLargeBoxControl(t, filepath.Join(sd, "control.sock"), hosts)
	b.client = &http.Client{
		Timeout: largeBoxLatchBound + 15*time.Second,
		Transport: &http.Transport{
			Proxy:             nil,
			DisableKeepAlives: true,
			TLSClientConfig:   &tls.Config{RootCAs: pool},
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				return (&net.Dialer{}).DialContext(ctx, "tcp", b.httpsAddr)
			},
		},
		CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
	}
	return b
}

func (b *largeBox) seedFlat(t *testing.T, h string) {
	t.Helper()
	db := provision.DefaultDBPath(b.dataDir, h)
	for path, body := range map[string]string{db: "db " + h, db + "-wal": "wal", filepath.Join(b.dataDir, h+".env"): "CARLOS_ADMIN_TOKEN=t\n"} {
		if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
}

// crashedMove leaves h as an agent killed mid-migration to nested leaves
// it: the instance directories made and the journal written, then
// nothing moved (after-journal), the database moved (mid-files), or
// everything moved and the row flipped with only the journal's removal
// left (before-remove).
func (b *largeBox) crashedMove(t *testing.T, h, crash string) {
	t.Helper()
	b.seedFlat(t, h)
	if err := layout.EnsureDirs(b.dataDir, b.socketDir, h, -1); err != nil {
		t.Fatal(err)
	}
	plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested)
	if err != nil {
		t.Fatal(err)
	}
	present := make([]bool, len(plan.Artifacts))
	for i, a := range plan.Artifacts {
		if a.Name == "socket" {
			continue
		}
		_, err := os.Lstat(a.Src)
		present[i] = err == nil
	}
	j := plan.Journal(present)
	if err := layout.WriteJournal(b.dataDir, j); err != nil {
		t.Fatal(err)
	}
	move := func(a layout.Artifact) {
		if err := os.Rename(a.Src, a.Dst); err != nil {
			t.Fatal(err)
		}
	}
	switch crash {
	case "mid-files":
		move(plan.Artifacts[0])
	case "before-remove":
		for i, a := range plan.Artifacts {
			if present[i] {
				move(a)
			}
		}
		if err := b.reg.SetInstancePaths(h, plan.From.DB, plan.From.Socket, plan.To.DB, plan.To.Socket); err != nil {
			t.Fatal(err)
		}
		if err := layout.UpdatePhase(b.dataDir, j, layout.PhaseDone); err != nil {
			t.Fatal(err)
		}
	}
}

// heldMove leaves h held down after a refused move to nested: its hold
// marker, and the reason, which persists across the restart (a second
// name for its database, which a flat tenant can plant).
func (b *largeBox) heldMove(t *testing.T, h string, i int) {
	t.Helper()
	b.seedFlat(t, h)
	if err := os.Link(provision.DefaultDBPath(b.dataDir, h), filepath.Join(b.dataDir, ".x"+strconv.Itoa(i))); err != nil {
		t.Fatal(err)
	}
	dir := filepath.Join(b.dataDir, "layout")
	if err := os.MkdirAll(dir, 0o700); err != nil {
		t.Fatal(err)
	}
	marker, _ := json.Marshal(map[string]any{"host": h, "reason": "db has 2 names", "since": time.Now().UTC()})
	if err := os.WriteFile(filepath.Join(dir, h+".hold"), marker, 0o600); err != nil {
		t.Fatal(err)
	}
}

// assertRecovered checks the files: every journal completed and removed,
// each journalled host's row and database at the nested layout, and every
// hold marker kept.
func (b *largeBox) assertRecovered(t *testing.T) {
	t.Helper()
	if hosts, err := layout.JournalHosts(b.dataDir); err != nil || len(hosts) != 0 {
		t.Errorf("journals after startup: %v %v", hosts, err)
	}
	for i := 0; i < largeBoxJournals; i++ {
		h := largeBoxHost(i)
		plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested)
		if err != nil {
			t.Fatal(err)
		}
		rt, err := b.reg.Get(h)
		if err != nil || rt == nil || rt.DBPath != plan.To.DB || rt.Socket != plan.To.Socket {
			t.Errorf("%s row after recovery: %+v %v", h, rt, err)
		}
		if body, err := os.ReadFile(plan.Artifacts[0].Dst); err != nil || string(body) != "db "+h {
			t.Errorf("%s database after recovery: %q %v", h, body, err)
		}
	}
	held, err := layout.HoldMarkers(b.dataDir)
	if err != nil || len(held) != largeBoxHolds {
		t.Errorf("hold markers after startup: %v %v", held, err)
	}
}

// assertNoWrongService reads the backend's record: no wake admitted before
// the latch opened, no held host started, and no journalled host started
// at its pre-recovery socket.
func (b *largeBox) assertNoWrongService(t *testing.T, a *largeBoxAgent) {
	t.Helper()
	if v := a.events("violation"); len(v) != 0 {
		t.Errorf("backend violations: %v", v)
	}
	for _, e := range append(a.events("acquire"), a.events("start")...) {
		for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ {
			if e.arg(0) == largeBoxHost(i) {
				t.Errorf("held host %s reached the backend: %v", e.arg(0), e)
			}
		}
	}
	for _, e := range a.events("start") {
		for i := 0; i < largeBoxJournals; i++ {
			if h := largeBoxHost(i); e.arg(0) == h && e.arg(1) == filepath.Join(b.socketDir, h+".sock") {
				t.Errorf("%s started at its pre-recovery socket", h)
			}
		}
	}
}

type largeBoxAnswer struct {
	status int
	body   string
	header http.Header
	err    error
	at     time.Time // when the answer was read
}

// get asks the edge for host's root over TLS. sidecar marks the request
// as the host's sidecar's (activator.SidecarHeader).
func (b *largeBox) get(host string, sidecar bool) largeBoxAnswer {
	req, err := http.NewRequest(http.MethodGet, "https://"+host+"/", nil)
	if err != nil {
		return largeBoxAnswer{err: err, at: time.Now()}
	}
	if sidecar {
		req.Header.Set(activator.SidecarHeader, "1")
	}
	resp, err := b.client.Do(req)
	if err != nil {
		return largeBoxAnswer{err: err, at: time.Now()}
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	return largeBoxAnswer{status: resp.StatusCode, body: string(body), header: resp.Header, err: err, at: time.Now()}
}

func (b *largeBox) fire(host string, sidecar bool) chan largeBoxAnswer {
	ch := make(chan largeBoxAnswer, 1)
	go func() { ch <- b.get(host, sidecar) }()
	return ch
}

func (b *largeBox) answer(t *testing.T, ch chan largeBoxAnswer) largeBoxAnswer {
	t.Helper()
	select {
	case got := <-ch:
		return got
	case <-time.After(10 * time.Second):
		t.Fatal("request did not finish")
		return largeBoxAnswer{}
	}
}

func freeLoopbackAddr(t *testing.T) string {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer ln.Close()
	return ln.Addr().String()
}

// writeLargeBoxCerts puts one self-signed certificate for hosts into the
// edge's autocert directory cache, under each host's name, so the edge in
// production mode serves them TLS without ordering anything.
func writeLargeBoxCerts(t *testing.T, dir string, hosts []string) *x509.CertPool {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &x509.Certificate{
		SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "large box"}, DNSNames: hosts,
		NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour),
		KeyUsage:    x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
		ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, BasicConstraintsValid: true, IsCA: true,
	}
	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	if err != nil {
		t.Fatal(err)
	}
	keyDER, err := x509.MarshalECPrivateKey(key)
	if err != nil {
		t.Fatal(err)
	}
	pemBytes := append(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...)
	for _, h := range hosts {
		if err := os.WriteFile(filepath.Join(dir, h), pemBytes, 0o600); err != nil {
			t.Fatal(err)
		}
	}
	cert, err := x509.ParseCertificate(der)
	if err != nil {
		t.Fatal(err)
	}
	pool := x509.NewCertPool()
	pool.AddCert(cert)
	return pool
}

// --- instance control ---------------------------------------------------------

// largeBoxControl is the real instance-control server with a cold systemd
// behind its Systemctl seam: every call waits for warm, then the box's
// leftover units are listed active and each stop takes largeBoxUnitStop.
type largeBoxControl struct {
	warm  chan struct{}
	units string
	stops atomic.Int32
}

func startLargeBoxControl(t *testing.T, path string, hosts []string) *largeBoxControl {
	t.Helper()
	c := &largeBoxControl{warm: make(chan struct{})}
	var units strings.Builder
	for _, h := range hosts {
		fmt.Fprintf(&units, "%s loaded active running carlos instance\n", instancectl.UnitName(instancectl.RoleRun, h))
	}
	c.units = units.String()
	srv := &instancectl.Server{
		Systemctl:  c.systemctl,
		PeerOK:     func(net.Conn) error { return nil },
		CgroupRoot: t.TempDir(),
		Build:      version.Build,
		Logf:       func(string, ...any) {},
	}
	ln, err := net.Listen("unix", path)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { defer close(done); _ = srv.Serve(ctx, ln) }()
	t.Cleanup(func() { cancel(); <-done })
	return c
}

func (c *largeBoxControl) systemctl(ctx context.Context, args ...string) (string, error) {
	select {
	case <-c.warm:
	case <-ctx.Done():
		return "", ctx.Err()
	}
	switch args[0] {
	case "list-units":
		return c.units, nil
	case "list-jobs":
		return "", nil
	case "stop":
		select {
		case <-time.After(largeBoxUnitStop):
		case <-ctx.Done():
			return "", ctx.Err()
		}
		c.stops.Add(1)
		return "", nil
	case "show":
		return "inactive\n", nil
	}
	return "", fmt.Errorf("systemctl %v: not part of a sweep", args)
}

// --- the agent process ----------------------------------------------------------

type largeBoxAgent struct {
	cmd        *exec.Cmd
	began      time.Time
	listen     time.Duration // process start to the gate probe's first answer
	eventsPath string
	log        string
	exited     chan struct{} // closed once the process has been waited for
	exitErr    error         // Wait's answer, read after exited closes
	b          *largeBox
}

// startAgent starts the agent and returns it once the gate's probe is
// answered and its TLS listener accepts. The edge's -http and -https take
// an address, and :0 would bind, but the edge never says which port it
// got, so the test picks free ports and hands them over. A port taken in
// between makes the edge's listener fail and the agent exit; only that is
// retried on fresh ports (largeBoxRetryable). A deadline missed is the
// failure this test exists to catch, and is never retried.
func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent {
	t.Helper()
	for attempt := 1; ; attempt++ {
		b.httpAddr, b.httpsAddr = freeLoopbackAddr(t), freeLoopbackAddr(t)
		a := b.launchAgent(t)
		outcome, err := a.awaitListening()
		if outcome == largeBoxUp {
			outcome, err = a.awaitTLS()
		}
		if outcome == largeBoxUp {
			return a
		}
		agentLog, _ := os.ReadFile(a.log)
		if largeBoxRetryable(outcome, string(agentLog), b.httpAddr, b.httpsAddr) && attempt < largeBoxBindAttempts {
			t.Logf("attempt %d: %v; a port the test handed the agent was taken before it bound it, starting again on fresh ports", attempt, err)
			continue
		}
		t.Fatal(err)
	}
}

// largeBoxOutcome is how waiting for the agent to come up ended.
type largeBoxOutcome int

const (
	largeBoxUp     largeBoxOutcome = iota
	largeBoxLate                   // a deadline passed: a failure, never retried
	largeBoxExited                 // the agent exited before it was up
)

// largeBoxRetryable reports whether a start may be retried on fresh
// ports: only when the agent exited, and its log says the edge could not
// bind one of the addresses the test handed it because it was in use.
// Edge listener failures are logged as "edge: http listener <addr>: ..."
// (edge.RunWithActivatorReady).
func largeBoxRetryable(o largeBoxOutcome, agentLog string, addrs ...string) bool {
	if o != largeBoxExited {
		return false
	}
	for _, line := range strings.Split(agentLog, "\n") {
		if !strings.Contains(line, "address already in use") {
			continue
		}
		for _, addr := range addrs {
			if strings.Contains(line, " listener "+addr+": ") {
				return true
			}
		}
	}
	return false
}

func TestLargeBoxRetriesOnlyItsOwnBindFailure(t *testing.T) {
	const plain, tls = "127.0.0.1:40001", "127.0.0.1:40002"
	bind := "edge: https listener " + tls + ": listen tcp " + tls + ": bind: address already in use\n"
	for _, c := range []struct {
		name string
		o    largeBoxOutcome
		log  string
		want bool
	}{
		{"own bind failure", largeBoxExited, "edge: serving 300 route(s)\n" + bind, true},
		{"deadline missed, bind failure logged too", largeBoxLate, bind, false},
		{"exited for another reason", largeBoxExited, "edge: tunnels: invalid\n", false},
		{"someone else's address", largeBoxExited, "edge: http listener 127.0.0.1:1: listen tcp 127.0.0.1:1: bind: address already in use\n", false},
		{"in use, but not a listener line", largeBoxExited, "agent: something: address already in use\n", false},
	} {
		if got := largeBoxRetryable(c.o, c.log, plain, tls); got != c.want {
			t.Errorf("%s: retry = %v, want %v", c.name, got, c.want)
		}
	}
}

func (b *largeBox) launchAgent(t *testing.T) *largeBoxAgent {
	t.Helper()
	cfgDir := t.TempDir()
	cfg := largeBoxConfig{
		DataDir: b.dataDir, SocketDir: b.socketDir, StoreDir: b.storeDir, AcmeDir: b.acmeDir,
		Control: filepath.Join(b.socketDir, "control.sock"), Events: filepath.Join(cfgDir, "events"),
		HTTP: b.httpAddr, HTTPS: b.httpsAddr,
	}
	raw, err := json.Marshal(cfg)
	if err != nil {
		t.Fatal(err)
	}
	cfgPath := filepath.Join(cfgDir, "agent.json")
	if err := os.WriteFile(cfgPath, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	exe, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	a := &largeBoxAgent{eventsPath: cfg.Events, log: filepath.Join(cfgDir, "agent.log"), exited: make(chan struct{}), b: b}
	logFile, err := os.Create(a.log)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { logFile.Close() })
	a.cmd = exec.Command(exe, "-test.run=^TestStartupLargeBoxEndToEnd$", "-test.count=1")
	// A clean environment: nothing of the developer's or the runner's
	// (a deployment bucket, AWS credentials, CARLOS_*) reaches the agent.
	a.cmd.Env = []string{"HOME=" + b.dataDir, "TMPDIR=" + os.TempDir(), "CARLOS_DATA_DIR=" + b.dataDir, largeBoxAgentEnv + "=" + cfgPath}
	a.cmd.Stdout, a.cmd.Stderr = logFile, logFile
	a.began = time.Now()
	if err := a.cmd.Start(); err != nil {
		t.Fatal(err)
	}
	go func() {
		a.exitErr = a.cmd.Wait()
		close(a.exited)
	}()
	t.Cleanup(func() {
		select {
		case <-a.exited:
		default:
			_ = a.cmd.Process.Kill()
			<-a.exited
		}
		if t.Failed() {
			if out, err := os.ReadFile(a.log); err == nil {
				if len(out) > 16<<10 {
					out = out[len(out)-16<<10:]
				}
				t.Logf("agent log (tail):\n%s", out)
			}
		}
	})
	return a
}

// awaitListening probes as the update gate does (systemapply.go
// productionSysOps.probeEdge): GET / on the edge's plain-HTTP listener,
// three-second client timeout, no redirect followed, and any HTTP answer
// passes. It records how long after the process started the first answer
// came; an answer past largeBoxListenBound, or none by then, is late.
func (a *largeBoxAgent) awaitListening() (largeBoxOutcome, error) {
	probe := &http.Client{
		Timeout:       3 * time.Second,
		Transport:     &http.Transport{Proxy: nil, DisableKeepAlives: true},
		CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
	}
	deadline := a.began.Add(largeBoxListenBound)
	for {
		resp, err := probe.Get("http://" + a.b.httpAddr + "/")
		if err == nil {
			resp.Body.Close()
			a.listen = time.Since(a.began)
			if a.listen > largeBoxListenBound {
				return largeBoxLate, fmt.Errorf("gate probe answered %s after the agent started, over %s", a.listen, largeBoxListenBound)
			}
			return largeBoxUp, nil
		}
		if time.Now().After(deadline) {
			return largeBoxLate, fmt.Errorf("gate probe unanswered %s after the agent started: %v", time.Since(a.began), err)
		}
		select {
		case <-a.exited:
			return largeBoxExited, fmt.Errorf("agent exited before listening: %v", a.exitErr)
		default:
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// awaitTLS waits, boundedly, for the edge's TLS listener: the edge binds
// it in its own goroutine, beside the plain-HTTP one the gate probes, so
// it can come up a moment later. Not part of the gate's measurement.
func (a *largeBoxAgent) awaitTLS() (largeBoxOutcome, error) {
	deadline := time.Now().Add(largeBoxListenBound)
	for {
		c, err := net.DialTimeout("tcp", a.b.httpsAddr, time.Second)
		if err == nil {
			c.Close()
		}
		select {
		case <-a.exited:
			return largeBoxExited, fmt.Errorf("agent exited before its TLS listener was up: %v", a.exitErr)
		default:
		}
		if err == nil {
			return largeBoxUp, nil
		}
		if time.Now().After(deadline) {
			return largeBoxLate, fmt.Errorf("TLS listener not up %s after the gate probe: %v", largeBoxListenBound, err)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func (a *largeBoxAgent) signal(t *testing.T, sig os.Signal) {
	t.Helper()
	if err := a.cmd.Process.Signal(sig); err != nil {
		t.Fatal(err)
	}
}

func (a *largeBoxAgent) waitExit(t *testing.T, bound time.Duration) {
	t.Helper()
	select {
	case <-a.exited:
		if a.exitErr != nil {
			t.Fatalf("agent exit: %v", a.exitErr)
		}
	case <-time.After(bound):
		t.Fatalf("agent did not exit within %s of the signal", bound)
	}
}

// assertNeverStarted: startup never finished, so the activator never
// started and nothing reached the backend or ran.
func (a *largeBoxAgent) assertNeverStarted(t *testing.T) {
	t.Helper()
	for _, kind := range []string{"ready", "activator-start", "acquire", "start", "worker-start", "served"} {
		if evs := a.events(kind); len(evs) != 0 {
			t.Errorf("%s with startup unfinished: %v", kind, evs)
		}
	}
}

// assertAfter waits briefly for the event kind with args and fails unless
// it came after ref.
func (a *largeBoxAgent) assertAfter(t *testing.T, ref largeBoxEvent, kind string, args ...string) {
	t.Helper()
	e := a.waitEvent(t, 10*time.Second, kind, args...)
	if e.at.Before(ref.at) {
		t.Errorf("%s %v came %s before %s", kind, args, ref.at.Sub(e.at), ref.kind)
	}
}

// assertCleanShutdown reads the agent's own account of its shutdown: the
// startup work joined within runAgent's bound, and no goroutine of the
// startup sequence, the activator or the fake instances left after it.
func (a *largeBoxAgent) assertCleanShutdown(t *testing.T) {
	t.Helper()
	j, ok := a.event("joined")
	if !ok {
		t.Fatal("agent did not report its startup join")
	}
	took, err := time.ParseDuration(j.arg(1))
	if j.arg(0) != "true" || err != nil || took > largeBoxJoinBound {
		t.Fatalf("startup join: %v", j)
	}
	t.Logf("startup work joined %s after the edge returned", took)
	if _, ok := a.event("leak-free"); !ok {
		t.Fatal("goroutines of the startup sequence outlived shutdown; see the agent log")
	}
}

type largeBoxEvent struct {
	at   time.Time
	kind string
	args []string
}

func (e largeBoxEvent) arg(i int) string {
	if i < len(e.args) {
		return e.args[i]
	}
	return ""
}

func (a *largeBoxAgent) all() []largeBoxEvent {
	f, err := os.Open(a.eventsPath)
	if err != nil {
		return nil
	}
	defer f.Close()
	var out []largeBoxEvent
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		fields := strings.Fields(sc.Text())
		if len(fields) < 2 {
			continue
		}
		ns, err := strconv.ParseInt(fields[0], 10, 64)
		if err != nil {
			continue
		}
		out = append(out, largeBoxEvent{at: time.Unix(0, ns), kind: fields[1], args: fields[2:]})
	}
	return out
}

func (a *largeBoxAgent) events(kind string) []largeBoxEvent {
	var out []largeBoxEvent
	for _, e := range a.all() {
		if e.kind == kind {
			out = append(out, e)
		}
	}
	return out
}

func (a *largeBoxAgent) event(kind string, args ...string) (largeBoxEvent, bool) {
	for _, e := range a.events(kind) {
		match := true
		for i, want := range args {
			if e.arg(i) != want {
				match = false
			}
		}
		if match {
			return e, true
		}
	}
	return largeBoxEvent{}, false
}

func (a *largeBoxAgent) waitEvent(t *testing.T, bound time.Duration, kind string, args ...string) largeBoxEvent {
	t.Helper()
	deadline := time.Now().Add(bound)
	for {
		if e, ok := a.event(kind, args...); ok {
			return e
		}
		if time.Now().After(deadline) {
			t.Fatalf("agent did not report %s %v within %s", kind, args, bound)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// --- inside the agent process ---------------------------------------------------

type largeBoxEvents struct {
	mu sync.Mutex
	f  *os.File
}

func (e *largeBoxEvents) log(kind string, args ...any) {
	var b strings.Builder
	fmt.Fprintf(&b, "%d %s", time.Now().UnixNano(), kind)
	for _, a := range args {
		fmt.Fprintf(&b, " %v", a)
	}
	b.WriteByte('\n')
	e.mu.Lock()
	defer e.mu.Unlock()
	_, _ = e.f.WriteString(b.String())
}

// runLargeBoxAgent is the agent: runAgent's legacy-box startup over the
// fixture, then the edge, then runAgent's shutdown. It returns the exit
// code.
func runLargeBoxAgent(cfgPath string) int {
	raw, err := os.ReadFile(cfgPath)
	if err != nil {
		log.Printf("largebox agent: %v", err)
		return 2
	}
	var cfg largeBoxConfig
	if err := json.Unmarshal(raw, &cfg); err != nil {
		log.Printf("largebox agent: %v", err)
		return 2
	}
	f, err := os.OpenFile(cfg.Events, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
	if err != nil {
		log.Printf("largebox agent: %v", err)
		return 2
	}
	defer f.Close()
	ev := &largeBoxEvents{f: f}

	agentCtx, cancelAgent := context.WithCancel(context.Background())
	defer cancelAgent()
	startupCtx, cancelStartup := context.WithCancel(agentCtx)
	defer cancelStartup()

	dd := cfg.DataDir
	regPath := filepath.Join(dd, "registry.db")
	reg, err := registry.Open(regPath)
	if err != nil {
		log.Printf("largebox agent: %v", err)
		return 2
	}
	defer reg.Close()

	control := instancectl.NewClient(cfg.Control)
	nestedOK := nestedGate(func() error { return instancefs.SelfCheck(dd) }, instancefs.MountNoSymfollow, control.Ready)
	started := make(chan struct{})
	backend := &largeBoxBackend{ev: ev, started: started}
	svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg)
	backend.svc = svc
	svc.Routes = reg
	svc.Startup = started
	openStore := func(context.Context) (release.Store, error) { return release.NewDirStore(cfg.StoreDir), nil }

	startupDone := activatorStart{
		Ready: func() {
			ev.log("ready") // before the close, so no acquire can be stamped earlier
			close(started)
		},
		Sweep: func(ctx context.Context) error {
			sw, err := control.Sweep(ctx)
			if err != nil {
				log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
			} else if len(sw.Units) > 0 {
				log.Printf("agent: swept %d instance unit(s) left from a previous agent", len(sw.Units))
			}
			ev.log("swept", len(sw.Units), err == nil)
			return err
		},
		Recover: func(ctx context.Context, swept bool) *layout.Pass {
			pass := newLayoutPass(reg, svc, dd, nestedOK, swept)
			pass.SocketDir, pass.SocketsGID = cfg.SocketDir, -1 // the fixture's run dir; no carlos-sockets group
			intent := startupRecoveryIntent(ctx, dd, openStore, "largebox")
			recoverLayoutAtStart(ctx, pass, intent)
			ev.log("recovered", intent)
			return pass
		},
		Routes:  reg.List,
		Route:   reg.Get,
		Keys:    largeBoxKeys{ev},
		Slot:    svc.PrepareScheduledToken,
		Publish: armedLayoutPass.Store,
		// The checkpoint: when Start runs, every hold marker's host must
		// already be fenced again and every journal complete.
		Start: func() {
			held, _ := layout.HoldMarkers(dd)
			fenced := 0
			for _, h := range held {
				if svc.FenceGeneration(h) != 0 {
					fenced++
				}
			}
			journals, _ := layout.JournalHosts(dd)
			ev.log("activator-start", "held="+strconv.Itoa(len(held)), "fenced="+strconv.Itoa(fenced), "journals="+strconv.Itoa(len(journals)))
			svc.Start()
		},
	}.Run(startupCtx)
	go func() {
		<-startupDone
		ev.log("startup-done")
	}()
	// runAgent's worker boot pass, which waits on the same latch.
	go func() {
		select {
		case <-started:
		case <-startupCtx.Done():
			return
		}
		routes, err := reg.List()
		if err != nil {
			return
		}
		runWorkerBootPass(routes, func(rt registry.Route) error {
			err := svc.WakeWorker(rt)
			ev.log("worker-boot", rt.Host, err == nil)
			return err
		})
	}()

	// Admissions from inside the agent that do not pass runAgent's own
	// gates, so they meet the activator's latch on their own: a worker
	// wake as an operator command or the sidecar-wake leg makes it, and
	// one scheduler tick through the real runner and deliverer with a
	// delivery already due.
	if rt, err := reg.Get(largeBoxWorker); err == nil && rt != nil {
		go func() {
			ev.log("worker-direct-begin", rt.Host)
			err := svc.WakeWorker(*rt)
			ev.log("worker-direct", rt.Host, err == nil)
		}()
	}
	runner := &schedule.Runner{
		Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
		Now: time.Now, Timeout: largeBoxLatchBound + 15*time.Second, Box: "largebox", Build: "test",
	}
	runner.Tick(agentCtx) // delivers in the background, and returns
	ev.log("tick-done")

	edge.RunWithActivatorReady([]string{"-registry", regPath, "-acme-cache", cfg.AcmeDir, "-http", cfg.HTTP, "-https", cfg.HTTPS},
		largeBoxActivator{Service: svc, ev: ev}, edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes),
		func(e *edge.Edge) {
			reload := func() { _ = e.Reload() }
			layoutReload.Store(&reload)
		})

	// runAgent's shutdown, in its order.
	cancelStartup()
	joinCtx, joinCancel := context.WithTimeout(context.Background(), largeBoxJoinBound)
	began := time.Now()
	joined := joinAgentStartup(joinCtx, startupDone)
	joinCancel()
	ev.log("joined", joined, time.Since(began))
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	svc.Shutdown(ctx)
	cancel()
	cancelAgent()

	deadline := time.Now().Add(2 * time.Second)
	for {
		left := largeBoxStartupGoroutines()
		if len(left) == 0 {
			ev.log("leak-free")
			return 0
		}
		if time.Now().After(deadline) {
			log.Printf("largebox agent: %d goroutine(s) outlived shutdown:\n\n%s", len(left), strings.Join(left, "\n\n"))
			ev.log("leak", len(left))
			return 0
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// largeBoxStartupGoroutines lists every goroutine but this one that runs,
// or was started by, the startup sequence, the layout pass, the control
// client, the activator, the scheduler, or this agent's own fakes, and
// every server connection still open (a request held at the latch is
// answered only once the activator shuts down, after the edge has stopped
// waiting for it). The edge's process-lifetime loops (its registry watch,
// its SIGHUP reader) are not startup's and are left alone.
func largeBoxStartupGoroutines() []string {
	buf := make([]byte, 8<<20)
	n := runtime.Stack(buf, true)
	blocks := strings.Split(string(buf[:n]), "\n\n")
	markers := []string{
		"host.activatorStart", "host.startActivator", "host.backfillInstanceKeys", "host.keyBackfillInSlot",
		"host.recoverLayoutAtStart", "host.startupRecoveryIntent", "host.runWorkerBootPass",
		"host.runLargeBoxAgent", "host.(*largeBoxBackend)", "host.largeBoxActivator", "host.(*activatorDeliverer)",
		"layout.(*Pass)", "instancectl.(*Client)", "activator.(*Service)", "schedule.(*Runner)",
		"net/http.(*conn).serve",
	}
	var left []string
	for _, g := range blocks[1:] { // [0] is the calling goroutine
		for _, m := range markers {
			if strings.Contains(g, m) {
				left = append(left, g)
				break
			}
		}
	}
	return left
}

// largeBoxActivator is the activator as the edge sees it, reporting each
// instance request as it reaches it.
type largeBoxActivator struct {
	*activator.Service
	ev *largeBoxEvents
}

func (a largeBoxActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
	if rt.Kind == "instance" {
		a.ev.log("enter", rt.Host, rt.Socket)
	}
	return a.Service.ServeHTTP(w, r, rt, proxy)
}

// largeBoxBackend is activator.Backend with no S3 and no processes: a wake
// serves an app on the row's socket that names the host and the socket it
// was started at, and a worker reports ready at once, as a child does
// over the agent socket. It records every acquire, start and request
// served, and any acquire the startup latch should have prevented.
type largeBoxBackend struct {
	ev      *largeBoxEvents
	started <-chan struct{}
	svc     *activator.Service
}

func (b *largeBoxBackend) Acquire(_ context.Context, rt registry.Route) (activator.Lease, error) {
	select {
	case <-b.started:
	default:
		b.ev.log("violation", "acquire-before-ready", rt.Host)
	}
	b.ev.log("acquire", rt.Host)
	return nopLease{}, nil
}

func (b *largeBoxBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
func (b *largeBoxBackend) Seed(context.Context, registry.Route) error            { return nil }

func (b *largeBoxBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) {
	b.ev.log("start", rt.Host, rt.Socket)
	if err := os.MkdirAll(filepath.Dir(rt.Socket), 0o750); err != nil {
		return nil, "", err
	}
	ln, err := net.Listen("unix", rt.Socket)
	if err != nil {
		return nil, "", err
	}
	body := "host=" + rt.Host + " socket=" + rt.Socket
	p := &largeBoxProc{done: make(chan struct{})}
	p.srv = &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		b.ev.log("served", rt.Host, r.URL.Path)
		io.WriteString(w, body)
	})}
	go func() { defer close(p.done); _ = p.srv.Serve(ln) }()
	return p, "", nil
}

func (b *largeBoxBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
func (b *largeBoxBackend) HasSidecar(registry.Route) bool                         { return false }
func (b *largeBoxBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
	return nil, errors.New("largeBoxBackend: no co-spawned sidecars on this box")
}

func (b *largeBoxBackend) StartWorker(_ context.Context, rt registry.Route, _ map[string]string) (activator.Proc, string, error) {
	b.ev.log("worker-start", rt.Host)
	go b.svc.NoteWorkerReady(rt.Host, "")
	return &largeBoxWorkerProc{}, "", nil
}

type largeBoxProc struct {
	srv    *http.Server
	done   chan struct{}
	exited atomic.Bool
}

func (p *largeBoxProc) Stop(context.Context) error {
	_ = p.srv.Close()
	<-p.done
	p.exited.Store(true)
	return nil
}

func (p *largeBoxProc) Exited() bool { return p.exited.Load() }

type largeBoxWorkerProc struct{ exited atomic.Bool }

func (p *largeBoxWorkerProc) Stop(context.Context) error { p.exited.Store(true); return nil }
func (p *largeBoxWorkerProc) Exited() bool               { return p.exited.Load() }

// largeBoxKeys is the exec backend's key writer as the key pass sees it:
// every row it is handed already has its key.
type largeBoxKeys struct{ ev *largeBoxEvents }

func (k largeBoxKeys) BackfillInstanceKey(_ context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) {
	k.ev.log("key", rt.Host)
	return activator.InstanceKeyPresent, nil
}