package activator

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// launchBackend is the swap fixture's fakeSwapBackend (lease, restore,
// raise, switch) with StartStaged on a REAL ExecBackend, flat or nested, so
// a staged spawn runs the production preparation: the admin token, the
// config read, the control socket's bind and, nested, the launch file and
// the root start call. The activator's row is mapped onto the real
// backend's row (rt) of the chosen layout. Its KillGen stops the
// generation's handle, as execStagedBackend's does.
type launchBackend struct {
	*fakeSwapBackend
	eb  *ExecBackend
	rt  registry.Route
	ctl *exeControl // nested only: what root was asked to start

	lmu     sync.Mutex
	entered []uint64 // generation numbers that reached the real backend
	procs   map[uint64]Proc
	builds  []string // the builds the real backend returned a process for
}

func (b *launchBackend) StartStaged(ctx context.Context, _ registry.Route, g Gen) (Gen, Proc, error) {
	b.lmu.Lock()
	b.entered = append(b.entered, g.N)
	b.lmu.Unlock()
	got, p, err := b.eb.StartStaged(ctx, b.rt, g)
	if p != nil {
		b.lmu.Lock()
		b.procs[g.N] = p
		b.builds = append(b.builds, g.Build)
		b.lmu.Unlock()
	}
	return got, p, err
}

func (b *launchBackend) KillGen(ctx context.Context, _ registry.Route, n uint64) error {
	b.lmu.Lock()
	p := b.procs[n]
	b.lmu.Unlock()
	if p == nil {
		return nil
	}
	_ = p.Stop(ctx)
	if !p.Exited() {
		return ErrStillRunning
	}
	return nil
}

func (b *launchBackend) gens() []uint64 {
	b.lmu.Lock()
	defer b.lmu.Unlock()
	return slices.Clone(b.entered)
}

func (b *launchBackend) stopAll() {
	b.lmu.Lock()
	procs := make([]Proc, 0, len(b.procs))
	for _, p := range b.procs {
		procs = append(procs, p)
	}
	b.lmu.Unlock()
	for _, p := range procs {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		_ = p.Stop(ctx)
		cancel()
	}
}

// launched is the evidence that a build's process was started: the real
// backend returned a process for it; flat, the marker its script writes the
// moment it runs; nested, root was asked to start the unit with it.
func (b *launchBackend) launched(t *testing.T, id release.StagedIdentity, marker string) bool {
	t.Helper()
	b.lmu.Lock()
	started := slices.Contains(b.builds, id.Version)
	b.lmu.Unlock()
	if started {
		return true
	}
	if b.ctl != nil {
		b.ctl.mu.Lock()
		defer b.ctl.mu.Unlock()
		for _, x := range b.ctl.got {
			if x.Path == id.StagedPath {
				return true
			}
		}
		return false
	}
	_, err := os.Stat(marker)
	return err == nil
}

// prepGone checks that what generation n's preparation made is gone after
// a refusal at the launch: its control socket, and nested, its launch file.
func (b *launchBackend) prepGone(t *testing.T, n uint64) {
	t.Helper()
	var sock string
	p := b.eb.pathsFor(b.rt)
	switch p.layout {
	case LayoutNested:
		sock = nestedGenSocketIn(p.runDir, n)
		launch := filepath.Join(p.dataDir, instancectl.GenLaunchFileName(n))
		if _, err := os.Stat(launch); !os.IsNotExist(err) {
			t.Fatalf("generation %d's launch file %s after a refusal at the launch: %v, want it removed", n, launch, err)
		}
	default:
		sock = b.eb.flatGenSocket(b.rt.Host, n)
	}
	if _, err := os.Stat(sock + ".control"); !os.IsNotExist(err) {
		t.Fatalf("generation %d's control socket after a refusal at the launch: %v, want it unbound", n, err)
	}
}

// newLaunchBackend builds the real backend for layout and wraps fb.
func newLaunchBackend(t *testing.T, layout string, fb *fakeSwapBackend) *launchBackend {
	t.Helper()
	b := &launchBackend{fakeSwapBackend: fb, procs: map[uint64]Proc{}}
	switch layout {
	case "flat":
		eb := testExecBackend(t)
		eb.RunDirOverride = t.TempDir()
		host := "launch.example.com"
		stageLive(t, eb, host, "v0")
		b.eb, b.rt = eb, registry.Route{Host: host, Kind: "instance", Hibernate: true,
			DBPath: filepath.Join(eb.DataDir, host+".db"), Socket: filepath.Join(eb.RunDirOverride, host+".sock")}
	case "nested":
		eb, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{})
		eb.GenContainment = true
		b.ctl = &exeControl{}
		eb.Control = b.ctl
		b.eb, b.rt = eb, rt
	default:
		t.Fatalf("layout %q", layout)
	}
	if got, _ := b.eb.Layout(b.rt); (layout == "flat") != (got == LayoutFlat) || (layout == "nested") != (got == LayoutNested) {
		t.Fatalf("fixture layout %v, want %s", got, layout)
	}
	t.Cleanup(b.stopAll)
	return b
}

// markerScript is a staged build that writes marker the moment it runs.
func markerScript(t *testing.T, dir, ver, marker string) release.StagedIdentity {
	t.Helper()
	return stagedScript(t, dir, ver, "#!/bin/sh\necho ran > '"+marker+"'\nexec sleep 30\n")
}

// onCall runs f on the call numbered want (counting from 1) and on no
// other.
func onCall(want int32, f func()) func(string) {
	var n atomic.Int32
	return func(string) {
		if n.Add(1) == want {
			f()
		}
	}
}

