package activator

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"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"
)

// Settings spec §4.2, Task 5c: every unclaimed wake settles ONE executable
// identity after the restore, records it before the seed and the spawn,
// and runs only that file; legacy sidecars run their generation's recorded
// file by path whatever CARLOS_SWAP says.

// identityFake is fakeBackend's identity capability. A test opts in
// (setIdentity); until then identityFor answers nil, so every test written
// before Task 5c keeps today's Start and Seed and their events.
type identityFake struct {
	on bool
	// live is the build live/<host> names, "" when the link is missing;
	// LiveIdentity then materializes materializeTo first.
	live          string
	materializeTo string
	materializes  int
	captures      int
	// unversioned makes the live link name a file outside the version
	// store (Version "").
	unversioned bool
	// onLiveIdentity, when set, runs inside LiveIdentity before it resolves
	// the link (under the fake's lock): a repoint that lands between a held
	// wake's wait and its capture.
	onLiveIdentity func(*identityFake)
	exe            []exeCall
	last           release.StagedIdentity
	// buildIDs is every build BuildIdentity was asked to name, and
	// buildIDErr its answer when set (a staged copy that cannot be read).
	buildIDs   []string
	buildIDErr error
}

// BuildIdentity is the fake's buildIdentifier: build's staged file as
// fakeIdentity names it, or buildIDErr.
func (b *fakeBackend) BuildIdentity(_ registry.Route, build string) (release.StagedIdentity, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ident.buildIDs = append(b.ident.buildIDs, build)
	if b.ident.buildIDErr != nil {
		return release.StagedIdentity{}, b.ident.buildIDErr
	}
	return fakeIdentity(build), nil
}

func (b *fakeBackend) buildIdentities() []string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return slices.Clone(b.ident.buildIDs)
}

type exeCall struct {
	n      uint64
	perGen bool
	id     release.StagedIdentity
}

const unversionedPath = "/srv/hand-placed/app"

func (b *fakeBackend) setIdentity(live string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ident.on, b.ident.live = true, live
}

func (b *fakeBackend) setLive(v string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ident.live = v
}

func (b *fakeBackend) identityFor(registry.Route) identityBackend {
	b.mu.Lock()
	defer b.mu.Unlock()
	if !b.ident.on {
		return nil
	}
	return b
}

// LiveVersion makes New wire a.liveVer to the fake's link. "" (nothing
// opted in) is exactly the stub New used before.
func (b *fakeBackend) LiveVersion(string) (string, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.ident.live, nil
}

func (b *fakeBackend) LiveIdentity(_ context.Context, rt registry.Route) (release.StagedIdentity, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if f := b.ident.onLiveIdentity; f != nil {
		f(&b.ident)
	}
	if b.ident.unversioned {
		b.ident.captures++
		return release.StagedIdentity{StagedPath: unversionedPath, Digest: "sha-of-hand-placed"}, nil
	}
	if b.ident.live == "" {
		if b.ident.materializeTo == "" {
			return release.StagedIdentity{}, fmt.Errorf("no live binary for %s", rt.Host)
		}
		b.ident.live = b.ident.materializeTo
		b.ident.materializes++
	}
	b.ident.captures++
	return fakeIdentity(b.ident.live), nil
}

func (b *fakeBackend) StartGenExe(ctx context.Context, rt registry.Route, n uint64, perGen bool, id release.StagedIdentity) (Proc, Gen, error) {
	// The exec seam, as the real backend's: the exec gate and the wake's
	// pending record (amadan #272), before anything starts.
	if err := runExecGate(ctx); err != nil {
		return nil, Gen{}, err
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.starts++
	b.order = append(b.order, "start "+id.Version)
	b.startedRoutes = append(b.startedRoutes, rt)
	b.ident.exe = append(b.ident.exe, exeCall{n: n, perGen: perGen, id: id})
	b.ident.last = id
	sock := rt.Socket
	if perGen {
		sock = fmt.Sprintf("%s.g%d", rt.Socket, n)
	}
	g := Gen{N: n, Socket: sock, Control: sock + ".control"}
	if b.startErr != nil {
		return nil, g, b.startErr
	}
	proc := &fakeProc{label: "instance", rec: b, stuck: b.procStuck}
	b.procs = append(b.procs, proc)
	g.Build, g.StagedPath, g.Digest, g.Demand = id.Version, id.StagedPath, id.Digest, id.Demand
	return proc, g, nil
}

func (b *fakeBackend) SeedExe(ctx context.Context, rt registry.Route, id release.StagedIdentity) error {
	if _, err := runPendingRecord(ctx); err != nil {
		return err
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.seeds++
	b.order = append(b.order, "seed "+id.Version)
	b.seedRoutes = append(b.seedRoutes, rt)
	b.ident.last = id
	return b.seedErr
}

var _ identityBackend = (*fakeBackend)(nil)

// liveStarts counts Start calls: the live-link spawn the identity path
// replaces.
func (b *fakeBackend) liveStarts() int {
	n := 0
	for _, ev := range b.events() {
		if ev == "start:instance" {
			n++
		}
	}
	return n
}

// lastExeVersion is the build of the last file StartGenExe or SeedExe ran.
func (b *fakeBackend) lastExeVersion() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.ident.last.Version
}

// lastExe is the last StartGenExe call.
func (b *fakeBackend) lastExe() (exeCall, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.ident.exe) == 0 {
		return exeCall{}, false
	}
	return b.ident.exe[len(b.ident.exe)-1], true
}

