package activator

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

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

func workerAttempt(cause Cause, mode SwapMode, onto release.StagedIdentity) Attempt {
	return Attempt{Host: "bot.x.internal", Cause: cause, Build: onto.Version, StagedPath: onto.StagedPath,
		Digest: onto.Digest, Demand: onto.Demand, Mode: mode, SchedRev: 1}
}

func one(string) uint64 { return 1 }

// waitForWorkerStarts waits until the backend has made n worker children:
// a spawn call is recorded before its exec, so b.calls() alone runs ahead
// of the child a test wants to act on.
func waitForWorkerStarts(t *testing.T, b *stagedFake, n int) {
	t.Helper()
	waitFor(t, "a worker child", func() bool { return b.workerStarts() == n })
}

// Spec §7.9: the attempt respawns on the SAME entry. The lease is never
// released or re-taken, and the new child execs the attempt's staged build.
func TestReplaceWorkerKeepsTheEntryAndItsLease(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	acquires, _, _ := b.counts()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	if res.Outcome != SwapDone || res.Gen.Build != "v2" || res.Gen.Digest != fakeIdentity("v2").Digest || !res.Gen.Worker {
		t.Fatalf("ReplaceWorker = %+v", res)
	}
	if got, _, _ := b.counts(); got != acquires {
		t.Fatalf("lease acquires %d -> %d: a socketless attempt must keep the lease (spec §7.9)", acquires, got)
	}
	if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v2") {
		t.Fatalf("spawns %v, want the wake, then v2 by its staged path", calls)
	}
	if got := a.entryServing(host); got != fakeIdentity("v2") {
		t.Fatalf("serving %+v", got)
	}
	if _, ok := a.Current(host); !ok {
		t.Fatal("Current hides the worker after its attempt finished")
	}
}

// Spec §6.5: a failed release in overlap mode respawns the good build in
// its place, on the same entry, and reports the failure.
func TestReplaceWorkerFallsBackToTheGoodBuild(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	good := fakeIdentity("v1")
	go func() {
		waitForWorkerStarts(t, b, 2)
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
	if res.Outcome != SwapFailed || res.Gen.Build != "v1" {
		t.Fatalf("ReplaceWorker = %+v, want failed with the good v1 serving", res)
	}
	if calls := b.calls(); len(calls) != 3 || *calls[2] != good {
		t.Fatalf("spawns %v, want the good build last", calls)
	}
	if got := a.entryServing(host); got != good {
		t.Fatalf("serving %+v, want the good build", got)
	}
}

// Carried to Task 14 (from Task 11): a worker crash inside ReplaceWorker's
// observation is the attempt's to read. The sweep sees the dead child first
// and asks CoalesceRecycle, which the scheduler answers true while the
// host's attempt runs (rollout.Manager.CoalesceRecycle). So the sweep's
// crash restart does not respawn the attempt's build behind it, and the
// observation reads the crash as the build's failure and falls back to the
// good build. Without the hook the sweep respawns v2 and the attempt never
// sees its child fail.
func TestACrashInsideAReplaceWorkerObservationIsCoalesced(t *testing.T) {
	// A 2 s window polls every 250 ms (settlePoll), so the sweep below runs
	// well before the observation's next look at the child.
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	var running atomic.Bool
	var asked atomic.Int32
	a.CoalesceRecycle = func(h string) bool {
		if h == host {
			asked.Add(1)
		}
		return running.Load()
	}
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	good := fakeIdentity("v1")
	go func() {
		waitForWorkerStarts(t, b, 2)
		time.Sleep(100 * time.Millisecond)
		b.lastWorker().crash()
		a.sweepOnce() // the crash restart's arm, inside the observation
	}()
	running.Store(true)
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
	running.Store(false)
	if asked.Load() == 0 {
		t.Fatal("the sweep never asked CoalesceRecycle about the crashed child")
	}
	if res.Outcome != SwapFailed || res.Gen.Build != "v1" {
		t.Fatalf("ReplaceWorker = %+v, want failed with the good v1 serving", res)
	}
	calls := b.calls()
	if len(calls) != 3 || *calls[1] != fakeIdentity("v2") || *calls[2] != good {
		t.Fatalf("spawns %v, want the wake, v2 once, then the good build: the sweep respawned behind the attempt", calls)
	}
	if got := a.entryServing(host); got != good {
		t.Fatalf("serving %+v, want the good build", got)
	}
}

// Spec §6.4, §8.3: a hold build that ran never falls back. The instance
// stays down.
func TestReplaceWorkerInHoldModeStaysDown(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	good := fakeIdentity("v1")
	go func() {
		waitForWorkerStarts(t, b, 2)
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeHold, fakeIdentity("v2")), &good, one)
	if res.Outcome != SwapDown {
		t.Fatalf("ReplaceWorker = %+v, want down", res)
	}
	if len(b.calls()) != 2 {
		t.Fatalf("spawns %v: a hold build fell back", b.calls())
	}
	waitFor(t, "the entry to stop", func() bool { return !a.Owns(host) })
}