// Final review F1: every staged instance spawn's composed gate is asked at
// the backend's final launch point, AFTER the backend's own preparation:
// flat, after the token, the config read and the control socket's bind,
// right before the process starts; nested, after the launch file, right
// before root is asked to start the unit. A lease lost, a fence set or the
// attempt superseded during that preparation starts nothing, removes what
// the preparation made, records no execution evidence and is abandonment,
// never a failure of the build. The four callers: a planned wake's B, its
// fallback, a hold's pinned build and an overlap successor, on both
// layouts. Before the fix the gate ran only before StartStaged, and each of
// these launched.
func TestTheExecGateIsAskedAtTheBackendsLaunchPoint(t *testing.T) {
	for _, layout := range []string{"flat", "nested"} {
		t.Run(layout+"/planned B, lease lost", func(t *testing.T) {
			b := newLaunchBackend(t, layout, &fakeSwapBackend{fakeBackend: &fakeBackend{}})
			dir := t.TempDir()
			mark := filepath.Join(t.TempDir(), "v2.ran")
			next := markerScript(t, dir, "v2", mark)
			good := markerScript(t, dir, "v1", filepath.Join(t.TempDir(), "v1.ran"))
			a, evs := launchActivator(t, b, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
			b.eb.launchHook = onCall(1, func() { b.lastRep().setErr(errors.New("lease renewal failed")) })
			_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
			if !errors.Is(err, errSwapLeaseLost) || classifyStep(nil, err) == stepFailed {
				t.Fatalf("wake = %v, want the lost lease, abandonment", err)
			}
			gens := b.gens()
			if len(gens) != 1 {
				t.Fatalf("the backend was entered for generations %v, want B's alone (the refusal is inside it)", gens)
			}
			if b.launched(t, next, mark) {
				t.Fatal("B was launched after the lease was lost during its preparation")
			}
			b.prepGone(t, gens[0])
			time.Sleep(50 * time.Millisecond)
			if k := evs.kinds(); len(k) != 0 {
				t.Fatalf("events %v: a refusal at the launch was recorded", k)
			}
		})
		t.Run(layout+"/planned fallback, fenced", func(t *testing.T) {
			defer func(k time.Duration) { firstWakeKillBound = k }(firstWakeKillBound)
			firstWakeKillBound = 500 * time.Millisecond
			b := newLaunchBackend(t, layout, &fakeSwapBackend{fakeBackend: &fakeBackend{}})
			dir := t.TempDir()
			markB, markGood := filepath.Join(t.TempDir(), "v2.ran"), filepath.Join(t.TempDir(), "v1.ran")
			next, good := markerScript(t, dir, "v2", markB), markerScript(t, dir, "v1", markGood)
			a, evs := launchActivator(t, b, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
			const host = "one.example.com"
			a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
			b.eb.launchHook = onCall(2, func() {
				a.mu.Lock()
				a.fences = map[string]uint64{host: 1}
				a.mu.Unlock()
			})
			if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
				t.Fatal("served after the fallback was refused")
			}
			if gens := b.gens(); len(gens) != 2 {
				t.Fatalf("the backend was entered for generations %v, want B's and the fallback's", gens)
			}
			waitFor(t, "B's launch evidence", func() bool { return b.launched(t, next, markB) })
			time.Sleep(100 * time.Millisecond)
			if b.launched(t, good, markGood) {
				t.Fatal("the good build was launched on a host fenced during its preparation")
			}
			b.prepGone(t, b.gens()[1])
			// B ran and failed: that is recorded. The refused fallback is not
			// a fallback that served.
			evs.wait(t, GenFailed)
			if k := evs.kinds(); slices.Contains(k, GenFellBack) {
				t.Fatalf("events %v: a fallback refused at its launch was reported as served", k)
			}
		})
		t.Run(layout+"/hold's pinned build, lease lost", func(t *testing.T) {
			f := newSwapFixture(t)
			b := newLaunchBackend(t, layout, f.b)
			f.a.backend = b
			sink := &lockedSink{}
			f.a.Logs = sink
			mark := filepath.Join(t.TempDir(), "v2.ran")
			id := markerScript(t, t.TempDir(), "v2", mark)
			at := f.attempt(ModeHold)
			at.StagedPath, at.Digest = id.StagedPath, id.Digest
			b.eb.launchHook = onCall(1, func() { f.b.lastRep().setErr(errors.New("lease renewal failed")) })
			res := f.a.Swap(context.Background(), at, f.newest)
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapLeaseLost.Error()) {
				t.Fatalf("Swap = %+v, want abandoned: the lost lease", res)
			}
			gens := b.gens()
			if len(gens) != 1 {
				t.Fatalf("the backend was entered for generations %v, want the pin's alone", gens)
			}
			if b.launched(t, id, mark) {
				t.Fatal("the hold's build was launched after the lease was lost during its preparation")
			}
			b.prepGone(t, gens[0])
			if got := sink.errors(); len(got) != 0 {
				t.Fatalf("member error events %q for an abandoned hold", got)
			}
		})
		t.Run(layout+"/overlap successor, superseded", func(t *testing.T) {
			f := newSwapFixture(t)
			b := newLaunchBackend(t, layout, f.b)
			f.a.backend = b
			mark := filepath.Join(t.TempDir(), "v2.ran")
			id := markerScript(t, t.TempDir(), "v2", mark)
			at := f.attempt(ModeOverlap)
			at.StagedPath, at.Digest = id.StagedPath, id.Digest
			b.eb.launchHook = onCall(1, func() { f.rev.Add(1) })
			res := f.a.Swap(context.Background(), at, f.newest)
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
				t.Fatalf("Swap = %+v, want abandoned: superseded", res)
			}
			gens := b.gens()
			if len(gens) != 1 {
				t.Fatalf("the backend was entered for generations %v, want the successor's alone", gens)
			}
			if b.launched(t, id, mark) {
				t.Fatal("the successor was launched after its attempt was superseded during its preparation")
			}
			b.prepGone(t, gens[0])
			if g, members := f.a.genStateForTest(f.host); g.N != f.incumbent.N || len(members) != 0 {
				t.Fatalf("current %+v, members %v: want the incumbent serving and an empty operation set", g, members)
			}
		})
	}
}