func (b *fakeBackend) identityCounts() (captures, materializes int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.ident.captures, b.ident.materializes
}

// identityActivator is testActivator with the identity capability on and
// live/<host> naming live.
func identityActivator(t *testing.T, live string) (*Service, *fakeBackend) {
	t.Helper()
	a, b, _, _ := testActivator(t, time.Hour)
	b.setIdentity(live)
	return a, b
}

// identityRoute is testRoute with its socket and database in the test's
// own directory: a flag-on wake links the canonical path for real.
func identityRoute(t *testing.T, host string) registry.Route {
	t.Helper()
	rt := testRoute(host)
	dir := t.TempDir()
	rt.Socket, rt.DBPath = filepath.Join(dir, "app.sock"), filepath.Join(dir, "app.db")
	return rt
}

func seeded(rt registry.Route) registry.Route {
	rt.Seed = `["seed"]`
	return rt
}

// recordInto is a BeforeExec that writes "record <build>" into b's event
// log before l's own record, so a test reads one ordered log.
func recordInto(b *fakeBackend, l *execLog) func(context.Context, string, string) error {
	return func(ctx context.Context, host, build string) error {
		b.record("record " + build)
		return l.hook(ctx, host, build)
	}
}

// identityEvents is the record, seed and instance start events in b's log,
// in order: "record <v>", "seed <v>" and "start <v>" for the identity
// path, "seed" and "start:instance" for today's.
func identityEvents(b *fakeBackend) []string {
	var out []string
	for _, ev := range b.events() {
		if strings.HasPrefix(ev, "record ") || strings.HasPrefix(ev, "seed") || strings.HasPrefix(ev, "start ") || ev == "start:instance" {
			out = append(out, ev)
		}
	}
	return out
}

func (a *Service) entryGen(host string) (Gen, bool) {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	if e == nil || e.state != actAwake {
		return Gen{}, false
	}
	return e.gen, true
}

// --- the wake ---

func TestTheBuildStateIsWrittenBeforeTheWake(t *testing.T) {
	t.Parallel()
	a, b := identityActivator(t, "v1")
	l := &execLog{spawns: b.startCount}
	a.BeforeExec = l.hook
	rt := identityRoute(t, "one.example.com")
	done, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatal(err)
	}
	done()
	calls, at, locs := l.seen()
	if len(calls) != 1 || calls[0] != "one.example.com@v1" || at[0] != 0 {
		t.Fatalf("records %v at spawn counts %v, want one.example.com@v1 before the spawn", calls, at)
	}
	if locs[0] != (BuildStateLocation{Bucket: "test-bucket", Prefix: HibernatePrefix("", rt.Host)}) {
		t.Fatalf("location = %+v, want the wake's lease", locs[0])
	}
	x, ok := b.lastExe()
	if b.startCount() != 1 || b.liveStarts() != 0 || !ok || x.id != fakeIdentity("v1") || x.perGen || x.n == 0 {
		t.Fatalf("starts %d, live-link starts %d, exe %+v: want one flag-off start of the recorded v1", b.startCount(), b.liveStarts(), x)
	}
	if g, _ := a.entryGen(rt.Host); g.StagedPath != fakeIdentity("v1").StagedPath || g.Build != "v1" || g.N != x.n {
		t.Fatalf("serving generation %+v, want the recorded file's", g)
	}
}

func TestAFailedWriteBlocksTheSpawn(t *testing.T) {
	t.Parallel()
	a, b := identityActivator(t, "v1")
	b.restoreFresh = true
	a.BeforeExec = (&execLog{fail: failAll}).hook
	_, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com")))
	if !errors.Is(err, errBuildStateUnwritten) {
		t.Fatalf("err = %v, want errBuildStateUnwritten", err)
	}
	if b.startCount() != 0 || b.seedCount() != 0 {
		t.Fatalf("spawned %d, seeded %d with no build state", b.startCount(), b.seedCount())
	}
	waitFor(t, "the lease to go back", func() bool { return b.releaseCount() == 1 })
}

// Review Focus 1, Astra I1 #5: root repoints the live link while the build
// state is being written. The wake still execs exactly the file it
// recorded, never the link, and records once.
func TestTheWakeExecsTheIdentityItRecorded(t *testing.T) {
	a, b := identityActivator(t, "v1")
	b.restoreFresh = true
	l := &execLog{}
	a.BeforeExec = func(ctx context.Context, host, build string) error {
		b.setLive("v2") // root's repoint lands mid-write
		return l.hook(ctx, host, build)
	}
	done, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com")))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if calls, _, _ := l.seen(); len(calls) != 1 || calls[0] != "one.example.com@v1" {
		t.Fatalf("calls = %v; want one record of v1", calls)
	}
	if got := identityEvents(b); !slices.Equal(got, []string{"seed v1", "start v1"}) {
		t.Fatalf("events %v, want the recorded v1 seeded and started by path", got)
	}
	if c, _ := b.identityCounts(); b.liveStarts() != 0 || c != 1 {
		t.Fatalf("live-link starts %d, identity captures %d: want one capture and no live-link exec", b.liveStarts(), c)
	}
}

// Astra I1 #5: the seed runs the recorded identity, after the record.
func TestTheSeedRunsTheRecordedIdentity(t *testing.T) {
	a, b := identityActivator(t, "v1")
	b.restoreFresh = true
	l := &execLog{}
	a.BeforeExec = recordInto(b, l)
	done, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com")))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if got := identityEvents(b); !slices.Equal(got, []string{"record v1", "seed v1", "start v1"}) {
		t.Fatalf("events = %v; want record, seed and start, all v1, in that order", got)
	}
}