// Spec §6.5: a failed restart respawns the build it replaced, and writes
// nothing against it.
func TestReplaceWorkerRestartFailureRespawnsTheServingBuild(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	serving := a.entryServing(host)
	go func() {
		waitForWorkerStarts(t, b, 2)
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRestart, ModeOverlap, serving), nil, one)
	if res.Outcome != SwapFailed {
		t.Fatalf("ReplaceWorker = %+v", res)
	}
	waitFor(t, "the serving build back", func() bool { return len(b.calls()) == 3 && !a.entryRespawning(host) })
	if calls := b.calls(); *calls[2] != serving {
		t.Fatalf("respawned %+v, want the serving build", calls[2])
	}
}

// Spec §6.5: a clean exit inside the 10 s is a one-shot's success.
func TestReplaceWorkerOneShotCleanExitIsReady(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatal(err)
	}
	go func() {
		waitForWorkerStarts(t, b, 2)
		b.lastWorker().exit()
	}()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	if res.Outcome != SwapDone {
		t.Fatalf("ReplaceWorker = %+v, want done", res)
	}
	if !res.CleanExit {
		t.Fatalf("ReplaceWorker = %+v: a one-shot that exited cleanly must say so, so the scheduler records its build good (spec 6.5, 8.5)", res)
	}
}

// Spec §6.5, §8.5: only a build that exited cleanly is a one-shot's success.
// One still running after the settle window is up and serving (done, but
// the scheduler records it good when the activator acknowledges it, not
// here), and a crash is a failure.
func TestReplaceWorkerRunningChildIsNotACleanExit(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, _, _, _, _ := stagedActivator(t)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatal(err)
	}
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	if res.Outcome != SwapDone || res.CleanExit {
		t.Fatalf("ReplaceWorker = %+v, want done and not a clean exit", res)
	}
}

// Astra R3 flag triage T10: a first-wake worker whose build fails and whose
// fallback boots only after the ordinary wake bound is a success for
// WakeWorker, as it is for a request: the wait extends once to arrival +
// firstWakeBudget (spec 8.2). Without the extension the caller was told
// "timed out" for a wake that went on to succeed. A wake that is not a
// first wake still gives up at the ordinary bound.
func TestWakeWorkerWaitsOutAFirstWakeFallback(t *testing.T) {
	old := wakeWaitBound
	wakeWaitBound = 100 * time.Millisecond
	t.Cleanup(func() { wakeWaitBound = old })
	for _, tc := range []struct {
		name      string
		firstWake bool
		wantErr   bool
	}{
		{"first wake", true, false},
		{"not a first wake", false, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			shrinkFirstWake(t, 900*time.Millisecond, 200*time.Millisecond, 50*time.Millisecond)
			a, b, _, _, _ := stagedActivator(t)
			a.gens = true
			a.backend = &plannedHostKill{stagedFake: b} // the failed child's host cgroup proves empty
			dir := t.TempDir()
			good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
			var starts atomic.Int32
			a.Planner = func(context.Context, registry.Route) WakePlan {
				if !tc.firstWake {
					// An ordinary wake of the good build; its start is slow.
					return WakePlan{Boot: good, Mode: ModeOverlap}
				}
				return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
			}
			b.mu.Lock()
			b.workerStartHook = func(registry.Route) {
				n := starts.Add(1)
				switch {
				case tc.firstWake && n == 1:
					// The failing build dies inside its settle window.
					time.AfterFunc(20*time.Millisecond, func() { b.lastWorker().crash() })
				case tc.firstWake && n == 2, !tc.firstWake && n == 1:
					time.Sleep(250 * time.Millisecond) // the good build boots slowly
				}
			}
			b.mu.Unlock()
			err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal"))
			if tc.wantErr {
				if !errors.Is(err, errWakeTimeout) {
					t.Fatalf("WakeWorker = %v, want errWakeTimeout at the ordinary bound", err)
				}
				return
			}
			if err != nil {
				t.Fatalf("WakeWorker = %v: a first wake's successful fallback was reported as a failure", err)
			}
		})
	}
}