// launchActivator is planActivator over a launchBackend.
func launchActivator(t *testing.T, b *launchBackend, plan WakePlan) (*Service, *genLog) {
	t.Helper()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
	a.gens = true
	evs := &genLog{}
	a.OnGen = evs.add
	a.Planner = func(context.Context, registry.Route) WakePlan { return plan }
	a.genHealth = func(context.Context, Gen) error { return nil }
	t.Cleanup(func() { a.Shutdown(context.Background()) })
	return a, evs
}

// Merge triage T9: on a first wake whose health window is short (under the
// verdict floor, so B would not be tried and the good build would boot in
// its place), a refusal by B's own gate is abandonment: nothing boots, B is
// not reported untried, and nothing is recorded. verdictFloorGate asks the
// gate first, so a build held since the plan (or a lost lease, a fence) is
// never mistaken for a lack of time. Here B's selection check passes at
// checkPlan and refuses at B's exec. With the floor asked first, the good
// build booted untried and the wake reported GenHealthy.
func TestAGateRefusalOnAShortWindowAbandonsRatherThanGoingUntried(t *testing.T) {
	shrinkVerdictFloor(t, time.Hour) // every window is under the floor
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, evs := planActivator(t, WakePlan{})
	var asked atomic.Int32
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good,
			Check: func(build string) error {
				if build == "v2" && asked.Add(1) >= 2 {
					return errors.New("v2 failed on a sibling")
				}
				return nil
			}}
	}
	_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if !errors.Is(err, errPlanStale) || errors.Is(err, errNoTimeForVerdict) {
		t.Fatalf("wake = %v, want the gate's refusal (errPlanStale), never a lack of time", err)
	}
	if got := classifyStep(nil, err); got != stepStale {
		t.Fatalf("classifyStep = %v, want stepStale: abandonment", got)
	}
	if gens := b.started(); len(gens) != 0 {
		t.Fatalf("spawned %+v: the good build booted untried after B's gate refused", gens)
	}
	time.Sleep(50 * time.Millisecond)
	if k := evs.kinds(); len(k) != 0 {
		t.Fatalf("events %v: an abandoned short-window wake reported something (an untried B reads as GenHealthy)", k)
	}
}

// The exec gate rides on the context through SplitBackend to its exec half,
// which asks it at its launch (genbackend.go's contract): a refusal there
// reaches the caller unchanged.
func TestSplitBackendCarriesTheExecGate(t *testing.T) {
	exec := &recordingGenExec{fakeBackend: &fakeBackend{}}
	s := &SplitBackend{Exec: exec, Unit: &UnitBackend{}}
	refused := errors.New("refused at the launch")
	ctx := withExecGate(context.Background(), func() error { return refused })
	_, p, err := s.StartStaged(ctx, registry.Route{Host: "e.example.com"}, Gen{N: 7, StagedPath: "/opt/carlos/versions/v2/app"})
	if !errors.Is(err, refused) || p != nil || exec.started != 0 {
		t.Fatalf("StartStaged = %v, %v (started %d): want the gate's refusal and nothing started", p, err, exec.started)
	}
}