// Astra I1 #5: a planned wake records its boot build before the seed, seeds
// that build (never the live link), and startPinnedAt does not write it
// again.
func TestAPlannedWakeRecordsBeforeTheSeed(t *testing.T) {
	t.Parallel()
	next := stagedID(t, t.TempDir(), "v3")
	a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap})
	b.fakeBackend.setIdentity("v1")
	b.restoreFresh = true
	l := &execLog{}
	a.BeforeExec = recordInto(b.fakeBackend, l)
	done, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com")))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if got := identityEvents(b.fakeBackend); !slices.Equal(got, []string{"record v3", "seed v3"}) {
		t.Fatalf("events = %v; want the planned build recorded before its seed", got)
	}
	if calls, _, _ := l.seen(); len(calls) != 1 {
		t.Fatalf("calls = %v; startPinnedAt re-recorded the build the wake already recorded", calls)
	}
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v3" {
		t.Fatalf("spawned %+v, want the planned v3", gens)
	}
	if c, _ := b.identityCounts(); c != 0 {
		t.Fatalf("%d live-link captures: a planned wake runs its plan", c)
	}
}

// Ruling R-REPLAN: step 2b asks a planned wake's selection gate before it
// chooses the identity, so a replan lands first and the seed and the boot
// run one build, recorded once.
func TestAReplanBeforeTheSeedSeedsAndBootsOneBuild(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	stale, next := stagedID(t, dir, "v2"), stagedID(t, dir, "v3")
	a, b, _ := planActivator(t, WakePlan{})
	b.fakeBackend.setIdentity("v1")
	b.restoreFresh = true
	var plans atomic.Int32
	a.Planner = func(context.Context, registry.Route) WakePlan {
		if plans.Add(1) == 1 {
			return WakePlan{Boot: stale, Mode: ModeOverlap, Check: func(string) error { return errors.New("superseded") }}
		}
		return WakePlan{Boot: next, Mode: ModeOverlap}
	}
	l := &execLog{}
	a.BeforeExec = recordInto(b.fakeBackend, l)
	done, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com")))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if got := identityEvents(b.fakeBackend); !slices.Equal(got, []string{"record v3", "seed v3"}) {
		t.Fatalf("events %v, want the replanned v3 recorded and seeded", got)
	}
	if calls, _, _ := l.seen(); !slices.Equal(calls, []string{"one.example.com@v3"}) {
		t.Fatalf("records %v, want v3 once", calls)
	}
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v3" {
		t.Fatalf("spawned %+v, want the replanned v3", gens)
	}
}

// D6: a replan that only the exec's own selection gate makes (after step
// 2b settled the first plan's boot) is recorded by startPinnedAt before it
// execs. The first plan's boot never ran, so its pending record is never
// written (amadan #272).
func TestAReplanAfterTheRecordRecordsTheNewBuild(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	stale, next := stagedID(t, dir, "v2"), stagedID(t, dir, "v3")
	a, b, _ := planActivator(t, WakePlan{})
	b.fakeBackend.setIdentity("v1")
	a.Planner = laterReplan(stale, next)
	l := &execLog{spawns: func() int { return len(b.started()) }}
	a.BeforeExec = l.hook
	done, err := a.beginRequest(context.Background(), identityRoute(t, "one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	calls, at, _ := l.seen()
	if !slices.Equal(calls, []string{"one.example.com@v3"}) || at[0] != 0 {
		t.Fatalf("records %v at spawn counts %v, want only v3, before the replanned exec", calls, at)
	}
	if gens := b.started(); len(gens) != 1 || gens[0].Build != "v3" {
		t.Fatalf("spawned %+v, want the replanned v3", gens)
	}
}

// laterReplan plans stale first, whose selection gate passes once (step
// 2b) and refuses after (the exec's), and next on the replan.
func laterReplan(stale, next release.StagedIdentity) func(context.Context, registry.Route) WakePlan {
	var plans, checks atomic.Int32
	return func(context.Context, registry.Route) WakePlan {
		if plans.Add(1) == 1 {
			return WakePlan{Boot: stale, Mode: ModeOverlap, Check: func(string) error {
				if checks.Add(1) > 1 {
					return errors.New("superseded")
				}
				return nil
			}}
		}
		return WakePlan{Boot: next, Mode: ModeOverlap}
	}
}

// No hook (a box with no replica bucket): the wake still runs the file it
// captured, by path, and records nothing.
func TestANilHookRunsTheCapturedFileAndRecordsNothing(t *testing.T) {
	a, b := identityActivator(t, "v1")
	done, err := a.beginRequest(context.Background(), identityRoute(t, "one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if b.startCount() != 1 || b.lastExeVersion() != "v1" || b.liveStarts() != 0 {
		t.Fatalf("starts %d of %q (live-link %d), want the captured v1 by path", b.startCount(), b.lastExeVersion(), b.liveStarts())
	}
}

// Rulings R-EXEC-NOVERSION and -2: a live link outside the version store
// names no build. With no hook the wake is today's, in every layout: the
// live-link seed, start and sidecar, nothing published or recorded. With a
// hook it is refused.
func TestAnUnversionedLinkWithNoHookWakesAsToday(t *testing.T) {
	t.Parallel()
	a, b := sidecarActivator(t, "")
	b.fakeBackend.setIdentity("")
	b.mu.Lock()
	b.ident.unversioned = true
	b.mu.Unlock()
	b.restoreFresh = true
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	done, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "two.example.com")))
	if err != nil {
		t.Fatalf("an unversioned link with no hook: %v", err)
	}
	done()
	if got := identityEvents(b.fakeBackend); !slices.Equal(got, []string{"seed", "start:instance"}) {
		t.Fatalf("events %v, want today's live-link seed and start", got)
	}
	if _, ok := b.lastExe(); ok || len(b.sidecarsOn()) != 0 || b.sidecarStartCount() != 1 {
		t.Fatalf("exe starts %v, generation sidecars %d, live-link sidecars %d; want today's wake", ok, len(b.sidecarsOn()), b.sidecarStartCount())
	}
	if g, err := release.ReadGenerations(a.GenerationsFile); err == nil && g != nil && len(g.Paths["two.example.com"]) != 0 {
		t.Fatalf("published %v for a file no build names", g.Paths)
	}

	a, b = sidecarActivator(t, "")
	b.fakeBackend.setIdentity("")
	b.mu.Lock()
	b.ident.unversioned = true
	b.mu.Unlock()
	a.BeforeExec = (&execLog{}).hook
	if _, err := a.beginRequest(context.Background(), identityRoute(t, "two.example.com")); !errors.Is(err, errBuildStateUnwritten) {
		t.Fatalf("err = %v, want an unversioned build refused while a build state is kept", err)
	}
	if b.startCount() != 0 || b.sidecarStartCount() != 0 {
		t.Fatalf("%d starts, %d sidecars of a build no build state names", b.startCount(), b.sidecarStartCount())
	}
}