// Spec §6.3, §7.4, §8.1: the attempt's checks run again at the actual
// spawn, after the old child has retired, and again before its result is
// accepted. A newer release (or a sibling's failure, whose halt bumps the
// revision) that lands during the retire means the attempt's build never
// execs: the build it replaced comes back instead. One that lands during
// the observation means the result is abandoned, never done, so the build
// is not recorded good. Checked only before the retire, both slip through.
func TestReplaceWorkerReChecksAtTheSpawnAndBeforeTheResult(t *testing.T) {
	t.Run("superseded while the old child retires", func(t *testing.T) {
		shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
		a, b, _, _, _ := stagedActivator(t)
		host := "bot.x.internal"
		if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
			t.Fatal(err)
		}
		gate := make(chan struct{})
		b.lastWorker().setStopGate(gate)
		var rev atomic.Uint64
		rev.Store(1)
		results := make(chan SwapResult, 1)
		go func() {
			results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
				func(string) uint64 { return rev.Load() })
		}()
		waitFor(t, "the retire to begin", func() bool { return a.entryRespawning(host) })
		rev.Store(2) // a newer release lands while the old child drains
		close(gate)
		res := <-results
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
			t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded", res)
		}
		for _, c := range b.calls() {
			if c != nil && *c == fakeIdentity("v2") {
				t.Fatalf("spawns %v: the superseded build was exec'd", b.calls())
			}
		}
		waitFor(t, "a child back on the entry", func() bool { return len(b.calls()) == 2 && !a.entryRespawning(host) })
		if got := a.entryServing(host); got != fakeIdentity("v1") {
			t.Fatalf("serving %+v, want the build the attempt replaced", got)
		}
	})
	t.Run("superseded during the observation", func(t *testing.T) {
		shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
		a, b, _, _, _ := stagedActivator(t)
		host := "bot.x.internal"
		if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
			t.Fatal(err)
		}
		var rev atomic.Uint64
		rev.Store(1)
		go func() {
			waitForWorkerStarts(t, b, 2)
			rev.Store(2)
		}()
		res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
			func(string) uint64 { return rev.Load() })
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
			t.Fatalf("ReplaceWorker = %+v, want abandoned: a superseded result is never done", res)
		}
	})
}

// PF11: the attempt's checks also run in main's exec-time gate, inside
// startWorker's before-exec hook, after the preparation (here the config
// admission) and the publication. A newer release that lands after the
// respawn's first check but before the exec means the attempt's build never
// execs, and the build it replaced comes back.
func TestReplaceWorkerReChecksAtTheExecItself(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	var rev atomic.Uint64
	rev.Store(1)
	b.mu.Lock()
	b.admitHook = func() { rev.Store(2) } // lands during the respawn's preparation
	b.mu.Unlock()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
		func(string) uint64 { return rev.Load() })
	if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
		t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded at the exec", res)
	}
	waitFor(t, "the restore", func() bool { return b.workerStarts() == 2 && !a.entryRespawning(host) })
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving %+v, want the build the attempt replaced (v2 must never have exec'd)", got)
	}
}

// PF11: and once more before the child is installed. A newer release that
// lands while the attempt's child is being exec'd sheds that child; the
// build it replaced comes back, and the result is abandoned, never done.
func TestReplaceWorkerReChecksBeforeInstalling(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	var rev atomic.Uint64
	rev.Store(1)
	var hooked atomic.Bool
	b.mu.Lock()
	b.workerStartHook = func(registry.Route) {
		if hooked.CompareAndSwap(false, true) {
			rev.Store(2) // lands after the exec gate, before the install check
		}
	}
	b.mu.Unlock()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil,
		func(string) uint64 { return rev.Load() })
	if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
		t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded before the install", res)
	}
	waitFor(t, "the restore", func() bool { return b.workerStarts() == 3 && !a.entryRespawning(host) })
	b.mu.Lock()
	shed := b.workers[1]
	b.mu.Unlock()
	if !shed.isStopped() {
		t.Fatal("the superseded attempt's child was left running")
	}
	if got := a.entryServing(host); got != fakeIdentity("v1") {
		t.Fatalf("serving %+v, want the build the attempt replaced", got)
	}
}