// gatedCgroups gives b a cgroup manager on a temp-dir subtree (the
// fixture delegatedBackend uses) whose log calls flip once a limit is
// written: a log line naming a memory.max is the manager's preparation
// (the parent's tier, the child's limits) under way.
func gatedCgroups(t *testing.T, b *ExecBackend, host string, flip func()) string {
	t.Helper()
	root := t.TempDir()
	for _, f := range []string{"cgroup.procs", "cgroup.subtree_control"} {
		if err := os.WriteFile(filepath.Join(root, f), nil, 0o644); err != nil {
			t.Fatal(err)
		}
	}
	var once sync.Once
	m, err := cgroup.SetupAt(root, func(format string, args ...any) {
		t.Logf(format, args...)
		if flip != nil && strings.Contains(format, "memory.max=") {
			once.Do(flip)
		}
	})
	if err != nil {
		t.Fatal(err)
	}
	parent := filepath.Join(root, cgroup.ExecName(host))
	if err := os.MkdirAll(parent, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(parent, "cgroup.controllers"), []byte("cpu io memory pids\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	b.Cgroups = m
	return parent
}

// workerGateService is a Service over a REAL ExecBackend worker spawn
// (realWorkerBackend) whose cgroup manager is gatedCgroups' temp-dir
// subtree: a worker start runs the production preparation (the build's
// publication, the cgroup and its limits) and its start attempts. It
// returns the worker's row, the file its build writes its $0 to the moment
// it runs, and the host's cgroup directory.
func workerGateService(t *testing.T, flip func()) (*Service, *ExecBackend, registry.Route, string, string) {
	t.Helper()
	eb := testExecBackend(t)
	host := "bot.example.com"
	out := filepath.Join(t.TempDir(), "argv0")
	stageArgv0Worker(t, eb, host, "v1", out)
	dir := gatedCgroups(t, eb, host, flip)
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, &realWorkerBackend{fakeBackend: &fakeBackend{}, exec: eb}, reg)
	a.GenerationsFile = release.GenerationsPath(t.TempDir())
	t.Cleanup(func() { a.Shutdown(context.Background()) })
	return a, eb, workerRoute(host), out, dir
}

// Astra R3 §5 item 4 (ruling 206, "park worker launch hook"): a worker
// start's ownership gate is the cgroup manager's before-start hook, as an
// instance's is, so it is asked AFTER the cgroup preparation (the host's
// cgroup and its limits) and before the start. Ownership lost during that
// preparation starts nothing: no process, no $0 written, nothing placed in
// the cgroup, the gate's own error back, and the build's pending
// publication dropped. Before the fix the gate rode the before-exec hook,
// asked once before the cgroup preparation, and the worker started.
func TestAWorkerStartIsGatedAfterItsCgroupPreparation(t *testing.T) {
	t.Parallel()
	var refuse atomic.Bool
	a, _, rt, out, dir := workerGateService(t, func() { refuse.Store(true) })
	lost := errors.New("lease renewal failed")
	var asked atomic.Int32
	gate := func() error {
		asked.Add(1)
		if refuse.Load() {
			return lost
		}
		return nil
	}
	p, _, _, err := a.startWorker(context.Background(), rt, nil, nil, gate)
	if p != nil {
		_ = p.Stop(context.Background())
	}
	if !errors.Is(err, lost) || p != nil {
		t.Fatalf("startWorker = %v, %v; want no process and the gate's refusal", p, err)
	}
	if !refuse.Load() || asked.Load() == 0 {
		t.Fatalf("the cgroup preparation ran: %v; the gate was asked %d times", refuse.Load(), asked.Load())
	}
	time.Sleep(200 * time.Millisecond)
	if _, err := os.Stat(out); err == nil {
		t.Fatal("the worker ran after its ownership was lost during the cgroup preparation")
	}
	if b, err := os.ReadFile(filepath.Join(dir, "cgroup.procs")); err == nil && len(b) != 0 {
		t.Fatalf("the worker's cgroup was given pid %s after a refusal at its start", b)
	}
	if got := readGenerations(t, a.GenerationsFile); len(got[rt.Host]) != 0 {
		t.Fatalf("generations file %v names a build that never ran", got)
	}
}

// The same gate is the manager's before hook at the PID-move fallback a
// flat start takes after a failed placement (cgroup.StartGated's
// contract), so ownership lost by then starts nothing. The temp-dir subtree
// is no cgroupfs: the shim's checks refuse the placement before any fork,
// so the gate (asked after a placement) is first asked at the fallback, and
// refuses there. That a placement which got past the gate and then failed
// asks it again at the fallback is internal/cgroup's
// TestFlatPlacementFailureFallsBackToPIDMove: the hook passed through is
// this same gate. Before the clone3 start left (amadan #272) this asked
// twice, at the clone3 start and its retry.
func TestAWorkerStartFallbackAsksTheGate(t *testing.T) {
	t.Parallel()
	a, eb, rt, out, dir := workerGateService(t, nil)
	var asked atomic.Int32
	eb.launchHook = func(string) { asked.Add(1) }
	superseded := errors.New("the attempt was superseded")
	gate := func() error {
		if asked.Load() >= 1 {
			return superseded
		}
		return nil
	}
	p, _, _, err := a.startWorker(context.Background(), rt, nil, nil, gate)
	if p != nil {
		_ = p.Stop(context.Background())
	}
	if n := asked.Load(); n != 1 {
		t.Fatalf("the gate was asked %d times at the starts, want once: the PID-move fallback's", n)
	}
	if b, err := os.ReadFile(filepath.Join(dir, "cgroup.procs")); err == nil && len(b) != 0 {
		t.Fatalf("the worker's cgroup was given pid %s after a refusal at its start", b)
	}
	if !errors.Is(err, superseded) || p != nil {
		t.Fatalf("startWorker = %v, %v; want no process and the gate's refusal", p, err)
	}
	time.Sleep(200 * time.Millisecond)
	if _, err := os.Stat(out); err == nil {
		t.Fatal("the retry started the worker after its gate refused")
	}
}

// Final review wave 2 (F1 completion, Astra round 2 finding 1): a flat
// staged spawn's exec gate is the cgroup manager's before-start hook, so
// it is asked AFTER the cgroup preparation (the parent's tier, the child
// cgroup and its limits, their log lines) and the placement, before EACH
// exec, a flat start's PID-move fallback included. Ownership lost during
// that preparation, or between a failed placement and its fallback,
// starts nothing: no process, no marker, the control socket unbound, the
// generation's child cgroup removed, abandonment with no execution
// evidence. Before the fix the gate ran once, before the cgroup
// preparation, and each of these launched.
func TestTheExecGateIsAskedAtEachCgroupStart(t *testing.T) {
	t.Run("overlap successor, superseded during the cgroup preparation", func(t *testing.T) {
		f := newSwapFixture(t)
		b := newLaunchBackend(t, "flat", f.b)
		b.eb.GenContainment = true
		parent := gatedCgroups(t, b.eb, b.rt.Host, func() { f.rev.Add(1) })
		cgroup.FakePlacementForTest(t) // a temp dir is no cgroupfs, and a generation is never moved by PID
		f.a.backend = b
		mark := filepath.Join(t.TempDir(), "v2.ran")
		id := markerScript(t, t.TempDir(), "v2", mark)
		at := f.attempt(ModeOverlap)
		at.StagedPath, at.Digest = id.StagedPath, id.Digest
		res := f.a.Swap(context.Background(), at, f.newest)
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) || res.Executed {
			t.Fatalf("Swap = %+v, want abandoned (superseded), nothing executed", res)
		}
		gens := b.gens()
		if len(gens) != 1 {
			t.Fatalf("the backend was entered for generations %v, want the successor's alone", gens)
		}
		time.Sleep(100 * time.Millisecond)
		if b.launched(t, id, mark) {
			t.Fatal("the successor was launched after its attempt was superseded during the cgroup preparation")
		}
		b.prepGone(t, gens[0])
		// The preparation did run (the flip is its log line), and nothing was
		// placed in the child it made. (On cgroupfs the empty child is
		// removed; in this fixture its limit files are ordinary files, so the
		// rmdir is refused, as CleanupHost's comment explains.)
		if b, err := os.ReadFile(filepath.Join(parent, cgroup.GenChild(gens[0]), "cgroup.procs")); err == nil && len(b) != 0 {
			t.Fatalf("generation %d's child cgroup was given pid %s after a refusal at its start", gens[0], b)
		}
	})
	t.Run("hold's pinned build, lease lost during the cgroup preparation", func(t *testing.T) {
		f := newSwapFixture(t)
		b := newLaunchBackend(t, "flat", f.b)
		gatedCgroups(t, b.eb, b.rt.Host, func() { f.b.lastRep().setErr(errors.New("lease renewal failed")) })
		f.a.backend = b
		mark := filepath.Join(t.TempDir(), "v2.ran")
		id := markerScript(t, t.TempDir(), "v2", mark)
		at := f.attempt(ModeHold)
		at.StagedPath, at.Digest = id.StagedPath, id.Digest
		res := f.a.Swap(context.Background(), at, f.newest)
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapLeaseLost.Error()) || res.Executed {
			t.Fatalf("Swap = %+v, want abandoned (the lost lease), nothing executed", res)
		}
		time.Sleep(100 * time.Millisecond)
		if b.launched(t, id, mark) {
			t.Fatal("the hold's build was launched after the lease was lost during the cgroup preparation")
		}
		b.prepGone(t, b.gens()[0])
	})
	t.Run("overlap successor, a failed placement is not retried by PID move", func(t *testing.T) {
		f := newSwapFixture(t)
		b := newLaunchBackend(t, "flat", f.b)
		b.eb.GenContainment = true
		// The temp-dir subtree is no cgroupfs: the shim's checks refuse the
		// placement before any fork, so the gate, asked only after a
		// placement (amadan #272, Ruling 13), is never asked. A generation is
		// never started and then moved by PID (readiness round 2 Task 7), so
		// there is no retry and no gate at all: the start is refused as the
		// box's condition.
		parent := gatedCgroups(t, b.eb, b.rt.Host, nil)
		var asked atomic.Int32
		b.eb.launchHook = func(string) { asked.Add(1) }
		f.a.backend = b
		mark := filepath.Join(t.TempDir(), "v2.ran")
		id := markerScript(t, t.TempDir(), "v2", mark)
		at := f.attempt(ModeOverlap)
		at.StagedPath, at.Digest = id.StagedPath, id.Digest
		res := f.a.Swap(context.Background(), at, f.newest)
		if n := asked.Load(); n != 0 {
			t.Fatalf("the gate was asked %d times, want never: no placement, and no retry", n)
		}
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, cgroup.ErrLimitNotApplied.Error()) || res.Executed {
			t.Fatalf("Swap = %+v, want abandoned (limits not applied), nothing executed", res)
		}
		time.Sleep(100 * time.Millisecond)
		if b.launched(t, id, mark) {
			t.Fatal("the successor was launched after its placement failed")
		}
		gens := b.gens()
		b.prepGone(t, gens[0])
		// No PID move. In this fixture the child's limit files are ordinary
		// files, so the refusal's rmdir leaves the child in place: it must
		// still be there with its memory.max written (else the check below
		// proves nothing), and with no cgroup.procs at all, which is the
		// file a start-and-move writes the PID into.
		child := filepath.Join(parent, cgroup.GenChild(gens[0]))
		if _, err := os.Stat(filepath.Join(child, "memory.max")); err != nil {
			t.Fatalf("generation %d's child cgroup has no limit written (%v): the start never reached its placement", gens[0], err)
		}
		if got, err := os.ReadFile(filepath.Join(child, "cgroup.procs")); !os.IsNotExist(err) {
			t.Fatalf("generation %d's child cgroup has a cgroup.procs (%q, %v): a PID move wrote it", gens[0], got, err)
		}
	})
}