// rootLikeControl refuses a staged executable with no version, as root's
// validateStagedExe answers 403, so a wake that sends one fails here as it
// would on a box.
type rootLikeControl struct{ identityControl }

var errRootRefusesUnversioned = errors.New("instance-control: 403: the staged executable names no version")

func (c rootLikeControl) StartExe(ctx context.Context, host string, role instancectl.Role, max int64, x instancectl.StagedExe) (string, error) {
	if x.Version == "" {
		c.rec("refused startexe")
		return "", errRootRefusesUnversioned
	}
	return c.identityControl.StartExe(ctx, host, role, max, x)
}
func (c rootLikeControl) SeedExe(ctx context.Context, host string, max int64, x instancectl.StagedExe) (instancectl.SeedResponse, error) {
	if x.Version == "" {
		c.rec("refused seedexe")
		return instancectl.SeedResponse{}, errRootRefusesUnversioned
	}
	return c.identityControl.SeedExe(ctx, host, max, x)
}
func (c rootLikeControl) StartGenExe(ctx context.Context, host string, n uint64, max int64, x instancectl.StagedExe) (string, error) {
	if x.Version == "" {
		c.rec("refused startgenexe")
		return "", errRootRefusesUnversioned
	}
	return c.identityControl.StartGenExe(ctx, host, n, max, x)
}
func (c rootLikeControl) StartSidecarExe(ctx context.Context, host string, max int64, x instancectl.StagedExe) (string, error) {
	if x.Version == "" {
		c.rec("refused startsidecarexe")
		return "", errRootRefusesUnversioned
	}
	return c.identityControl.StartSidecarExe(ctx, host, max, x)
}

// leaselessExec is a real ExecBackend with the bucket's lease and restore
// faked, so a whole doWake runs over its real nested start and seed.
type leaselessExec struct {
	*ExecBackend
	fb *fakeBackend
}

func (b *leaselessExec) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
	return b.fb.Acquire(ctx, rt)
}
func (b *leaselessExec) Restore(context.Context, registry.Route) (bool, error) { return true, nil }

// Ruling R-EXEC-NOVERSION-2 on the real nested backend: an unversioned live
// link with no hook seeds and starts through root's plain verbs, as today,
// where the staged verbs would name no version and root would refuse them.
func TestANestedUnversionedLinkWithNoHookWakesAsToday(t *testing.T) {
	ctl := &fakeControl{inv: "inv1", seed: instancectl.SeedResponse{InvocationID: "s1", Result: "success", Status: "0", Code: "exited"}}
	unit := &fakeUnit{}
	eb, rt := nestedBackend(t, ctl, unit)
	eb.Control = rootLikeControl{identityControl{ctl}}
	hand := filepath.Join(t.TempDir(), "app")
	if err := os.WriteFile(hand, []byte("#!/bin/sh\n"), 0o755); err != nil {
		t.Fatal(err)
	}
	repointLiveLink(t, eb, rt.Host, hand)
	if id, err := eb.LiveIdentity(context.Background(), rt); err != nil || id.Version != "" {
		t.Fatalf("setup: live identity %+v, %v; want an unversioned file", id, err)
	}
	rt.Seed = `["seed"]`
	// The seed unit's cgroup is empty after it (fakeUnit's zero state); the
	// run unit comes up inside root's start, where the app would bind.
	var stopServing func()
	ctl.onStart = func() {
		unit.set("active", "inv1", true, true)
		stopServing = serveOK(t, rt.Socket)
	}
	defer func() {
		if stopServing != nil {
			stopServing()
		}
	}()
	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}, &leaselessExec{ExecBackend: eb, fb: &fakeBackend{}}, reg)
	done, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatalf("a nested unversioned link with no hook did not wake: %v\nroot was asked:\n%s", err, ctl.joined())
	}
	done()
	got := ctl.joined()
	if !strings.Contains(got, "seed "+rt.Host) || !strings.Contains(got, "start "+rt.Host+" run") || strings.Contains(got, "exe") {
		t.Fatalf("root was asked:\n%s\nwant today's plain seed and run start", got)
	}
	unit.set("inactive", "inv1", false, false)
}