// PF11 (plan 3's termination proof is never dropped): a respawn whose old
// child is not proven gone execs nothing, the attempt's build and a restore
// alike, and the attempt is abandoned.
func TestReplaceWorkerNeverExecsBesideAnUnprovenOldChild(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	a.gens = true
	hk := &plannedHostKill{stagedFake: b}
	a.backend = hk
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	hk.bad.Store(true) // the host cgroup still holds a process
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	if res.Outcome != SwapAbandoned {
		t.Fatalf("ReplaceWorker = %+v, want abandoned", res)
	}
	waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
	time.Sleep(50 * time.Millisecond)
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("%d worker children: something exec'd beside an unproven old child", n)
	}
}

// Spec §7.4: a respawn whose gate finds the host no longer owned (here the
// lease lost while the old child drained) spawns nothing at all, the
// replaced build included. Turning every refusal into a restore ran the old
// build against a database the box no longer owned.
func TestReplaceWorkerWithoutOwnershipSpawnsNothing(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	gate := make(chan struct{})
	b.lastWorker().setStopGate(gate)
	results := make(chan SwapResult, 1)
	go func() {
		results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	}()
	waitFor(t, "the retire to begin", func() bool { return a.entryRespawning(host) })
	b.lastRep().setErr(errors.New("lease renewal failed")) // ownership lost mid-retire
	close(gate)
	if res := <-results; res.Outcome != SwapAbandoned {
		t.Fatalf("ReplaceWorker = %+v, want abandoned", res)
	}
	waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(host) })
	if calls := b.calls(); len(calls) != 1 {
		t.Fatalf("spawns %v: something exec'd after the lease was lost", calls)
	}
}

// A superseded attempt touches nothing.
func TestReplaceWorkerSupersededSpawnsNothing(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatal(err)
	}
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, func(string) uint64 { return 2 })
	if res.Outcome != SwapAbandoned || len(b.calls()) != 1 {
		t.Fatalf("ReplaceWorker = %+v, spawns %v", res, b.calls())
	}
}

// PF9: an attempt whose build could not be published before its exec
// (ErrBuildNotPublished) ran nothing of the build. It is abandonment, a
// staging refusal, never a failure of the build.
func TestReplaceWorkerRefusedForPublicationIsAbandoned(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	blocker := filepath.Join(t.TempDir(), "not-a-dir")
	if err := os.WriteFile(blocker, nil, 0o644); err != nil {
		t.Fatal(err)
	}
	a.mu.Lock()
	a.GenerationsFile = filepath.Join(blocker, "generations.json") // unwritable from here on
	a.mu.Unlock()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	if res.Outcome != SwapAbandoned {
		t.Fatalf("ReplaceWorker = %+v, want abandoned: an unpublished build never ran", res)
	}
	waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("%d worker children: an unpublished build was exec'd", n)
	}
}

// Carried from Task 9: e.unexpectedExit is never cleared by the arms that
// set it. A stale mark from an earlier crash must not turn an intentional
// stop during the attempt's observation into a verdict: the stop cancels
// the observation, and the attempt is abandoned, never done.
func TestReplaceWorkerAnIntentionalStopDuringTheObservationIsNoVerdict(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	a.mu.Lock()
	a.entries[host].unexpectedExit = true // left by an earlier crash restart
	a.mu.Unlock()
	go func() {
		waitForWorkerStarts(t, b, 2)
		waitFor(t, "the observation", func() bool {
			a.mu.Lock()
			defer a.mu.Unlock()
			e := a.entries[host]
			return e != nil && e.firstWatch != nil
		})
		a.mu.Lock()
		a.beginStopLocked(a.entries[host], StopIdle) // an operator's stop
		a.mu.Unlock()
	}()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
	if res.Outcome != SwapAbandoned {
		t.Fatalf("ReplaceWorker = %+v, want abandoned: an intentional stop records nothing", res)
	}
}