// liveGate is a LiveExecGate for tests: it records every build it is asked
// about and refuses the builds in refuse, as a declined box's failure holds
// would (internal/host).
type liveGate struct {
	mu     sync.Mutex
	asked  []string
	refuse map[string]bool
}

var errLiveHeld = errors.New("a failure record holds this build")

func (g *liveGate) gate(_ registry.Route, build string) error {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.asked = append(g.asked, build)
	if g.refuse[build] {
		return fmt.Errorf("%w: %s", errLiveHeld, build)
	}
	return nil
}

func (g *liveGate) builds() []string {
	g.mu.Lock()
	defer g.mu.Unlock()
	return slices.Clone(g.asked)
}

// repointLive points b's live/<host> at bin, as root's adopt pass does.
func repointLive(t *testing.T, b *ExecBackend, host, bin string) {
	t.Helper()
	tmp := b.liveLink(host) + ".tmp"
	os.Remove(tmp)
	if err := os.Symlink(bin, tmp); err != nil {
		t.Error(err)
		return
	}
	if err := os.Rename(tmp, b.liveLink(host)); err != nil {
		t.Error(err)
	}
}

// flatLiveFixture is a flat row on a real ExecBackend whose live link names
// v1, with v2 staged beside it. Each build writes its own $0 to out the
// moment it runs, so a test can tell which bytes were exec'd and by which
// path.
func flatLiveFixture(t *testing.T) (b *ExecBackend, rt registry.Route, v1, v2, out string) {
	t.Helper()
	b = testExecBackend(t)
	b.RunDirOverride = t.TempDir()
	host := "live.example.com"
	out = filepath.Join(t.TempDir(), "argv0")
	v2 = stageArgv0Worker(t, b, host, "v2", out)
	v1 = stageArgv0Worker(t, b, host, "v1", out)
	rt = registry.Route{Host: host, Kind: "instance", Hibernate: true,
		DBPath: filepath.Join(b.DataDir, host+".db"), Socket: filepath.Join(b.RunDirOverride, host+".sock")}
	return b, rt, v1, v2, out
}