// A refused record starts nothing on any wake kind: no seed, no instance,
// no worker, no planned build, and the member reads the fence's wording.
func TestARefusingHookStartsNothing(t *testing.T) {
	t.Parallel()
	a, b := identityActivator(t, "v1")
	b.restoreFresh, b.sidecarStaged = true, true
	a.BeforeExec = (&execLog{fail: failAll}).hook
	_, err := a.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com")))
	if !errors.Is(err, errBuildStateUnwritten) {
		t.Fatalf("err = %v, want errBuildStateUnwritten", err)
	}
	if got := memberSafeWakeReason(err); got != memberSafeWakeReason(ErrFenced) {
		t.Fatalf("a member reads %q, want the fence's wording", got)
	}
	if b.startCount() != 0 || b.seedCount() != 0 || b.sidecarStartCount() != 0 {
		t.Fatalf("started %d, seeded %d, sidecars %d after a refused record", b.startCount(), b.seedCount(), b.sidecarStartCount())
	}

	sa, sb, _, _, _ := stagedActivator(t)
	sb.fakeBackend.setIdentity("v1")
	sa.BeforeExec = (&execLog{fail: failAll}).hook
	if err := sa.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); !errors.Is(err, errBuildStateUnwritten) {
		t.Fatalf("WakeWorker = %v, want errBuildStateUnwritten", err)
	}
	// The record is written at the worker's exec (amadan #272), so the
	// spawn is entered and refused there: no child.
	if sb.workerStarts() != 0 {
		t.Fatalf("%d children after a refused record", sb.workerStarts())
	}

	next := stagedID(t, t.TempDir(), "v3")
	pa, pb, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap})
	pb.fakeBackend.setIdentity("v1")
	pb.restoreFresh = true
	pa.BeforeExec = (&execLog{fail: failAll}).hook
	if _, err := pa.beginRequest(context.Background(), seeded(identityRoute(t, "one.example.com"))); !errors.Is(err, errBuildStateUnwritten) {
		t.Fatalf("planned wake err = %v, want errBuildStateUnwritten", err)
	}
	if len(pb.started()) != 0 || pb.seedCount() != 0 {
		t.Fatalf("planned: spawned %+v, seeded %d after a refused record", pb.started(), pb.seedCount())
	}
}

// Astra I3 #3: identity capability is per route. On a SplitBackend whose
// exec half pins executables, a unit-backed route still wakes by today's
// Start and Seed, records nothing, and never reaches StartGenExe; an exec
// row on the same service takes the identity path.
func TestAUnitRouteOnASplitBackendKeepsTodaysWake(t *testing.T) {
	exec, unit := &fakeBackend{}, &fakeBackend{}
	exec.setIdentity("v1")
	unit.setIdentity("v1")
	// struct{ Backend } hides the unit fake's identity methods, as
	// UnitBackend has none.
	split := &SplitBackend{Exec: exec, Unit: struct{ Backend }{unit}}
	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}, split, reg) // as agent.go builds it
	l := &execLog{}
	a.BeforeExec = l.hook
	unit.restoreFresh = true
	ur := seeded(identityRoute(t, "u.example.com"))
	ur.Backing = registry.BackingUnit
	done, err := a.beginRequest(context.Background(), ur)
	if err != nil {
		t.Fatalf("a unit route's wake failed on the split backend: %v", err)
	}
	done()
	calls, _, _ := l.seen()
	if unit.liveStarts() != 1 || unit.seedCount() != 1 || exec.startCount() != 0 || len(calls) != 0 {
		t.Fatalf("unit live starts %d, seeds %d, exec starts %d, records %v; want today's unit wake, nothing recorded",
			unit.liveStarts(), unit.seedCount(), exec.startCount(), calls)
	}
	if got := identityEvents(unit); !slices.Equal(got, []string{"seed", "start:instance"}) {
		t.Fatalf("unit events %v, want today's seed and start", got)
	}
	done, err = a.beginRequest(context.Background(), identityRoute(t, "e.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if calls, _, _ := l.seen(); exec.lastExeVersion() != "v1" || len(calls) != 1 || calls[0] != "e.example.com@v1" {
		t.Fatalf("the exec row did not take the identity path: %q, %v", exec.lastExeVersion(), calls)
	}
}

// Astra I2 #6: a cold default instance with no live link still wakes: the
// Materialize hook runs before the identity is captured, and the identity
// is the build it materialized.
func TestADefaultWakeMaterializesBeforeItCapturesTheIdentity(t *testing.T) {
	a, b := identityActivator(t, "")
	b.mu.Lock()
	b.ident.materializeTo = "v1"
	b.mu.Unlock()
	l := &execLog{}
	a.BeforeExec = l.hook
	done, err := a.beginRequest(context.Background(), identityRoute(t, "one.example.com"))
	if err != nil {
		t.Fatalf("a cold wake failed before materializing: %v", err)
	}
	done()
	calls, _, _ := l.seen()
	if _, m := b.identityCounts(); m != 1 || len(calls) != 1 || calls[0] != "one.example.com@v1" || b.lastExeVersion() != "v1" {
		t.Fatalf("materialized %d, recorded %v, ran %q", m, calls, b.lastExeVersion())
	}
}