// Spec §7.9: with CARLOS_SWAP unset (no generations, no rollout manager), a
// worker's captured serving identity is still published on its wake and on
// every respawn, so `carlos sidecars ls` shows what runs. Only the rollout
// manager wrote it, and a flag-off box showed "unknown", or a stale build
// after the flag was turned off.
func TestAWorkersServingIdentityIsPublishedWithTheFlagOff(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	var mu sync.Mutex
	var got []string
	a.OnWorkerServing = func(host string, id release.StagedIdentity) {
		mu.Lock()
		got = append(got, host+"="+id.Version)
		mu.Unlock()
	}
	published := func() int { mu.Lock(); defer mu.Unlock(); return len(got) }
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	waitFor(t, "the wake's identity", func() bool { return published() == 1 })
	b.lastWorker().crash()
	a.sweepOnce() // the always-on worker respawns
	waitFor(t, "the respawn's identity", func() bool { return published() == 2 })
	mu.Lock()
	defer mu.Unlock()
	if want := host + "=" + a.entryServing(host).Version; got[1] != want {
		t.Fatalf("published %v, want the respawned child's %q last", got, want)
	}
}

// Spec §7.9: a delayed publication of an older worker incarnation never
// overwrites a newer one. Here the wake's publication (incarnation 1) is
// replayed after the respawn published incarnation 2: it is dropped. An
// unconditional write left the older version on display indefinitely.
func TestAnOlderWorkerPublicationIsDropped(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	var mu sync.Mutex
	var got []string
	a.OnWorkerServing = func(_ string, id release.StagedIdentity) {
		mu.Lock()
		got = append(got, id.Version)
		mu.Unlock()
	}
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	a.mu.Lock()
	e := a.entries[host]
	firstID, firstInc := e.workerServing, e.workerInc
	a.mu.Unlock()
	b.lastWorker().crash()
	a.sweepOnce()
	waitFor(t, "the respawn's publication", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 2 })
	a.publishWorkerServing(e, firstID, firstInc) // the older one, delayed
	mu.Lock()
	defer mu.Unlock()
	if len(got) != 2 {
		t.Fatalf("published %v: an older incarnation's publication was delivered after a newer one", got)
	}
}

// Spec §7.9 (P5 review finding 8): a child exec'd from an external live-link
// target has no captured version. Its publication still goes out, empty,
// so the row stops naming the build the previous child ran. Skipping empty
// versions left ServingVersion=v1 on display indefinitely.
func TestAnUnknownVersionClearsTheServingVersion(t *testing.T) {
	t.Parallel()
	a, _, _, _, _ := stagedActivator(t)
	var mu sync.Mutex
	var got []string
	a.OnWorkerServing = func(_ string, id release.StagedIdentity) {
		mu.Lock()
		got = append(got, id.Version)
		mu.Unlock()
	}
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	waitFor(t, "the wake's identity", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 1 })
	a.mu.Lock()
	e := a.entries[host]
	e.workerInc++ // a replacement exec'd from an external live-link target
	inc := e.workerInc
	a.mu.Unlock()
	a.publishWorkerServing(e, release.StagedIdentity{StagedPath: "/srv/external/app"}, inc)
	mu.Lock()
	defer mu.Unlock()
	if len(got) != 2 || got[1] != "" {
		t.Fatalf("published %q, want the empty version last: the old one must be cleared", got)
	}
}

// Spec §7.4 (P5 review finding 1): every respawn passes the ownership check
// at its exec, a fallback included. Here B execs, the lease is lost, and
// then B crashes: the good build is NOT respawned. Without the check the
// fallback ran A against a database the box no longer owned.
func TestAWorkerFallbackWithoutOwnershipSpawnsNothing(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	good := fakeIdentity("v1")
	go func() {
		waitForWorkerStarts(t, b, 2)
		b.lastRep().setErr(errors.New("lease renewal failed")) // ownership lost after B's exec
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
	if res.Outcome != SwapFailed || res.Gen.Build != "" {
		t.Fatalf("ReplaceWorker = %+v, want failed with nothing serving", res)
	}
	if calls := b.calls(); len(calls) != 2 {
		t.Fatalf("spawns %v: the fallback exec'd after the lease was lost", calls)
	}
}

// Spec §6.5, §8.1 (P5 review finding 5): a restart counts only once a NEW
// child is installed. Here the restart's exec fails after the old child
// stopped cleanly: the entry still holds that child, with the same
// artifact, and its clean exit read as a one-shot's success, so the
// scheduler acknowledged config that never ran.
func TestAFailedRestartExecIsNeverDone(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	serving := a.entryServing(host)
	b.mu.Lock()
	b.workerErr = errors.New("exec: text file busy")
	b.mu.Unlock()
	res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRestart, ModeOverlap, serving), nil, one)
	if res.Outcome == SwapDone {
		t.Fatalf("ReplaceWorker = %+v: a restart whose child never ran was done", res)
	}
	if res.Outcome != SwapAbandoned {
		t.Fatalf("ReplaceWorker = %+v: a failed exec is a staging refusal, never a failure of the build", res)
	}
	waitFor(t, "the respawns to finish", func() bool { return !a.entryRespawning(host) })
}