// Task 9 (Astra R1 #6, the R2/R3 declined residuals; plan review #6): on a
// box that declined CARLOS_SWAP=overlap and honours failure records, the
// flat live-link spawn resolves live/<host> ONCE, asks LiveExecGate about
// that build at the launch (after the cgroup's preparation), and execs the
// staged path it resolved, so a repoint landing between the gate and the
// exec cannot start a build the gate never saw. A refusal starts nothing.
// With the gate nil the spawn execs the live link exactly as before.
func TestALiveExecGateGatesTheFlatLiveLinkSpawn(t *testing.T) {
	t.Parallel()
	t.Run("gated build is the exec'd build", func(t *testing.T) {
		b, rt, v1, v2, out := flatLiveFixture(t)
		g := &liveGate{refuse: map[string]bool{"v2": true}}
		b.LiveExecGate = g.gate
		// Root repoints at held v2 after the gate has been asked about v1 and
		// before the exec.
		b.launchHook = onCall(1, func() { repointLive(t, b, rt.Host, v2) })
		p, ver, err := b.spawnChildFrom(context.Background(), rt, rt.Socket, 0, "", cgroup.Limit{})
		if err != nil {
			t.Fatalf("spawn = %v", err)
		}
		defer p.Stop(context.Background())
		if got := strings.TrimSpace(waitForFile(t, out, 5*time.Second)); got != v1 {
			t.Fatalf("$0 = %q, want the staged v1 the gate was asked about (%s)", got, v1)
		}
		if ver != "v1" || !slices.Equal(g.builds(), []string{"v1"}) {
			t.Fatalf("spawn version %q, gate asked %v; want v1 and [v1]", ver, g.builds())
		}
	})
	t.Run("a held build starts nothing", func(t *testing.T) {
		b, rt, _, v2, out := flatLiveFixture(t)
		repointLive(t, b, rt.Host, v2)
		g := &liveGate{refuse: map[string]bool{"v2": true}}
		b.LiveExecGate = g.gate
		p, _, err := b.spawnChildFrom(context.Background(), rt, rt.Socket, 0, "", cgroup.Limit{})
		if p != nil {
			_ = p.Stop(context.Background())
		}
		if !errors.Is(err, errLiveHeld) || p != nil {
			t.Fatalf("spawn = %v, %v; want no process and the gate's refusal", p, err)
		}
		time.Sleep(200 * time.Millisecond)
		if _, err := os.Stat(out); err == nil {
			t.Fatal("held v2 ran")
		}
		if _, err := os.Stat(rt.Socket + ".control"); !os.IsNotExist(err) {
			t.Fatalf("control socket after a refusal: %v, want it unbound", err)
		}
		if !slices.Equal(g.builds(), []string{"v2"}) {
			t.Fatalf("gate asked %v, want [v2]", g.builds())
		}
	})
	t.Run("no gate is today's spawn", func(t *testing.T) {
		b, rt, _, _, out := flatLiveFixture(t)
		p, ver, err := b.spawnChildFrom(context.Background(), rt, rt.Socket, 0, "", cgroup.Limit{})
		if err != nil {
			t.Fatalf("spawn = %v", err)
		}
		defer p.Stop(context.Background())
		if got := strings.TrimSpace(waitForFile(t, out, 5*time.Second)); got != b.liveLink(rt.Host) || ver != "v1" {
			t.Fatalf("$0 = %q, version %q; want today's exec of %s, v1", got, ver, b.liveLink(rt.Host))
		}
	})
}

// Task 9: a nested live-link start asks LiveExecGate about the build the
// live link names, right before root is asked to start the unit. A refusal
// asks root for nothing.
func TestALiveExecGateGatesTheNestedLiveLinkStart(t *testing.T) {
	ctl := &fakeControl{inv: "inv1"}
	b, rt := nestedBackend(t, ctl, &fakeUnit{})
	g := &liveGate{refuse: map[string]bool{"v1": true}}
	b.LiveExecGate = g.gate
	_, _, err := b.Start(context.Background(), rt)
	if !errors.Is(err, errLiveHeld) {
		t.Fatalf("Start = %v, want the gate's refusal", err)
	}
	if strings.Contains(ctl.joined(), "start ") {
		t.Fatalf("root was asked to start the unit after the gate refused: %q", ctl.joined())
	}
	if !slices.Equal(g.builds(), []string{"v1"}) {
		t.Fatalf("gate asked %v, want [v1]", g.builds())
	}
	if _, err := os.Stat(rt.Socket + ".control"); !os.IsNotExist(err) {
		t.Fatalf("control socket after a refusal: %v, want it unbound", err)
	}
}

// Task 9: a worker spawned from the live link (onto nil) asks LiveExecGate
// about the build it resolved, before its publication and its exec; a
// refusal publishes and runs nothing. A recycle onto the captured build
// (onto set) is not a move onto the live link, and is not asked.
func TestALiveExecGateGatesAWorkerSpawnFromTheLiveLink(t *testing.T) {
	t.Parallel()
	b := testExecBackend(t)
	host := "bot.example.com"
	out := filepath.Join(t.TempDir(), "argv0")
	v1 := stageArgv0Worker(t, b, host, "v1", out)
	stageArgv0Worker(t, b, host, "v2", out)
	g := &liveGate{refuse: map[string]bool{"v2": true}}
	b.LiveExecGate = g.gate
	published := 0
	ctx := withBeforeExec(context.Background(), func(release.StagedIdentity) error { published++; return nil })
	p, _, err := b.StartWorkerStaged(ctx, workerRoute(host), nil, nil)
	if p != nil {
		_ = p.Stop(context.Background())
	}
	if !errors.Is(err, errLiveHeld) || p != nil || published != 0 {
		t.Fatalf("StartWorkerStaged = %v, %v (published %d); want the gate's refusal, nothing published or run", p, err, published)
	}
	time.Sleep(200 * time.Millisecond)
	if _, err := os.Stat(out); err == nil {
		t.Fatal("held v2 ran as the worker")
	}
	sum, err := release.FileSHA256(v1)
	if err != nil {
		t.Fatal(err)
	}
	onto := release.StagedIdentity{Version: "v1", StagedPath: v1, Digest: sum}
	p, id, err := b.StartWorkerStaged(ctx, workerRoute(host), nil, &onto)
	if err != nil {
		t.Fatalf("recycle onto the captured v1 = %v", err)
	}
	defer p.Stop(context.Background())
	if id != onto || strings.TrimSpace(waitForFile(t, out, 5*time.Second)) != v1 {
		t.Fatalf("recycled %+v, want the captured v1", id)
	}
	if !slices.Equal(g.builds(), []string{"v2"}) {
		t.Fatalf("gate asked %v, want only the live-link spawn's [v2]", g.builds())
	}
}