// CARLOS_SWAP on, no plan: the recorded file runs as generation n on its
// own socket, exactly where startGen would have put it.
func TestAFlagOnUnplannedWakeRunsTheRecordedFileAtGenN(t *testing.T) {
	a, b := identityActivator(t, "v1")
	a.gens = true
	l := &execLog{}
	a.BeforeExec = l.hook
	rt := identityRoute(t, "one.example.com")
	done, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatal(err)
	}
	done()
	x, ok := b.lastExe()
	if !ok || !x.perGen || x.n == 0 || x.id != fakeIdentity("v1") || b.liveStarts() != 0 {
		t.Fatalf("exe %+v (live-link starts %d), want the recorded v1 at its own generation", x, b.liveStarts())
	}
	if calls, _, _ := l.seen(); len(calls) != 1 || calls[0] != "one.example.com@v1" {
		t.Fatalf("records %v", calls)
	}
	g, _ := a.entryGen(rt.Host)
	if g.N != x.n || g.Socket != fmt.Sprintf("%s.g%d", rt.Socket, x.n) || g.StagedPath != fakeIdentity("v1").StagedPath {
		t.Fatalf("serving generation %+v, want generation %d of the recorded file", g, x.n)
	}
}

// Ruling R-COLDPUBLISH: the generations file names the captured build
// before its build state does, so root cannot sweep the file between the
// capture and the exec. A refused wake's pending build is dropped again.
func TestTheColdWakePublishesBeforeItRecords(t *testing.T) {
	t.Parallel()
	a, b := identityActivator(t, "v1")
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	path := fakeIdentity("v1").StagedPath
	names := func() bool {
		g, err := release.ReadGenerations(a.GenerationsFile)
		return err == nil && g != nil && slices.Contains(g.Paths["one.example.com"], path)
	}
	var publishedFirst atomic.Bool
	l := &execLog{}
	a.BeforeExec = func(ctx context.Context, host, build string) error {
		publishedFirst.Store(names())
		return l.hook(ctx, host, build)
	}
	done, err := a.beginRequest(context.Background(), identityRoute(t, "one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if !publishedFirst.Load() {
		t.Fatal("the build state was written before the generations file named the build")
	}
	waitFor(t, "the serving generation to stay published", names)
	if b.startCount() != 1 {
		t.Fatalf("starts = %d", b.startCount())
	}

	a, _ = identityActivator(t, "v1")
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	a.BeforeExec = (&execLog{fail: failAll}).hook
	if _, err := a.beginRequest(context.Background(), identityRoute(t, "one.example.com")); !errors.Is(err, errBuildStateUnwritten) {
		t.Fatalf("err = %v", err)
	}
	waitFor(t, "the refused wake's pending build to be dropped", func() bool { return !names() })
}

// A hold swap's pinned wake records the pin before the seed, seeds the pin
// (never the live link), and startPinnedAt does not record it again.
func TestAPinnedWakeSeedsThePin(t *testing.T) {
	f := newSwapFixture(t)
	f.b.fakeBackend.setIdentity("v1")
	f.b.restoreFresh = true
	f.a.mu.Lock()
	f.a.entries[f.host].route.Seed = `["seed"]`
	f.a.mu.Unlock()
	l := &execLog{}
	f.a.BeforeExec = recordInto(f.b.fakeBackend, l)
	res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
	if res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	// start:instance is the fixture's incumbent, woken before the test
	// opted into the identity capability.
	if got := identityEvents(f.b.fakeBackend); !slices.Equal(got, []string{"start:instance", "record v2", "seed v2"}) {
		t.Fatalf("events %v, want the pin recorded, then seeded", got)
	}
	if calls, _, _ := l.seen(); len(calls) != 1 {
		t.Fatalf("records %v: the pin was recorded twice", calls)
	}
	if c, _ := f.b.identityCounts(); c != 0 {
		t.Fatalf("%d live-link captures: a pinned wake runs its pin", c)
	}
}

// A worker wake execs the identity it recorded at step 2b (onto), so a
// repoint during the record cannot change the worker's build, and its
// exec-time hook does not write the same build again.
func TestAWorkerWakeExecsTheIdentityItRecorded(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	b.fakeBackend.setIdentity("v1")
	l := &execLog{}
	a.BeforeExec = func(ctx context.Context, host, build string) error {
		b.fakeBackend.setLive("v2")
		return l.hook(ctx, host, build)
	}
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := b.calls(); len(got) != 1 || got[0] == nil || *got[0] != fakeIdentity("v1") {
		t.Fatalf("staged spawns %v, want one onto the recorded v1", got)
	}
	if calls, _, _ := l.seen(); len(calls) != 1 || calls[0] != host+"@v1" {
		t.Fatalf("records %v, want one of v1", calls)
	}
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving %+v, want the recorded v1", got)
	}
}

// --- legacy sidecars (ruling R-EXEC-SIDECAR) ---

// sidecarActivator is a flag-off service whose backend can start a
// sidecar from a generation (planGenBackend) and pin an identity, with a
// sidecar staged for every host.
func sidecarActivator(t *testing.T, live string) (*Service, *planGenBackend) {
	t.Helper()
	b := newPlanGenBackend()
	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)
	b.sidecarStaged = true
	if live != "" {
		b.fakeBackend.setIdentity(live)
	}
	return a, b
}