// Spec §8.1 (P5 review finding 7): a ReplaceWorker fallback holds its
// publication until the scheduler has recorded B's failure and calls
// PublishHeldServing. Published from the respawn, a blocked callback sat
// in front of the SwapFailed result, and so of the failure record.
func TestAWorkerFallbackPublishesOnlyWhenReleased(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	var mu sync.Mutex
	var got []string
	block := make(chan struct{})
	a.OnWorkerServing = func(_ string, id release.StagedIdentity) {
		mu.Lock()
		got = append(got, id.Version)
		n := len(got)
		mu.Unlock()
		if n > 2 {
			<-block // a slow registry write for the fallback's identity (the wake's v1, then B's v2, come first)
		}
	}
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatal(err)
	}
	good := fakeIdentity("v1")
	go func() {
		waitForWorkerStarts(t, b, 2)
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	results := make(chan SwapResult, 1)
	go func() {
		results <- a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), &good, one)
	}()
	select {
	case res := <-results:
		if res.Outcome != SwapFailed || res.Gen.Build != "v1" {
			t.Fatalf("ReplaceWorker = %+v", res)
		}
	case <-time.After(5 * time.Second):
		close(block)
		t.Fatal("the failure result waited on the fallback's publication")
	}
	mu.Lock()
	n := len(got)
	mu.Unlock()
	if n != 2 {
		t.Fatalf("published %v before the failure was recorded", got)
	}
	close(block)
	a.PublishHeldServing(host) // the scheduler, after done()
	mu.Lock()
	defer mu.Unlock()
	if len(got) != 3 || got[2] != "v1" {
		t.Fatalf("published %v, want the fallback's v1 once released", got)
	}
}

// Spec §8.1 (P5 review finding 7): on a planned worker wake that falls back,
// the failure transition (GenFellBack, whose OnGen records B's failure)
// completes before the fallback's identity is published, and a blocked
// publication does not hold it up. Publishing before afterWake left B
// unheld while the callback blocked.
func TestAFallbackWakeRecordsTheFailureBeforeItPublishes(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond)
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	a.backend = &plannedHostKill{stagedFake: b} // the failed child's host cgroup proves empty
	evs := &genLog{}
	a.OnGen = evs.add
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}
	}
	block := make(chan struct{})
	var sawFailure, called atomic.Bool
	a.OnWorkerServing = func(string, release.StagedIdentity) {
		sawFailure.Store(slices.Contains(evs.kinds(), GenFellBack))
		called.Store(true)
		<-block
	}
	go func() {
		waitForWorkerStarts(t, b, 1)
		time.Sleep(50 * time.Millisecond)
		b.lastWorker().crash()
	}()
	go a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) //nolint:errcheck
	evs.wait(t, GenFellBack)                               // arrives while the publication is still blocked
	waitFor(t, "the publication", func() bool { return called.Load() })
	close(block)
	if !sawFailure.Load() {
		t.Fatal("the fallback's identity was published before its failure transition")
	}
}