// Task 9 (brief (b)): a declined worker whose live link moves to a held
// build is not stopped by the sweep's converge. The activator asks the gate
// before it retires the old child, so a refusal leaves the worker running
// its current build instead of stopping it into a refused spawn.
func TestADeclinedWorkerIsNotConvergedOntoAHeldBuild(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	g := &liveGate{refuse: map[string]bool{"v2": true}}
	a.SetLiveExecGate(g.gate)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	a.sweepOnce()
	time.Sleep(100 * time.Millisecond)
	if calls := b.calls(); len(calls) != 1 || a.entryRespawning(host) {
		t.Fatalf("spawns %v (respawning %v): the worker was converged onto held v2", calls, a.entryRespawning(host))
	}
	if b.lastWorker().Exited() {
		t.Fatal("the converge stopped the worker serving v1")
	}
	if got := a.entrySpawnVersion(host); got != "v1" {
		t.Fatalf("spawn version %q, want v1", got)
	}
	if !slices.Contains(g.builds(), "v2") {
		t.Fatalf("gate asked %v, want v2 asked", g.builds())
	}
	// Once the link names a build the gate allows, the converge runs as
	// before.
	ver.Store("v3")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the converge onto v3", func() bool { return a.entrySpawnVersion(host) == "v3" && !a.entryRespawning(host) })
}

// Task 9 (brief (c)): a declined worker that crashes while its live link
// names a held build respawns its captured build, however long the drift
// lasts: the recycle's converge is refused before anything is stopped.
func TestADeclinedWorkerCrashRecyclesItsCapturedBuildWhenTheLiveLinkIsHeld(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	a.SetLiveExecGate((&liveGate{refuse: map[string]bool{"v2": true}}).gate)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	for i := 1; i <= 5; i++ {
		b.lastWorker().kill()
		clk.advance(sidecarRestartMax + time.Second)
		a.sweepOnce()
		waitForRestart(t, a, b.fakeBackend, host, i+1)
	}
	for i, c := range b.calls()[1:] {
		if c == nil || *c != fakeIdentity("v1") {
			t.Fatalf("respawn %d asked for %v; want the captured v1 every time", i+1, c)
		}
	}
	if got := a.entrySpawnVersion(host); got != "v1" {
		t.Fatalf("spawn version %q, want v1", got)
	}
}

// Task 9 (brief (c)): a worker with no captured build (a backend without
// the staged capability) whose live link names a held build is not
// respawned from it: a crashed one stays down until a wake starts it, and
// one whose max_run falls due keeps running.
func TestADeclinedUncapturedWorkerIsNotRespawnedOntoAHeldBuild(t *testing.T) {
	t.Parallel()
	t.Run("crashed", func(t *testing.T) {
		a, b, clk, _ := testActivator(t, time.Hour)
		a.liveVer = func(string) (string, error) { return "v2", nil }
		a.SetLiveExecGate((&liveGate{refuse: map[string]bool{"v2": true}}).gate)
		host := "bot.x.internal"
		if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
			t.Fatalf("WakeWorker: %v", err)
		}
		b.lastWorker().kill()
		for range 3 {
			a.sweepOnce()
			clk.advance(sidecarRestartMax + time.Second)
		}
		time.Sleep(100 * time.Millisecond)
		if n := b.workerStarts(); n != 1 || a.entryRespawning(host) {
			t.Fatalf("workerStarts = %d (respawning %v): respawned onto held v2", n, a.entryRespawning(host))
		}
		if !a.Owns(host) {
			t.Fatal("the entry went: it stays, down, for a wake to start")
		}
	})
	t.Run("max_run", func(t *testing.T) {
		a, b, clk, _ := testActivator(t, time.Hour)
		a.liveVer = func(string) (string, error) { return "v2", nil }
		a.SetLiveExecGate((&liveGate{refuse: map[string]bool{"v2": true}}).gate)
		host := "bot.x.internal"
		rt := alwaysOnWorkerRoute(host)
		rt.MaxRunSeconds = 30
		if err := a.WakeWorker(rt); err != nil {
			t.Fatalf("WakeWorker: %v", err)
		}
		for range 3 {
			clk.advance(31 * time.Second)
			a.sweepOnce()
		}
		time.Sleep(100 * time.Millisecond)
		if n := b.workerStarts(); n != 1 || a.entryRespawning(host) {
			t.Fatalf("workerStarts = %d (respawning %v): recycled onto held v2", n, a.entryRespawning(host))
		}
		if b.lastWorker().Exited() {
			t.Fatal("the running child was stopped by a recycle the gate refused")
		}
	})
}