// crashSidecar kills host's legacy sidecar and runs the supervisor.
func crashSidecar(t *testing.T, a *Service, host string) {
	t.Helper()
	a.mu.Lock()
	e := a.entries[host]
	var p *fakeProc
	if e != nil {
		p, _ = e.sidecar.(*fakeProc)
	}
	a.mu.Unlock()
	if p == nil {
		t.Fatalf("%s has no sidecar to crash", host)
	}
	p.kill()
	a.sweepOnce()
}

// Flag off: the co-spawned sidecar runs the recorded file by path, not
// whatever the live link names by then, and the wake's record covers it.
func TestAFlagOffSidecarRunsTheRecordedFile(t *testing.T) {
	a, b := sidecarActivator(t, "v1")
	l := &execLog{}
	a.BeforeExec = func(ctx context.Context, host, build string) error {
		b.fakeBackend.setLive("v2")
		return l.hook(ctx, host, build)
	}
	done, err := a.beginRequest(context.Background(), identityRoute(t, "one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	sc := b.sidecarsOn()
	if len(sc) != 1 || sc[0].StagedPath != fakeIdentity("v1").StagedPath || sc[0].Build != "v1" || b.sidecarStartCount() != 0 {
		t.Fatalf("sidecars on %+v, %d live-link sidecar starts; want the recorded v1 by path", sc, b.sidecarStartCount())
	}
	if calls, _, _ := l.seen(); len(calls) != 1 {
		t.Fatalf("records %v: the co-spawn wrote the wake's build again", calls)
	}
}

// A supervisor restart is a new exec against the instance's data: it
// records its generation's build first, and runs that file.
func TestASidecarRestartRecordsItsBuild(t *testing.T) {
	t.Parallel()
	a, b := sidecarActivator(t, "v1")
	l := &execLog{spawns: func() int { return len(b.sidecarsOn()) }}
	a.BeforeExec = l.hook
	host := "one.example.com"
	done, err := a.beginRequest(context.Background(), identityRoute(t, host))
	if err != nil {
		t.Fatal(err)
	}
	done()
	b.fakeBackend.setLive("v2") // a repoint since the wake
	crashSidecar(t, a, host)
	waitFor(t, "the sidecar restart", func() bool { return len(b.sidecarsOn()) == 2 && a.sidecarSettled(host) })
	calls, at, _ := l.seen()
	if !slices.Equal(calls, []string{host + "@v1", host + "@v1"}) || at[1] != 1 {
		t.Fatalf("records %v at sidecar counts %v, want the restart's v1 before its exec", calls, at)
	}
	if sc := b.sidecarsOn(); sc[1].StagedPath != fakeIdentity("v1").StagedPath {
		t.Fatalf("restarted on %+v, want the generation's recorded file", sc[1])
	}
}

// A refused sidecar record starts no sidecar and is logged as a sidecar
// start failure; the instance keeps serving, at the wake and at a restart.
func TestARefusedSidecarRecordStartsNoSidecar(t *testing.T) {
	t.Parallel()
	a, b := sidecarActivator(t, "v1")
	l := &execLog{}
	a.BeforeExec = l.hook
	host := "one.example.com"
	done, err := a.beginRequest(context.Background(), identityRoute(t, host))
	if err != nil {
		t.Fatal(err)
	}
	done()
	l.mu.Lock()
	l.fail = failAll
	l.mu.Unlock()
	crashSidecar(t, a, host)
	waitFor(t, "the refused restart", func() bool {
		calls, _, _ := l.seen()
		return len(calls) == 2 && a.sidecarSettled(host)
	})
	if n := len(b.sidecarsOn()); n != 1 || b.sidecarStartCount() != 0 {
		t.Fatalf("%d generation sidecars, %d live-link sidecars after a refused record", n, b.sidecarStartCount())
	}
	if _, ok := a.entryGen(host); !ok {
		t.Fatal("the instance stopped serving over its sidecar's record")
	}

	// At the wake: a replanned build is recorded again by its sidecar (the
	// wake's context names the first plan's, which never ran and so is never
	// written), and that record is refused.
	dir := t.TempDir()
	stale, next := stagedID(t, dir, "v2"), stagedID(t, dir, "v3")
	pa, pb, _ := planActivator(t, WakePlan{})
	pb.fakeBackend.setIdentity("v1")
	pb.sidecarStaged = true
	pa.Planner = laterReplan(stale, next)
	var n atomic.Int32
	pl := &execLog{fail: func(string) bool { return n.Add(1) == 2 }}
	pa.BeforeExec = pl.hook
	done, err = pa.beginRequest(context.Background(), identityRoute(t, host))
	if err != nil {
		t.Fatalf("the wake failed over its sidecar's record: %v", err)
	}
	done()
	if calls, _, _ := pl.seen(); !slices.Equal(calls, []string{host + "@v3", host + "@v3"}) {
		t.Fatalf("records %v, want the replanned spawn's and the sidecar's", calls)
	}
	if sc := pb.sidecarsOn(); len(sc) != 0 || pb.sidecarStartCount() != 0 {
		t.Fatalf("sidecars %+v (%d live): a sidecar ran unrecorded", sc, pb.sidecarStartCount())
	}
	if g, ok := pa.entryGen(host); !ok || g.Build != "v3" {
		t.Fatalf("serving %+v (%v), want v3", g, ok)
	}
}

// An instance whose generation names no staged file (awake from before
// this binary): its sidecar resolves the live link once, records that
// build, and runs that file by path.
func TestASidecarOnAnUnstagedGenerationResolvesRecordsAndRunsByPath(t *testing.T) {
	t.Parallel()
	a, b := sidecarActivator(t, "")
	host := "one.example.com"
	done, err := a.beginRequest(context.Background(), identityRoute(t, host))
	if err != nil {
		t.Fatal(err)
	}
	done()
	if g, _ := a.entryGen(host); g.StagedPath != "" || b.sidecarStartCount() != 1 {
		t.Fatalf("setup: generation %+v, %d live-link sidecars", g, b.sidecarStartCount())
	}
	b.fakeBackend.setIdentity("v1")
	l := &execLog{}
	a.BeforeExec = l.hook
	b.lastSidecar().kill()
	a.sweepOnce()
	waitFor(t, "the sidecar restart", func() bool { return len(b.sidecarsOn()) == 1 && a.sidecarSettled(host) })
	if sc := b.sidecarsOn(); sc[0].StagedPath != fakeIdentity("v1").StagedPath || sc[0].Build != "v1" || sc[0].Tier != cgroup.TierDefault {
		t.Fatalf("restarted on %+v, want the resolved v1 by path at its own tier (GenTier)", sc[0])
	}
	if calls, _, _ := l.seen(); !slices.Equal(calls, []string{host + "@v1"}) {
		t.Fatalf("records %v, want the resolved v1", calls)
	}
	if c, _ := b.identityCounts(); c != 1 || b.sidecarStartCount() != 1 {
		t.Fatalf("captures %d, live-link sidecars %d: want one capture and no new live-link start", c, b.sidecarStartCount())
	}
}

// Flag on: a sidecar attached late to the incumbent and its replacement
// after an overlap switch each record their generation's build before
// their exec (the replacement runs off any wake's context, so it writes).
func TestASidecarReplacementRecordsTheNewBuild(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	f.b.sidecarStaged = true
	l := &execLog{spawns: func() int { return len(f.b.sidecarGenPaths()) }}
	f.a.BeforeExec = l.hook
	f.a.sweepOnce()
	waitFor(t, "late attachment", func() bool { return f.a.sidecarSettled(f.host) && len(f.b.sidecarGenPaths()) == 1 })
	if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	waitFor(t, "the replacement", func() bool { return f.a.sidecarSettled(f.host) && len(f.b.sidecarGenPaths()) == 2 })
	calls, at, _ := l.seen()
	want := []string{f.host + "@v1", f.host + "@v2", f.host + "@v2"}
	if !slices.Equal(calls, want) || at[0] != 0 || at[2] != 1 {
		t.Fatalf("records %v at sidecar counts %v, want %v: the attachment's v1, the successor's v2, then the replacement's v2 before its exec", calls, at, want)
	}
	if paths := f.b.sidecarGenPaths(); paths[1] != f.staged {
		t.Fatalf("replacement ran %q, want %q", paths[1], f.staged)
	}
}

// A sidecar restart asks ownership again after its record: a lease lost
// while the build state was written execs nothing.
func TestASidecarExecsOnlyIfStillOwnedAfterItsRecord(t *testing.T) {
	t.Parallel()
	a, b := sidecarActivator(t, "v1")
	host := "one.example.com"
	var lose atomic.Bool
	l := &execLog{}
	a.BeforeExec = func(ctx context.Context, h, build string) error {
		if lose.Load() {
			b.lastRep().setErr(errors.New("lease lost"))
		}
		return l.hook(ctx, h, build)
	}
	done, err := a.beginRequest(context.Background(), identityRoute(t, host))
	if err != nil {
		t.Fatal(err)
	}
	done()
	lose.Store(true)
	crashSidecar(t, a, host)
	waitFor(t, "the restart's record", func() bool {
		calls, _, _ := l.seen()
		return len(calls) == 2 && a.sidecarSettled(host)
	})
	if n := len(b.sidecarsOn()); n != 1 || b.sidecarStartCount() != 0 {
		t.Fatalf("%d generation sidecars, %d live-link: a sidecar exec'd after its lease was lost", n, b.sidecarStartCount())
	}
}

// A worker wake whose token cannot be minted fails after step 2b published
// its build: the wake's pending build is dropped with it, and the lease
// goes back.
func TestAWorkerTokenMintFailureDropsThePendingBuild(t *testing.T) {
	a, b, _, _, _ := stagedActivator(t)
	b.fakeBackend.setIdentity("v1")
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	old := randRead
	randRead = func([]byte) (int, error) { return 0, errors.New("entropy pool exhausted") }
	defer func() { randRead = old }()
	host := "bot.x.internal"
	if err := a.WakeWorker(testWorkerRoute(host)); err == nil || !strings.Contains(err.Error(), "mint worker token") {
		t.Fatalf("WakeWorker = %v, want the mint failure", err)
	}
	waitFor(t, "the pending build to be dropped", func() bool {
		g, err := release.ReadGenerations(a.GenerationsFile)
		return err == nil && g != nil && !slices.Contains(g.Paths[host], fakeIdentity("v1").StagedPath)
	})
	if n := b.releaseCount(); n != 1 || b.workerStarts() != 0 {
		t.Fatalf("releases %d, workers %d, want the lease back and nothing spawned", n, b.workerStarts())
	}
}