// Final review F2 (spec §§6.4/8.3, P5's hold-everywhere invariant): "not
// installed" is not evidence that a build never ran. A hold attempt whose
// build was exec'd, and then superseded before its install (the install
// check sheds the child), may have migrated the database: the build it
// replaced is NOT restored, the entry stops, and the attempt is abandoned,
// not a failure of the build. Superseded at its exec gate instead, the hold
// build provably never ran, so the replaced build comes back as before. An
// overlap attempt superseded after its exec still restores, as before.
func TestASupersededWorkerAttemptRestoresOnlyWhatProvablyNeverRan(t *testing.T) {
	for _, tc := range []struct {
		name     string
		mode     SwapMode
		executed bool // superseded after the exec (true) or at the exec gate
		restores bool
	}{
		{"hold, superseded after its exec", ModeHold, true, false},
		{"hold, superseded at its exec gate", ModeHold, false, true},
		{"overlap, superseded after its exec", ModeOverlap, true, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond)
			a, b, _, _, _ := stagedActivator(t)
			host := "bot.x.internal"
			if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
				t.Fatal(err)
			}
			var rev atomic.Uint64
			rev.Store(1)
			var hooked atomic.Bool
			b.mu.Lock()
			if tc.executed {
				b.workerStartHook = func(registry.Route) {
					if hooked.CompareAndSwap(false, true) {
						rev.Store(2) // after the exec, before the install check
					}
				}
			} else {
				b.admitHook = func() {
					if hooked.CompareAndSwap(false, true) {
						rev.Store(2) // during the preparation, before the exec gate
					}
				}
			}
			b.mu.Unlock()
			res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, tc.mode, fakeIdentity("v2")), nil,
				func(string) uint64 { return rev.Load() })
			if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) {
				t.Fatalf("ReplaceWorker = %+v, want abandoned: superseded (never a failure of the build)", res)
			}
			wantStarts := 1 // the wake
			if tc.executed {
				wantStarts++ // the attempt's child, shed
			}
			if tc.restores {
				wantStarts++ // the replaced build, back
				waitFor(t, "the restore", func() bool { return b.workerStarts() == wantStarts && !a.entryRespawning(host) })
				if got := a.entryServing(host); got != fakeIdentity("v1") {
					t.Fatalf("serving %+v, want the build the attempt replaced", got)
				}
				return
			}
			waitFor(t, "the entry to stop", func() bool { return !a.Owns(host) })
			time.Sleep(100 * time.Millisecond)
			if n := b.workerStarts(); n != wantStarts {
				t.Fatalf("%d worker children, want %d: the replaced build was restored after the hold build ran", n, wantStarts)
			}
			for _, c := range b.calls()[wantStarts-1:] {
				if c != nil && *c == fakeIdentity("v1") {
					t.Fatalf("spawns %v: the replaced build was respawned after the hold build ran", b.calls())
				}
			}
		})
	}
}

// Round 2 Task 6 (recon B §1): a one-shot's clean exit inside its
// observation is a good build (spec §6.5) only when the sweep parked it
// after its own fence and lease checks. Any other stop of the exited child
// (a lost lease, a fence) leaves the entry not awake, and that answer was
// forgiven for every exited one-shot, so the build was recorded good with no
// ownership behind it. Those stops are abandoned; the sweep's park is done.
func TestAnExitedOneShotIsGoodOnlyAfterACleanPark(t *testing.T) {
	for _, tc := range []struct {
		name  string
		stop  func(t *testing.T, a *Service, b *stagedFake, host string)
		clean bool
	}{
		{"parked by the sweep", func(t *testing.T, a *Service, b *stagedFake, host string) {
			a.sweepOnce()
		}, true},
		{"lease lost, then the sweep", func(t *testing.T, a *Service, b *stagedFake, host string) {
			b.lastRep().setErr(errors.New("renewal failed"))
			a.sweepOnce()
		}, false},
		{"fenced", func(t *testing.T, a *Service, b *stagedFake, host string) {
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			if _, err := a.FenceAndStop(ctx, host, 1); err != nil {
				t.Errorf("FenceAndStop: %v", err)
			}
		}, false},
		{"parked, then fenced before the result", func(t *testing.T, a *Service, b *stagedFake, host string) {
			a.sweepOnce()
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			if _, err := a.FenceAndStop(ctx, host, 1); err != nil {
				t.Errorf("FenceAndStop: %v", err)
			}
		}, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			// A 2 s window polls every 250 ms (settlePoll): the stop below
			// lands well before the observation's first look at the exit.
			shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
			a, b, _, _, _ := stagedActivator(t)
			host := "bot.x.internal"
			if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
				t.Fatal(err)
			}
			done := make(chan struct{})
			go func() {
				defer close(done)
				waitForWorkerStarts(t, b, 2)
				waitFor(t, "the observation", func() bool {
					a.mu.Lock()
					defer a.mu.Unlock()
					e := a.entries[host]
					return e != nil && e.firstWatch != nil
				})
				b.lastWorker().exit() // status 0: the one-shot finished
				tc.stop(t, a, b, host)
			}()
			res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one)
			<-done
			if tc.clean {
				if res.Outcome != SwapDone || !res.CleanExit {
					t.Fatalf("ReplaceWorker = %+v, want done with a clean exit: the sweep parked a finished one-shot", res)
				}
				return
			}
			if res.Outcome != SwapAbandoned || res.CleanExit {
				t.Fatalf("ReplaceWorker = %+v, want abandoned: no clean park with no fence stands behind the exit, so its build must not be recorded good", res)
			}
		})
	}
}