// Task 9 fix round 1 (ruling 1): the legacy co-spawned sidecar execs the
// live link too, and its wake tail, late attachment and the sweep's
// supervision respawn it while the instance keeps running. On a declined box
// with the gate set, root's repoint of the link at a held build must not
// start that build as the sidecar: the start resolves the link (no hash),
// asks the gate about that build, and execs the path it resolved. A refusal
// starts nothing (the supervisor's backoff keeps it down). With the gate nil
// the sidecar execs the link, as before.
func TestALiveExecGateGatesTheLegacySidecarStart(t *testing.T) {
	t.Parallel()
	sidecarEnv := func(t *testing.T, b *ExecBackend, rt registry.Route) {
		t.Helper()
		if err := os.WriteFile(b.pathsFor(rt).sidecarEnv, nil, 0o600); err != nil {
			t.Fatal(err)
		}
	}
	t.Run("flat, gated build is the exec'd build", func(t *testing.T) {
		b, rt, v1, _, out := flatLiveFixture(t)
		sidecarEnv(t, b, rt)
		g := &liveGate{refuse: map[string]bool{"v2": true}}
		b.LiveExecGate = g.gate
		p, err := b.StartSidecar(context.Background(), rt)
		if err != nil {
			t.Fatalf("StartSidecar = %v", err)
		}
		defer p.Stop(context.Background())
		if got := strings.TrimSpace(waitForFile(t, out, 5*time.Second)); got != v1 {
			t.Fatalf("sidecar $0 = %q, want the staged v1 the gate was asked about (%s)", got, v1)
		}
		if !slices.Equal(g.builds(), []string{"v1"}) {
			t.Fatalf("gate asked %v, want [v1]", g.builds())
		}
	})
	t.Run("flat, a held build starts nothing", func(t *testing.T) {
		b, rt, _, v2, out := flatLiveFixture(t)
		sidecarEnv(t, b, rt)
		repointLive(t, b, rt.Host, v2)
		b.LiveExecGate = (&liveGate{refuse: map[string]bool{"v2": true}}).gate
		p, err := b.StartSidecar(context.Background(), rt)
		if p != nil {
			_ = p.Stop(context.Background())
		}
		if !errors.Is(err, errLiveHeld) || p != nil {
			t.Fatalf("StartSidecar = %v, %v; want no process and the gate's refusal", p, err)
		}
		time.Sleep(200 * time.Millisecond)
		if _, err := os.Stat(out); err == nil {
			t.Fatal("held v2 ran as the sidecar")
		}
	})
	t.Run("flat, no gate is today's", func(t *testing.T) {
		b, rt, _, _, out := flatLiveFixture(t)
		sidecarEnv(t, b, rt)
		p, err := b.StartSidecar(context.Background(), rt)
		if err != nil {
			t.Fatalf("StartSidecar = %v", err)
		}
		defer p.Stop(context.Background())
		if got := strings.TrimSpace(waitForFile(t, out, 5*time.Second)); got != b.liveLink(rt.Host) {
			t.Fatalf("sidecar $0 = %q, want today's exec of %s", got, b.liveLink(rt.Host))
		}
	})
	t.Run("nested, a held build asks root for nothing", func(t *testing.T) {
		ctl := &fakeControl{inv: "inv1"}
		b, rt := nestedBackend(t, ctl, &fakeUnit{})
		g := &liveGate{refuse: map[string]bool{"v1": true}}
		b.LiveExecGate = g.gate
		p, err := b.StartSidecar(context.Background(), rt)
		if !errors.Is(err, errLiveHeld) || p != nil {
			t.Fatalf("StartSidecar = %v, %v; want the gate's refusal", p, err)
		}
		if strings.Contains(ctl.joined(), "start ") {
			t.Fatalf("root was asked to start the sidecar after the gate refused: %q", ctl.joined())
		}
		if !slices.Equal(g.builds(), []string{"v1"}) {
			t.Fatalf("gate asked %v, want [v1]", g.builds())
		}
	})
}

// Task 9 fix round 1 (ruling 2): the sweep's drift cycle of an awake
// instance, or a hibernating worker, stops it so its next wake runs the new
// build. On a declined box a held build's next wake is refused, so the gate
// is asked first: a refusal keeps it running and spends no budget. A build
// the gate allows cycles as before.
func TestADeclinedSweepDoesNotCycleOntoAHeldBuild(t *testing.T) {
	t.Parallel()
	t.Run("instance", func(t *testing.T) {
		a, b, clk, _ := testActivator(t, time.Hour)
		var ver atomic.Value
		ver.Store("v1")
		a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
		b.execVerFn = func() string { return ver.Load().(string) }
		a.SetLiveExecGate((&liveGate{refuse: map[string]bool{"v2": true}}).gate)
		release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
		if err != nil {
			t.Fatalf("beginRequest: %v", err)
		}
		release()
		ver.Store("v2")
		a.sweepOnce()
		clk.advance(ConvergeGrace + time.Second)
		a.sweepOnce()
		a.sweepOnce()
		time.Sleep(50 * time.Millisecond)
		if a.entryCount() != 1 || b.lastProc().isStopped() {
			t.Fatal("the sweep stopped an instance serving v1 into a wake onto held v2")
		}
		ver.Store("v3")
		a.sweepOnce()
		clk.advance(ConvergeGrace + time.Second)
		a.sweepOnce()
		waitFor(t, "the cycle onto v3", func() bool { return a.entryCount() == 0 })
	})
	t.Run("hibernating worker", func(t *testing.T) {
		a, b, clk, _ := testActivator(t, time.Hour)
		var ver atomic.Value
		ver.Store("v1")
		a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
		b.execVerFn = func() string { return ver.Load().(string) }
		a.SetLiveExecGate((&liveGate{refuse: map[string]bool{"v2": true}}).gate)
		host := "bot.x.internal"
		if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
			t.Fatalf("WakeWorker: %v", err)
		}
		ver.Store("v2")
		a.sweepOnce()
		clk.advance(ConvergeGrace + time.Second)
		a.sweepOnce()
		a.sweepOnce()
		time.Sleep(50 * time.Millisecond)
		if !a.Owns(host) || b.lastWorker().Exited() {
			t.Fatal("the sweep stopped a worker serving v1 into a wake onto held v2")
		}
	})
}