// awaitExtensionSeam installs wakeBoundExtended for one test: the first
// extension of a waiter's bound lets the wake fail (release) and waits until
// that failed wake has deleted its entry, the window in which a waiter that
// re-read the entries map started a second wake.
func awaitExtensionSeam(t *testing.T, a *Service, release chan struct{}) *atomic.Int32 {
	t.Helper()
	var calls atomic.Int32
	old := wakeBoundExtended
	wakeBoundExtended = func() {
		if calls.Add(1) != 1 {
			return
		}
		close(release)
		waitFor(t, "the failed wake to delete its entry", func() bool { return a.entryCount() == 0 })
	}
	t.Cleanup(func() { wakeBoundExtended = old })
	return &calls
}

// Round 2 Task 6 (recon B §2): a first wake's waiter extends its bound once
// (spec §8.2) and then waits again on the SAME ready. It used to loop back to
// the entries map: a wake that failed inside the extension had deleted its
// entry, so the waiter found nothing and started a second wake, a second
// lease acquire and a second spawn (#184's shape).
func TestWakeWorkerAWakeFailingInsideTheExtensionStartsNoSecond(t *testing.T) {
	old := wakeWaitBound
	wakeWaitBound = 50 * time.Millisecond
	t.Cleanup(func() { wakeWaitBound = old })
	a, b, _, _, _ := stagedActivator(t)
	a.gens = true
	a.backend = &plannedHostKill{stagedFake: b}
	next := stagedID(t, t.TempDir(), "v2")
	a.Planner = func(context.Context, registry.Route) WakePlan {
		return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true}
	}
	boom := errors.New("exec refused")
	release := make(chan struct{})
	calls := awaitExtensionSeam(t, a, release)
	b.mu.Lock()
	b.workerStartHook = func(registry.Route) {
		<-release
		b.mu.Lock()
		b.workerErr = boom
		b.mu.Unlock()
	}
	b.mu.Unlock()
	err := a.WakeWorker(testWorkerRoute("bot.x.internal"))
	if calls.Load() == 0 {
		t.Fatal("the waiter never extended its bound: the test did not reach the window")
	}
	if !errors.Is(err, boom) {
		t.Fatalf("WakeWorker = %v, want the wake's own error", err)
	}
	if acquires, _, _ := b.counts(); acquires != 1 {
		t.Fatalf("%d lease acquires, want 1: a waiter whose wake failed inside the extension started a second wake", acquires)
	}
}

// The same window for a request (begin): the failed wake reports to its
// waiter, and nothing wakes the instance a second time.
func TestBeginAWakeFailingInsideTheExtensionStartsNoSecond(t *testing.T) {
	old := wakeWaitBound
	wakeWaitBound = 50 * time.Millisecond
	t.Cleanup(func() { wakeWaitBound = old })
	next := stagedID(t, t.TempDir(), "v2")
	a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true})
	boom := errors.New("GET / answered 500")
	release := make(chan struct{})
	calls := awaitExtensionSeam(t, a, release)
	a.genHealth = func(context.Context, Gen) error {
		<-release
		return boom
	}
	_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if calls.Load() == 0 {
		t.Fatal("the waiter never extended its bound: the test did not reach the window")
	}
	if err == nil || errors.Is(err, errWakeTimeout) {
		t.Fatalf("beginRequest = %v, want the failed wake's error", err)
	}
	if acquires, _, _ := b.counts(); acquires != 1 {
		t.Fatalf("%d lease acquires, want 1: a waiter whose wake failed inside the extension started a second wake", acquires)
	}
	if gens := b.started(); len(gens) != 1 {
		t.Fatalf("spawned %+v, want one generation", gens)
	}
}